threading infra

include/hammy/job.h

@@ -2,16 +2,58 @@
 #define HAMMY_JOB_H
 
 #include <concord/discord.h>
+#include <stdbool.h>
+#include <stddef.h>
 #include <stdint.h>
+
 #include <hammy/types.h>
 
+// A single slash-command option. Flattened out of the interaction event.
+// Both strings are owned by the job.
+typedef struct hammy_arg_t {
+    char* name;
+    char* value;
+} hammy_arg_t;
+
+// A unit of deferred work handed from the gateway thread to a worker.
+//
+// OWNERSHIP: a job is created by the gateway thread and, on a successful
+// hammy_pool_push(), ownership transfers to the pool. After that the creating
+// thread MUST NOT touch it again. The worker that pops it owns it and destroys
+// it. On a failed push the caller still owns it and must destroy it itself.
 struct hammy_job_t {
-    char* token; // Interaction token
+    char* token; // Interaction token - owning
     u64snowflake id; // Interaction ID
-    u64snowflake user; // Interaction User ID
-    char* command;
-    char** args; // Might be useful to parse into a struct someday? TODO
-    int64_t queued_at; // Staleness checks
+    u64snowflake appId; // Application ID, needed to edit the original response
+    u64snowflake user; // Invoking User ID, rate limiting, logging, etc.
+    char* command; // Command name - owning
+    hammy_arg_t* args; // Array of size nArgs. Owning, array and contents.
+    size_t nArgs;
+    int64_t queuedAt; // Staleness checks, in ms. From discord_timestamp().
 };
 
+// Deep-copies everything the job needs out of the interaction event, so the
+// event may be freed by Concord the moment the handler returns.
+// Returns NULL on allocation failure.
+hammy_job_t* hammy_job_create(struct discord* client, const struct discord_interaction* event);
+ 
+// Frees the job and everything it owns. NULLs the passing reference.
+// Safe to call with NULL or with a pointer to NULL.
+bool hammy_job_destroy(hammy_job_t** job);
+ 
+// Looks up an option by name. Returns NULL if absent. Result is owned by the job.
+const char* hammy_job_get_arg(const hammy_job_t* job, const char* name);
+ 
+// Milliseconds elapsed since the job was created.
+int64_t hammy_job_age_ms(const hammy_job_t* job, struct discord* client);
+ 
+// Runs the job to completion and sends the reply. Called from a worker thread,
+// so client MUST be that worker's own clone, never the gateway client.
+// Does not destroy the job.
+void hammy_job_run(hammy_job_t* job, struct discord* client);
+ 
+// Edits the (already deferred) interaction response with a plain text body.
+// Used by hammy_job_run() and by the pool's error paths.
+void hammy_job_reply(const hammy_job_t* job, struct discord* client, const char* content);
+
 #endif

include/hammy/pool.h

@@ -1,22 +1,69 @@
 #ifndef HAMMY_POOL_H
 #define HAMMY_POOL_H
 
-#include <hammy/types.h>
+#include <concord/discord.h>
+
 #include <pthread.h>
 #include <stdint.h>
+#include <stdbool.h>
+#include <stddef.h>
+
+#include <hammy/types.h>
+
+// TODO: Change this accordingly; also probably make it configurable without rebuilding at some point
+#define HAMMY_POOL_DEFAULT_WORKERS 2
+#define HAMMY_POOL_DEFAULT_CAPACITY 64
+
+// Technically, discord allows 15 minutes... but hell no. 30 seconds.
+#define HAMMY_JOB_MAX_AGE_MS 30000
+
+typedef enum {
+    HAMMY_PUSH_OK = 0, // Queued. Ownership transferred to the pool.
+    HAMMY_PUSH_FULL = 1, // At capacity. Caller still owns the job.
+    HAMMY_PUSH_SHUTDOWN = 2 // Pool is closing. Caller still owns the job.
+} hammy_push_result_t;
 
 struct hammy_pool_t {
     pthread_mutex_t lock;
     pthread_cond_t notEmpty;
-    hammy_job_t** jobs; // ring buffer
-    size_t head;
-    size_t tail;
-    size_t count;
+
+    hammy_job_t** jobs; // Ring buffer of cap job pointers - owning while queued
+    size_t head; // Next slot to pop
+    size_t tail; // Next slot to push
+    size_t count; // Live entries - tracked so full != empty ambiguity doesn't exist
     size_t cap;
+
+    size_t busy; // Workers currently processing jobs.
     bool shutdown;
 
     hammy_worker_t* workers;
     size_t nWorkers;
 };
 
