move rncxx threading (#50702)

Summary:
Pull Request resolved: https://github.com/facebook/react-native/pull/50702

Changelog: [Internal]

Reviewed By: rshest

Differential Revision: D72865847

fbshipit-source-id: 0477670ca1f99089aece9b21feb2482b56f43936

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 <functional>
+
+namespace facebook::react {
+
+void MessageQueueThreadImpl::runOnQueue(std::function<void()>&& runnable) {
+  if (!taskDispatchThread_.isRunning()) {
+    return;
+  }
+  taskDispatchThread_.runAsync(
+      [runnable = std::move(runnable)]() noexcept { runnable(); });
+}
+
+void MessageQueueThreadImpl::runOnQueueSync(std::function<void()>&& 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

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 <cxxreact/MessageQueueThread.h>
+#include <react/threading/TaskDispatchThread.h>
+#include <memory>
+
+namespace facebook::react {
+
+using MessageQueueThreadFactory =
+    std::function<std::shared_ptr<MessageQueueThread>()>;
+
+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<void()>&& 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<void()>&& 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

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 <folly/system/ThreadName.h>
+#include <chrono>
+#include <future>
+#include <utility>
+
+#ifdef ANDROID
+#include <fbjni/fbjni.h>
+#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<std::mutex> 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<void> 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<std::mutex> 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

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 <chrono>
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <queue>
+#include <thread>
+
+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<void()>;
+  using TimePoint = std::chrono::time_point<std::chrono::system_clock>;
+
+  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<Task> queue_;
+  std::atomic<bool> running_{true};
+  std::string threadName_;
+  std::thread thread_;
+};
+
+} // namespace facebook::react