Skip to content

Commit 41e39b4

Browse files
srajkosrajko
committed
Run async libgit2 calls on custom threadpool
1 parent ebc73a8 commit 41e39b4

8 files changed

Lines changed: 169 additions & 1 deletion

File tree

‎generate/templates/manual/include/async_libgit2_queue_worker.h‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
#ifndef ASYNC_LIBGIT2_QUEUE_WORKER_H
2+
#define ASYNC_LIBGIT2_QUEUE_WORKER_H
3+
4+
#include <nan.h>
5+
#include <uv.h>
6+
#include "../include/thread_pool.h"
7+
#include "../include/nodegit.h"
8+
9+
10+
// Runs WorkComplete of the scheduled AsyncWorker,
11+
// and destroys it. This is run in the uv_default_loop event loop.
12+
NAN_INLINE void AsyncLibgit2Complete (void* data) {
13+
Nan::AsyncWorker *worker = static_cast<Nan::AsyncWorker*>(data);
14+
worker->WorkComplete();
15+
worker->Destroy();
16+
}
17+
18+
// Runs Execute of the scheduled AyncWorker on the dedicated libgit2 thread /
19+
// event loop, and schedules the WorkComplete callback to run on the
20+
// uv_default_loop event loop
21+
NAN_INLINE void AsyncLibgit2Execute (void *vworker) {
22+
// execute the worker
23+
Nan::AsyncWorker *worker = static_cast<Nan::AsyncWorker*>(vworker);
24+
worker->Execute();
25+
}
26+
27+
// Schedules the AsyncWorker to run on the dedicated libgit2 thread / event loop,
28+
// and on completion AsyncLibgit2Complete on the default loop
29+
NAN_INLINE void AsyncLibgit2QueueWorker (Nan::AsyncWorker* worker) {
30+
libgit2ThreadPool.QueueWork(AsyncLibgit2Execute, AsyncLibgit2Complete, worker);
31+
}
32+
33+
#endif

‎generate/templates/manual/include/nodegit.h‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
#ifndef NODEGIT_H
2+
#define NODEGIT_H
3+
4+
#include "thread_pool.h"
5+
6+
extern ThreadPool libgit2ThreadPool;
7+
8+
#endif

‎generate/templates/manual/include/thread_pool.h‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
#ifndef THREAD_POOL_H
2+
#define THREAD_POOL_H
3+
4+
#include <uv.h>
5+
#include <queue>
6+
7+
class ThreadPool {
8+
typedef void (*Callback) (void *);
9+
struct Work {
10+
Callback workCallback;
11+
Callback loopCallback;
12+
void *data;
13+
14+
Work(Callback workCallback, Callback loopCallback, void *data)
15+
: workCallback(workCallback), loopCallback(loopCallback), data(data) {
16+
}
17+
};
18+
19+
// work to be performed on the threadpool
20+
std::queue<Work> workQueue;
21+
uv_mutex_t workMutex;
22+
uv_sem_t workSemaphore;
23+
24+
// completion callbacks to be performed on the loop
25+
std::queue<Work> loopQueue;
26+
uv_mutex_t loopMutex;
27+
uv_async_t loopAsync;
28+
29+
static void RunEventQueue(void *threadPool);
30+
void RunEventQueue();
31+
static void RunLoopCallbacks(uv_async_t* handle);
32+
void RunLoopCallbacks();
33+
public:
34+
// Initializes thread pool and spins up the requested number of threads
35+
// The provided loop will be used for completion callbacks, whenever
36+
// queued work is completed
37+
ThreadPool(int numberOfThreads, uv_loop_t *loop);
38+
// Queues work on the thread pool, followed by completion call scheduled
39+
// on the loop provided in the constructor.
40+
// QueueWork should be called on the loop provided in the constructor.
41+
void QueueWork(Callback workCallback, Callback loopCallback, void *data);
42+
};
43+
44+
#endif