+// Creates the pool and starts n_workers threads, each with its own
+// discord_clone() of client. Pass 0 for either size to take the defaults.
+// Returns NULL on failure; no threads are left running in that case.
+hammy_pool_t* hammy_pool_create(struct discord* client, size_t nWorkers, size_t queueCap);
+ 
+// Enqueues a job. See hammy_push_result_t for who owns the job afterwards.
+// Never blocks on anything but the (uncontended, short) queue mutex, so it is
+// safe to call from the gateway thread.
+hammy_push_result_t hammy_pool_push(hammy_pool_t* pool, hammy_job_t* job);
+ 
+// Signals all workers to finish the queue and exit, then joins them.
+// Idempotent. Queued jobs are still run, so an in-flight command still gets a
+// reply; use hammy_pool_shutdown_now() if you would rather drop them.
+void hammy_pool_shutdown(hammy_pool_t* pool);
+ 
+// As above but discards anything still queued (each dropped job gets an
+// apology reply if it can be sent quickly).
+void hammy_pool_shutdown_now(hammy_pool_t* pool);
+ 
+// Frees the pool. Runs hammy_pool_shutdown() first if it has not happened yet.
+// NULLs the passing reference.
+bool hammy_pool_destroy(hammy_pool_t** pool);
+ 
+// Snapshot of queue depth and busy workers, for a /stats command or logging.
+void hammy_pool_stats(hammy_pool_t* pool, size_t* outQueued, size_t* outBusy);
+
 #endif

include/hammy/worker.h

@@ -7,9 +7,18 @@
 
 struct hammy_worker_t {
     pthread_t thread;
-    struct discord* clientCopy; // Clone of the client for concord threading safety
-    hammy_pool_t* pool;
-    int id;
+    struct discord* clientCopy; // Clone of the client for concord threading safety - owning
+    hammy_pool_t* pool; // Non-owning back-reference
+    int id; // Log logging mainly
+    bool started; // For joining
 };
 
+// Clones client, spawns the thread. Returns false on clone or spawn failure,
+// leaving the worker safe to pass to hammy_worker_join().
+bool hammy_worker_start(hammy_worker_t* worker, hammy_pool_t* pool, struct discord* client, int id);
+ 
+// Joins the thread if it was started and cleans up the clone.
+// The caller must have already set pool->shutdown and broadcast, or this hangs.
+void hammy_worker_join(hammy_worker_t* worker);
+
 #endif

src/hammy/job.c

