fix(thread-safety): eliminate races in log mutex, watcher, and index threads (#207)
- http_server: replace lazy log mutex init with 3-state UNINIT/INITING/INITED CAS pattern; cbm_ui_log_init() called from main before thread creation, also still invoked lazily from cbm_ui_log_append() so early single-threaded logs are not dropped - watcher: protect projects hash table with cbm_mutex_t (watch/unwatch/touch/watch_count/free); poll_once snapshots project pointers under lock then iterates without it so git I/O and indexing don't block watch/unwatch - watcher: handle state_new() OOM in watch path - compat_thread: add cbm_thread_detach() for POSIX and Windows; both join() and detach() clear the handle on success for consistent lifecycle tracking - http_server: detach index job threads to prevent handle leaks
This commit is contained in:
@@ -56,6 +56,15 @@ int cbm_thread_join(cbm_thread_t *t) {
|
||||
return CBM_NOT_FOUND;
|
||||
}
|
||||
CloseHandle(t->handle);
|
||||
t->handle = NULL;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int cbm_thread_detach(cbm_thread_t *t) {
|
||||
if (t->handle) {
|
||||
CloseHandle(t->handle);
|
||||
t->handle = NULL;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -74,7 +83,19 @@ int cbm_thread_create(cbm_thread_t *t, size_t stack_size, void *(*fn)(void *), v
|
||||
}
|
||||
|
||||
int cbm_thread_join(cbm_thread_t *t) {
|
||||
return pthread_join(t->handle, NULL);
|
||||
int rc = pthread_join(t->handle, NULL);
|
||||
if (rc == 0) {
|
||||
memset(&t->handle, 0, sizeof(t->handle));
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
|
||||
int cbm_thread_detach(cbm_thread_t *t) {
|
||||
int rc = pthread_detach(t->handle);
|
||||
if (rc == 0) {
|
||||
memset(&t->handle, 0, sizeof(t->handle));
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
|
||||
#endif
|
||||
|
||||
@@ -39,6 +39,9 @@ int cbm_thread_create(cbm_thread_t *t, size_t stack_size, void *(*fn)(void *), v
|
||||
/* Wait for thread to finish. Returns 0 on success. */
|
||||
int cbm_thread_join(cbm_thread_t *t);
|
||||
|
||||
/* Detach thread so resources are freed on exit. Returns 0 on success. */
|
||||
int cbm_thread_detach(cbm_thread_t *t);
|
||||
|
||||
/* ── Mutex ────────────────────────────────────────────────────── */
|
||||
|
||||
#ifdef _WIN32
|
||||
|
||||
@@ -375,6 +375,9 @@ int main(int argc, char **argv) {
|
||||
}
|
||||
|
||||
/* Create and start watcher in background thread */
|
||||
/* Initialize log mutex before any threads are created */
|
||||
cbm_ui_log_init();
|
||||
|
||||
cbm_store_t *watch_store = cbm_store_open_memory();
|
||||
g_watcher = cbm_watcher_new(watch_store, watcher_index_fn, NULL);
|
||||
|
||||
|
||||
+31
-5
@@ -142,16 +142,41 @@ static char g_log_ring[LOG_RING_SIZE][LOG_LINE_MAX];
|
||||
static int g_log_head = 0;
|
||||
static int g_log_count = 0;
|
||||
static cbm_mutex_t g_log_mutex;
|
||||
static atomic_int g_log_mutex_init = 0;
|
||||
|
||||
enum {
|
||||
CBM_LOG_MUTEX_UNINIT = 0,
|
||||
CBM_LOG_MUTEX_INITING = 1,
|
||||
CBM_LOG_MUTEX_INITED = 2
|
||||
};
|
||||
static atomic_int g_log_mutex_init = CBM_LOG_MUTEX_UNINIT;
|
||||
|
||||
/* Safe for concurrent callers: only publishes INITED after cbm_mutex_init()
|
||||
* has completed. Callers that lose the CAS race spin until init finishes. */
|
||||
void cbm_ui_log_init(void) {
|
||||
int state = atomic_load(&g_log_mutex_init);
|
||||
if (state == CBM_LOG_MUTEX_INITED)
|
||||
return;
|
||||
|
||||
state = CBM_LOG_MUTEX_UNINIT;
|
||||
if (atomic_compare_exchange_strong(&g_log_mutex_init, &state, CBM_LOG_MUTEX_INITING)) {
|
||||
cbm_mutex_init(&g_log_mutex);
|
||||
atomic_store(&g_log_mutex_init, CBM_LOG_MUTEX_INITED);
|
||||
return;
|
||||
}
|
||||
|
||||
/* Another thread is initializing — spin until done */
|
||||
while (atomic_load(&g_log_mutex_init) != CBM_LOG_MUTEX_INITED) {
|
||||
cbm_usleep(1000); /* 1ms */
|
||||
}
|
||||
}
|
||||
|
||||
/* Called from a log hook — appends a line to the ring buffer (thread-safe) */
|
||||
void cbm_ui_log_append(const char *line) {
|
||||
if (!line)
|
||||
return;
|
||||
if (!atomic_load(&g_log_mutex_init)) {
|
||||
cbm_mutex_init(&g_log_mutex);
|
||||
atomic_store(&g_log_mutex_init, 1);
|
||||
}
|
||||
/* Ensure mutex is initialized (safe for early single-threaded logging
|
||||
* and concurrent calls via atomic_exchange once-init pattern). */
|
||||
cbm_ui_log_init();
|
||||
cbm_mutex_lock(&g_log_mutex);
|
||||
snprintf(g_log_ring[g_log_head], LOG_LINE_MAX, "%s", line);
|
||||
g_log_head = (g_log_head + 1) % LOG_RING_SIZE;
|
||||
@@ -793,6 +818,7 @@ static void handle_index_start(struct mg_connection *c, struct mg_http_message *
|
||||
mg_http_reply(c, 500, g_cors_json, "{\"error\":\"thread creation failed\"}");
|
||||
return;
|
||||
}
|
||||
cbm_thread_detach(&tid); /* Don't leak thread handle */
|
||||
|
||||
mg_http_reply(c, 202, g_cors_json, "{\"status\":\"indexing\",\"slot\":%d,\"path\":\"%s\"}",
|
||||
slot, job->root_path);
|
||||
|
||||
@@ -32,6 +32,9 @@ void cbm_http_server_run(cbm_http_server_t *srv);
|
||||
/* Check if the server started successfully (listener bound). */
|
||||
bool cbm_http_server_is_running(const cbm_http_server_t *srv);
|
||||
|
||||
/* Initialize the log ring buffer mutex. Must be called once before any threads. */
|
||||
void cbm_ui_log_init(void);
|
||||
|
||||
/* Append a log line to the UI ring buffer (called from log hook). */
|
||||
void cbm_ui_log_append(const char *line);
|
||||
|
||||
|
||||
+62
-2
@@ -20,6 +20,7 @@
|
||||
#include "foundation/log.h"
|
||||
#include "foundation/hash_table.h"
|
||||
#include "foundation/compat.h"
|
||||
#include "foundation/compat_thread.h"
|
||||
#include "foundation/compat_fs.h"
|
||||
#include "foundation/str_util.h"
|
||||
|
||||
@@ -50,6 +51,7 @@ struct cbm_watcher {
|
||||
cbm_index_fn index_fn;
|
||||
void *user_data;
|
||||
CBMHashTable *projects; /* name → project_state_t* */
|
||||
cbm_mutex_t projects_lock;
|
||||
atomic_int stopped;
|
||||
};
|
||||
|
||||
@@ -236,6 +238,7 @@ cbm_watcher_t *cbm_watcher_new(cbm_store_t *store, cbm_index_fn index_fn, void *
|
||||
w->index_fn = index_fn;
|
||||
w->user_data = user_data;
|
||||
w->projects = cbm_ht_create(CBM_SZ_32);
|
||||
cbm_mutex_init(&w->projects_lock);
|
||||
atomic_init(&w->stopped, 0);
|
||||
return w;
|
||||
}
|
||||
@@ -244,8 +247,11 @@ void cbm_watcher_free(cbm_watcher_t *w) {
|
||||
if (!w) {
|
||||
return;
|
||||
}
|
||||
cbm_mutex_lock(&w->projects_lock);
|
||||
cbm_ht_foreach(w->projects, free_state_entry, NULL);
|
||||
cbm_ht_free(w->projects);
|
||||
cbm_mutex_unlock(&w->projects_lock);
|
||||
cbm_mutex_destroy(&w->projects_lock);
|
||||
free(w);
|
||||
}
|
||||
|
||||
@@ -264,6 +270,7 @@ void cbm_watcher_watch(cbm_watcher_t *w, const char *project_name, const char *r
|
||||
}
|
||||
|
||||
/* Remove old entry first (key points to state's project_name) */
|
||||
cbm_mutex_lock(&w->projects_lock);
|
||||
project_state_t *old = cbm_ht_get(w->projects, project_name);
|
||||
if (old) {
|
||||
cbm_ht_delete(w->projects, project_name);
|
||||
@@ -271,7 +278,13 @@ void cbm_watcher_watch(cbm_watcher_t *w, const char *project_name, const char *r
|
||||
}
|
||||
|
||||
project_state_t *s = state_new(project_name, root_path);
|
||||
if (!s) {
|
||||
cbm_mutex_unlock(&w->projects_lock);
|
||||
cbm_log_warn("watcher.watch.oom", "project", project_name, "path", root_path);
|
||||
return;
|
||||
}
|
||||
cbm_ht_set(w->projects, s->project_name, s);
|
||||
cbm_mutex_unlock(&w->projects_lock);
|
||||
cbm_log_info("watcher.watch", "project", project_name, "path", root_path);
|
||||
}
|
||||
|
||||
@@ -279,10 +292,16 @@ void cbm_watcher_unwatch(cbm_watcher_t *w, const char *project_name) {
|
||||
if (!w || !project_name) {
|
||||
return;
|
||||
}
|
||||
bool removed = false;
|
||||
cbm_mutex_lock(&w->projects_lock);
|
||||
project_state_t *s = cbm_ht_get(w->projects, project_name);
|
||||
if (s) {
|
||||
cbm_ht_delete(w->projects, project_name);
|
||||
state_free(s);
|
||||
removed = true;
|
||||
}
|
||||
cbm_mutex_unlock(&w->projects_lock);
|
||||
if (removed) {
|
||||
cbm_log_info("watcher.unwatch", "project", project_name);
|
||||
}
|
||||
}
|
||||
@@ -291,18 +310,23 @@ void cbm_watcher_touch(cbm_watcher_t *w, const char *project_name) {
|
||||
if (!w || !project_name) {
|
||||
return;
|
||||
}
|
||||
cbm_mutex_lock(&w->projects_lock);
|
||||
project_state_t *s = cbm_ht_get(w->projects, project_name);
|
||||
if (s) {
|
||||
/* Reset backoff — poll immediately on next cycle */
|
||||
s->next_poll_ns = 0;
|
||||
}
|
||||
cbm_mutex_unlock(&w->projects_lock);
|
||||
}
|
||||
|
||||
int cbm_watcher_watch_count(const cbm_watcher_t *w) {
|
||||
if (!w) {
|
||||
return 0;
|
||||
}
|
||||
return (int)cbm_ht_count(w->projects);
|
||||
cbm_mutex_lock(&((cbm_watcher_t *)w)->projects_lock);
|
||||
int count = (int)cbm_ht_count(w->projects);
|
||||
cbm_mutex_unlock(&((cbm_watcher_t *)w)->projects_lock);
|
||||
return count;
|
||||
}
|
||||
|
||||
/* ── Single poll cycle ──────────────────────────────────────────── */
|
||||
@@ -411,17 +435,53 @@ static void poll_project(const char *key, void *val, void *ud) {
|
||||
s->next_poll_ns = ctx->now + ((int64_t)s->interval_ms * US_PER_MS);
|
||||
}
|
||||
|
||||
/* Callback to snapshot project state pointers into an array. */
|
||||
typedef struct {
|
||||
project_state_t **items;
|
||||
int count;
|
||||
int cap;
|
||||
} snapshot_ctx_t;
|
||||
|
||||
static void snapshot_project(const char *key, void *val, void *ud) {
|
||||
(void)key;
|
||||
snapshot_ctx_t *sc = ud;
|
||||
if (val && sc->count < sc->cap) {
|
||||
sc->items[sc->count++] = val;
|
||||
}
|
||||
}
|
||||
|
||||
int cbm_watcher_poll_once(cbm_watcher_t *w) {
|
||||
if (!w) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Snapshot project pointers under lock, then poll without holding it.
|
||||
* This keeps the critical section small — poll_project does git I/O
|
||||
* and may invoke index_fn which runs the full pipeline. */
|
||||
cbm_mutex_lock(&w->projects_lock);
|
||||
int n = cbm_ht_count(w->projects);
|
||||
if (n == 0) {
|
||||
cbm_mutex_unlock(&w->projects_lock);
|
||||
return 0;
|
||||
}
|
||||
project_state_t **snap = malloc(n * sizeof(project_state_t *));
|
||||
if (!snap) {
|
||||
cbm_mutex_unlock(&w->projects_lock);
|
||||
return 0;
|
||||
}
|
||||
snapshot_ctx_t sc = {.items = snap, .count = 0, .cap = n};
|
||||
cbm_ht_foreach(w->projects, snapshot_project, &sc);
|
||||
cbm_mutex_unlock(&w->projects_lock);
|
||||
|
||||
poll_ctx_t ctx = {
|
||||
.w = w,
|
||||
.now = now_ns(),
|
||||
.reindexed = 0,
|
||||
};
|
||||
cbm_ht_foreach(w->projects, poll_project, &ctx);
|
||||
for (int i = 0; i < sc.count; i++) {
|
||||
poll_project(NULL, snap[i], &ctx);
|
||||
}
|
||||
free(snap);
|
||||
return ctx.reindexed;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user