add Go-like unbuffered channels

2 files changed, 119 insertions, 1 deletions

diff --git a/coroutine.h b/coroutine.h
index 4b16b1f..d02d0f9 100644
--- a/coroutine.h
+++ b/coroutine.h
@@ -19,6 +19,9 @@
  *
  * See also: coroutine_rpc.h is a request/response system built on top
  * of coroutine.{h,c}.
+ *
+ * See also: coroutine_chan.h is a 1-way channel system built on top
+ * of coroutine.{h,c}.
  */
 #ifndef _COROUTINE_H_
 #define _COROUTINE_H_
@@ -66,7 +69,7 @@ typedef size_t cid_t;
  * Specifically, coroutine_add() and
  * cr_{yield,pause_and_yield,exit,end}() are explicitly forbidden to
  * call from within a coroutine before cr_begin() (note that the
- * cr_rpc_*() macros call these functions).
+ * cr_rpc_*() and cr_chan_*() macros call these functions).
  */
 typedef void (*cr_fn_t)(void *args);
 #define COROUTINE __attribute__ ((noreturn)) void
diff --git a/coroutine_chan.h b/coroutine_chan.h
new file mode 100644
index 0000000..f31586a
--- /dev/null
+++ b/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;                                 \
+			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;                                 \
+			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_ */