WIP: Pluggable thread pool

node.gyp

@@ -363,6 +363,7 @@
         'src/node_util.cc',
         'src/node_v8.cc',
         'src/node_stat_watcher.cc',
+        'src/node_threadpool.cc',
         'src/node_watchdog.cc',
         'src/node_worker.cc',
         'src/node_zlib.cc',
@@ -421,6 +422,7 @@
         'src/node_persistent.h',
         'src/node_platform.h',
         'src/node_root_certs.h',
+        'src/node_threadpool.h',
         'src/node_version.h',
         'src/node_watchdog.h',
         'src/node_wrap.h',
@@ -963,6 +965,7 @@
         'test/cctest/test_node_postmortem_metadata.cc',
         'test/cctest/test_environment.cc',
         'test/cctest/test_platform.cc',
+        'test/cctest/test_threadpool.cc',
         'test/cctest/test_traced_value.cc',
         'test/cctest/test_util.cc',
         'test/cctest/test_url.cc'

src/node.cc

@@ -24,6 +24,7 @@
 #include "node_javascript.h"
 #include "node_code_cache.h"
 #include "node_platform.h"
+#include "node_threadpool.h"
 #include "node_version.h"
 #include "node_internals.h"
 #include "node_revert.h"
@@ -283,8 +284,20 @@ class NodeTraceStateObserver :
   v8::TracingController* controller_;
 };
 
