Skip to content

Commit d10dcee

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

10 files changed

Lines changed: 238 additions & 1 deletion

File tree

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

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
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+
// schedule AsyncLibgit2Complete on the default loop
27+
defaultLoopQueuer.QueueWork(AsyncLibgit2Complete, vworker);
28+
}
29+
30+
// Schedules the AsyncWorker to run on the dedicated libgit2 thread / event loop.
31+
NAN_INLINE void AsyncLibgit2QueueWorker (Nan::AsyncWorker* worker) {
32+
libgit2ThreadPool.QueueWork(AsyncLibgit2Execute, worker);
33+
}
34+
35+
#endif

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

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
#ifndef LOOP_QUEUER_H
2+
#define LOOP_QUEUER_H
3+
4+
#include <uv.h>
5+
#include <vector>
6+
7+
// Allows easy scheduling of calls on a libuv loop.
8+
// Uses uv_async_send internally but does not coalesce calls.
9+
// Instead, allocates multiple uv_async_t handles to accomodate multiple
10+
// overlapping schedulings.
11+
class LoopQueuer {
12+
typedef void (*Callback)(void *);
13+
14+
struct Handle {
15+
volatile bool used;
16+
uv_async_t async;
17+
18+
Callback callback;
19+
void *data;
20+
};
21+
22+
uv_mutex_t handlesMutex;
23+
std::vector<Handle *> handles;
24+
uv_loop_t *loop;
25+
26+
Handle *NewHandle();
27+
Handle *GetUnusedHandle();
28+
static void AsyncCallback(uv_async_t *async);
29+
30+
public:
31+
LoopQueuer(uv_loop_t *loop)
32+
: loop(loop) {
33+
uv_mutex_init(&handlesMutex);
34+
}
35+
36+
void QueueWork(Callback callback, void *data);
37+
};
38+
39+
#endif

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

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

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

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
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 callback;
11+
void *data;
12+
13+
Work(Callback callback, void *data)
14+
: callback(callback), data(data) {
15+
}
16+
};
17+
18+
std::queue<Work> queue;
19+
uv_mutex_t queueMutex;
20+
uv_sem_t threadSemaphore;
21+
uv_async_t keepNodeAlive;
22+
23+
enum {
24+
CLOSED,
25+
OPEN
26+
} handleState;
27+
28+
static void RunEventQueue(void *threadPool);
29+
void RunEventQueue();
30+
public:
31+
ThreadPool(int numberOfThreads);
32+
void QueueWork(Callback callback, void *data);
33+
};
34+
35+
#endif

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

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
#include "../include/loop_queuer.h"
2+
3+
LoopQueuer::Handle *LoopQueuer::NewHandle() {
4+
Handle *handle = new Handle();
5+
6+
handle->used = true;
7+
uv_async_init(loop, &handle->async, AsyncCallback);
8+
// unreference the handle so it's not preventing node from terminating
9+
uv_unref((uv_handle_t *)&handle->async);
10+
handle->async.data = handle;
11+
12+
return handle;
13+
}
14+
15+
LoopQueuer::Handle *LoopQueuer::GetUnusedHandle() {
16+
uv_mutex_lock(&handlesMutex);
17+
// most of the time, we will find an existing unused handle and reuse it
18+
// this was written with the assumption that the number of handles
19+
// needed concurrently is small / the vector would not grow very large
20+
for (auto it = handles.begin(); it != handles.end(); it++) {
21+
if (!(*it)->used) {
22+
Handle *handle = *it;
23+
handle->used = true;
24+
uv_mutex_unlock(&handlesMutex);
25+
return handle;
26+
}
27+
}
28+
29+
// if we have not found an existing handle, create a new one
30+
Handle *handle = NewHandle();
31+
handles.push_back(handle);
32+
uv_mutex_unlock(&handlesMutex);
33+
return handle;
34+
}
35+
36+
void LoopQueuer::QueueWork(Callback callback, void *data) {
37+
Handle *handle = GetUnusedHandle();
38+
handle->callback = callback;
39+
handle->data = data;
40+
// reference handle so it prevents node from terminating
41+
// until work is complete
42+
uv_ref((uv_handle_t *)&handle->async);
43+
uv_async_send(&handle->async);
44+
}
45+
46+
// this is performed on the requested libuv loop
47+
void LoopQueuer::AsyncCallback(uv_async_t *async) {
48+
Handle *handle = (Handle *)async->data;
49+
Callback callback = handle->callback;
50+
void *data = handle->data;
51+
52+
(*callback)(data);
53+
54+
uv_unref((uv_handle_t *)&handle->async);
55+
handle->used = false;
56+
}

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

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
#include "../include/thread_pool.h"
2+
3+
// Initializes thread pool and spins up the requested number of threads
4+
ThreadPool::ThreadPool(int numberOfThreads) {
5+
uv_mutex_init(&queueMutex);
6+
uv_sem_init(&threadSemaphore, 0);
7+
uv_async_init(uv_default_loop(), &keepNodeAlive, NULL);
8+
uv_unref((uv_handle_t *)&keepNodeAlive);
9+
10+
for(int i=0; i<numberOfThreads; i++) {
11+
uv_thread_t thread;
12+
uv_thread_create(&thread, RunEventQueue, this);
13+
}
14+
}
15+
16+
// Queues work on the thread pool
17+
void ThreadPool::QueueWork(Callback callback, void *data) {
18+
uv_mutex_lock(&queueMutex);
19+
if(handleState == CLOSED) {
20+
// there is work on the thread pool - reference the handle so
21+
// node doesn't terminate
22+
uv_ref((uv_handle_t *)&keepNodeAlive);
23+
handleState = OPEN;
24+
}
25+
queue.push(Work(callback, data));
26+
uv_mutex_unlock(&queueMutex);
27+
uv_sem_post(&threadSemaphore);
28+
}
29+
30+
void ThreadPool::RunEventQueue(void *threadPool) {
31+
static_cast<ThreadPool *>(threadPool)->RunEventQueue();
32+
}
33+
34+
void ThreadPool::RunEventQueue() {
35+
for ( ; ; ) {
36+
// wait until there is work to do
37+
uv_sem_wait(&threadSemaphore);
38+
uv_mutex_lock(&queueMutex);
39+
// the semaphore should guarantee that queue is not empty
40+
Work work = queue.front();
41+
queue.pop();
42+
uv_mutex_unlock(&queueMutex);
43+
44+
// perform the queued work
45+
(*work.callback)(work.data);
46+
47+
uv_mutex_lock(&queueMutex);
48+
if(queue.empty() && handleState == OPEN) {
49+
// the queue is empty - unreference the handle so node can terminate
50+
uv_unref((uv_handle_t *)&keepNodeAlive);
51+
handleState = CLOSED;
52+
}
53+
uv_mutex_unlock(&queueMutex);
54+
}
55+
}

‎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: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222
"src/convenient_patch.cc",
2323
"src/convenient_hunk.cc",
2424
"src/str_array_converter.cc",
25+
"src/thread_pool.cc",
26+
"src/loop_queuer.cc",
2527
{% each %}
2628
{% if type != "enum" %}
2729
"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: 4 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,9 @@ void OpenSSL_ThreadSetup() {
8081
CRYPTO_set_id_callback(OpenSSL_IDCallback);
8182
}
8283

84+
LoopQueuer defaultLoopQueuer(uv_default_loop());
85+
ThreadPool libgit2ThreadPool(10);
86+
8387
extern "C" void init(Local<v8::Object> target) {
8488
// Initialize thread safety in openssl and libssh2
8589
OpenSSL_ThreadSetup();

0 commit comments

Comments
 (0)