From 0c58083375df5f46267d1450177c04089814d944 Mon Sep 17 00:00:00 2001 From: Peter Abbondanzo Date: Sun, 11 Oct 2026 17:43:35 -0700 Subject: [PATCH 1/2] Fix dispatcher callback ownership and shutdown races (#58995) 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 From d13b25014c87ecab74dc3b787a03584234590805 Mon Sep 17 00:00:00 2001 From: Peter Abbondanzo Date: Sun, 11 Oct 2026 17:43:35 -0700 Subject: [PATCH 2/2] Coordinate inspector and transport shutdown Summary: Retain the inspector until serialized target and socket cleanup has completed, reject late work during shutdown, and join the dispatcher from its external owner. Keep transport close requests callable from IO callbacks without self-joining or holding the socket lock across a join. Serialize writes and pings on the IO executor, discard expired implementation references, and make external-owner destruction finish cleanup when a closed callback throws. Add focused inspector lifetime and real loopback transport regressions, including active-IO destruction, nested close, late connection completions, and invalid descriptors. Destruction of a socket wrapper on its own IO thread remains unsupported. Changelog: [General][Fixed] - Fix native inspector ownership and transport shutdown races Differential Revision: D124675718 --- .../react/devsupport/inspector/Inspector.cpp | 45 +- .../react/devsupport/inspector/Inspector.h | 7 + .../tests/InspectorShutdownTests.cpp | 150 +++++++ .../http/platform/cxx/WebSocketClient.cpp | 194 +++++--- .../react/http/platform/cxx/WebSocketClient.h | 7 + .../http/platform/cxx/WebSocketClientImpl.h | 63 +++ .../react/http/tests/WebSocketClientTests.cpp | 413 ++++++++++++++++++ .../react/runtime/ReactHost.cpp | 6 +- 8 files changed, 808 insertions(+), 77 deletions(-) create mode 100644 packages/react-native/ReactCxxPlatform/react/devsupport/inspector/tests/InspectorShutdownTests.cpp create mode 100644 packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClientImpl.h create mode 100644 packages/react-native/ReactCxxPlatform/react/http/tests/WebSocketClientTests.cpp diff --git a/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.cpp b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.cpp index 0ba040caee6c..62e51c0191b1 100644 --- a/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.cpp +++ b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.cpp @@ -135,18 +135,36 @@ Inspector::Inspector( httpClientFactory_(std::move(httpClientFactory)) {} Inspector::~Inspector() noexcept { - if (pageId_) { - jsinspector_modern::getInspectorInstance().removePage(*pageId_); - taskDispatchThread_.runSync([weakThis = weak_from_this()]() { - if (auto strongThis = weakThis.lock()) { - strongThis->pageId_.reset(); - strongThis->target_.reset(); - } - }); + shutdown(); +} + +void Inspector::shutdown() noexcept { + CHECK(!taskDispatchThread_.isOnThread()); + if (shuttingDown_.exchange(true)) { + return; } + taskDispatchThread_.runSync([this]() { + taskDispatchThread_.cancelPendingTasks(); + if (pageId_) { + jsinspector_modern::getInspectorInstance().removePage(*pageId_); + pageId_.reset(); + } + if (packagerConnection_) { + packagerConnection_->closeQuietly(); + packagerConnection_.reset(); + } + target_.reset(); + hostDelegate_.reset(); + }); taskDispatchThread_.quit(); } +void Inspector::runOnInspectorThreadSync( + TaskDispatchThread::TaskFn&& callback) { + CHECK(!shuttingDown_); + taskDispatchThread_.runSync(std::move(callback)); +} + void Inspector::connectDebugger(const std::string& inspectorUrl) noexcept { invokeElsePost([weakThis = weak_from_this(), inspectorUrl]() { auto self = weakThis.lock(); @@ -205,11 +223,20 @@ void Inspector::ensureHostTarget( void Inspector::invokeElsePost( TaskDispatchThread::TaskFn&& callback, std::chrono::milliseconds delayMs) { + if (shuttingDown_) { + return; + } if (taskDispatchThread_.isOnThread() && delayMs == std::chrono::milliseconds::zero()) { callback(); } else { - taskDispatchThread_.runAsync(std::move(callback), delayMs); + taskDispatchThread_.runAsync( + [weakThis = weak_from_this(), callback = std::move(callback)]() { + if (auto self = weakThis.lock(); self && !self->shuttingDown_) { + callback(); + } + }, + delayMs); } } diff --git a/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.h b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.h index c110ad6aea4c..7238d16c0a52 100644 --- a/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.h +++ b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.h @@ -14,6 +14,7 @@ #include #include #include +#include #include #include #include @@ -37,6 +38,11 @@ class Inspector : public InspectorThread, public std::enable_shared_from_this shuttingDown_{false}; TaskDispatchThread taskDispatchThread_{"InspectorThread"}; std::string appName_; std::string deviceName_; diff --git a/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/tests/InspectorShutdownTests.cpp b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/tests/InspectorShutdownTests.cpp new file mode 100644 index 000000000000..67b69ac7c9b1 --- /dev/null +++ b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/tests/InspectorShutdownTests.cpp @@ -0,0 +1,150 @@ +/* + * Copyright (c) Meta Platforms, Inc. and affiliates. + * + * This source code is licensed under the MIT license found in the + * LICENSE file in the root directory of this source tree. + */ + +#include +#include +#include +#include +#include + +namespace facebook::react { + +class InspectorShutdownTest : public ::testing::Test { + protected: + struct SocketState { + IWebSocketClient::OnConnectCallback onConnect; + IWebSocketClient::OnClosedCallback onClosed; + IWebSocketClient::OnMessageCallback onMessage; + std::weak_ptr inspector; + std::thread::id closeThread; + bool ownedDuringClose{false}; + int closes{0}; + }; + + class Socket final : public IWebSocketClient { + public: + explicit Socket(std::shared_ptr state) + : state_(std::move(state)) {} + + void setOnClosedCallback(OnClosedCallback&& callback) noexcept override { + state_->onClosed = std::move(callback); + } + + void setOnMessageCallback(OnMessageCallback&& callback) noexcept override { + state_->onMessage = std::move(callback); + } + + void connect(const std::string& /*url*/, OnConnectCallback&& callback) + override { + state_->onConnect = std::move(callback); + } + + void close(const std::string& reason) override { + ++state_->closes; + state_->closeThread = std::this_thread::get_id(); + state_->ownedDuringClose = !state_->inspector.expired(); + if (state_->onClosed) { + state_->onClosed(reason); + } + } + + void send(const std::string& /*message*/) override {} + void ping() override {} + + private: + std::shared_ptr state_; + }; + + void SetUp() override { + socket = std::make_shared(); + inspector = std::make_shared( + "shutdown-test", + "test-device", + [state = socket]() { return std::make_unique(state); }, + []() { return std::unique_ptr{}; }); + socket->inspector = inspector; + } + + void TearDown() override { + if (inspector) { + inspector->shutdown(); + } + } + + std::shared_ptr socket; + std::shared_ptr inspector; +}; + +TEST_F(InspectorShutdownTest, ShutdownDisposesPageAndSocketWhileOwned) { + auto pageCount = jsinspector_modern::getInspectorInstance().getPages().size(); + inspector->ensureHostTarget({}, {}); + EXPECT_NE(inspector->inspectorTarget(), nullptr); + inspector->connectDebugger("http://localhost:8081/inspector/device"); + std::thread::id inspectorThread; + inspector->runOnInspectorThreadSync( + [&]() { inspectorThread = std::this_thread::get_id(); }); + ASSERT_TRUE(socket->onConnect); + EXPECT_EQ( + jsinspector_modern::getInspectorInstance().getPages().size(), + pageCount + 1); + + inspector->shutdown(); + + EXPECT_EQ(inspector->inspectorTarget(), nullptr); + EXPECT_EQ( + jsinspector_modern::getInspectorInstance().getPages().size(), pageCount); + EXPECT_EQ(socket->closes, 1); + EXPECT_EQ(socket->closeThread, inspectorThread); + EXPECT_TRUE(socket->ownedDuringClose); + EXPECT_EQ(inspector.use_count(), 1); + inspector->shutdown(); + EXPECT_EQ(socket->closes, 1); + inspector.reset(); + EXPECT_TRUE(socket->inspector.expired()); +} + +TEST_F( + InspectorShutdownTest, + ShutdownDropsDelayedOwnersAndRejectsLateCallbacks) { + InspectorThread& dispatch = *inspector; + bool ran = false; + dispatch.invokeElsePost( + [owner = inspector, &ran]() { ran = true; }, std::chrono::hours(1)); + + inspector->shutdown(); + + EXPECT_FALSE(ran); + EXPECT_EQ(inspector.use_count(), 1); + dispatch.invokeElsePost([&]() { ran = true; }); + dispatch.invokeElsePost([&]() { ran = true; }, std::chrono::hours(1)); + inspector->connectDebugger("http://localhost:8081/inspector/device"); + EXPECT_FALSE(ran); + EXPECT_FALSE(socket->onConnect); + inspector.reset(); + EXPECT_TRUE(socket->inspector.expired()); +} + +TEST_F( + InspectorShutdownTest, + LateTransportFailureCannotReconnectAfterShutdown) { + inspector->connectDebugger("http://localhost:8081/inspector/device"); + inspector->runOnInspectorThreadSync([]() {}); + ASSERT_TRUE(socket->onConnect); + auto onConnect = socket->onConnect; + auto onMessage = socket->onMessage; + + inspector->shutdown(); + onConnect(false, "late failure"); + onMessage(R"({"event":"getPages"})"); + + EXPECT_EQ(socket->closes, 1); + EXPECT_EQ(inspector.use_count(), 1); + inspector.reset(); + EXPECT_TRUE(socket->inspector.expired()); +} + +} // namespace facebook::react diff --git a/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.cpp b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.cpp index eb236fe450a2..489f0b07cb95 100644 --- a/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.cpp +++ b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.cpp @@ -5,63 +5,39 @@ * LICENSE file in the root directory of this source tree. */ -#include "WebSocketClient.h" +#include "WebSocketClientImpl.h" -#include -#include -#include -#include +#include +#include #include #include -#include -#include -#include +#include namespace facebook::react { -struct WebSocketClient::Impl final : public std::enable_shared_from_this { - void onResolveCompleted( - boost::system::error_code ec, - const boost::asio::ip::tcp::resolver::results_type& results); - - void onConnectionCompleted(boost::system::error_code ec); - - void onHandshakeCompleted(boost::system::error_code ec); - - void listen(); - - void write(); - - void onConnectCallback(bool connected, const std::string& error); - - // Callbacks and Uri - OnConnectCallback onConnectCallback_; - OnClosedCallback onClosedCallback_; - OnMessageCallback onMessageCallback_; - std::optional uri_; - - // Boost Beast WebSocket Client - boost::asio::io_context ioContext_; - boost::asio::ip::tcp::resolver resolver_{ioContext_}; - boost::beast::multi_buffer buffer_; - folly::Synchronized< - boost::beast::websocket::stream> - ws_{boost::beast::websocket::stream{ - ioContext_}}; - - // Input and Output handling - std::mutex mutexOut_; - std::queue messagesOut_; - std::atomic isWriting_{false}; - std::atomic isConnected_{false}; - std::atomic isClosing_{false}; -}; - WebSocketClient::WebSocketClient() noexcept : impl_(std::make_shared()) {} WebSocketClient::~WebSocketClient() { - WebSocketClient::close("Force close as WebSocketClient being destroyed"); + react_native_assert( + !impl_->isOnIOThread() && + "WebSocketClient must be destroyed on its external owner thread"); + try { + WebSocketClient::close("Force close as WebSocketClient being destroyed"); + } catch (...) { + try { + impl_->isClosing_ = true; + impl_->ioContext_.stop(); + std::lock_guard lock(threadMutex_); + if (thread_) { + thread_->join(); + thread_.reset(); + } + impl_->closeSocket(); + } catch (...) { + std::terminate(); + } + } } void WebSocketClient::setOnClosedCallback( @@ -77,6 +53,7 @@ void WebSocketClient::setOnMessageCallback( void WebSocketClient::connect( const std::string& url, OnConnectCallback&& callback) { + std::lock_guard lock(threadMutex_); if (thread_) { react_native_assert(false && "WebSocketClient::connect() called twice"); return; @@ -98,33 +75,78 @@ void WebSocketClient::connect( }); // Start the I/O thread - static int32_t s_threadId = 0; + static std::atomic s_threadId{0}; thread_ = std::make_unique( - [weakImpl = std::weak_ptr(impl_), tid = s_threadId++]() { + [weakImpl = std::weak_ptr(impl_), tid = s_threadId.fetch_add(1)]() { folly::setThreadName("WebSocket" + std::to_string(tid)); if (auto impl = weakImpl.lock()) { - impl->ioContext_.run(); + impl->run(); } }); } +void WebSocketClient::Impl::run() { + auto restoreMarker = + folly::makeGuard([previous = activeIO_]() { activeIO_ = previous; }); + activeIO_ = this; + ioContext_.run(); +} + void WebSocketClient::close(const std::string& reason) { - if (!impl_->isClosing_.exchange(true)) { - if (impl_->onClosedCallback_) { - impl_->onClosedCallback_(reason); - } - auto ws = impl_->ws_.wlock(); - if (ws->is_open()) { - ws->async_close( - boost::beast::websocket::close_reason(reason), [](auto&&) {}); - } else { - ws->next_layer().close(); + std::unique_lock lock(threadMutex_, std::defer_lock); + closeImpl(reason, [this, &lock]() { + if (impl_->isOnIOThread()) { + return false; } + lock.lock(); if (thread_) { thread_->join(); + thread_.reset(); + } + return true; + }); +} + +void WebSocketClient::closeImpl( + const std::string& reason, + const std::function& join) { + if (!impl_->isClosing_.exchange(true)) { + boost::asio::post( + impl_->ioContext_, [weakImpl = std::weak_ptr(impl_), reason]() { + if (auto impl = weakImpl.lock()) { + impl->closeOnIOThread(reason); + } + }); + if (impl_->onClosedCallback_) { + impl_->onClosedCallback_(reason); } } - thread_ = nullptr; + if (join()) { + // The loop may have exited before the posted close could execute. + impl_->closeSocket(); + } +} + +void WebSocketClient::Impl::closeOnIOThread(const std::string& reason) { + pendingPings_ = 0; + resolver_.cancel(); + auto ws = ws_.wlock(); + if (ws->is_open()) { + ws->async_close( + boost::beast::websocket::close_reason(reason), [](auto&&) {}); + } else { + boost::system::error_code ec; + // NOLINTNEXTLINE(bugprone-unused-return-value): Boost may return void. + (void)ws->next_layer().close(ec); + } +} + +void WebSocketClient::Impl::closeSocket() { + resolver_.cancel(); + auto ws = ws_.wlock(); + boost::system::error_code ec; + // NOLINTNEXTLINE(bugprone-unused-return-value): Boost may return void. + (void)ws->next_layer().close(ec); } void WebSocketClient::send(const std::string& message) { @@ -132,28 +154,60 @@ void WebSocketClient::send(const std::string& message) { std::lock_guard lock(impl_->mutexOut_); impl_->messagesOut_.emplace(message); } - impl_->write(); + boost::asio::post(impl_->ioContext_, [weakImpl = std::weak_ptr(impl_)]() { + if (auto impl = weakImpl.lock()) { + impl->write(); + } + }); } void WebSocketClient::ping() { - auto ws = impl_->ws_.wlock(); - // Send a ping message + boost::asio::post(impl_->ioContext_, [weakImpl = std::weak_ptr(impl_)]() { + if (auto impl = weakImpl.lock()) { + impl->requestPing(); + } + }); +} + +void WebSocketClient::Impl::requestPing() { + if (isClosing_) { + return; + } + ++pendingPings_; + ping(); +} + +void WebSocketClient::Impl::ping() { + if (isClosing_ || isPinging_ || pendingPings_ == 0) { + return; + } + isPinging_ = true; + --pendingPings_; + auto ws = ws_.wlock(); ws->async_ping( - {}, [weakImpl = std::weak_ptr(impl_)](boost::beast::error_code ec) { + {}, [weakImpl = weak_from_this()](boost::beast::error_code ec) { auto impl = weakImpl.lock(); - if (!impl || impl->isClosing_) { + if (!impl) { + return; + } + impl->isPinging_ = false; + if (impl->isClosing_) { + impl->pendingPings_ = 0; return; } if (ec) { LOG(ERROR) << "Error pinging websocket: " << ec.message(); - return; } + impl->ping(); }); } void WebSocketClient::Impl::onResolveCompleted( boost::system::error_code ec, const boost::asio::ip::tcp::resolver::results_type& results) { + if (isClosing_) { + return; + } if (ec) { // TODO: Handle retry logic here onConnectCallback(false, "Failed to resolve host"); @@ -176,6 +230,9 @@ void WebSocketClient::Impl::onResolveCompleted( void WebSocketClient::Impl::onConnectionCompleted( boost::system::error_code ec) { + if (isClosing_) { + return; + } if (ec) { // TODO: Handle retry logic here onConnectCallback(false, "Failed to connect"); @@ -210,6 +267,9 @@ void WebSocketClient::Impl::onConnectionCompleted( } void WebSocketClient::Impl::onHandshakeCompleted(boost::system::error_code ec) { + if (isClosing_) { + return; + } if (ec) { // TODO: Handle retry logic here onConnectCallback(false, "Failed to handshake"); diff --git a/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.h b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.h index 7c12c2acbd28..e06da3e49297 100644 --- a/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.h +++ b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.h @@ -8,7 +8,9 @@ #pragma once #include +#include #include +#include #include #include @@ -29,6 +31,7 @@ class WebSocketClient : public IWebSocketClient { void connect(const std::string &url, OnConnectCallback &&onConnectCallback = nullptr) override; + // Close may run in IO callbacks; destroy only from an external owner thread. void close(const std::string &reason) override; void send(const std::string &message) override; @@ -36,10 +39,14 @@ class WebSocketClient : public IWebSocketClient { void ping() override; private: + friend class WebSocketClientTestPeer; struct Impl; + void closeImpl(const std::string &reason, const std::function &join); + // Instance data and IO thread const std::shared_ptr impl_; + std::mutex threadMutex_; std::unique_ptr thread_; }; diff --git a/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClientImpl.h b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClientImpl.h new file mode 100644 index 000000000000..40560fa9f5a7 --- /dev/null +++ b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClientImpl.h @@ -0,0 +1,63 @@ +/* + * Copyright (c) Meta Platforms, Inc. and affiliates. + * + * This source code is licensed under the MIT license found in the + * LICENSE file in the root directory of this source tree. + */ + +#pragma once + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace facebook::react { + +struct WebSocketClient::Impl final : public std::enable_shared_from_this { + void onResolveCompleted(boost::system::error_code ec, const boost::asio::ip::tcp::resolver::results_type &results); + void onConnectionCompleted(boost::system::error_code ec); + void onHandshakeCompleted(boost::system::error_code ec); + void listen(); + void write(); + void onConnectCallback(bool connected, const std::string &error); + void run(); + bool isOnIOThread() const noexcept + { + return activeIO_ == this; + } + void requestPing(); + void ping(); + void closeOnIOThread(const std::string &reason); + void closeSocket(); + + inline static thread_local const Impl *activeIO_{nullptr}; + std::size_t pendingPings_{0}; + bool isPinging_{false}; + + OnConnectCallback onConnectCallback_; + OnClosedCallback onClosedCallback_; + OnMessageCallback onMessageCallback_; + std::optional uri_; + + boost::asio::io_context ioContext_; + boost::asio::ip::tcp::resolver resolver_{ioContext_}; + boost::beast::multi_buffer buffer_; + folly::Synchronized> ws_{ + boost::beast::websocket::stream{ioContext_}}; + + std::mutex mutexOut_; + std::queue messagesOut_; + std::atomic isWriting_{false}; + std::atomic isConnected_{false}; + std::atomic isClosing_{false}; +}; + +} // namespace facebook::react diff --git a/packages/react-native/ReactCxxPlatform/react/http/tests/WebSocketClientTests.cpp b/packages/react-native/ReactCxxPlatform/react/http/tests/WebSocketClientTests.cpp new file mode 100644 index 000000000000..a291f46f786f --- /dev/null +++ b/packages/react-native/ReactCxxPlatform/react/http/tests/WebSocketClientTests.cpp @@ -0,0 +1,413 @@ +/* + * Copyright (c) Meta Platforms, Inc. and affiliates. + * + * This source code is licensed under the MIT license found in the + * LICENSE file in the root directory of this source tree. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "WebSocketClientImpl.h" + +namespace facebook::react { + +class WebSocketClientTestPeer { + public: + static auto impl(const WebSocketClient& client) { + return client.impl_; + } + + static bool hasThread(const WebSocketClient& client) { + return client.thread_ != nullptr; + } + + static void close( + WebSocketClient& client, + const std::string& reason, + const std::function& join) { + client.closeImpl(reason, join); + } +}; + +TEST(WebSocketClientTest, DestructionJoinsAfterThrowingClosedCallback) { + auto client = std::make_unique(); + auto impl = WebSocketClientTestPeer::impl(*client); + auto work = boost::asio::make_work_guard(impl->ioContext_); + auto entered = std::make_shared>(); + auto enteredFuture = entered->get_future(); + boost::asio::post(impl->ioContext_, [entered] { entered->set_value(); }); + int closes = 0; + client->connect( + "ws://127.0.0.1:0", + [](bool /*connected*/, const std::string& /*reason*/) {}); + ASSERT_EQ( + enteredFuture.wait_for(std::chrono::seconds(5)), + std::future_status::ready); + EXPECT_TRUE(WebSocketClientTestPeer::hasThread(*client)); + client->setOnClosedCallback([&](const std::string& /*reason*/) { + ++closes; + throw std::runtime_error("closed callback failure"); + }); + + client.reset(); + + EXPECT_EQ(closes, 1); + EXPECT_TRUE(impl->isClosing_); + EXPECT_TRUE(impl->ioContext_.stopped()); +} + +TEST(WebSocketClientTest, CloseDoesNotHoldSocketLockAcrossJoin) { + WebSocketClient client; + auto impl = WebSocketClientTestPeer::impl(client); + bool callbackRan = false; + boost::asio::post(impl->ioContext_, [&]() { + auto ws = impl->ws_.tryWLock(); + EXPECT_TRUE(ws); + EXPECT_TRUE(impl->isClosing_); + callbackRan = true; + }); + + WebSocketClientTestPeer::close(client, "test close", [&]() { + { + auto ws = impl->ws_.tryWLock(); + EXPECT_TRUE(ws); + if (!ws) { + return false; + } + } + impl->run(); + return true; + }); + + EXPECT_TRUE(callbackRan); +} + +TEST(WebSocketClientTest, CloseIsReentrantAndNotifiesOnceBeforeJoining) { + WebSocketClient client; + auto impl = WebSocketClientTestPeer::impl(client); + int closes = 0; + int joins = 0; + client.setOnClosedCallback([&](const std::string& reason) { + EXPECT_EQ(reason, "first close"); + ++closes; + WebSocketClientTestPeer::close(client, "nested close", [&]() { + ++joins; + EXPECT_GE(impl->ioContext_.poll(), 1); + return true; + }); + }); + + WebSocketClientTestPeer::close(client, "first close", [&]() { + EXPECT_EQ(closes, 1); + ++joins; + return true; + }); + WebSocketClientTestPeer::close(client, "later close", [&]() { + ++joins; + return true; + }); + + EXPECT_EQ(closes, 1); + EXPECT_EQ(joins, 3); +} + +TEST(WebSocketClientTest, CloseRequestDefersSocketCleanupUntilOwnerJoin) { + WebSocketClient client; + auto impl = WebSocketClientTestPeer::impl(client); + { + auto ws = impl->ws_.wlock(); + boost::system::error_code ec; + // NOLINTNEXTLINE(bugprone-unused-return-value): Boost may return void. + (void)ws->next_layer().open(boost::asio::ip::tcp::v4(), ec); + ASSERT_FALSE(ec); + } + WebSocketClientTestPeer::close( + client, "request only", []() { return false; }); + { + auto ws = impl->ws_.wlock(); + EXPECT_TRUE(ws->next_layer().is_open()); + } + + bool joined = false; + WebSocketClientTestPeer::close(client, "owner join", [&]() { + joined = true; + return true; + }); + + EXPECT_TRUE(joined); + auto ws = impl->ws_.wlock(); + EXPECT_FALSE(ws->next_layer().is_open()); +} + +TEST(WebSocketClientTest, CloseDropsLateConnectionCompletions) { + WebSocketClient client; + auto impl = WebSocketClientTestPeer::impl(client); + int connects = 0; + impl->onConnectCallback_ = [&](bool, const std::string&) { ++connects; }; + client.close("cancel connection"); + + for (auto ec : + {boost::system::error_code{}, + boost::system::error_code(boost::asio::error::operation_aborted)}) { + impl->onResolveCompleted(ec, {}); + impl->onConnectionCompleted(ec); + impl->onHandshakeCompleted(ec); + } + + EXPECT_EQ(connects, 0); + EXPECT_FALSE(impl->isConnected_); +} + +TEST(WebSocketClientTest, CloseWithoutRunningLoopDoesNotRetainImpl) { + auto client = std::make_unique(); + auto impl = WebSocketClientTestPeer::impl(*client); + std::weak_ptr weakImpl = impl; + client->close("no IO loop"); + impl.reset(); + client.reset(); + EXPECT_TRUE(weakImpl.expired()); +} + +TEST(WebSocketClientTest, DestructorDoesNotThrowForInvalidSocketDescriptor) { + EXPECT_EXIT( + { + { + WebSocketClient client; + auto impl = WebSocketClientTestPeer::impl(client); + auto ws = impl->ws_.wlock(); + boost::system::error_code ec; + // NOLINTNEXTLINE(bugprone-unused-return-value): void overload. + (void)ws->next_layer().open(boost::asio::ip::tcp::v4(), ec); + if (ec || ::close(ws->next_layer().native_handle()) != 0) { + std::_Exit(1); + } + } + std::_Exit(0); + }, + ::testing::ExitedWithCode(0), + ""); +} + +TEST(WebSocketClientTest, LoopbackHandshakeMessagesPingAndGracefulClose) { + boost::asio::io_context serverIO; + boost::asio::ip::tcp::acceptor acceptor( + serverIO, {boost::asio::ip::make_address("127.0.0.1"), 0}); + boost::beast::websocket::stream server( + serverIO); + boost::beast::flat_buffer buffer; + std::promise pinged; + int pings = 0; + server.control_callback([&](auto type, auto) { + if (type == boost::beast::websocket::frame_type::ping && ++pings == 2) { + pinged.set_value(); + } + }); + std::promise serverClosed; + std::string received; + acceptor.async_accept(server.next_layer(), [&](auto ec) { + if (ec) { + serverClosed.set_value(ec); + return; + } + server.async_accept([&](auto ec) { + if (ec) { + serverClosed.set_value(ec); + return; + } + server.async_read(buffer, [&](auto ec, auto) { + if (ec) { + serverClosed.set_value(ec); + return; + } + received = boost::beast::buffers_to_string(buffer.data()); + server.async_write(boost::asio::buffer(received), [&](auto ec, auto) { + if (ec) { + serverClosed.set_value(ec); + return; + } + buffer.consume(buffer.size()); + server.async_read( + buffer, [&](auto ec, auto) { serverClosed.set_value(ec); }); + }); + }); + }); + }); + std::thread serverThread([&]() { serverIO.run(); }); + WebSocketClient client; + client.setOnClosedCallback( + [&](const std::string&) { client.close("nested external close"); }); + std::promise connected; + std::promise message; + client.setOnMessageCallback( + [&](const std::string& value) { message.set_value(value); }); + std::ostringstream url; + url << "ws://127.0.0.1:" << acceptor.local_endpoint().port() + << "/test?token=1"; + client.send("before handshake"); + client.connect(url.str(), [&](bool success, const std::string&) { + connected.set_value(success); + }); + auto connectedFuture = connected.get_future(); + auto messageFuture = message.get_future(); + auto pingFuture = pinged.get_future(); + EXPECT_EQ( + connectedFuture.wait_for(std::chrono::seconds(5)), + std::future_status::ready); + if (connectedFuture.wait_for(std::chrono::seconds(0)) == + std::future_status::ready) { + EXPECT_TRUE(connectedFuture.get()); + } + EXPECT_EQ( + messageFuture.wait_for(std::chrono::seconds(5)), + std::future_status::ready); + if (messageFuture.wait_for(std::chrono::seconds(0)) == + std::future_status::ready) { + EXPECT_EQ(messageFuture.get(), "before handshake"); + } + std::promise enteredIO; + std::promise releaseIO; + auto enteredFuture = enteredIO.get_future(); + auto releaseFuture = releaseIO.get_future(); + auto impl = WebSocketClientTestPeer::impl(client); + boost::asio::post(impl->ioContext_, [&]() { + enteredIO.set_value(); + releaseFuture.wait(); + }); + EXPECT_EQ( + enteredFuture.wait_for(std::chrono::seconds(5)), + std::future_status::ready); + client.ping(); + client.ping(); + releaseIO.set_value(); + EXPECT_EQ( + pingFuture.wait_for(std::chrono::seconds(5)), std::future_status::ready); + client.close("loopback done"); + auto closeFuture = serverClosed.get_future(); + EXPECT_EQ( + closeFuture.wait_for(std::chrono::seconds(5)), std::future_status::ready); + if (closeFuture.wait_for(std::chrono::seconds(0)) == + std::future_status::ready) { + EXPECT_EQ(closeFuture.get(), boost::beast::websocket::error::closed); + } + serverIO.stop(); + serverThread.join(); + EXPECT_EQ(received, "before handshake"); + EXPECT_EQ(pings, 2); +} + +class WebSocketClientCallbackCloseTest : public ::testing::Test { + protected: + void verifyClose(bool fromMessage) { + boost::asio::io_context serverIO; + boost::asio::ip::tcp::acceptor acceptor( + serverIO, {boost::asio::ip::make_address("127.0.0.1"), 0}); + boost::beast::websocket::stream server( + serverIO); + boost::beast::flat_buffer buffer; + std::string payload = "close from message"; + std::promise serverClosed; + auto readForClose = [&]() { + server.async_read( + buffer, [&](auto ec, auto) { serverClosed.set_value(ec); }); + }; + acceptor.async_accept(server.next_layer(), [&](auto ec) { + if (ec) { + serverClosed.set_value(ec); + return; + } + server.async_accept([&](auto ec) { + if (ec) { + serverClosed.set_value(ec); + return; + } + if (fromMessage) { + server.async_write(boost::asio::buffer(payload), [&](auto ec, auto) { + if (ec) { + serverClosed.set_value(ec); + return; + } + readForClose(); + }); + } else { + readForClose(); + } + }); + }); + std::thread serverThread([&]() { serverIO.run(); }); + auto client = std::make_unique(); + auto impl = WebSocketClientTestPeer::impl(*client); + std::weak_ptr weakImpl = impl; + int notifications = 0; + client->setOnClosedCallback([&](const std::string&) { + ++notifications; + client->close("nested callback close"); + }); + std::promise callbackReturned; + auto callbackFuture = callbackReturned.get_future(); + if (fromMessage) { + client->setOnMessageCallback([&](const std::string& message) { + EXPECT_EQ(message, payload); + EXPECT_TRUE(impl->isOnIOThread()); + client->close("message callback close"); + callbackReturned.set_value(true); + }); + } + std::ostringstream url; + url << "ws://127.0.0.1:" << acceptor.local_endpoint().port() + << "/callback-close"; + client->connect(url.str(), [&](bool success, const std::string&) { + if (!fromMessage || !success) { + EXPECT_TRUE(impl->isOnIOThread()); + client->close("connect callback close"); + callbackReturned.set_value(success); + } + }); + const bool requestedClose = + callbackFuture.wait_for(std::chrono::seconds(5)) == + std::future_status::ready; + EXPECT_TRUE(requestedClose); + if (requestedClose) { + EXPECT_TRUE(callbackFuture.get()); + EXPECT_TRUE(WebSocketClientTestPeer::hasThread(*client)); + } + if (!fromMessage || !requestedClose) { + client->close("external owner close"); + EXPECT_FALSE(WebSocketClientTestPeer::hasThread(*client)); + } + client.reset(); + EXPECT_EQ(notifications, 1); + auto closeFuture = serverClosed.get_future(); + EXPECT_EQ( + closeFuture.wait_for(std::chrono::seconds(5)), + std::future_status::ready); + if (closeFuture.wait_for(std::chrono::seconds(0)) == + std::future_status::ready) { + EXPECT_EQ(closeFuture.get(), boost::beast::websocket::error::closed); + } + serverIO.stop(); + serverThread.join(); + impl.reset(); + EXPECT_TRUE(weakImpl.expired()); + } +}; + +TEST_F( + WebSocketClientCallbackCloseTest, + CloseFromConnectKeepsThreadForOwnerJoin) { + verifyClose(false); +} + +TEST_F( + WebSocketClientCallbackCloseTest, + CloseFromMessageKeepsThreadForOwnerDestruction) { + verifyClose(true); +} + +} // namespace facebook::react diff --git a/packages/react-native/ReactCxxPlatform/react/runtime/ReactHost.cpp b/packages/react-native/ReactCxxPlatform/react/runtime/ReactHost.cpp index 61a36ce073c0..bc61f65da76d 100644 --- a/packages/react-native/ReactCxxPlatform/react/runtime/ReactHost.cpp +++ b/packages/react-native/ReactCxxPlatform/react/runtime/ReactHost.cpp @@ -108,6 +108,9 @@ ReactHost::ReactHost( ReactHost::~ReactHost() noexcept { destroyReactInstance(); + if (inspector_) { + inspector_->shutdown(); + } } void ReactHost::createReactInstance() { @@ -304,7 +307,8 @@ void ReactHost::createReactInstance() { void ReactHost::destroyReactInstance() { if (inspector_ != nullptr) { - reactInstance_->unregisterFromInspector(); + inspector_->runOnInspectorThreadSync( + [this]() { reactInstance_->unregisterFromInspector(); }); } if (surfaceManager_) { surfaceManager_->stopAllSurfaces();