Copilot commented on code in PR #13038: URL: https://github.com/apache/gluten/pull/13038#discussion_r4329994172
########## cpp/velox/compute/HookedExecutor.cc: ########## @@ -0,0 +1,246 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "compute/HookedExecutor.h" + +#include <folly/ScopeGuard.h> +#include <glog/logging.h> +#include <condition_variable> +#include <mutex> +#include <unordered_map> +#include <utility> +#include <vector> + +#include "threads/ThreadInitializer.h" +#include "utils/Exception.h" +#include "velox/common/process/StackTrace.h" + +namespace gluten { +namespace { + +class HookedExecutor final : public folly::Executor { + public: + HookedExecutor( + folly::Executor* parent, + std::string name, + ThreadInitializer* initializer, + bool debug, + std::chrono::milliseconds joinTimeout, + bool releaseQueuedOnStop) + : parent_(parent), + name_(std::move(name)), + initializer_(initializer), + debug_(debug), + joinTimeout_(joinTimeout), + releaseQueuedOnStop_(releaseQueuedOnStop) {} + + ~HookedExecutor() override { + if (releaseQueuedOnStop_) { + releaseQueued(); + } + if (!join()) { + LOG(WARNING) << "Timed out waiting for hooked executor " << name_ << " to finish after " << joinTimeout_.count() + << " ms."; + dumpOutstandingTasks(); + } + } Review Comment: Potential deadlock if `~HookedExecutor()` is invoked from within one of its own wrapped tasks running on the parent executor thread: `join()` waits for `state_->tasks.empty()`, but the currently-running task cannot complete while the thread is blocked in the destructor. To fix this, add a self-destruction guard (e.g., thread-local tracking of the currently executing HookedExecutor task) and relax the join predicate when the current thread is executing one of these tasks (e.g., wait until only the current task remains), or avoid blocking in that case. ########## cpp/velox/compute/HookedExecutor.cc: ########## @@ -0,0 +1,246 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "compute/HookedExecutor.h" + +#include <folly/ScopeGuard.h> +#include <glog/logging.h> +#include <condition_variable> +#include <mutex> +#include <unordered_map> +#include <utility> +#include <vector> + +#include "threads/ThreadInitializer.h" +#include "utils/Exception.h" +#include "velox/common/process/StackTrace.h" + +namespace gluten { +namespace { + +class HookedExecutor final : public folly::Executor { + public: + HookedExecutor( + folly::Executor* parent, + std::string name, + ThreadInitializer* initializer, + bool debug, + std::chrono::milliseconds joinTimeout, + bool releaseQueuedOnStop) + : parent_(parent), + name_(std::move(name)), + initializer_(initializer), + debug_(debug), + joinTimeout_(joinTimeout), + releaseQueuedOnStop_(releaseQueuedOnStop) {} + + ~HookedExecutor() override { + if (releaseQueuedOnStop_) { + releaseQueued(); + } + if (!join()) { + LOG(WARNING) << "Timed out waiting for hooked executor " << name_ << " to finish after " << joinTimeout_.count() + << " ms."; + dumpOutstandingTasks(); + } + } + + uint8_t getNumPriorities() const override { + return parent_ == nullptr ? 1 : parent_->getNumPriorities(); + } + + void add(folly::Func func) override { + submit(std::move(func), 0, false); + } + + void addWithPriority(folly::Func func, int8_t priority) override { + submit(std::move(func), priority, true); + } + + private: + struct Task { + explicit Task(folly::Func callable) : func(std::move(callable)) {} + + folly::Func func; + bool running{false}; + uint64_t id{0}; + std::chrono::steady_clock::time_point enqueueTime; + int8_t priority{0}; + std::string submitStacktrace; + }; + + struct State { + std::mutex mutex; + std::condition_variable cv; + uint64_t nextTaskId{0}; + bool stopping{false}; + std::unordered_map<uint64_t, std::shared_ptr<Task>> tasks; + }; + + static void markDone(const std::shared_ptr<State>& state, uint64_t taskId) { + std::lock_guard<std::mutex> lock(state->mutex); + state->tasks.erase(taskId); + state->cv.notify_all(); + } + + static void releaseTask(const std::shared_ptr<State>& state, const std::shared_ptr<Task>& task) { + folly::Func func; + { + std::lock_guard<std::mutex> lock(state->mutex); + if (!task->func || task->running) { + return; + } + func = std::move(task->func); + } + func = nullptr; + markDone(state, task->id); + } + + void submit(folly::Func func, int8_t priority, bool withPriority) { + GLUTEN_CHECK(parent_ != nullptr, "Parent executor is null."); + GLUTEN_CHECK(func, "Submitted callable is null."); + auto task = std::make_shared<Task>(std::move(func)); + task->priority = priority; + if (debug_) { + task->enqueueTime = std::chrono::steady_clock::now(); + task->submitStacktrace = facebook::velox::process::StackTrace().toString(); + } + { + std::lock_guard<std::mutex> lock(state_->mutex); + if (state_->stopping) { + return; + } + task->id = state_->nextTaskId++; + state_->tasks.emplace(task->id, task); + } + + // A rejected submission must not leave an outstanding callable behind. + auto rollback = folly::makeGuard([&] { releaseTask(state_, task); }); + auto wrapped = wrap(task); + if (withPriority) { + parent_->addWithPriority(std::move(wrapped), priority); + } else { + parent_->add(std::move(wrapped)); + } + rollback.dismiss(); + } + + folly::Func wrap(const std::shared_ptr<Task>& task) { + auto state = state_; + auto* initializer = initializer_; + auto taskName = name_ + std::to_string(task->id); + return [state = std::move(state), task, initializer, taskName = std::move(taskName)]() mutable { + folly::Func func; + { + std::lock_guard<std::mutex> lock(state->mutex); + if (!task->func) { + // Teardown owns completion accounting for released callables. + return; + } + task->running = true; + func = std::move(task->func); + } + + auto done = folly::makeGuard([&] { + // Also release captures if thread initialization throws. + func = nullptr; + markDone(state, task->id); + }); + GLUTEN_CHECK(initializer != nullptr, "ThreadInitializer is null."); + initializer->initialize(taskName); + auto destroy = folly::makeGuard([&] { initializer->destroy(taskName); }); + // Captures must die before the thread context and outstanding-task entry. + auto localFunc = std::move(func); + localFunc(); + }; + } + + void releaseQueued() { + std::vector<std::pair<uint64_t, folly::Func>> queued; + { + std::lock_guard<std::mutex> lock(state_->mutex); + state_->stopping = true; + for (const auto& [taskId, task] : state_->tasks) { + if (!task->running && task->func) { + queued.emplace_back(taskId, std::move(task->func)); + } + } + } + // Capture destruction can destroy readers or submit more work. Do not hold + // the registry lock, and keep these tasks outstanding until it is finished. + for (auto& [taskId, func] : queued) { + func = nullptr; + } + { + std::lock_guard<std::mutex> lock(state_->mutex); + for (const auto& [taskId, func] : queued) { + state_->tasks.erase(taskId); + } + state_->cv.notify_all(); + } + } + + bool join() { + std::unique_lock<std::mutex> lock(state_->mutex); + return state_->cv.wait_for(lock, joinTimeout_, [&] { return state_->tasks.empty(); }); + } Review Comment: Potential deadlock if `~HookedExecutor()` is invoked from within one of its own wrapped tasks running on the parent executor thread: `join()` waits for `state_->tasks.empty()`, but the currently-running task cannot complete while the thread is blocked in the destructor. To fix this, add a self-destruction guard (e.g., thread-local tracking of the currently executing HookedExecutor task) and relax the join predicate when the current thread is executing one of these tasks (e.g., wait until only the current task remains), or avoid blocking in that case. ########## cpp/velox/tests/HookedExecutorTest.cc: ########## @@ -0,0 +1,341 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "compute/HookedExecutor.h" + +#include <gtest/gtest.h> +#include <atomic> +#include <deque> +#include <functional> +#include <future> +#include <mutex> +#include <stdexcept> +#include <thread> + +#include "threads/ThreadInitializer.h" + +namespace gluten { +namespace { + +using namespace std::chrono_literals; + +class QueuedExecutor final : public folly::Executor { + public: + void add(folly::Func func) override { + if (reject) { + throw std::runtime_error("Rejected submission"); + } + std::lock_guard<std::mutex> lock(mutex_); + tasks_.push_back(std::move(func)); + } + + void addWithPriority(folly::Func func, int8_t priority) override { + lastPriority = priority; + add(std::move(func)); + } + + uint8_t getNumPriorities() const override { + return 3; + } + + folly::Func take() { + std::lock_guard<std::mutex> lock(mutex_); + auto func = std::move(tasks_.front()); + tasks_.pop_front(); + return func; + } + + size_t size() { + std::lock_guard<std::mutex> lock(mutex_); + return tasks_.size(); + } + + bool reject{false}; + int8_t lastPriority{0}; + + private: + std::mutex mutex_; + std::deque<folly::Func> tasks_; +}; + +struct HookCounts { + std::atomic<int> initialized{0}; + std::atomic<int> destroyed{0}; +}; + +class CountingInitializer final : public ThreadInitializer { + public: + explicit CountingInitializer(std::shared_ptr<HookCounts> counts, bool failInitialization = false) + : counts_(std::move(counts)), failInitialization_(failInitialization) {} + + void initialize(const std::string&) override { + ++counts_->initialized; + if (failInitialization_) { + throw std::runtime_error("Thread initialization failed"); + } + } + + void destroy(const std::string&) override { + ++counts_->destroyed; + } + + private: + std::shared_ptr<HookCounts> counts_; + bool failInitialization_; +}; + +struct DestructionCallback { + explicit DestructionCallback(std::function<void()> callback) : callback(std::move(callback)) {} + + ~DestructionCallback() { + callback(); + } + + std::function<void()> callback; +}; + +class HookedExecutorTest : public testing::Test { + protected: + std::unique_ptr<folly::Executor> + makeExecutor(bool releaseQueuedOnStop = true, std::chrono::milliseconds timeout = 10s, bool debug = false) { + return makeHookedExecutor(&parent_, "test", initializer_.get(), debug, timeout, releaseQueuedOnStop); + } + + QueuedExecutor parent_; + std::shared_ptr<HookCounts> counts_{std::make_shared<HookCounts>()}; + std::unique_ptr<ThreadInitializer> initializer_{std::make_unique<CountingInitializer>(counts_)}; +}; + +TEST_F(HookedExecutorTest, releasesQueuedCapturesWithoutScheduling) { + for (bool debug : {false, true}) { + auto executor = makeExecutor(true, 10s, debug); + auto resource = std::make_shared<int>(42); + std::weak_ptr<int> weak = resource; + executor->add([resource] { FAIL() << "Cancelled callable ran: " << *resource; }); + resource.reset(); + + executor.reset(); + EXPECT_TRUE(weak.expired()); + EXPECT_EQ(counts_->initialized, 0); + ASSERT_EQ(parent_.size(), 1); + parent_.take()(); + EXPECT_EQ(counts_->destroyed, 0); + } +} + +TEST_F(HookedExecutorTest, cancelledWrappersDoNotUseDestroyedInitializer) { + auto executor = makeExecutor(); + executor->add([] { FAIL() << "Cancelled callable ran"; }); + executor.reset(); + initializer_.reset(); + parent_.take()(); + EXPECT_EQ(counts_->initialized, 0); + EXPECT_EQ(counts_->destroyed, 0); +} + +TEST_F(HookedExecutorTest, releasesCompletedCapturesBeforeWrapperDestruction) { + auto executor = makeExecutor(); + auto resource = std::make_shared<int>(42); + std::weak_ptr<int> weak = resource; + executor->add([resource] { EXPECT_EQ(*resource, 42); }); + resource.reset(); + auto wrapper = parent_.take(); + wrapper(); + EXPECT_TRUE(weak.expired()); + EXPECT_EQ(counts_->initialized, 1); + EXPECT_EQ(counts_->destroyed, 1); + executor.reset(); +} + +TEST_F(HookedExecutorTest, forwardsPrioritiesAndCancelsPriorityTasks) { + auto executor = makeExecutor(); + EXPECT_EQ(executor->getNumPriorities(), 3); + executor->addWithPriority([] {}, 2); + EXPECT_EQ(parent_.lastPriority, 2); + parent_.take()(); + executor->addWithPriority([] { FAIL() << "Cancelled priority callable ran"; }, 1); + executor.reset(); + parent_.take()(); + EXPECT_EQ(counts_->initialized, 1); + EXPECT_EQ(counts_->destroyed, 1); +} + +TEST_F(HookedExecutorTest, cancellationIsScopedToOneRuntime) { + auto first = makeExecutor(); + auto second = makeExecutor(); + first->add([] { FAIL() << "Cancelled callable ran"; }); + bool ran = false; + second->add([&] { ran = true; }); + first.reset(); + parent_.take()(); + parent_.take()(); + EXPECT_TRUE(ran); + second.reset(); + EXPECT_EQ(counts_->initialized, 1); +} + +TEST_F(HookedExecutorTest, computeAndSpillKeepQueuedWorkOnTimeout) { + auto executor = makeExecutor(false, 0ms); + auto resource = std::make_shared<int>(42); + std::weak_ptr<int> weak = resource; + executor->add([resource] { EXPECT_EQ(*resource, 42); }); + resource.reset(); + executor.reset(); + EXPECT_FALSE(weak.expired()); + parent_.take()(); + EXPECT_TRUE(weak.expired()); + EXPECT_EQ(counts_->initialized, 1); + EXPECT_EQ(counts_->destroyed, 1); +} + +TEST_F(HookedExecutorTest, queuedCaptureDestructionCanSubmitMoreWork) { + auto executor = makeExecutor(); + auto* raw = executor.get(); + auto resource = std::make_shared<int>(42); + std::weak_ptr<int> weak = resource; + auto onDestroy = std::make_shared<DestructionCallback>( + [raw, resource] { raw->add([resource] { FAIL() << "Callable submitted during teardown ran: " << *resource; }); }); + executor->add([onDestroy] { FAIL() << "Cancelled callable ran"; }); + resource.reset(); + onDestroy.reset(); + executor.reset(); + EXPECT_TRUE(weak.expired()); + EXPECT_EQ(parent_.size(), 1); + parent_.take()(); +} + +TEST_F(HookedExecutorTest, waitsForRunningWorkAndDiscardsItsLateSubmissions) { + auto executor = makeExecutor(); + auto* raw = executor.get(); + std::promise<void> started; + std::promise<void> resume; + auto resumeFuture = resume.get_future(); + executor->add([&] { + started.set_value(); + resumeFuture.wait(); + raw->add([] { FAIL() << "Late I/O submission ran"; }); + }); Review Comment: The new teardown/join logic is susceptible to self-deadlock when the executor is destroyed from inside one of its own running tasks. Please add a unit test that resets/destroys the HookedExecutor from within a wrapped task (executed via the parent executor), and asserts teardown completes without hanging (and that queued tasks are handled per `releaseQueuedOnStop`). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
