aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/connection.c45
-rw-r--r--src/main.c125
-rw-r--r--src/meson.build5
-rw-r--r--src/pipeline.c316
4 files changed, 491 insertions, 0 deletions
diff --git a/src/connection.c b/src/connection.c
new file mode 100644
index 0000000..ecbb7fa
--- /dev/null
+++ b/src/connection.c
@@ -0,0 +1,45 @@
+#include "connection.h"
+#include "pipeline.h"
+#include <stdlib.h>
+
+struct ptx_connection *ptx_connection_new(void)
+{
+ struct ptx_connection *conn = malloc(sizeof(struct ptx_connection));
+ if (!conn) return NULL;
+
+ /* initialize these in case we goto cleanup later */
+ conn->cli_read = NULL;
+ conn->cli_write = NULL;
+ conn->srv_read = NULL;
+ conn->srv_write = NULL;
+
+ /* do initialization */
+
+ if (!(conn->cli_read = ptx_pipeline_new(4))) goto cleanup;
+ if (!(conn->cli_write = ptx_pipeline_new(4))) goto cleanup;
+ if (!(conn->srv_read = ptx_pipeline_new(4))) goto cleanup;
+ if (!(conn->srv_write = ptx_pipeline_new(4))) goto cleanup;
+
+ return conn;
+
+cleanup:
+ ptx_connection_free(conn);
+
+ return NULL;
+}
+
+void ptx_connection_free(struct ptx_connection *conn)
+{
+ /* NOTE: this function might be called with a partially-constructed or inconsistent object */
+
+ if (!conn) return;
+
+ /* do cleanup */
+
+ ptx_pipeline_free(conn->cli_read);
+ ptx_pipeline_free(conn->cli_write);
+ ptx_pipeline_free(conn->srv_read);
+ ptx_pipeline_free(conn->srv_write);
+
+ free(conn);
+}
diff --git a/src/main.c b/src/main.c
new file mode 100644
index 0000000..83ed688
--- /dev/null
+++ b/src/main.c
@@ -0,0 +1,125 @@
+#include <stdio.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdarg.h>
+
+#include "pipeline.h"
+
+#define UNUSED(_v) ((void)(_v))
+
+PIPELINE_STAGE static int identity_stage_handle(const ptx_pipeline_ctx_t *ctx, const void *data, size_t sz)
+{
+ return ptx_pipeline_ctx_next(ctx, data, sz);
+}
+
+PIPELINE_INIT static int debug_stage_init(const char *name, void **ppuser, va_list init_args)
+{
+ static const char *const messages[] = { "no continue :(", "continue :)" };
+
+ UNUSED(name);
+
+ int config = !!va_arg(init_args, int);
+ *ppuser = (void *)(uintptr_t)config;
+
+ printf("Debug stage %s (init): %s\n", name, messages[config]);
+
+ return 0;
+}
+
+PIPELINE_CLEANUP static void debug_stage_cleanup(const char *name, void *puser)
+{
+ printf("Debug stage %s (cleanup): %p\n", name, puser);
+}
+
+PIPELINE_STAGE static int debug_stage_handle(const ptx_pipeline_ctx_t *ctx, const void *data, size_t sz)
+{
+ bool cont = !!(uintptr_t)(*ptx_pipeline_ctx_get_user(ctx));
+ printf("Debug stage %s:\n Data: %p\n Size: %zu\n", ptx_pipeline_ctx_get_name(ctx), data, sz);
+ return cont ? ptx_pipeline_ctx_next(ctx, data, sz) : 0;
+}
+
+static const struct ptx_pipeline_stage_funcs identity_stage = {
+ .init = NULL,
+ .handler = &identity_stage_handle,
+ .cleanup = NULL
+};
+
+static const struct ptx_pipeline_stage_funcs debug_stage = {
+ .init = &debug_stage_init,
+ .handler = &debug_stage_handle,
+ .cleanup = &debug_stage_cleanup
+};
+
+/* mockup:
+ * Pipeline signature: D -> D*1
+ * int simple_1to1_stage(struct pipeline_ctx *ctx, void *data, size_t len) {
+ * void *out = ... transform data ...;
+ * size_t out_len = ^^^;
+ * return ptx_pipeline_next(ctx, out, out_len);
+ * }
+ *
+ * Pipeline signature: D -> D*(0,n)
+ * int complex_1tomany_stage(struct pipeline_ctx *ctx, void *data, size_t len) {
+ * int err;
+ * for (void *out; has_data(data, len); out = next_data(data, &len)) {
+ * if (!out) {
+ * return -1; ... next_data failed somehow ...
+ * }
+ *
+ * if ((err = ptx_pipeline_next(ctx, out, out_len)) < 0) {
+ * return err;
+ * }
+ * }
+ * return 0;
+ * }
+ *
+ * Pipeline signature: D -> D*1
+ * int stateful_stage(struct pipeline_ctx *ctx, void *data, size_t len) {
+ * my_stage_t *stg = ctx->me;
+ * stg->stat_in += len;
+ * return ptx_pipeline_next(ctx, data, len);
+ * }
+ *
+ * Pipeline signature: D -> void (i.e., never calls next)
+ * int terminal_stage(struct pipeline_ctx *ctx, void *data, size_t len) {
+ * my_handler_t *handler = ctx->me;
+ * handler->do_handle(data, len);
+ * return 0; ... return values should indicate pipeline (i.e., parsing) errors, not application errors ...
+ * }
+ */
+
+int main(void) {
+ ptx_pipeline_t *pl = ptx_pipeline_new(1);
+ if (!pl) {
+ fputs("Allocation failure: pipeline\n", stderr);
+ return 1;
+ }
+
+ if (ptx_pipeline_add_before(pl, &identity_stage, "identity1", NULL) < 0) {
+ fputs("Failed to add identity1 to pipeline\n", stderr);
+ return 1;
+ }
+
+ if (ptx_pipeline_add_after(pl, &debug_stage, "debug1", "identity1", true) < 0) {
+ fputs("Failed to add debug1 to pipeline\n", stderr);
+ return 1;
+ }
+
+ if (ptx_pipeline_add_after(pl, &debug_stage, "debug2", "debug1", false) < 0) {
+ fputs("Failed to add debug2 to pipeline\n", stderr);
+ return 1;
+ }
+
+ printf("Handle returned: %d\n", ptx_pipeline_handle(pl, "sussy", 10));
+
+ if (ptx_pipeline_remove_stage(pl, "debug1") < 1) {
+ fputs("Failed to remove debug1 from pipeline\n", stderr);
+ return 1;
+ }
+
+ printf("Handle returned: %d\n", ptx_pipeline_handle(pl, "sussy2", 20));
+
+ ptx_pipeline_free(pl);
+
+ return 0;
+}
diff --git a/src/meson.build b/src/meson.build
new file mode 100644
index 0000000..d1a3033
--- /dev/null
+++ b/src/meson.build
@@ -0,0 +1,5 @@
+prog_sources = files(
+ 'pipeline.c',
+ 'connection.c',
+ 'main.c'
+)
diff --git a/src/pipeline.c b/src/pipeline.c
new file mode 100644
index 0000000..8b54cca
--- /dev/null
+++ b/src/pipeline.c
@@ -0,0 +1,316 @@
+#include "pipeline.h"
+
+#include <string.h>
+#include <stdlib.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <limits.h>
+
+struct ptx__pipeline_stage_tag
+{
+ struct ptx_pipeline_stage_funcs impl;
+ char *name;
+ void *user;
+};
+
+struct ptx__pipeline_ctx_tag
+{
+ ptx_pipeline_t *pipeline;
+ size_t curidx;
+};
+
+struct ptx__pipeline_tag
+{
+ size_t stages_len, stages_cap;
+ struct ptx__pipeline_stage_tag *stages;
+ bool handling;
+};
+
+/* returns:
+ * -1 : allocation failed
+ * 0 : nothing was allocated
+ * 1 : pipeline grown */
+static int ptx__pipeline_reserve(ptx_pipeline_t *pl, size_t req)
+{
+ assert(pl);
+ assert(req < SSIZE_MAX);
+
+ if (req < pl->stages_cap) return 0;
+
+ struct ptx__pipeline_stage_tag *temp = realloc(pl->stages, req * sizeof(struct ptx__pipeline_stage_tag));
+ if (!temp) return -1;
+
+ pl->stages = temp;
+ pl->stages_cap = req;
+
+ return 1;
+}
+
+/* allocates a new pipeline, with space for `initial_size' new stages. */
+PTX_API ptx_pipeline_t *ptx_pipeline_new(size_t initial_size)
+{
+ ptx_pipeline_t *pl = malloc(sizeof(ptx_pipeline_t));
+ if (!pl) return NULL;
+
+ pl->stages_len = 0;
+ pl->stages_cap = 0;
+ pl->stages = NULL;
+ pl->handling = false;
+
+ ptx__pipeline_reserve(pl, initial_size < 1 ? 1 : initial_size);
+
+ return pl;
+}
+
+static void ptx__cleanup_pipeline_stage(struct ptx__pipeline_stage_tag *stage)
+{
+ assert(stage);
+
+ if (stage->impl.cleanup) {
+ (*stage->impl.cleanup)(stage->name, stage->user);
+ }
+
+ free(stage->name);
+
+ /* clobber the structure just cuz */
+ memset(stage, 0, sizeof(*stage));
+}
+
+/* frees an existing pipeline stage. note that this will call the cleanup function in the stages as well. */
+PTX_API void ptx_pipeline_free(ptx_pipeline_t *pl)
+{
+ if (!pl) return;
+
+ for (size_t idx = 0; idx < pl->stages_len; ++idx) {
+ ptx__cleanup_pipeline_stage(pl->stages + idx);
+ }
+
+ free(pl->stages);
+ free(pl);
+}
+
+static int ptx__init_pipeline_stage(
+ struct ptx__pipeline_stage_tag *PTX_RESTRICT pstage,
+ const struct ptx_pipeline_stage_funcs *PTX_RESTRICT stage_impl,
+ const char *name,
+ va_list setup_args)
+{
+ assert(pstage);
+ assert(stage_impl);
+ assert(name);
+
+ char *namedup = NULL;
+ void *user = NULL;
+
+ /* I'd rather try to duplicate the name first. Otherwise, we'd have to clean up the
+ * stage's user data right after initializing it. */
+ namedup = strdup(name);
+ if (!namedup) goto cleanup;
+
+ if (stage_impl->init && (*stage_impl->init)(name, &user, setup_args) < 0) {
+ /* setup failed :((( */
+ goto cleanup;
+ }
+
+ /* yay it worked :) */
+ pstage->user = user;
+ pstage->name = namedup;
+ memcpy(&pstage->impl, stage_impl, sizeof(*stage_impl));
+
+ return 0;
+
+cleanup:
+ free(namedup);
+ return -1;
+}
+
+/* this will grow a pipeline to fit a new stage.
+ * this function also shifts entries in `pl->stages' such that `to_add' is an empty stage. */
+static int ptx__pipeline_prepare_insert_stage_at(ptx_pipeline_t *pipeline, size_t to_add)
+{
+ assert(to_add <= pipeline->stages_len);
+ if (ptx__pipeline_reserve(pipeline, pipeline->stages_len + 1) < 0) {
+ return -1;
+ }
+
+ /* don't perform the move if we wouldn't be moving anything. */
+ if (to_add < pipeline->stages_len) {
+ memmove(pipeline->stages + to_add + 1,
+ pipeline->stages + to_add,
+ sizeof(struct ptx__pipeline_stage_tag) * (pipeline->stages_len - to_add));
+ }
+
+ /* increment the length of the array here for consistency.
+ * NOTE: this function still leaves the pipeline in an inconsistent state:
+ * there is an uninitialized stage at `to_add' :( */
+ ++pipeline->stages_len;
+
+ return 0;
+}
+
+static ssize_t ptx__pipeline_find_stage(ptx_pipeline_t *ptx, const char *name)
+{
+ assert(ptx);
+ assert(name);
+
+ for (size_t idx = 0; idx < ptx->stages_len; ++idx) {
+ if (!strcmp(ptx->stages[idx].name, name)) {
+ assert(idx <= SSIZE_MAX);
+ return (ssize_t)idx;
+ }
+ }
+
+ return -1;
+}
+
+static int ptx__pipeline_add_at(
+ ptx_pipeline_t *pipeline,
+ const struct ptx_pipeline_stage_funcs *stage_impl,
+ const char *name,
+ size_t idx,
+ va_list init_args)
+{
+ struct ptx__pipeline_stage_tag temp_stage;
+ memset(&temp_stage, 0, sizeof(temp_stage));
+
+ /* first, initialize the stage itself */
+ if (ptx__init_pipeline_stage(&temp_stage, stage_impl, name, init_args) < 0) {
+ return -1;
+ }
+
+ /* next, grow the pipeline to fit the new stage */
+ if (ptx__pipeline_prepare_insert_stage_at(pipeline, idx) < 0) {
+ /* uh oh, it failed. clean up the stage we just initialized */
+ ptx__cleanup_pipeline_stage(&temp_stage);
+ return -1;
+ }
+
+ /* finally, put the stage into the array */
+ memcpy(pipeline->stages + idx, &temp_stage, sizeof(temp_stage));
+ return 0;
+}
+
+/* adds `stage' with `name' to `pipeline' after `after' *
+ * returns: number of stages added to pipeline (0 or 1), or -1 if there was an allocation or initialization error. */
+PTX_API int ptx_pipeline_add_after(
+ ptx_pipeline_t *pipeline,
+ const struct ptx_pipeline_stage_funcs *stage,
+ const char *name,
+ const char *after,
+ ...)
+{
+ assert(pipeline);
+ assert(!pipeline->handling);
+
+ size_t idx;
+
+ if (after) {
+ ssize_t sidx = ptx__pipeline_find_stage(pipeline, after);
+ if (sidx < 0) return 0;
+
+ idx = (size_t)(sidx + 1); /* after */
+ } else {
+ idx = pipeline->stages_len;
+ }
+
+ va_list init_args;
+ va_start(init_args, after);
+
+ int ret = ptx__pipeline_add_at(pipeline, stage, name, idx, init_args);
+
+ va_end(init_args);
+ return ret;
+}
+
+/* adds `stage' with `name' to `pipeline' before `before'
+ * returns: number of stages added to pipeline (0 or 1), or -1 if there was an allocation or initialization error. */
+PTX_API int ptx_pipeline_add_before(
+ ptx_pipeline_t *pipeline,
+ const struct ptx_pipeline_stage_funcs *stage,
+ const char *name,
+ const char *before,
+ ...)
+{
+ assert(pipeline);
+ assert(!pipeline->handling);
+
+ size_t idx;
+
+ if (before) {
+ ssize_t sidx = ptx__pipeline_find_stage(pipeline, before);
+ if (sidx < 0) return 0;
+
+ idx = (size_t)sidx; /* before */
+ } else {
+ idx = 0;
+ }
+
+ va_list init_args;
+ va_start(init_args, before);
+
+ int ret = ptx__pipeline_add_at(pipeline, stage, name, idx, init_args);
+
+ va_end(init_args);
+ return ret;
+}
+
+/* returns: number of stages removed from pipeline */
+PTX_API int ptx_pipeline_remove_stage(ptx_pipeline_t *pipeline, const char *name)
+{
+ assert(pipeline);
+ assert(!pipeline->handling);
+
+ ssize_t idx = ptx__pipeline_find_stage(pipeline, name);
+ if (idx < 0) return 0;
+
+ ptx__cleanup_pipeline_stage(pipeline->stages + idx);
+ memmove(pipeline->stages + idx, pipeline->stages + idx + 1, sizeof(pipeline->stages[0]) * (pipeline->stages_len - (size_t)idx - 1));
+
+ --pipeline->stages_len;
+ return 1;
+}
+
+PTX_API void **ptx_pipeline_ctx_get_user(const ptx_pipeline_ctx_t *ctx)
+{
+ assert(ctx);
+ return &ctx->pipeline->stages[ctx->curidx].user;
+}
+
+PTX_API const char *ptx_pipeline_ctx_get_name(const ptx_pipeline_ctx_t *ctx)
+{
+ assert(ctx);
+ return ctx->pipeline->stages[ctx->curidx].name;
+}
+
+PTX_API int ptx_pipeline_ctx_next(const ptx_pipeline_ctx_t *ctx, const void *nextdata, size_t nextsize)
+{
+ assert(ctx->curidx + 1 < ctx->pipeline->stages_len);
+
+ ptx_pipeline_ctx_t new_ctx;
+ new_ctx.pipeline = ctx->pipeline;
+ new_ctx.curidx = ctx->curidx + 1;
+
+ assert(new_ctx.pipeline->stages[new_ctx.curidx].impl.handler);
+
+ return (*new_ctx.pipeline->stages[new_ctx.curidx].impl.handler)(&new_ctx, nextdata, nextsize);
+}
+
+PTX_API int ptx_pipeline_handle(ptx_pipeline_t *pipeline, const void *data, size_t sz)
+{
+ assert(pipeline);
+ assert(!pipeline->handling);
+
+ if (pipeline->stages_len == 0) return 0; /* pipeline does nothing... */
+
+ pipeline->handling = true;
+
+ ptx_pipeline_ctx_t ctx;
+ ctx.pipeline = pipeline;
+ ctx.curidx = 0;
+
+ assert(pipeline->stages[0].impl.handler);
+ int ret = (*pipeline->stages[0].impl.handler)(&ctx, data, sz);
+
+ pipeline->handling = 0;
+ return ret;
+}
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2025-12-06 08:29:48 +0000