From d61245dc1944d3cfb385ccdb5f781fc266108dc9 Mon Sep 17 00:00:00 2001 From: Peter Abbondanzo Date: Mon, 12 Oct 2026 13:05:37 -0700 Subject: [PATCH] Synchronize native Fantom stub queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: Fantom’s native `StubQueue` can be written by multiple threads while it is being drained, but its queue access was unsynchronized. Protect queue storage with a mutex while keeping callback invocation and destruction outside the lock, so callbacks can reenter the queue. This is a tester-only change. The single-consumer and queue-lifetime requirements are unchanged; startup readiness and shared runtime teardown are out of scope. Changelog: [Internal] - Synchronize native Fantom stub queue access Differential Revision: D124675719 --- .../tester/src/stubs/StubQueue.cpp | 28 ++- .../tester/src/stubs/StubQueue.h | 10 +- .../tester/tests/StubQueueTests.cpp | 195 ++++++++++++++++++ 3 files changed, 220 insertions(+), 13 deletions(-) create mode 100644 private/react-native-fantom/tester/tests/StubQueueTests.cpp diff --git a/private/react-native-fantom/tester/src/stubs/StubQueue.cpp b/private/react-native-fantom/tester/src/stubs/StubQueue.cpp index 4ba08956bcf8..1016b3f51e93 100644 --- a/private/react-native-fantom/tester/src/stubs/StubQueue.cpp +++ b/private/react-native-fantom/tester/src/stubs/StubQueue.cpp @@ -10,7 +10,11 @@ namespace facebook::react { void StubQueue::runOnQueue(std::function&& func) { - callbackQueue_.push(func); + auto callback = std::make_unique>(std::move(func)); + { + std::lock_guard lock(mutex_); + callbackQueue_.push(std::move(callback)); + } } void StubQueue::runOnQueueSync(std::function&& runnable) { @@ -23,23 +27,29 @@ void StubQueue::quitSynchronous() { } void StubQueue::flush() { - while (!callbackQueue_.empty()) { - tick(); + while (tick()) { } } bool StubQueue::hasPendingCallbacks() { + std::lock_guard lock(mutex_); return !callbackQueue_.empty(); } -void StubQueue::tick() { - if (!callbackQueue_.empty()) { - auto callback = callbackQueue_.front(); - callbackQueue_.pop(); - if (callback) { - callback(); +bool StubQueue::tick() { + std::unique_ptr> callback; + { + std::lock_guard lock(mutex_); + if (callbackQueue_.empty()) { + return false; } + callback = std::move(callbackQueue_.front()); + callbackQueue_.pop(); + } + if (*callback) { + (*callback)(); } + return true; } } // namespace facebook::react diff --git a/private/react-native-fantom/tester/src/stubs/StubQueue.h b/private/react-native-fantom/tester/src/stubs/StubQueue.h index 83da413309d6..0592c06e2de7 100644 --- a/private/react-native-fantom/tester/src/stubs/StubQueue.h +++ b/private/react-native-fantom/tester/src/stubs/StubQueue.h @@ -8,8 +8,8 @@ #pragma once #include -#include -#include +#include +#include #include namespace facebook::react { @@ -24,12 +24,14 @@ class StubQueue : public MessageQueueThread { bool hasPendingCallbacks(); + // Draining belongs to the single consumer thread. void flush(); private: - void tick(); + bool tick(); - std::queue> callbackQueue_; + std::mutex mutex_; + std::queue>> callbackQueue_; }; } // namespace facebook::react diff --git a/private/react-native-fantom/tester/tests/StubQueueTests.cpp b/private/react-native-fantom/tester/tests/StubQueueTests.cpp new file mode 100644 index 000000000000..5b348b3a1521 --- /dev/null +++ b/private/react-native-fantom/tester/tests/StubQueueTests.cpp @@ -0,0 +1,195 @@ +/* + * 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 + +namespace facebook::react { + +TEST(StubQueueTest, FlushPreservesOrderAndAllowsReentrantEnqueue) { + StubQueue queue; + std::vector callbacks; + queue.runOnQueue([&] { + callbacks.push_back(1); + queue.runOnQueue([&] { callbacks.push_back(3); }); + queue.flush(); + }); + queue.runOnQueue([&] { callbacks.push_back(2); }); + + queue.flush(); + + EXPECT_EQ(callbacks, (std::vector{1, 2, 3})); + EXPECT_FALSE(queue.hasPendingCallbacks()); +} + +TEST(StubQueueTest, FlushDoesNotCopySmallNoexceptCallable) { + struct State { + int copies{0}; + int calls{0}; + }; + struct Callback { + explicit Callback(std::shared_ptr state) : state(std::move(state)) {} + Callback(const Callback& other) noexcept : state(other.state) { + ++state->copies; + } + Callback& operator=(const Callback&) = default; + void operator()() const { + ++state->calls; + } + std::shared_ptr state; + }; + StubQueue queue; + auto state = std::make_shared(); + queue.runOnQueue(Callback(state)); + state->copies = 0; + + queue.flush(); + + EXPECT_EQ(state->calls, 1); + EXPECT_EQ(state->copies, 0); +} + +TEST(StubQueueTest, CallableTransfersAllowReentrantCopyAndDestruction) { + std::promise result; + auto completed = result.get_future(); + std::thread worker([result = std::move(result)]() mutable { + StubQueue queue; + struct State { + StubQueue* queue{nullptr}; + bool armed{false}; + int calls{0}; + }; + struct Callback { + explicit Callback(std::shared_ptr state) + : state(std::move(state)) {} + Callback(const Callback& other) noexcept : state(other.state) { + probe(); + } + Callback(Callback&&) noexcept = default; + Callback& operator=(const Callback&) = default; + Callback& operator=(Callback&&) noexcept = delete; + ~Callback() { + probe(); + } + void probe() const noexcept { + if (state && state->armed) { + (void)state->queue->hasPendingCallbacks(); + } + } + void operator()() const { + ++state->calls; + } + std::shared_ptr state; + }; + auto state = std::make_shared(); + state->queue = &queue; + std::function callback = Callback(state); + state->armed = true; + queue.runOnQueue(std::move(callback)); + callback = nullptr; + queue.flush(); + result.set_value(state->calls); + }); + auto status = completed.wait_for(std::chrono::seconds(5)); + if (status == std::future_status::ready) { + worker.join(); + } else { + // NOLINTNEXTLINE(facebook-hte-BadCall-detach): failure-only private state. + worker.detach(); + } + ASSERT_EQ(status, std::future_status::ready); + EXPECT_EQ(completed.get(), 1); +} + +TEST(StubQueueTest, FlushSkipsEmptyCallbacks) { + StubQueue queue; + int calls = 0; + queue.runOnQueue({}); + queue.runOnQueue([&] { ++calls; }); + + queue.flush(); + + EXPECT_EQ(calls, 1); + EXPECT_FALSE(queue.hasPendingCallbacks()); +} + +TEST(StubQueueTest, ConcurrentProducersExecuteExactlyOnceInProducerOrder) { + StubQueue queue; + constexpr int producerCount = 8; + constexpr int callbacksPerProducer = 10000; + const auto consumer = std::this_thread::get_id(); + std::promise start; + auto started = start.get_future().share(); + std::atomic completedProducers{0}; + int executed = 0; + std::vector callbacks(producerCount, 0); + std::vector producers; + producers.reserve(producerCount); + for (int producer = 0; producer < producerCount; ++producer) { + producers.emplace_back([&, producer] { + started.wait(); + for (int index = 0; index < callbacksPerProducer; ++index) { + queue.runOnQueue([&, producer, index] { + EXPECT_EQ(std::this_thread::get_id(), consumer); + EXPECT_EQ(callbacks[producer], index); + ++callbacks[producer]; + ++executed; + }); + } + ++completedProducers; + }); + } + + start.set_value(); + while (completedProducers.load() != producerCount) { + queue.flush(); + std::this_thread::yield(); + } + for (auto& producer : producers) { + producer.join(); + } + queue.flush(); + + EXPECT_EQ(executed, producerCount * callbacksPerProducer); + EXPECT_EQ(callbacks, std::vector(producerCount, callbacksPerProducer)); + EXPECT_FALSE(queue.hasPendingCallbacks()); +} + +TEST(StubQueueTest, SynchronousWorkRunsAfterPendingCallbacks) { + StubQueue queue; + std::vector callbacks; + queue.runOnQueue([&] { callbacks.push_back(1); }); + + queue.runOnQueueSync([&] { callbacks.push_back(2); }); + + EXPECT_EQ(callbacks, (std::vector{1, 2})); +} + +TEST(StubQueueTest, ThrowingCallbackLeavesLaterWorkDrainable) { + StubQueue queue; + std::vector callbacks; + queue.runOnQueue([&] { + callbacks.push_back(1); + throw std::runtime_error("queued failure"); + }); + queue.runOnQueue([&] { callbacks.push_back(2); }); + + EXPECT_THROW(queue.flush(), std::runtime_error); + EXPECT_TRUE(queue.hasPendingCallbacks()); + queue.flush(); + + EXPECT_EQ(callbacks, (std::vector{1, 2})); + EXPECT_FALSE(queue.hasPendingCallbacks()); +} + +} // namespace facebook::react