+static struct {
+  // Returns zero on success
+  int Initialize(void) {
+    tp_.reset(new threadpool::Threadpool());
+    tp_->Initialize();
+    return uv_replace_executor(tp_->GetExecutor());
+  }
+
+  std::unique_ptr<threadpool::Threadpool> tp_;
+} node_threadpool;
+
 static struct {
 #if NODE_USE_V8_PLATFORM
+  // TODO(davisjam): Pass in the node_threadpool.
   void Initialize(int thread_pool_size) {
     tracing_agent_.reset(new tracing::Agent());
     auto controller = tracing_agent_->GetTracingController();
@@ -3338,6 +3351,10 @@ int Start(int argc, char** argv) {
   V8::SetEntropySource(crypto::EntropySource);
 #endif  // HAVE_OPENSSL
 
+  // This needs to run before any work is queued to the libuv executor.
+  CHECK_EQ(0, node_threadpool.Initialize());
+
+  // TODO(davisjam): Pass the v8_platform the node_threadpool.
   v8_platform.Initialize(
       per_process_opts->v8_thread_pool_size);
   V8::Initialize();

src/node_crypto.cc

@@ -4584,7 +4584,9 @@ bool ECDH::IsKeyPairValid() {
 struct CryptoJob : public ThreadPoolWork {
   Environment* const env;
   std::unique_ptr<AsyncWrap> async_wrap;
-  inline explicit CryptoJob(Environment* env) : ThreadPoolWork(env), env(env) { SetOptionsType(UV_WORK_USER_CPU); }
+  inline explicit CryptoJob(Environment* env) : ThreadPoolWork(env), env(env) {
+    SetOptionsType(UV_WORK_USER_CPU);
+  }
   inline void AfterThreadPoolWork(int status) final;
   virtual void AfterThreadPoolWork() = 0;
   static inline void Run(std::unique_ptr<CryptoJob> job, Local<Value> wrap);

src/node_internals.h

@@ -502,6 +502,7 @@ class InternalCallbackScope {
   bool closed_ = false;
 };
 
+// TODO(davisjam): Update to use node_threadpool.
 class ThreadPoolWork {
  public:
   explicit inline ThreadPoolWork(Environment* env) : env_(env) {
@@ -512,10 +513,18 @@ class ThreadPoolWork {
   }
   inline virtual ~ThreadPoolWork() = default;
 
-  inline void SetOptionsType(uv_work_type type) { work_req_options_.type = type; }
-  inline void SetOptionsPriority(int priority) { work_req_options_.priority = priority; }
-  inline void SetOptionsCancelable(int cancelable) { work_req_options_.cancelable = cancelable; }
-  inline void SetOptionsData(void *data) { work_req_options_.data = data; }
+  inline void SetOptionsType(uv_work_type type) {
+    work_req_options_.type = type;
+  }
+  inline void SetOptionsPriority(int priority) {
+    work_req_options_.priority = priority;
+  }
+  inline void SetOptionsCancelable(int cancelable) {
+    work_req_options_.cancelable = cancelable;
+  }
+  inline void SetOptionsData(void* data) {
+    work_req_options_.data = data;
+  }
 
   inline void ScheduleWork();
   inline int CancelWork();
@@ -531,6 +540,10 @@ class ThreadPoolWork {
 
 void ThreadPoolWork::ScheduleWork() {
   env_->IncreaseWaitingRequestCounter();
+  // TODO(davisjam): Should we route to node TP instead?
+  // I don't think so.
+  // These are pending user requests with a CB for the event loop.
+  // So it makes sense for libuv to handle them start-to-finish.
   int status = uv_executor_queue_work(
       env_->event_loop(),
       &work_req_,

src/node_platform.cc

@@ -125,6 +125,8 @@ class WorkerThreadsTaskRunner::DelayedTaskScheduler {
   static void RunTask(uv_timer_t* timer) {
     DelayedTaskScheduler* scheduler =
         ContainerOf(&DelayedTaskScheduler::loop_, timer->loop);
+    // This adds the Task to the TP queue.
+    // TODO(davisjam): Plug in TP implementation.
     scheduler->pending_worker_tasks_->Push(scheduler->TakeTimerTask(timer));
   }
 
@@ -330,6 +332,7 @@ void NodePlatform::DrainTasks(Isolate* isolate) {
 
   do {
     // Worker tasks aren't associated with an Isolate.
+    // TODO(davisjam): This will require some dancing with the TP.
     worker_thread_task_runner_->BlockingDrain();
   } while (per_isolate->FlushForegroundTasksInternal());
 }
@@ -371,6 +374,7 @@ bool PerIsolatePlatformData::FlushForegroundTasksInternal() {
 }
 
 void NodePlatform::CallOnWorkerThread(std::unique_ptr<v8::Task> task) {
+  // TODO(davisjam): Plug in TP implementation
   worker_thread_task_runner_->PostTask(std::move(task));
 }
 

src/node_platform.h

@@ -96,11 +96,14 @@ class PerIsolatePlatformData :
 };
 
 // This acts as the single worker thread task runner for all Isolates.
+// API is modeled on v8::TaskRunner.
 class WorkerThreadsTaskRunner {
  public:
   explicit WorkerThreadsTaskRunner(int thread_pool_size);
 
+  // Add task to queue for eventual Run()
   void PostTask(std::unique_ptr<v8::Task> task);
+  // Add task to queue after at least delay_in_seconds
   void PostDelayedTask(std::unique_ptr<v8::Task> task,
                        double delay_in_seconds);
 
@@ -110,6 +113,7 @@ class WorkerThreadsTaskRunner {
   int NumberOfWorkerThreads() const;
 
  private:
+  // Push'd directly by PostTask() and indirectly by PostDelayedTask.
   TaskQueue<v8::Task> pending_worker_tasks_;
 
   class DelayedTaskScheduler;

src/node_threadpool.cc

@@ -0,0 +1,225 @@
+#include "node_threadpool.h"
+#include "node_internals.h"
+
+#include "env-inl.h"
+#include "debug_utils.h"
+#include "util.h"
+#include <algorithm>
+
+// TODO(davisjam): DO NOT MERGE. Only for debugging.
+// TODO(davisjam): There must be a better way to do this.
+#define DEBUG_LOG 1
+#undef DEBUG_LOG
+
+#ifdef DEBUG_LOG
+#include <stdio.h>
+#define LOG_0(fmt) fprintf(stderr, fmt)
+#define LOG_1(fmt, a1) fprintf(stderr, fmt, a1)
+#define LOG_2(fmt, a1, a2) fprintf(stderr, fmt, a1, a2)
+#define LOG_3(fmt, a1, a2, a3) fprintf(stderr, fmt, a1, a2, a3)
+#define LOG_4(fmt, a1, a2, a3, a4) fprintf(stderr, fmt, a1, a2, a3, a4)
+#define LOG_5(fmt, a1, a2, a3, a4, a5) fprintf(stderr, fmt, a1, a2, a3, a4, a5)
+#else
+#define LOG_0(fmt) (void) 0
+#define LOG_1(fmt, a1) (void) 0
+#define LOG_2(fmt, a1, a2) (void) 0
+#define LOG_3(fmt, a1, a2, a3) (void) 0
+#define LOG_4(fmt, a1, a2, a3, a4) (void) 0
+#define LOG_5(fmt, a1, a2, a3, a4, a5) (void) 0
+#endif
+
+namespace node {
+namespace threadpool {
+
+/**************
+ * Worker
+ ***************/
+
+Worker::Worker() {
+}
+
+void Worker::Start(TaskQueue* queue) {
+  CHECK_EQ(0, uv_thread_create(&self_, _Run, reinterpret_cast<void *>(queue)));
+}
+
+void Worker::Join(void) {
+  CHECK_EQ(0, uv_thread_join(&self_));
+}
+
+void Worker::_Run(void* data) {
+  TaskQueue* queue = static_cast<TaskQueue*>(data);
+  while (std::unique_ptr<Task> task = queue->BlockingPop()) {
+    task->Run();
+  }
+}
+
+/**************
+ * LibuvTask
+ ***************/
+
+LibuvTask::LibuvTask(Threadpool* tp,
+                     uv_work_t* req,
+                     const uv_work_options_t* opts)
+  : Task(), tp_(tp), req_(req) {
+  req_ = req;
+
+  // Copy opts.
+  if (opts) {
+    switch (opts->type) {
+    case UV_WORK_FS:
+      details_.type = TaskDetails::FS;
+      break;
+    case UV_WORK_DNS:
+      details_.type = TaskDetails::DNS;
+      break;
+    case UV_WORK_USER_IO:
+      details_.type = TaskDetails::IO;
+      break;
+    case UV_WORK_USER_CPU:
+      details_.type = TaskDetails::CPU;
+      break;
+    default:
+      details_.type = TaskDetails::UNKNOWN;
+    }
+
+    details_.priority = opts->priority;
+    details_.cancelable = opts->cancelable;
+  } else {
+    details_.type = TaskDetails::UNKNOWN;
+    details_.priority = -1;
+    details_.cancelable = false;
+  }
+
+  LOG_1("LibuvTask::LibuvTask: type %d\n", details_.type);
+}
+
+LibuvTask::~LibuvTask(void) {
+  LOG_1("LibuvTask::Run: Task %p done\n", req_);
+  tp_->GetExecutor()->done(req_);
+}
+
+void LibuvTask::Run() {
+  LOG_1("LibuvTask::Run: Running Task %p\n", req_);
+  req_->work_cb(req_);
+}
+
+/**************
+ * TaskQueue
+ ***************/
+
+TaskQueue::TaskQueue()
+  : queue_(), stopped_(false), lock_(), tasks_available_() {
+}
+
+bool TaskQueue::Push(std::unique_ptr<Task> task) {
+  Mutex::ScopedLock scoped_lock(lock_);
+
+  if (stopped_) {
+    return false;
+  }
+
+  queue_.push(std::move(task));
+  tasks_available_.Signal(scoped_lock);
+  return true;
+}
+
+std::unique_ptr<Task> TaskQueue::Pop(void) {
+  Mutex::ScopedLock scoped_lock(lock_);
+
+  if (queue_.empty()) {
+    return std::unique_ptr<Task>(nullptr);
+  }
+
+  std::unique_ptr<Task> task = std::move(queue_.front());
+  queue_.pop();
+  return task;
+}
+
+std::unique_ptr<Task> TaskQueue::BlockingPop(void) {
+  Mutex::ScopedLock scoped_lock(lock_);
+
+  while (queue_.empty() && !stopped_) {
+    tasks_available_.Wait(scoped_lock);
+  }
+
+  if (queue_.empty()) {
+    return std::unique_ptr<Task>(nullptr);
+  }
+
+  std::unique_ptr<Task> result = std::move(queue_.front());
+  queue_.pop();
+  return result;
+}
+
+void TaskQueue::Stop(void) {
+  Mutex::ScopedLock scoped_lock(lock_);
+  stopped_ = true;
+  tasks_available_.Broadcast(scoped_lock);
+}
+
+int TaskQueue::Length(void) const {
+  Mutex::ScopedLock scoped_lock(lock_);
+  return queue_.size();
+}
+
+/**************
+ * Threadpool
+ ***************/
+
+Threadpool::Threadpool(void)
+  : queue_(), workers_() {
+  executor_.init = uv_executor_init;
+  executor_.destroy = nullptr;
+  executor_.submit = uv_executor_submit;
+  executor_.cancel = nullptr;
+  executor_.data = this;
+}
+
+Threadpool::~Threadpool(void) {
+  // Block future Push's.
+  queue_.Stop();
+
+  // Wait for Workers to drain the queue.
+  for (auto& worker : workers_) {
+    worker->Join();
+  }
+}
+
+// TODO(davisjam): Return early on multiple initialization
+void Threadpool::Initialize(void) {
+  int n_workers = 4;  // TODO(davisjam):
+
+  for (int i = 0; i < n_workers; i++) {
+    std::unique_ptr<Worker> worker(new Worker());
+    worker->Start(&queue_);
+    workers_.push_back(std::move(worker));
+  }
+}
+
+void Threadpool::Post(std::unique_ptr<Task> task) {
+  LOG_1("Threadpool::Post: Got task of type %d\n",
+    task->details_.type);
+  queue_.Push(std::move(task));
+}
+
+int Threadpool::QueueLength(void) const {
+  return queue_.Length();
+}
+
+void Threadpool::uv_executor_init(uv_executor_t* executor) {
+}
+
+void Threadpool::uv_executor_submit(uv_executor_t* executor,
+                                    uv_work_t* req,
+                                    const uv_work_options_t* opts) {
+  Threadpool* threadpool = reinterpret_cast<Threadpool *>(executor->data);
+  LOG_0("Threadpool::uv_executor_submit: Got some work!\n");
+  threadpool->Post(std::unique_ptr<Task>(new LibuvTask(threadpool, req, opts)));
+}
+
+uv_executor_t* Threadpool::GetExecutor() {
+  return &executor_;
+}
+
+}  // namespace threadpool
+}  // namespace node