@@ -0,0 +1,147 @@
+#include <concord/discord.h>
+#include <concord/log.h>
+#include <inttypes.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <hammy/job.h>
+
+// strdup() is POSIX, so we'll keep a local and keep the code portable.
+static char* hammy_strdup(const char* src) {
+    if (!src) { return NULL; }
+
+    size_t len = strlen(src) + 1; // +1 for the null terminator
+    char* dst = (char*)malloc(len);
+    if (!dst) { return NULL; }
+
+    memcpy(dst, src, len);
+
+    return dst;
+}
+
+// Pulls the invoking user out of the event. Guild interactions carry it under
+// member->user, DM interactions under user directly.
+static u64snowflake hammy_job_extract_user(const struct discord_interaction* event) {
+    if (event->member && event->member->user) return event->member->user->id;
+    if (event->user) return event->user->id;
+ 
+    return 0;
+}
+
+hammy_job_t* hammy_job_create(struct discord* client, const struct discord_interaction* event) {
+    if (!client || !event) { return NULL; }
+
+    hammy_job_t* job = (hammy_job_t*)calloc(1, sizeof(*job));
+    if (!job) { return NULL; }
+
+    job->id = event->id;
+    job->appId = event->application_id;
+    job->user = hammy_job_extract_user(event);
+    job->token = hammy_strdup(event->token);
+    job->queuedAt = (int64_t)discord_timestamp(client);
+
+    if (!job->token) {
+        goto fail;
+    }
+
+    if (event->data && event->data->name) {
+        job->command = hammy_strdup(event->data->name);
+        if (!job->command) {
+            goto fail;
+        }
+    }
+
+    // Flatten the top-level options. Subcommand groups nest another options array inside an option
+    // Not handled yet, and worth revisiting before we need one; TODO
+    if (event->data && event->data->options && event->data->options->size > 0) {
+        size_t n = (size_t)event->data->options->size;
+
+        job->args = (hammy_arg_t*)calloc(n, sizeof(*job->args));
+        if (!job->args) {
+            goto fail;
+        }
+
+        for (size_t i = 0; i < n; i++) {
+            struct discord_application_command_interaction_data_option* opt = &event->data->options->array[i];
+
+            job->args[i].name = hammy_strdup(opt->name);
+            job->args[i].value = hammy_strdup(opt->value);
+
+            // A NULL value is legitimate for a flag-style option; a NULL name
+            // after a non-NULL source is an allocation failure.
+            if (opt->name && !job->args[i].name) goto fail;
+            if (opt->value && !job->args[i].value) goto fail;
+ 
+            job->nArgs++;
+        }
+    }
+
+    return job;
+
+fail:
+    hammy_job_destroy(&job);
+    return NULL;
+}
+
+bool hammy_job_destroy(hammy_job_t** job) {
+    if (!job || !*job) { return false; }
+
+    hammy_job_t* j = *job;
+
+    for (size_t i = 0; i < j->nArgs; i++) {
+        free(j->args[i].name);
+        free(j->args[i].value);
+    }
+
+    free(j->args);
+    free(j->token);
+    free(j->command);
+    free(j);
+
+    *job = NULL;
+
+    return true;
+}
+
+const char* hammy_job_get_arg(const hammy_job_t* job, const char* name) {
+    if (!job || !name) { return NULL; }
+
+    for (size_t i = 0; i < job->nArgs; i++) {
+        if (job->args[i].name && strcmp(job->args[i].name, name) == 0) {
+            return job->args[i].value;
+        }
+    }
+
+    return NULL;
+}
+
+int64_t hammy_job_age_ms(const hammy_job_t* job, struct discord* client) {
+    if (!job || !client) { return 0; }
+
+    return (int64_t)discord_timestamp(client) - job->queuedAt;
+}
+
+void hammy_job_reply(const hammy_job_t* job, struct discord* client, const char* content) {
+    if (!job || !client || !content) { return; }
+
+    // TODO: Embeds
+    struct discord_edit_original_interaction_response params = {
+        .content = (char*)content
+    };
+
+    CCORDcode code = discord_edit_original_interaction_response(client, job->appId, job->token, &params, NULL);
+    if (code != CCORD_OK) {
+        log_warn("[job] Failed to edit response for interaction %" PRIu64 ": %d", job->id, code);
+    }
+}
+
+void hammy_job_run(hammy_job_t* job, struct discord* client) {
+    if (!job || !client) { return; }
+
+    // TODO: look job->command up in the bot's command vector and call its
+    // handler with (job, client). Placeholder until command.h grows a
+    // dispatch entry point.
+    log_info("[job] Running command '%s' for interaction %" PRIu64, job->command ? job->command : "unknown", job->id);
+
+    hammy_job_reply(job, client, "This is a placeholder reply. The command handler is not yet implemented.");
+}

src/hammy/pool.c

