diff --git a/packages/react-native/ReactCxxPlatform/react/threading/MessageQueueThreadImpl.cpp b/packages/react-native/ReactCxxPlatform/react/threading/MessageQueueThreadImpl.cpp new file mode 100644 index 00000000000..acbdfaf6031 --- /dev/null +++ b/packages/react-native/ReactCxxPlatform/react/threading/MessageQueueThreadImpl.cpp @@ -0,0 +1,38 @@ +/* + * 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 "MessageQueueThreadImpl.h" + +#include + +namespace facebook::react { + +void MessageQueueThreadImpl::runOnQueue(std::function&& runnable) { + if (!taskDispatchThread_.isRunning()) { + return; + } + taskDispatchThread_.runAsync( + [runnable = std::move(runnable)]() noexcept { runnable(); }); +} + +void MessageQueueThreadImpl::runOnQueueSync(std::function&& runnable) { + if (!taskDispatchThread_.isRunning()) { + return; + } + if (taskDispatchThread_.isOnThread()) { + runnable(); + } else { + taskDispatchThread_.runSync( + [runnable = std::move(runnable)]() noexcept { runnable(); }); + } +} + +void MessageQueueThreadImpl::quitSynchronous() { + taskDispatchThread_.quit(); +} + +} // namespace facebook::react diff --git a/packages/react-native/ReactCxxPlatform/react/threading/MessageQueueThreadImpl.h b/packages/react-native/ReactCxxPlatform/react/threading/MessageQueueThreadImpl.h new file mode 100644 index 00000000000..6b3c68be952 --- /dev/null +++ b/packages/react-native/ReactCxxPlatform/react/threading/MessageQueueThreadImpl.h @@ -0,0 +1,51 @@ +/* + * 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 + +namespace facebook::react { + +using MessageQueueThreadFactory = + std::function()>; + +constexpr char MessageQueueThreadFactoryKey[] = "MessageQueueThreadFactoryKey"; + +/** + * MessageQueueThread implementation that uses a TaskDispatchThread for + * queueing and threading logic + */ +class MessageQueueThreadImpl : public MessageQueueThread { + public: + MessageQueueThreadImpl() noexcept = default; + + ~MessageQueueThreadImpl() noexcept override = default; + + /** Add a job to the queue asynchronously */ + void runOnQueue(std::function&& runnable) override; + + /** + * Add a job to the queue synchronously - call won't return until runnable + * has completed. Will run immediately if called from the looper thread. + * Should only be used for initialization. + */ + void runOnQueueSync(std::function&& runnable) override; + + /** + * Stop the message queue thread. Should only be used for cleanup - once it + * returns, no further work should run on the queue. + */ + void quitSynchronous() override; + + private: + TaskDispatchThread taskDispatchThread_{"MessageQueue"}; +}; + +} // namespace facebook::react diff --git a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp new file mode 100644 index 00000000000..d7d3a24ca78 --- /dev/null +++ b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp @@ -0,0 +1,112 @@ +/* + * 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 "TaskDispatchThread.h" + +#include +#include +#include +#include + +#ifdef ANDROID +#include +#endif + +namespace facebook::react { + +TaskDispatchThread::TaskDispatchThread(std::string threadName) noexcept + : threadName_(std::move(threadName)) { +#ifdef ANDROID + // Attaches the thread to JVM just in case anything calls out to Java + thread_ = std::thread([&]() { + facebook::jni::ThreadScope::WithClassLoader([&]() { loop(); }); + }); +#else + thread_ = std::thread(&TaskDispatchThread::loop, this); +#endif +} + +TaskDispatchThread::~TaskDispatchThread() noexcept { + quit(); +} + +bool TaskDispatchThread::isOnThread() noexcept { + return std::this_thread::get_id() == thread_.get_id(); +} + +bool TaskDispatchThread::isRunning() noexcept { + return running_; +} + +void TaskDispatchThread::runAsync( + TaskFn&& task, + std::chrono::milliseconds delayMs) noexcept { + if (!running_) { + return; + } + std::lock_guard guard(queueLock_); + auto dispatchTime = std::chrono::system_clock::now() + delayMs; + queue_.emplace(dispatchTime, std::move(task)); + loopCv_.notify_one(); +} + +void TaskDispatchThread::runSync(TaskFn&& task) noexcept { + std::promise promise; + runAsync([&]() { + if (running_) { + task(); + } + promise.set_value(); + }); + promise.get_future().wait(); +} + +void TaskDispatchThread::quit() noexcept { + if (!running_) { + return; + } + running_ = false; + loopCv_.notify_one(); + if (thread_.joinable()) { + if (!isOnThread()) { + thread_.join(); + } else { + thread_.detach(); + } + } +} + +void TaskDispatchThread::loop() noexcept { + if (!threadName_.empty()) { + folly::setThreadName(threadName_); + } + while (running_) { + std::unique_lock lock(queueLock_); + loopCv_.wait(lock, [&]() { return !running_ || !queue_.empty(); }); + while (!queue_.empty()) { + auto task = queue_.top(); + auto now = std::chrono::system_clock::now(); + if (task.dispatchTime > now) { + if (running_) { + loopCv_.wait_until(lock, task.dispatchTime); + } else { + // Shutting down, skip all the delayed tasks that are not to be + // executed yet + queue_.pop(); + } + continue; + } + + queue_.pop(); + lock.unlock(); + task.fn(); + lock.lock(); + } + } +} + +} // namespace facebook::react diff --git a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h new file mode 100644 index 00000000000..660550f1973 --- /dev/null +++ b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h @@ -0,0 +1,73 @@ +/* + * 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 + +namespace facebook::react { + +/** + * Representation of a thread looper which can add tasks to a queue and handle + * the synchronization of callers. + */ +class TaskDispatchThread { + public: + using TaskFn = std::function; + using TimePoint = std::chrono::time_point; + + TaskDispatchThread(std::string threadName = "") noexcept; + ~TaskDispatchThread() noexcept; + + /** Return true if the current thread is the same as this looper's thread. */ + bool isOnThread() noexcept; + + /** Return true until TaskDispatchThread.quit() is called */ + bool isRunning() noexcept; + + /** Add task to the queue and return immediately. */ + void runAsync( + TaskFn&& task, + std::chrono::milliseconds delayMs = + std::chrono::milliseconds::zero()) noexcept; + + /** Add task to the queue and wait until it has completed. */ + void runSync(TaskFn&& task) noexcept; + + /** Shut down and clean up the thread. */ + void quit() noexcept; + + protected: + struct Task { + TimePoint dispatchTime; + TaskFn fn; + + Task(TimePoint dispatchTime, TaskFn&& fn) + : dispatchTime(dispatchTime), fn(std::move(fn)) {} + + bool operator<(const Task& other) const { + // Have the earliest tasks be at the front of the queue. + return dispatchTime > other.dispatchTime; + } + }; + + void loop() noexcept; + + std::mutex queueLock_; + std::condition_variable loopCv_; + std::priority_queue queue_; + std::atomic running_{true}; + std::string threadName_; + std::thread thread_; +}; + +} // namespace facebook::react