Fix task queue

Author: mrousavy

cpp/RNMultithreadingInstaller.cpp

@@ -3,17 +3,30 @@
 #include <RNReanimated/Scheduler.h>
 #include <RNReanimated/ShareableValue.h>
 #include <RNReanimated/RuntimeManager.h>
+#include <RNReanimated/RuntimeDecorator.h>
+#include <RNReanimated/ErrorHandler.h>
+#include "MakeJSIRuntime.h"
 
 #define MAX_THREAD_COUNT 2
 
 namespace mrousavy {
 namespace multithreading {
 
-static ThreadPool pool(MAX_THREAD_COUNT);
+static ThreadPool pool(1);
+static std::unique_ptr<reanimated::RuntimeManager> manager;
 
 //reanimated::RuntimeManager manager;
 
 void install(jsi::Runtime& runtime) {
+  // Quickly setup the runtime - this is executed in parallel, and _might_ introduce race conditions if spawnThread is called before this finishes.
+  pool.enqueue([]() {
+    auto runtime = makeJSIRuntime();
+    reanimated::RuntimeDecorator::decorateRuntime(*runtime, "CUSTOM_THREAD");
+    manager = std::make_unique<reanimated::RuntimeManager>(std::move(runtime),
+                                                           std::shared_ptr<reanimated::ErrorHandler>(),
+                                                           std::shared_ptr<reanimated::Scheduler>());
+  });
+  
   // spawnThread(run: () => T): Promise<T>
   auto spawnThread = jsi::Function::createFromHostFunction(runtime,
                                                            jsi::PropNameID::forAscii(runtime, "spawnThread"),
@@ -37,10 +50,11 @@ void install(jsi::Runtime& runtime) {
           .call(runtime, jsi::JSError(runtime, message).value());
       };
       // TODO: Get correct RuntimeManager instance
-      auto run = reanimated::ShareableValue::adapt(runtime, arguments[0], nullptr);
+      auto run = reanimated::ShareableValue::adapt(runtime, arguments[0], manager.get());
 
-      pool.enqueue([&resolver, &rejecter, run](jsi::Runtime& runtime) {
+      pool.enqueue([&resolver, &rejecter, run]() {
         try {
+          auto& runtime = *manager->runtime;
           auto func = run->getValue(runtime).asObject(runtime).asFunction(runtime);
           auto result = func.callWithThis(runtime, func);
 

cpp/ThreadPool.cpp

@@ -18,13 +18,8 @@ namespace multithreading {
 ThreadPool::ThreadPool(size_t threads): stop(false) {
   for (size_t i = 0; i < threads; ++i) {
     workers.emplace_back([this, i] {
-      auto runtime = makeJSIRuntime();
-      std::stringstream stringstream;
-      stringstream << "THREAD #" << i;
-      reanimated::RuntimeDecorator::decorateRuntime(*runtime, stringstream.str());
-      
       while (true) {
-        task_t task;
+        std::function<void()> task;
 
         {
           std::unique_lock<std::mutex> lock(this->queue_mutex);
@@ -36,24 +31,29 @@ ThreadPool::ThreadPool(size_t threads): stop(false) {
           this->tasks.pop();
         }
 
-        task(*runtime);
+        task();
       }
     });
   }
 }
 
 // add new work item to the pool
-void ThreadPool::enqueue(task_t task) {
+std::future<void> ThreadPool::enqueue(std::function<void()> func) {
+  auto task = std::make_shared<std::packaged_task<void()>>(std::bind(std::forward<std::function<void()>>(func)));
+  std::future<void> res = task->get_future();
+  
   {
     std::unique_lock<std::mutex> lock(queue_mutex);
 
     // don't allow enqueueing after stopping the pool
     if (stop)
       throw std::runtime_error("enqueue on stopped ThreadPool");
 
-    tasks.emplace(task);
+    tasks.emplace([task](){ (*task)(); });
   }
+  
   condition.notify_one();
+  return res;
 }
 
 // the destructor joins all threads

cpp/ThreadPool.h

@@ -15,21 +15,19 @@
 #include <unordered_map>
 #include <jsi/jsi.h>
 
-typedef std::function<void (facebook::jsi::Runtime&)> task_t;
-
 namespace mrousavy {
 namespace multithreading {
 
 class ThreadPool {
 public:
   ThreadPool(size_t threadCount);
-  void enqueue(task_t task);
+  std::future<void> enqueue(std::function<void()> task);
   ~ThreadPool();
 private:
   // need to keep track of threads so we can join them
   std::vector<std::thread> workers;
   // the task queue
-  std::queue<task_t> tasks;
+  std::queue<std::function<void()>> tasks;
 
   // synchronization
   std::mutex queue_mutex;