@@ -0,0 +1,171 @@
+#include <concord/discord.h>
+#include <concord/log.h>
+#include <stdlib.h>
+
+#include <hammy/job.h>
+#include <hammy/pool.h>
+#include <hammy/worker.h>
+
+hammy_pool_t* hammy_pool_create(struct discord* client, size_t nWorkers, size_t queueCap) {
+    if (!client) { return NULL; }
+
+    if (nWorkers == 0) { nWorkers = HAMMY_POOL_DEFAULT_WORKERS; }
+    if (queueCap == 0) { queueCap = HAMMY_POOL_DEFAULT_CAPACITY; }
+
+    hammy_pool_t* pool = (hammy_pool_t*)calloc(1, sizeof(*pool));
+    if (!pool) { return NULL; }
+
+    pool->cap = queueCap;
+    pool->jobs = (hammy_job_t**)calloc(pool->cap, sizeof(*pool->jobs));
+    if (!pool->jobs) {
+        goto fail_jobs;
+    }
+
+    pool->workers = (hammy_worker_t*)calloc(nWorkers, sizeof(*pool->workers));
+    if (!pool->workers) {
+        goto fail_workers;
+    }
+
+    if (pthread_mutex_init(&pool->lock, NULL) != 0) {
+        goto fail_mutex;
+    }
+
+    if (pthread_cond_init(&pool->notEmpty, NULL) != 0) {
+        goto fail_cond;
+    }
+
+    // nWorkers counts STARTED threads, so a partial failure below still joins
+    // exactly the ones that exist.
+    for (size_t i = 0; i < nWorkers; i++) {
+        if (!hammy_worker_start(&pool->workers[i], pool, client, (int)i)) {
+            log_error("[pool] only %zu of %zu workers started, bailing!", i, nWorkers);
+            hammy_pool_shutdown(pool);
+            hammy_pool_destroy(&pool);
+
+            return NULL;
+        }
+
+        pool->nWorkers++;
+    }
+
+    log_info("[pool] Started %zu workers, queue cap %zu", pool->nWorkers, pool->cap);
+
+    return pool;
+
+// GOTOs
+fail_cond:
+    pthread_mutex_destroy(&pool->lock);
+fail_mutex:
+    free(pool->workers);
+fail_workers:
+    free(pool->jobs);
+fail_jobs:
+    free(pool);
+
+    return NULL;
+
+}
+
+hammy_push_result_t hammy_pool_push(hammy_pool_t* pool, hammy_job_t* job) {
+    if (!pool || !job) { return HAMMY_PUSH_SHUTDOWN; }
+
+    pthread_mutex_lock(&pool->lock);
+
+    if (pool->shutdown) {
+        pthread_mutex_unlock(&pool->lock);
+        return HAMMY_PUSH_SHUTDOWN;
+    }
+
+    if (pool->count == pool->cap) {
+        pthread_mutex_unlock(&pool->lock);
+        return HAMMY_PUSH_FULL;
+    }
+
+    pool->jobs[pool->tail] = job;
+    pool->tail = (pool->tail + 1) % pool->cap;
+    pool->count++;
+
+    // Signal inside the lock, wakeup cost kinda irrlevant compared to HTTP round trips
+    pthread_cond_signal(&pool->notEmpty);
+    pthread_mutex_unlock(&pool->lock);
+
+    return HAMMY_PUSH_OK;
+}
+
+static void hammy_pool_stop(hammy_pool_t* pool, bool drain) {
+    if (!pool) { return; }
+
+    pthread_mutex_lock(&pool->lock);
+
+    if (pool->shutdown) {
+        pthread_mutex_unlock(&pool->lock);
+        return;
+    }
+
+    pool->shutdown = true;
+
+    if (!drain) {
+        while (pool->count > 0) {
+            hammy_job_t* job = pool->jobs[pool->head];
+            pool->head = (pool->head + 1) % pool->cap;
+            pool->count--;
+
+            hammy_job_destroy(&job);
+        }
+    }
+
+    // Broadcast (NOT signal), every watier has to see the shutdown flag and exit, not just one.
+    pthread_cond_broadcast(&pool->notEmpty);
+    pthread_mutex_unlock(&pool->lock);
+
+    for (size_t i = 0; i < pool->nWorkers; i++) {
+        hammy_worker_join(&pool->workers[i]);
+    }
+
+    log_info("[pool] Shut down");
+}
+
+void hammy_pool_shutdown(hammy_pool_t* pool) {
+    hammy_pool_stop(pool, true);
+}
+ 
+void hammy_pool_shutdown_now(hammy_pool_t* pool) {
+    hammy_pool_stop(pool, false);
+}
+
+bool hammy_pool_destroy(hammy_pool_t** pool) {
+    if (!pool || !*pool) { return false; }
+
+    hammy_pool_t* p = *pool;
+
+    hammy_pool_shutdown(p); // No-op if already shut down
+
+    // Anything still queued after the drain is a but - free rather than leak it.
+    while (p->count > 0) {
+        hammy_job_t* job = p->jobs[p->head];
+        p->head = (p->head + 1) % p->cap;
+        p->count--;
+
+        hammy_job_destroy(&job);
+    }
+
+    pthread_cond_destroy(&p->notEmpty);
+    pthread_mutex_destroy(&p->lock);
+
+    free(p->workers);
+    free(p->jobs);
+    free(p);
+
+    *pool = NULL;
+
+    return true;
+}
+
+void hammy_pool_stats(hammy_pool_t* pool, size_t* outQueued, size_t* outBusy) {
+    if (!pool) { return; }
+
+    pthread_mutex_lock(&pool->lock);
+    if (outBusy) { *outBusy = pool->busy; }
+    if (outQueued) { *outQueued = pool->count; }
+    pthread_mutex_unlock(&pool->lock);
+}

