From e59df5b84727c69e4b9f3a1579132341af7ab4a1 Mon Sep 17 00:00:00 2001 From: Peter Abbondanzo Date: Sun, 11 Oct 2026 17:41:51 -0700 Subject: [PATCH] Fix dispatcher callback ownership and shutdown races Summary: Dispatch tasks through uniquely owned callable payloads so queue bookkeeping never copies or destroys populated captures while holding its mutex. Cancel queued work outside the queue lock, distinguish worker-thread shutdown requests from external-owner joining, and wake timed waits when the queue head changes. Add focused dispatcher admission, cancellation, callable-copy, reentrant destruction and joining regressions. Inspector ownership and tester readiness are separate dependent increments, not behavior introduced here. Changelog: [General][Fixed] - Fix native dispatcher callback ownership and shutdown synchronization Differential Revision: D124671180 --- .../react/threading/TaskDispatchThread.cpp | 87 ++++++------ .../react/threading/TaskDispatchThread.h | 13 +- .../tests/TaskDispatchThreadTests.cpp | 134 ++++++++++++++++++ 3 files changed, 187 insertions(+), 47 deletions(-) diff --git a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp index e90961d083dd..ff76847634cc 100644 --- a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp +++ b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp @@ -50,18 +50,15 @@ TaskDispatchThread::TaskDispatchThread( #else thread_ = std::thread(&TaskDispatchThread::loop, this); #endif + threadId_ = thread_.get_id(); } TaskDispatchThread::~TaskDispatchThread() noexcept { quit(); - - if (thread_.joinable()) { - thread_.join(); - } } bool TaskDispatchThread::isOnThread() noexcept { - return std::this_thread::get_id() == thread_.get_id(); + return !loopStopped_ && std::this_thread::get_id() == threadId_; } bool TaskDispatchThread::isRunning() noexcept { @@ -71,12 +68,13 @@ bool TaskDispatchThread::isRunning() noexcept { void TaskDispatchThread::runAsync( TaskFn&& task, std::chrono::milliseconds delayMs) noexcept { + auto ownedTask = std::make_unique(std::move(task)); std::lock_guard guard(mutex_); if (!running_) { return; } auto dispatchTime = std::chrono::system_clock::now() + delayMs; - queue_.emplace(dispatchTime, std::move(task)); + queue_.emplace(dispatchTime, std::move(ownedTask)); loopCv_.notify_one(); } @@ -97,61 +95,62 @@ void TaskDispatchThread::runSync(TaskFn&& task) noexcept { promise->get_future().wait(); } -void TaskDispatchThread::quit() noexcept { +void TaskDispatchThread::cancelPendingTasks() noexcept { + std::priority_queue discarded; { - /* - We should hold lock even when changing atomic variable to correctly - publish the modificatino to the waiting thread - from: https://en.cppreference.com/w/cpp/thread/condition_variable.html - */ - std::unique_lock lock(mutex_); - if (!running_.exchange(false)) { - return; - } + std::lock_guard lock(mutex_); + queue_.swap(discarded); } + loopCv_.notify_one(); +} +void TaskDispatchThread::quit() noexcept { + const bool onThread = isOnThread(); + { + std::lock_guard lock(mutex_); + running_ = false; + } loopCv_.notify_one(); - if (!isOnThread()) { - loopStoppedPromise_.get_future().wait(); + if (!onThread) { + { + std::lock_guard lock(joinMutex_); + if (thread_.joinable()) { + thread_.join(); + } + } } + cancelPendingTasks(); } void TaskDispatchThread::loop() noexcept { if (!threadName_.empty()) { folly::setThreadName(threadName_); } + std::unique_lock lock(mutex_); while (running_) { - std::unique_lock lock(mutex_); loopCv_.wait(lock, [&]() { return !running_ || !queue_.empty(); }); - - while (!queue_.empty()) { - auto task = queue_.top(); - auto now = std::chrono::system_clock::now(); - - if (running_) { - if (task.dispatchTime > now) { - // Wait until the scheduled task time, if delayed - loopCv_.wait_until(lock, task.dispatchTime); - continue; - } - } else { - // Shutting down, skip all the remaining tasks - queue_ = {}; - break; - } - - // We should check whether the task thread is still running at this - // point (which may not anymore be the case since the previous check) + if (!running_) { + break; + } + const auto dispatchTime = queue_.top().dispatchTime; + if (dispatchTime > std::chrono::system_clock::now()) { + loopCv_.wait_until(lock, dispatchTime, [this, dispatchTime]() { + return !running_ || queue_.empty() || + queue_.top().dispatchTime != dispatchTime; + }); + continue; + } + { + auto fn = std::move(queue_.top().fn); + queue_.pop(); + lock.unlock(); if (running_) { - lock.unlock(); - task.fn(); - lock.lock(); + (*fn)(); } - queue_.pop(); } + lock.lock(); } - - loopStoppedPromise_.set_value(); + loopStopped_ = true; } } // namespace facebook::react diff --git a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h index 2dd86991b9e5..4143c2845cea 100644 --- a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h +++ b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h @@ -7,10 +7,12 @@ #pragma once +#include #include #include #include #include +#include #include #include #include @@ -42,15 +44,18 @@ class TaskDispatchThread { /** Add task to the queue and wait until it has completed. */ void runSync(TaskFn &&task) noexcept; + void cancelPendingTasks() noexcept; + /** Shut down and clean up the thread. */ void quit() noexcept; protected: struct Task { TimePoint dispatchTime; - TaskFn fn; + // Moving the callback must preserve the heap's deadline ordering key. + mutable std::unique_ptr fn; - Task(TimePoint dispatchTime, TaskFn &&fn) : dispatchTime(dispatchTime), fn(std::move(fn)) {} + Task(TimePoint dispatchTime, std::unique_ptr &&fn) : dispatchTime(dispatchTime), fn(std::move(fn)) {} bool operator<(const Task &other) const { @@ -67,8 +72,10 @@ class TaskDispatchThread { std::atomic running_{true}; std::string threadName_; std::thread thread_; + std::thread::id threadId_; + std::mutex joinMutex_; - std::promise loopStoppedPromise_; + std::atomic loopStopped_{false}; }; } // namespace facebook::react diff --git a/packages/react-native/ReactCxxPlatform/react/threading/tests/TaskDispatchThreadTests.cpp b/packages/react-native/ReactCxxPlatform/react/threading/tests/TaskDispatchThreadTests.cpp index 90f5379eea06..70d31a1ee37a 100644 --- a/packages/react-native/ReactCxxPlatform/react/threading/tests/TaskDispatchThreadTests.cpp +++ b/packages/react-native/ReactCxxPlatform/react/threading/tests/TaskDispatchThreadTests.cpp @@ -10,6 +10,7 @@ #include #include #include +#include #include namespace facebook::react { @@ -216,4 +217,137 @@ TEST_F(TaskDispatchThreadTest, RunSyncDoesNotBlockWhenQuitCalledConcurrently) { localDispatcher.reset(); } } + +TEST_F(TaskDispatchThreadTest, ExternalQuitJoinsEvenAfterWorkerRequestedQuit) { + std::promise entered; + std::promise release; + auto released = release.get_future(); + dispatcher->runAsync([&]() { + dispatcher->quit(); + entered.set_value(); + released.wait(); + }); + auto ready = entered.get_future().wait_for(std::chrono::seconds(5)); + if (ready != std::future_status::ready) { + release.set_value(); + FAIL() << "Worker did not reach the shutdown barrier"; + } + + std::promise joined; + auto joinedFuture = joined.get_future(); + std::thread owner([&]() { + dispatcher->quit(); + joined.set_value(); + }); + EXPECT_EQ( + joinedFuture.wait_for(std::chrono::seconds(1)), + std::future_status::timeout); + release.set_value(); + owner.join(); + EXPECT_EQ( + joinedFuture.wait_for(std::chrono::seconds(0)), + std::future_status::ready); + EXPECT_FALSE(dispatcher->isRunning()); +} + +TEST_F(TaskDispatchThreadTest, ConcurrentExternalQuitWaitsForInFlightCallback) { + std::promise entered; + std::promise release; + auto released = release.get_future(); + std::atomic finished{false}; + dispatcher->runAsync([&]() { + entered.set_value(); + released.wait(); + finished = true; + }); + auto ready = entered.get_future().wait_for(std::chrono::seconds(5)); + if (ready != std::future_status::ready) { + release.set_value(); + FAIL() << "Worker did not reach the callback barrier"; + } + std::promise firstJoined; + std::promise secondJoined; + auto firstFuture = firstJoined.get_future(); + auto secondFuture = secondJoined.get_future(); + std::thread first([&]() { + dispatcher->quit(); + EXPECT_TRUE(finished); + firstJoined.set_value(); + }); + std::thread second([&]() { + dispatcher->quit(); + EXPECT_TRUE(finished); + secondJoined.set_value(); + }); + EXPECT_EQ( + firstFuture.wait_for(std::chrono::seconds(1)), + std::future_status::timeout); + EXPECT_EQ( + secondFuture.wait_for(std::chrono::seconds(1)), + std::future_status::timeout); + release.set_value(); + first.join(); + second.join(); + EXPECT_TRUE(finished); +} + +TEST( + TaskDispatchThreadShutdownTest, + DiscardedCapturesAreReleasedAfterJoinWithoutQueueLock) { + class Dispatcher final : public TaskDispatchThread { + public: + bool queueMutexAvailable() { + std::unique_lock lock(mutex_, std::try_to_lock); + return lock.owns_lock(); + } + } dispatcher; + bool released = false; + auto ownerThread = std::this_thread::get_id(); + auto token = std::shared_ptr(nullptr, [&](void*) { + EXPECT_TRUE(dispatcher.queueMutexAvailable()); + EXPECT_EQ(std::this_thread::get_id(), ownerThread); + released = true; + }); + dispatcher.runAsync([token]() {}, std::chrono::hours(1)); + token.reset(); + + dispatcher.quit(); + + EXPECT_TRUE(released); +} + +TEST_F(TaskDispatchThreadTest, DispatchDoesNotCopyCallbackOnWorker) { + struct State { + TaskDispatchThread* dispatcher{nullptr}; + std::atomic copiesOnWorker{0}; + std::promise ran; + }; + struct CopyCountingTask { + std::shared_ptr state; + + explicit CopyCountingTask(std::shared_ptr state) + : state(std::move(state)) {} + + CopyCountingTask(const CopyCountingTask& other) noexcept + : state(other.state) { + if (state->dispatcher->isOnThread()) { + ++state->copiesOnWorker; + } + } + + CopyCountingTask& operator=(const CopyCountingTask&) = default; + + void operator()() { + state->ran.set_value(); + } + }; + auto state = std::make_shared(); + state->dispatcher = dispatcher.get(); + auto ran = state->ran.get_future(); + dispatcher->runAsync(CopyCountingTask(state)); + ASSERT_EQ(ran.wait_for(std::chrono::seconds(5)), std::future_status::ready); + dispatcher->quit(); + EXPECT_EQ(state->copiesOnWorker.load(), 0); +} + } // namespace facebook::react