wip reorg

4 files changed, 353 insertions, 0 deletions

diff --git a/libcr_ipc/coroutine_chan.h b/libcr_ipc/coroutine_chan.h
new file mode 100644
index 0000000..a3621a6
--- /dev/null
+++ b/libcr_ipc/coroutine_chan.h
@@ -0,0 +1,115 @@
+/* coroutine_chan.h - Simple channel system for coroutine.{h,c}
+ *
+ * Copyright (C) 2024  Luke T. Shumaker <lukeshu@lukeshu.com>
+ * SPDX-Licence-Identifier: AGPL-3.0-or-later
+ */
+
+/**
+ * The cr_chan_* macros form a simple Go-like unbuffered channel
+ * mechanism built on top of the cr_pause_and_yeild() and cr_unpause()
+ * primitives.
+ */
+#ifndef _COROUTINE_CHAN_H_
+#define _COROUTINE_CHAN_H_
+
+#include <assert.h>
+
+#include "coroutine.h"
+
+/**
+ * cr_chan_t(val_t) returns the type definition for a channel on
+ * that transports values of type `val_t`.
+ */
+#define cr_chan_t(val_t) struct {   \
+	bool ok_to_write;           \
+	cid_t reader, *tail_reader; \
+	val_t *reader_val;          \
+                                    \
+	bool ok_to_read;            \
+	cid_t writer, *tail_writer; \
+	val_t *writer_val;          \
+	}
+
+/* These "functions" are preprocessor macros instead of real C
+ * functions so that the compiler can do type-checking instead of
+ * having these functions take `void*`.  */
+
+/**
+ * ch_chan_send(*ch, val) sends `val` over `ch`.
+ */
+#define cr_chan_send(ch, _val) do {                                 \
+		if ((ch)->ok_to_write) { /* non-blocking */         \
+			*((ch)->reader_val) = (_val);               \
+			(ch)->ok_to_write = false;                  \
+			cr_unpause((ch)->reader);                   \
+		} else { /* blocking */                             \
+			cid_t next = 0;                             \
+			if ((ch)->writer) {                         \
+				*((ch)->tail_writer) = cr_getcid(); \
+				(ch)->tail_writer = &next;          \
+				cr_pause_and_yield();               \
+			} else {                                    \
+				(ch)->writer = cr_getcid();         \
+				(ch)->tail_writer = &next;          \
+			}                                           \
+			assert((ch)->writer == cr_getcid());        \
+			(ch)->writer_val = &(_val);                 \
+			(ch)->ok_to_read = true;                    \
+			cr_pause_and_yield();                       \
+			assert((ch)->ok_to_read == false);          \
+			if (next) {                                 \
+				(ch)->writer = next;                \
+				cr_unpause(next);                   \
+			} else {                                    \
+				(ch)->writer = 0;                   \
+				(ch)->tail_writer = NULL;           \
+			}                                           \
+		}                                                   \
+	} while (0)
+
+/**
+ * cr_chan_recv(ch, val_p) reads a value from ch into `*val_p`.
+ */
+#define cr_chan_recv(ch, _val_p) do {                               \
+		if ((ch)->ok_to_read) { /* non-blocking */          \
+			*(_val_p) = *((ch)->writer_val);            \
+			(ch)->ok_to_read = false;                   \
+			cr_unpause((ch)->writer;                    \
+		} else { /* blocking */                             \
+			cid_t next = 0;                             \
+			if ((ch)->reader) {                         \
+				*((ch)->tail_reader) = cr_getcid(); \
+				(ch)->tail_reader = &next;          \
+				cr_pause_and_yield();               \
+			 } else {                                   \
+				(ch)->reader = cr_getcid();         \
+				(ch)->tail_reader = &next;          \
+			}                                           \
+			assert((ch)->reader == cr_getcid());        \
+			(ch)->reader_val = (_val_p);                \
+			(ch)->ok_to_write = true;                   \
+			cr_pause_and_yield();                       \
+			assert((ch)->ok_to_write == false);         \
+			if (next) {                                 \
+				(ch)->reader = next;                \
+				cr_unpause(next);                   \
+			} else {                                    \
+				(ch)->reader = 0;                   \
+				(ch)->tail_reader = NULL;           \
+			}                                           \
+		}                                                   \
+	} while (0)
+
+/**
+ * cr_chan_can_send(ch) returns whether cr_chan_send(ch, val) would
+ * run without blocking.
+ */
+#define cr_chan_can_send(ch) ((ch)->ok_to_write)
+
+/**
+ * cr_chan_can_recv(ch) returns whether cr_chan_recv(ch, &val) would
+ * run without blocking.
+ */
+#define cr_chan_can_recv(ch) ((ch)->ok_to_read)
+
+#endif /* _COROUTINE_CHAN_H_ */
diff --git a/libcr_ipc/coroutine_rpc.h b/libcr_ipc/coroutine_rpc.h
new file mode 100644
index 0000000..11d40d7
--- /dev/null
+++ b/libcr_ipc/coroutine_rpc.h
@@ -0,0 +1,108 @@
+/* coroutine_rpc.h - Simple request/response system for coroutine.{h,c}
+ *
+ * Copyright (C) 2024  Luke T. Shumaker <lukeshu@lukeshu.com>
+ * SPDX-Licence-Identifier: AGPL-3.0-or-later
+ */
+
+/**
+ * The cr_rpc_* macros form a simple request/response mechanism built
+ * on top of the cr_pause_and_yeild() and cr_unpause() primitives.
+ */
+#ifndef _COROUTINE_RPC_H_
+#define _COROUTINE_RPC_H_
+
+#include <assert.h>
+
+#include "coroutine.h"
+
+/**
+ * cr_rpc_t(req_t, resp_t) returns the type definition for a
+ * rcp-channel on which the requester submits a value of type `req_t`
+ * and the responder responds with a value of type `resp_t`.
+ *
+ * There may be multiple concurrent requesters, but only one
+ * concurrent responder.  If you need multiple concurrent responders,
+ * set up a system of coroutine_chan.h channels.
+ */
+#define cr_rpc_t(req_t, resp_t) struct {            \
+		cid_t   requester, *tail_requester; \
+		cid_t   responder;                  \
+		req_t  *req_p;                      \
+		resp_t *resp_p;                     \
+	}
+
+/* These "functions" are preprocessor macros instead of real C
+ * functions so that the compiler can do type-checking instead of
+ * having these functions take `void*`.  */ 
+
+/**
+ * ch_rpc_req(cr_rpc_t(req_t, resp_t) *ch, resp_t *resp, req_t req)
+ * submits the `req` request to `ch` puts the response in `*resp_p`.
+ *
+ * Blocking: Always; until the responder has called both
+ * cr_rpc_recv_req() and cr_rpc_send_resp().
+ */
+#define cr_rpc_req(ch, _resp_p, _req) do {                    \
+		cid_t next = 0;                               \
+		if ((ch)->requester) {                        \
+			*((ch)->tail_requester = cr_getcid(); \
+			(ch)->tail_requester = &next;         \
+			cr_pause_and_yield();                 \
+		} else {                                      \
+			(ch)->requester = cr_getcid();        \
+			(ch)->tail_requester = &next;         \
+		}                                             \
+		assert((ch)->requester == cr_getcid());       \
+		(ch)->req_p = &(_req);                        \
+		(ch)->resp_p = (_resp_p;                      \
+		if ((ch)->responder != 0)                     \
+			cr_unpause((ch)->responder);          \
+		cr_pause_and_yield();                         \
+		if (next) {                                   \
+			(ch)->requester = next;               \
+			cr_unpause(next);                     \
+		} else {                                      \
+			(ch)->requester = 0;                  \
+			(ch)->tail_requester = NULL;          \
+		}
+	} while (0)
+
+/**
+ * cr_rpc_have_req(cr_rpc_t(req_t, resp_t) *ch) allows a responder
+ * to check whether or not there is a request waiting to be received
+ * (with cr_rpc_recv_req()) without blocking if there is not.
+ *
+ * Blocking: Never.
+ */
+#define cr_rpc_have_req(ch) ((ch)->req_p != NULL)
+
+/**
+ * cr_rpc_recv_req(cr_rpc_t(req_t, resp_t) *ch, req_t *req_p) reads
+ * a request from ch into `*req_p`.
+ *
+ * If there is not a pending request on `ch`, blocks until there is.
+ *
+ * Blocking: Maybe.
+ */
+#define cr_rpc_recv_req(ch, _req_p) do {       \
+		(ch)->responder = cr_getcid(); \
+		if ((ch)->requester == 0)      \
+			cr_pause_and_yield();  \
+		*(_req_p) = *((ch)->req_p);    \
+		(ch)->req_p = NULL;            \
+	} while (0)
+
+/**
+ * cr_rpc_send_resp(cr_rpc_t(req_t, resp_t) *ch, resp_t resp) sends
+ * the reply to the most-recently-read request.
+ *
+ * Blocking: Never.
+ */
+#define cr_rpc_send_resp(ch, _resp) do {       \
+		(ch)->responder = 0;           \
+		*((ch)->resp_p) = (_resp);     \
+		(ch)->resp_p = NULL;           \
+		cr_unpause((ch)->requester);   \
+	} while (0)
+
+#endif /* _COROUTINE_RPC_H_ */
diff --git a/libcr_ipc/coroutine_sema.c b/libcr_ipc/coroutine_sema.c
new file mode 100644
index 0000000..a8b5ec4
--- /dev/null
+++ b/libcr_ipc/coroutine_sema.c
@@ -0,0 +1,95 @@
+/* coroutine_sema.h - Simple semaphores for coroutine.{h,c}
+ *
+ * Copyright (C) 2024  Luke T. Shumaker <lukeshu@lukeshu.com>
+ * SPDX-Licence-Identifier: AGPL-3.0-or-later
+ */
+
+#include <assert.h>
+
+#include "coroutine_sema.h"
+
+struct cid_list {
+	cid_t                    val;
+	struct cid_list         *next;
+};
+
+/* head->next->next->tail */
+
+struct _cr_sema {
+	int              cnt;
+
+	struct cid_list  *head, **tail;
+	/* locked indicates that a call from within a coroutine is is
+	 * messing with ->{head,tail}, so a signal handler can't read
+	 * it.  */
+	bool             locked;
+};
+
+/** Drain the sema->{head,tail} list.  Returns true if cr_getcid() was drained.  */
+static inline bool drain(volatile cr_sema_t *sema) {
+	assert(!sema->locked);
+	cid_t self = cr_getcid();
+
+	enum drain_result {
+		DRAINING,
+		DRAINED_SELF, /* stopped because drained `self` */
+		DRAINED_ALL,  /* stopped because sema->head == NULL */
+		DRAINED_SOME, /* stopped because sema->cnt == 0 */
+	} state = DRAINING;
+	do { 
+		sema->locked = true;
+		while (state == DRAINING) {
+			if (!sema->head) {
+				state = DRAINED_ALL;
+			} else if (!sema->cnt) {
+				state = DRAINED_SOME;
+			} else {
+				sema->cnt--;
+				cid_t cid = sema->head->val;
+				if (cid == self)
+					state = DRAINED_SELF;
+				else
+					cr_unpause(sema->head->val);
+				sema->head = sema->head->next;
+				if (!sema->head)
+					sema->tail = &sema->head;
+			}
+		}
+		sema->locked = false;
+		/* If there are still coroutines in sema->head, check
+		 * that sema->cnt wasn't incremented between `if
+		 * (!sema->cnt)` and `sema->locked = false`.  */
+	} while (state == DRAINED_SOME && cnt);
+	/* If state == DRAINED_SELF, then we better have been the last
+	 * item in the list!  */
+	assert(state != DRAINED_SELF || !sema->head);
+	return state == DRAINED_SELF;
+}
+
+void cr_sema_signal(volatile cr_sema_t *sema) {
+	sema->cnt++;
+	if (!sema->locked)
+		drain();
+}
+
+void cr_sema_wait(volatile cr_sema_t *sema) {
+	struct cid_list self = {
+		.val = cr_getcid(),
+		.next = NULL,
+	};
+
+	sema->locked = true;
+	if (!sema->tail)
+		sema->head = &self;
+	else
+		*(sema->tail) = &self;
+	sema->tail = &(self.next);
+	sema->locked = false;
+
+	if (drain())
+		/* DRAINED_SELF: (1) No need to pause+yield, (2) we
+		 * better have been the last item in the list!  */
+		assert(!self.next);
+	else
+		cr_pause_and_yield();
+}
diff --git a/libcr_ipc/coroutine_sema.h b/libcr_ipc/coroutine_sema.h
new file mode 100644
index 0000000..57d5855
--- /dev/null
+++ b/libcr_ipc/coroutine_sema.h
@@ -0,0 +1,35 @@
+/* coroutine_sema.h - Simple semaphores for coroutine.{h,c}
+ *
+ * Copyright (C) 2024  Luke T. Shumaker <lukeshu@lukeshu.com>
+ * SPDX-Licence-Identifier: AGPL-3.0-or-later
+ */
+
+#ifndef _COROUTINE_SEMA_H_
+#define _COROUTINE_SEMA_H_
+
+/**
+ * A cr_sema_t is a fair unbounded[1] counting semaphore.
+ *
+ * [1]: Well, INT_MAX
+ */
+typedef struct sema_t cr_sema_t;
+
+/**
+ * Increment the semaphore,
+ *
+ * @blocks never
+ * @yields never
+ * @run_in anywhere (coroutine, sighandler)
+ */
+void cr_sema_signal(cr_sema_t *sema);
+
+/**
+ * Wait until the semaphore is >0, then decrement it.
+ *
+ * @blocks maybe
+ * @yields maybe
+ * @may_run_in coroutine
+ */
+void cr_sema_wait(cr_sema_t *sema);
+
+#endif /* _COROUTINE_SEMA_H_ */