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) {}
+  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,11 +502,30 @@ class InternalCallbackScope {
   bool closed_ = false;
 };
 
+// TODO(davisjam): Update to use node_threadpool.
 class ThreadPoolWork {
  public:
-  explicit inline ThreadPoolWork(Environment* env) : env_(env) {}
+  explicit inline ThreadPoolWork(Environment* env) : env_(env) {
+    work_req_options_.type = UV_WORK_UNKNOWN;
+    work_req_options_.priority = -1;
+    work_req_options_.cancelable = 0;
+    work_req_options_.data = nullptr;
+  }
   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 ScheduleWork();
   inline int CancelWork();
 
@@ -516,13 +535,19 @@ class ThreadPoolWork {
  private:
   Environment* env_;
   uv_work_t work_req_;
+  uv_work_options_t work_req_options_;
 };
 
 void ThreadPoolWork::ScheduleWork() {
   env_->IncreaseWaitingRequestCounter();
-  int status = uv_queue_work(
+  // 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_,
+      &work_req_options_,
       [](uv_work_t* req) {
         ThreadPoolWork* self = ContainerOf(&ThreadPoolWork::work_req_, req);
         self->DoThreadPoolWork();

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;