Skeleton shutdown logic for thread pool and cancellation of work

nodegit · implausible · Oct 20, 2020 · Jul 31, 2020 · Jul 31, 2020 · Jul 31, 2020
commit 42e266ed124b1524acaa693a499c08f506db543d
diff --git a/generate/templates/manual/include/async_baton.h b/generate/templates/manual/include/async_baton.h
@@ -28,7 +28,7 @@ namespace nodegit {
       Nan::AsyncResource *GetAsyncResource();
 
     protected:
-      void ExecuteAsyncPerform(AsyncCallback asyncCallback, CompletionCallback onCompletion);
+      void ExecuteAsyncPerform(AsyncCallback asyncCallback, AsyncCallback asyncCancelCb, CompletionCallback onCompletion);
 
     private:
       void SignalCompletion();
@@ -53,17 +53,17 @@ namespace nodegit {
         : defaultResult(defaultResult) {
       }
 
-      ResultT ExecuteAsync(AsyncBaton::AsyncCallback asyncCallback, AsyncBaton::CompletionCallback onCompletion = nullptr) {
+      ResultT ExecuteAsync(AsyncBaton::AsyncCallback asyncCallback, AsyncBaton::AsyncCallback asyncCancelCb, AsyncBaton::CompletionCallback onCompletion = nullptr) {
         result = 0;
-        ExecuteAsyncPerform(asyncCallback, onCompletion);
+        ExecuteAsyncPerform(asyncCallback, asyncCancelCb, onCompletion);
         return result;
       }
   };
 
   class AsyncBatonWithNoResult : public AsyncBaton {
     public:
-      void ExecuteAsync(AsyncBaton::AsyncCallback asyncCallback, AsyncBaton::CompletionCallback onCompletion = nullptr) {
-        ExecuteAsyncPerform(asyncCallback, onCompletion);
+      void ExecuteAsync(AsyncBaton::AsyncCallback asyncCallback, AsyncBaton::AsyncCallback asyncCancelCb, AsyncBaton::CompletionCallback onCompletion = nullptr) {
+        ExecuteAsyncPerform(asyncCallback, asyncCancelCb, onCompletion);
       }
   };
 }

diff --git a/generate/templates/manual/include/async_worker.h b/generate/templates/manual/include/async_worker.h
@@ -9,8 +9,18 @@ namespace nodegit {
   public:
     AsyncWorker(Nan::Callback *callback, const char *resourceName);
 
+    // This must be implemented by every async worker
+    // so that the thread pool can lock separately
+    // from the execute method in the AsyncWorker
     virtual nodegit::LockMaster AcquireLocks() = 0;
 
+    // Ensure that the `HandleErrorCallback` will be called
+    // when the AsyncWork is complete
+    void Cancel();
+
+    // Retrieves the async resource attached to this AsyncWorker
+    // This is used to inform libgit2 callbacks what asyncResource
+    // they should use when working with any javascript
     Nan::AsyncResource *GetAsyncResource();
   };
 }

diff --git a/generate/templates/manual/include/context.h b/generate/templates/manual/include/context.h
@@ -15,15 +15,17 @@ namespace nodegit {
   public:
     Context(v8::Isolate *isolate);
 
-    void QueueWorker(nodegit::AsyncWorker *worker);
+    ~Context();
 
-    void SaveToPersistent(std::string key, const v8::Local<v8::Value> &value);
+    static Context *GetCurrentContext();
 
     v8::Local<v8::Value> GetFromPersistent(std::string key);
 
-    static Context *GetCurrentContext();
+    void QueueWorker(nodegit::AsyncWorker *worker);
 
-    ~Context();
+    void SaveToPersistent(std::string key, const v8::Local<v8::Value> &value);
+
+    void ShutdownThreadPool();
 
   private:
     v8::Isolate *isolate;

diff --git a/generate/templates/manual/include/thread_pool.h b/generate/templates/manual/include/thread_pool.h
@@ -14,7 +14,7 @@ namespace nodegit {
   class ThreadPool {
     public:
       typedef std::function<void()> Callback;
-      typedef std::function<void(Callback)> QueueCallbackFn;
+      typedef std::function<void(Callback, Callback)> QueueCallbackFn;
       typedef std::function<Callback(QueueCallbackFn, Callback)> OnPostCallbackFn;
 
       // Initializes thread pool and spins up the requested number of threads
@@ -41,6 +41,10 @@ namespace nodegit {
       // Called once at libgit2 initialization to setup contracts with libgit2
       static void InitializeGlobal();
 
+      // Will wait for all threads to terminate before returning
+      // It will also clean up any resources that the thread pool is keeping alive
+      void Shutdown();
+
     private:
       std::unique_ptr<ThreadPoolImpl> impl;
   };

diff --git a/generate/templates/manual/src/async_baton.cc b/generate/templates/manual/src/async_baton.cc
@@ -23,36 +23,39 @@ namespace nodegit {
     return asyncResource;
   }
 
-  void AsyncBaton::ExecuteAsyncPerform(AsyncCallback asyncCallback, CompletionCallback onCompletion) {
+  void AsyncBaton::ExecuteAsyncPerform(AsyncCallback asyncCallback, AsyncCallback asyncCancelCb, CompletionCallback onCompletion) {
     auto jsCallback = [asyncCallback, this]() {
       asyncCallback(this);
     };
+    auto cancelCallback = [asyncCancelCb, this]() {
+      asyncCancelCb(this);
+    };
 
     if (onCompletion) {
       this->onCompletion = [this, onCompletion]() {
         onCompletion(this);
       };
 
       ThreadPool::PostCallbackEvent(
-        [this, jsCallback](
+        [this, jsCallback, cancelCallback](
           ThreadPool::QueueCallbackFn queueCallback,
           ThreadPool::Callback callbackCompleted
         ) -> ThreadPool::Callback {
-          queueCallback(jsCallback);
+          queueCallback(jsCallback, cancelCallback);
           callbackCompleted();
 
           return []() {};
         }
       );
     } else {
       ThreadPool::PostCallbackEvent(
-        [this, jsCallback](
+        [this, jsCallback, cancelCallback](
           ThreadPool::QueueCallbackFn queueCallback,
           ThreadPool::Callback callbackCompleted
         ) -> ThreadPool::Callback {
           this->onCompletion = callbackCompleted;
 
-          queueCallback(jsCallback);
+          queueCallback(jsCallback, cancelCallback);
 
           return std::bind(&AsyncBaton::SignalCompletion, this);
         }

diff --git a/generate/templates/manual/src/async_worker.cc b/generate/templates/manual/src/async_worker.cc
@@ -5,6 +5,13 @@ namespace nodegit {
     : Nan::AsyncWorker(callback, resourceName)
   {}
 
+  void AsyncWorker::Cancel() {
+    // We use Nan::AsyncWorker's ErrorMessage flow
+    // to trigger `HandleErrorCallback` for cancellation
+    // of AsyncWork
+    SetErrorMessage("SHUTTING DOWN");
+  }
+
   Nan::AsyncResource *AsyncWorker::GetAsyncResource() {
     return async_resource;
   }

diff --git a/generate/templates/manual/src/context.cc b/generate/templates/manual/src/context.cc
@@ -1,27 +1,35 @@
 #include "../include/context.h"
 
 namespace nodegit {
+  std::map<v8::Isolate *, Context *> Context::contexts;
+
+  static void CleanupContext(void *data) {
+    Context *context = static_cast<Context *>(data);
+
+    context->ShutdownThreadPool();
+
+    delete context;
+  }
+
   Context::Context(v8::Isolate *isolate)
     : isolate(isolate), threadPool(10, node::GetCurrentEventLoop(isolate))
   {
     Nan::HandleScope scopoe;
     v8::Local<v8::Object> storage = Nan::New<v8::Object>();
     persistentStorage.Reset(storage);
     contexts[isolate] = this;
+    node::AddEnvironmentCleanupHook(isolate, CleanupContext, this);
   }
 
   Context::~Context() {
     contexts.erase(isolate);
   }
 
-  void Context::QueueWorker(nodegit::AsyncWorker *worker) {
-    threadPool.QueueWorker(worker);
-  }
-
-  void Context::SaveToPersistent(std::string key, const v8::Local<v8::Value> &value) {
+  Context *Context::GetCurrentContext() {
     Nan::HandleScope scope;
-    v8::Local<v8::Object> storage = Nan::New(persistentStorage);
-    Nan::Set(storage, Nan::New(key).ToLocalChecked(), value);
+    v8::Local<v8::Context> context = Nan::GetCurrentContext();
+    v8::Isolate *isolate = context->GetIsolate();
+    return contexts[isolate];
   }
 
   v8::Local<v8::Value> Context::GetFromPersistent(std::string key) {
@@ -31,12 +39,17 @@ namespace nodegit {
     return scope.Escape(value.ToLocalChecked());
   }
 
-  Context *Context::GetCurrentContext() {
+  void Context::QueueWorker(nodegit::AsyncWorker *worker) {
+    threadPool.QueueWorker(worker);
+  }
+
+  void Context::SaveToPersistent(std::string key, const v8::Local<v8::Value> &value) {
     Nan::HandleScope scope;
-    v8::Local<v8::Context> context = Nan::GetCurrentContext();
-    v8::Isolate *isolate = context->GetIsolate();
-    return contexts[isolate];
+    v8::Local<v8::Object> storage = Nan::New(persistentStorage);
+    Nan::Set(storage, Nan::New(key).ToLocalChecked(), value);
   }
 
-  std::map<v8::Isolate *, Context *> Context::contexts;
+  void Context::ShutdownThreadPool() {
+    threadPool.Shutdown();
+  }
 }