job system has callback when worker quits;

Author: bjornbytes

src/api/l_thread.c

@@ -92,6 +92,20 @@ typedef struct RunContext {
 static thread_local RunContext* contextPool;
 static thread_local lua_State* workerState;
 
+static void onWorkerQuit(void) {
+  if (workerState) {
+    lua_close(workerState);
+    workerState = NULL;
+  }
+
+  while (contextPool) {
+    RunContext* context = contextPool;
+    contextPool = context->next;
+    arr_free(&context->code);
+    lovrFree(context);
+  }
+}
+
 static bool luax_runlua(void** arg) {
   RunContext* context = *arg;
   lua_State* L = workerState;
@@ -282,7 +296,7 @@ int luaopen_lovr_thread(lua_State* L) {
   lua_newtable(L);
   lua_setfield(L, LUA_REGISTRYINDEX, "_lovrbytecode");
 
-  lovrThreadModuleInit(workers);
+  lovrThreadModuleInit(workers, onWorkerQuit);
   luax_atexit(L, lovrThreadModuleDestroy);
   return 1;
 }

src/core/job.c

@@ -16,9 +16,10 @@ static struct {
   atomic_uint head;
   atomic_uint tail;
   job jobs[MAX_JOBS];
-  void (*setupWorker)(uint32_t id);
   thrd_t workers[MAX_WORKERS];
   uint32_t workerCount;
+  fn_hook* workerInit;
+  fn_hook* workerQuit;
   cnd_t hasJob;
   mtx_t lock;
   bool quit;
@@ -32,8 +33,10 @@ static void runJob(void) {
 }
 
 static int workerLoop(void* arg) {
-  if (state.setupWorker) {
-    state.setupWorker((uint32_t) (uintptr_t) arg);
+  uint32_t id = (uint32_t) (uintptr_t) arg;
+
+  if (state.workerInit) {
+    state.workerInit(id);
   }
 
   for (;;) {
@@ -51,14 +54,20 @@ static int workerLoop(void* arg) {
   }
 
   mtx_unlock(&state.lock);
+
+  if (state.workerQuit) {
+    state.workerQuit(id);
+  }
+
   return 0;
 }
 
-bool job_init(uint32_t count, void (*setupWorker)(uint32_t id)) {
+bool job_init(uint32_t count, fn_hook* init, fn_hook* quit) {
   mtx_init(&state.lock, mtx_plain);
   cnd_init(&state.hasJob);
 
-  state.setupWorker = setupWorker;
+  state.workerInit = init;
+  state.workerQuit = quit;
   if (count > MAX_WORKERS) count = MAX_WORKERS;
   for (uint32_t i = 0; i < count; i++, state.workerCount++) {
     if (thrd_create(&state.workers[i], workerLoop, (void*) (uintptr_t) i) != thrd_success) {

src/core/job.h

@@ -4,8 +4,9 @@
 #pragma once
 
 typedef void fn_job(void* arg);
+typedef void fn_hook(uint32_t worker);
 
-bool job_init(uint32_t workerCount, void (*setupWorker)(uint32_t index));
+bool job_init(uint32_t workerCount, fn_hook* init, fn_hook* quit);
 void job_destroy(void);
 bool job_start(fn_job* fn, void* arg);
 void job_spin(void);

src/modules/thread/thread.c

@@ -39,24 +39,33 @@ static atomic_uint ref;
 
 static struct {
   uint32_t workers;
+  void (*onWorkerQuit)(void);
   mtx_t channelLock;
   map_t channels;
 } state;
 
-static void setupWorker(uint32_t id) {
+static void workerInit(uint32_t id) {
   lovrProfileSetThreadName("Worker");
   os_thread_set_name("Worker");
 }
 
-bool lovrThreadModuleInit(int32_t workers) {
+static void workerQuit(uint32_t id) {
+  if (state.onWorkerQuit) {
+    state.onWorkerQuit();
+  }
+}
+
+bool lovrThreadModuleInit(int32_t workers, void (*onWorkerQuit)(void)) {
   if (!lovrModuleAcquire(&ref)) return true;
+
   mtx_init(&state.channelLock, mtx_plain);
   map_init(&state.channels, 0);
 
   uint32_t cores = os_get_core_count();
   if (workers < 0) workers += cores;
   state.workers = MAX(workers, 0);
-  job_init(state.workers, setupWorker);
+  state.onWorkerQuit = onWorkerQuit;
+  job_init(state.workers, workerInit, workerQuit);
 
   lovrModuleReady(&ref);
   return true;

src/modules/thread/thread.h

@@ -14,7 +14,7 @@ union Variant;
 typedef struct Thread Thread;
 typedef struct Channel Channel;
 
-bool lovrThreadModuleInit(int32_t workers);
+bool lovrThreadModuleInit(int32_t workers, void (*onWorkerQuit)(void));
 void lovrThreadModuleDestroy(void);
 uint32_t lovrThreadGetWorkerCount(void);
 struct Channel* lovrThreadGetChannel(const char* name);