src/hammy/worker.c

@@ -0,0 +1,99 @@
+#include <concord/discord.h>
+#include <concord/log.h>
+#include <stdlib.h>
+ 
+#include <hammy/job.h>
+#include <hammy/pool.h>
+#include <hammy/worker.h>
+
+static void* hammy_worker_main(void* arg) {
+    hammy_worker_t* worker = (hammy_worker_t*)arg;
+    hammy_pool_t* pool = worker->pool;
+
+    log_info("[worker %d] Started", worker->id);
+
+    for (;;) {
+        pthread_mutex_lock(&pool->lock);
+
+        // while instead of if, because pthread_cond_wait() can spuriously wake up like an ass
+        while (pool->count == 0 && !pool->shutdown) {
+            pthread_cond_wait(&pool->notEmpty, &pool->lock);
+        }
+
+        if (pool->count == 0 && pool->shutdown) {
+            pthread_mutex_unlock(&pool->lock);
+            break;
+        }
+
+        hammy_job_t* job = pool->jobs[pool->head];
+        pool->head = (pool->head + 1) % pool->cap;
+        pool->count--;
+        pool->busy++;
+
+        pthread_mutex_unlock(&pool->lock);
+
+        // From here, the worker owns the job and is responsible for destroying it.
+        log_info("[worker %d] Processing job %llu from user %llu", worker->id, job->id, job->user);
+        int64_t age = hammy_job_age_ms(job, worker->clientCopy);
+
+        if (age > HAMMY_JOB_MAX_AGE_MS) {
+            log_warn("[worker %d] Dropping stale job '%s' (age %lld ms)", worker->id, job->command ? job->command : "unknown", (long long)age);
+            hammy_job_reply(job, worker->clientCopy, "Sorry, your command took too long to process and was dropped. Please try again.");
+        } else {
+            hammy_job_run(job, worker->clientCopy);
+        }
+
+        hammy_job_destroy(&job);
+
+        pthread_mutex_lock(&pool->lock);
+        pool->busy--;
+        pthread_mutex_unlock(&pool->lock);
+    }
+
+    log_info("[worker %d] Exiting", worker->id);
+
+    return NULL;
+}
+
+bool hammy_worker_start(hammy_worker_t* worker, hammy_pool_t* pool, struct discord* client, int id) {
+    if (!worker || !pool || !client) { return false; }
+
+    worker->pool = pool;
+    worker->id = id;
+    worker->clientCopy = NULL;
+    worker->started = false;
+
+    // According to the concord spec, each thread must have its own discord client, so we clone it here.
+    // However, concord's buffers, URLs, headers, etc. are NOT shared-safe. They're per-client.
+    worker->clientCopy = discord_clone(client);
+    if (!worker->clientCopy) {
+        log_error("[worker %d] Failed to clone client", id);
+        return false;
+    }
+
+    if (pthread_create(&worker->thread, NULL, &hammy_worker_main, worker) != 0) {
+        log_error("[worker %d] Failed to create thread", id);
+        discord_cleanup(worker->clientCopy);
+        worker->clientCopy = NULL;
+
+        return false;
+    }
+
+    worker->started = true;
+
+    return true;
+}
+
+void hammy_worker_join(hammy_worker_t* worker) {
+    if (!worker) { return; }
+
+    if (worker->started) {
+        pthread_join(worker->thread, NULL);
+        worker->started = false;
+    }
+
+    if (worker->clientCopy) {
+        discord_cleanup(worker->clientCopy);
+        worker->clientCopy = NULL;
+    }
+}

src/main.c

@@ -11,7 +11,7 @@ static void on_signal(int sig) {
 
 void on_ready(struct discord* client, const struct discord_ready* event) {
     (void)client;
-    log_info("Logged in as %s", event->user->username);
+    log_info("[master] Logged in as %s", event->user->username);
 }
 
 int main(void) {
@@ -27,7 +27,7 @@ int main(void) {
 
     hammy_bot_t* bot = hammy_bot_create();
     if (!bot) {
-        log_error("Hammy Bot creation returned NULL! Bailing!");
+        log_error("[master] Hammy Bot creation returned NULL! Bailing!");
         ccord_global_cleanup();
 
         return 1;