diff --git a/include/hammy/job.h b/include/hammy/job.h index be6bf56..865e54b 100644 --- a/include/hammy/job.h +++ b/include/hammy/job.h @@ -2,16 +2,58 @@ #define HAMMY_JOB_H #include +#include +#include #include + #include +// 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 diff --git a/include/hammy/pool.h b/include/hammy/pool.h index e4d8d14..3abf162 100644 --- a/include/hammy/pool.h +++ b/include/hammy/pool.h @@ -1,22 +1,69 @@ #ifndef HAMMY_POOL_H #define HAMMY_POOL_H -#include +#include + #include #include +#include +#include + +#include + +// 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 diff --git a/include/hammy/worker.h b/include/hammy/worker.h index 82591a5..e7fff9b 100644 --- a/include/hammy/worker.h +++ b/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 diff --git a/src/hammy/job.c b/src/hammy/job.c new file mode 100644 index 0000000..6c0171b --- /dev/null +++ b/src/hammy/job.c @@ -0,0 +1,147 @@ +#include +#include +#include +#include +#include + +#include + +// 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, ¶ms, 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."); +} diff --git a/src/hammy/pool.c b/src/hammy/pool.c new file mode 100644 index 0000000..231d5e5 --- /dev/null +++ b/src/hammy/pool.c @@ -0,0 +1,171 @@ +#include +#include +#include + +#include +#include +#include + +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); +} \ No newline at end of file diff --git a/src/hammy/worker.c b/src/hammy/worker.c new file mode 100644 index 0000000..b1a1bdf --- /dev/null +++ b/src/hammy/worker.c @@ -0,0 +1,99 @@ +#include +#include +#include + +#include +#include +#include + +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; + } +} diff --git a/src/main.c b/src/main.c index 9f3f0e4..62bd59c 100644 --- a/src/main.c +++ b/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;