‎generate/templates/manual/src/thread_pool.cc‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
#include "../include/thread_pool.h"
2+
3+
ThreadPool::ThreadPool(int numberOfThreads, uv_loop_t *loop) {
4+
uv_mutex_init(&workMutex);
5+
uv_sem_init(&workSemaphore, 0);
6+
7+
uv_async_init(loop, &loopAsync, RunLoopCallbacks);
8+
loopAsync.data = this;
9+
uv_unref((uv_handle_t *)&loopAsync);
10+
uv_mutex_init(&loopMutex);
11+
12+
for(int i=0; i<numberOfThreads; i++) {
13+
uv_thread_t thread;
14+
uv_thread_create(&thread, RunEventQueue, this);
15+
}
16+
}
17+
18+
void ThreadPool::QueueWork(Callback workCallback, Callback loopCallback, void *data) {
19+
// there is work on the thread pool - reference the handle so
20+
// node doesn't terminate
21+
uv_ref((uv_handle_t *)&loopAsync);
22+
uv_mutex_lock(&workMutex);
23+
workQueue.push(Work(workCallback, loopCallback, data));
24+
uv_mutex_unlock(&workMutex);
25+
uv_sem_post(&workSemaphore);
26+
}
27+
28+
void ThreadPool::RunEventQueue(void *threadPool) {
29+
static_cast<ThreadPool *>(threadPool)->RunEventQueue();
30+
}
31+
32+
void ThreadPool::RunEventQueue() {
33+
for ( ; ; ) {
34+
// wait until there is work to do
35+
uv_sem_wait(&workSemaphore);
36+
uv_mutex_lock(&workMutex);
37+
// the semaphore should guarantee that queue is not empty
38+
Work work = workQueue.front();
39+
workQueue.pop();
40+
uv_mutex_unlock(&workMutex);
41+
42+
// perform the queued work
43+
(*work.workCallback)(work.data);
44+
45+
// schedule the callback on the loop
46+
uv_mutex_lock(&loopMutex);
47+
loopQueue.push(work);
48+
uv_mutex_unlock(&loopMutex);
49+
uv_async_send(&loopAsync);
50+
}
51+
}
52+
53+
void ThreadPool::RunLoopCallbacks(uv_async_t* handle) {
54+
static_cast<ThreadPool *>(handle->data)->RunLoopCallbacks();
55+
}
56+
57+
void ThreadPool::RunLoopCallbacks() {
58+
// uv_async_send can coalesce calls, so we are not guaranteed one
59+
// RunLoopCallbacks per uv_async_send call
60+
// so we always process the entire loopQueue
61+
uv_mutex_lock(&loopMutex);
62+
while(!loopQueue.empty()) {
63+
Work work = loopQueue.front();
64+
loopQueue.pop();
65+
uv_mutex_unlock(&loopMutex);
66+
// perform the queued loop callback
67+
(*work.loopCallback)(work.data);
68+
uv_mutex_lock(&loopMutex);
69+
}
70+
uv_mutex_lock(&workMutex);
71+
// if both the workQueue and the loopQueue are empty, node doesn't need
72+
// to be prevented from terminating
73+
if(workQueue.empty()) {
74+
uv_unref((uv_handle_t *)&loopAsync);
75+
}
76+
uv_mutex_unlock(&workMutex);
77+
uv_mutex_unlock(&loopMutex);
78+
}

‎generate/templates/partials/async_function.cc‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -74,7 +74,7 @@ NAN_METHOD({{ cppClassName }}::{{ cppFunctionName }}) {
7474
{%endif%}
7575
{%endeach%}
7676

77-
Nan::AsyncQueueWorker(worker);
77+
AsyncLibgit2QueueWorker(worker);
7878
return;
7979
}
8080

‎generate/templates/templates/binding.gyp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
"src/convenient_patch.cc",
2323
"src/convenient_hunk.cc",
2424
"src/str_array_converter.cc",
25+
"src/thread_pool.cc",
2526
{% each %}
2627
{% if type != "enum" %}
2728
"src/{{ name }}.cc",

‎generate/templates/templates/class_content.cc‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ extern "C" {
1212
#include "../include/functions/copy.h"
1313
#include "../include/{{ filename }}.h"
1414
#include "nodegit_wrapper.cc"
15+
#include "../include/async_libgit2_queue_worker.h"
1516

1617
{% each dependencies as dependency %}
1718
#include "{{ dependency }}"

‎generate/templates/templates/nodegit.cc‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010

1111
#include "../include/init_ssh2.h"
1212
#include "../include/lock_master.h"
13+
#include "../include/nodegit.h"
1314
#include "../include/wrapper.h"
1415
#include "../include/promise_completion.h"
1516
#include "../include/functions/copy.h"
@@ -80,6 +81,8 @@ void OpenSSL_ThreadSetup() {
8081
CRYPTO_set_id_callback(OpenSSL_IDCallback);
8182
}
8283

84+
ThreadPool libgit2ThreadPool(10, uv_default_loop());
85+
8386
extern "C" void init(Local<v8::Object> target) {
8487
// Initialize thread safety in openssl and libssh2
8588
OpenSSL_ThreadSetup();

0 commit comments

Comments
 (0)