use channels to syncronize GAsync callbacks

channels are more idiomatic

CMakeLists.txt

@@ -4,6 +4,7 @@ project(coio)
 add_library(${PROJECT_NAME}
     coro.c
     coio.c
+    channel.c
     coio_glib.c)
 
 add_custom_target(${PROJECT_NAME}_files SOURCES coioimpl.h)
@@ -13,8 +14,14 @@ target_compile_options(${PROJECT_NAME} PRIVATE -W -Wall -Wextra -Werror)
 
 find_package(PkgConfig REQUIRED)
 pkg_check_modules(GLIB2 REQUIRED glib-2.0)
+pkg_check_modules(GIO2 REQUIRED gio-2.0)
 
 target_include_directories(${PROJECT_NAME} PUBLIC ${GLIB2_INCLUDE_DIRS})
 target_link_libraries(${PROJECT_NAME} ${GLIB2_LIBRARIES})
 
+target_include_directories(${PROJECT_NAME} PRIVATE ${GIO2_INCLUDE_DIRS})
+target_link_libraries(${PROJECT_NAME} ${GIO2_LIBRARIES})
+
+target_link_libraries(${PROJECT_NAME} m)
+
 target_include_directories(${PROJECT_NAME} PUBLIC .)

Makefile

@@ -16,10 +16,10 @@ $(LIB): $(OBJS)
 	$(AR) rvc $(LIB) $?
 
 testyield: testyield.c $(LIB)
-	$(CC) $(CFLAGS) -o $@ testyield.c $(LIB)
+	$(CC) $(CFLAGS) -o $@ testyield.c $(LIB) -lm
 
 testdelay: testdelay.c $(LIB)
-	$(CC) $(CFLAGS) -o $@ testdelay.c $(LIB)
+	$(CC) $(CFLAGS) -o $@ testdelay.c $(LIB) -lm
 
 test: testyield testdelay
 

channel.c

@@ -0,0 +1,393 @@
+/* Copyright (c) 2005 Russ Cox, MIT; see COPYRIGHT */
+/* Copyright (c) 2019 Moritz Bitsch */
+
+#include "coioimpl.h"
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+#include <assert.h>
+
+CoioChannel* coio_chancreate(int elemsize, int bufsize)
+{
+	CoioChannel* c;
+
+	c = malloc(sizeof *c+bufsize*elemsize);
+	if (c == NULL)
+	{
+		perror("chancreate malloc");
+		exit(1);
+	}
+	memset(c, 0, sizeof *c);
+	c->elemsize = elemsize;
+	c->bufsize = bufsize;
+	c->nbuf = 0;
+	c->buf = (unsigned char*)(c+1);
+	return c;
+}
+
+/* bug - work out races */
+void coio_chanfree(CoioChannel* c)
+{
+	if (c == NULL)
+		return;
+	free(c->name);
+	free(c->arecv.a);
+	free(c->asend.a);
+	free(c);
+}
+
+static void addarray(Altarray* a, CoioAlt* alt)
+{
+	if (a->n == a->m)
+	{
+		a->m += 16;
+		a->a = realloc(a->a, a->m*sizeof a->a[0]);
+	}
+	a->a[a->n++] = alt;
+}
+
+static void delarray(Altarray* a, int i)
+{
+	--a->n;
+	a->a[i] = a->a[a->n];
+}
+
+/*
+ * doesn't really work for things other than CHANSND and CHANRCV
+ * but is only used as arg to chanarray, which can handle it
+ */
+#define otherop(op)	(COIO_CHANSND+COIO_CHANRCV-(op))
+
+static Altarray* chanarray(CoioChannel* c, uint op)
+{
+	switch (op)
+	{
+	default:
+		return NULL;
+	case COIO_CHANSND:
+		return &c->asend;
+	case COIO_CHANRCV:
+		return &c->arecv;
+	}
+}
+
+static int altcanexec(CoioAlt* a)
+{
+	Altarray* ar;
+	CoioChannel* c;
+
+	if (a->op == COIO_CHANNOP)
+		return 0;
+	c = a->c;
+	if (c->bufsize == 0)
+	{
+		ar = chanarray(c, otherop(a->op));
+		return ar && ar->n;
+	}
+	else
+	{
+		switch (a->op)
+		{
+		default:
+			return 0;
+		case COIO_CHANSND:
+			return c->nbuf < c->bufsize;
+		case COIO_CHANRCV:
+			return c->nbuf > 0;
+		}
+	}
+}
+
+static void altqueue(CoioAlt* a)
+{
+	Altarray* ar;
+
+	ar = chanarray(a->c, a->op);
+	addarray(ar, a);
+}
+
+static void altdequeue(CoioAlt* a)
+{
+	unsigned int i;
+	Altarray* ar;
+
+	ar = chanarray(a->c, a->op);
+	if (ar == NULL)
+	{
+		fprintf(stderr, "bad use of altdequeue op=%d\n", a->op);
+		abort();
+	}
+
+	for (i=0; i<ar->n; i++)
+		if (ar->a[i] == a)
+		{
+			delarray(ar, i);
+			return;
+		}
+	fprintf(stderr, "cannot find self in altdq\n");
+	abort();
+}
+
+static void altalldequeue(CoioAlt* a)
+{
+	int i;
+
+	for (i=0; a[i].op!=COIO_CHANEND && a[i].op!=COIO_CHANNOBLK; i++)
+		if (a[i].op != COIO_CHANNOP)
+			altdequeue(&a[i]);
+}
+
+static void amove(void* dst, void* src, uint n)
+{
+	if (dst)
+	{
+		if (src == NULL)
+			memset(dst, 0, n);
+		else
+			memmove(dst, src, n);
+	}
+}
+
+/*
+ * Actually move the data around.  There are up to three
+ * players: the sender, the receiver, and the channel itself.
+ * If the channel is unbuffered or the buffer is empty,
+ * data goes from sender to receiver.  If the channel is full,
+ * the receiver removes some from the channel and the sender
+ * gets to put some in.
+ */
+static void altcopy(CoioAlt* s, CoioAlt* r)
+{
+	CoioAlt* t;
+	CoioChannel* c;
+	unsigned char* cp;
+
+	/*
+	 * Work out who is sender and who is receiver
+	 */
+	if (s == NULL && r == NULL)
+		return;
+	assert(s != NULL);
+	c = s->c;
+	if (s->op == COIO_CHANRCV)
+	{
+		t = s;
+		s = r;
+		r = t;
+	}
+	assert(s==NULL || s->op == COIO_CHANSND);
+	assert(r==NULL || r->op == COIO_CHANRCV);
+
+	/*
+	 * Channel is empty (or unbuffered) - copy directly.
+	 */
+	if (s && r && c->nbuf == 0)
+	{
+		amove(r->v, s->v, c->elemsize);
+		return;
+	}
+
+	/*
+	 * Otherwise it's always okay to receive and then send.
+	 */
+	if (r)
+	{
+		cp = c->buf + c->off*c->elemsize;
+		amove(r->v, cp, c->elemsize);
+		--c->nbuf;
+		if (++c->off == c->bufsize)
+			c->off = 0;
+	}
+	if (s)
+	{
+		cp = c->buf + (c->off+c->nbuf)%c->bufsize*c->elemsize;
+		amove(cp, s->v, c->elemsize);
+		++c->nbuf;
+	}
+}
+
+static void altexec(CoioAlt* a)
+{
+	int i;
+	Altarray* ar;
+	CoioAlt* other;
+	CoioChannel* c;
+
+	c = a->c;
+	ar = chanarray(c, otherop(a->op));
+	if (ar && ar->n)
+	{
+		i = rand()%ar->n;
+		other = ar->a[i];
+		altcopy(a, other);
+		altalldequeue(other->xalt);
+		other->xalt[0].xalt = other;
+		coio_ready(other->task);
+	}
+	else
+		altcopy(a, NULL);
+}
+
+#define dbgalt 0
+int coio_chanalt(CoioAlt* a)
+{
+	int i, j, ncan, n, canblock;
+	CoioChannel* c;
+	CoioTask* t;
+
+	//needstack(512);
+	for (i=0; a[i].op != COIO_CHANEND && a[i].op != COIO_CHANNOBLK; i++)
+		;
+	n = i;
+	canblock = a[i].op == COIO_CHANEND;
+
+	t = coio_current;
+	for (i=0; i<n; i++)
+	{
+		a[i].task = t;
+		a[i].xalt = a;
+	}
+	if (dbgalt) printf("alt ");
+	ncan = 0;
+	for (i=0; i<n; i++)
+	{
+		c = a[i].c;
+		if (dbgalt) printf(" %c:", "esrnb"[a[i].op]);
+		if (dbgalt)
+		{
+			if (c->name) printf("%s", c->name);
+			else printf("%p", c);
+		}
+		if (altcanexec(&a[i]))
+		{
+			if (dbgalt) printf("*");
+			ncan++;
+		}
+	}
+	if (ncan)
+	{
+		j = rand()%ncan;
+		for (i=0; i<n; i++)
+		{
+			if (altcanexec(&a[i]))
+			{
+				if (j-- == 0)
+				{
+					if (dbgalt)
+					{
+						c = a[i].c;
+						printf(" => %c:", "esrnb"[a[i].op]);
+						if (c->name) printf("%s", c->name);
+						else printf("%p", c);
+						printf("\n");
+					}
+					altexec(&a[i]);
+					return i;
+				}
+			}
+		}
+	}
+	if (dbgalt)printf("\n");
+
+	if (!canblock)
+		return -1;
+
+	for (i=0; i<n; i++)
+	{
+		if (a[i].op != COIO_CHANNOP)
+			altqueue(&a[i]);
+	}
+
+	coio_transfer();
+
+	/*
+	 * the guy who ran the op took care of dequeueing us
+	 * and then set a[0].alt to the one that was executed.
+	 */
+	return a[0].xalt - a;
+}
+
+static int _chanop(CoioChannel* c, int op, void* p, int canblock)
+{
+	CoioAlt a[2];
+
+	a[0].c = c;
+	a[0].op = op;
+	a[0].v = p;
+	a[1].op = canblock ? COIO_CHANEND : COIO_CHANNOBLK;
+	if (coio_chanalt(a) < 0)
+		return -1;
+	return 1;
+}
+
+int coio_chansend(CoioChannel* c, void* v)
+{
+	return _chanop(c, COIO_CHANSND, v, 1);
+}
+
+int coio_channbsend(CoioChannel* c, void* v)
+{
+	return _chanop(c, COIO_CHANSND, v, 0);
+}
+
+int coio_chanrecv(CoioChannel* c, void* v)
+{
+	return _chanop(c, COIO_CHANRCV, v, 1);
+}
+
+int coio_channbrecv(CoioChannel* c, void* v)
+{
+	return _chanop(c, COIO_CHANRCV, v, 0);
+}
+
+int coio_chansendp(CoioChannel* c, void* v)
+{
+	return _chanop(c, COIO_CHANSND, (void*)&v, 1);
+}
+
+void* coio_chanrecvp(CoioChannel* c)
+{
+	void* v;
+
+	_chanop(c, COIO_CHANRCV, (void*)&v, 1);
+	return v;
+}
+
+int coio_channbsendp(CoioChannel* c, void* v)
+{
+	return _chanop(c, COIO_CHANSND, (void*)&v, 0);
+}
+
+void* coio_channbrecvp(CoioChannel* c)
+{
+	void* v;
+
+	_chanop(c, COIO_CHANRCV, (void*)&v, 0);
+	return v;
+}
+
+int coio_chansendul(CoioChannel* c, ulong val)
+{
+	return _chanop(c, COIO_CHANSND, &val, 1);
+}
+
+ulong coio_chanrecvul(CoioChannel* c)
+{
+	ulong val;
+
+	_chanop(c, COIO_CHANRCV, &val, 1);
+	return val;
+}
+
+int coio_channbsendul(CoioChannel* c, ulong val)
+{
+	return _chanop(c, COIO_CHANSND, &val, 0);
+}
+
+ulong coio_channbrecvul(CoioChannel* c)
+{
+	ulong val;
+
+	_chanop(c, COIO_CHANRCV, &val, 0);
+	return val;
+}

coio.c

@@ -18,6 +18,7 @@
 #include <stdlib.h>
 #include <unistd.h>
 #include <time.h>
+#include <math.h>
 #include <stdio.h>
 
 #include "coioimpl.h"
@@ -31,7 +32,7 @@
 CoioTaskList 	coio_ready_list = {0, 0};
 CoioTaskList 	coio_sleeping = {0, 0};
 coro_context    coio_sched_ctx;
-CoioTask*       coio_current;
+CoioTask*       coio_current = NULL;
 unsigned long   coio_taskcount = 0;
 
 static int msleep(uvlong ms)
@@ -61,7 +62,7 @@ _process_events()
 		if (now >= t->timeout) {
 			ms = 0;
 		} else {
-			ms = (t->timeout - now) / 1000000;
+			ms = (t->timeout - now);
 		}
 	}
 	/* TODO:do I/O polling instead of usleep */
@@ -152,7 +153,7 @@ coio_timeout(CoioTask* task, int ms)
 	CoioTask*       t;
 
 	if (ms >= 0) {
-		task->timeout = coio_now() + (ms * 1000000);
+		task->timeout = coio_now() + ms;
 		for (t = coio_sleeping.head; t != NULL && t->timeout && t->timeout < task->timeout; t = t->next);
 	} else {
 		task->timeout = 0;
@@ -190,7 +191,7 @@ coio_delay(int ms)
 	uvlong 		when;
 	when = coio_timeout(coio_current, ms);
 	coio_transfer();
-	return (coio_now() - when) / 1000000;
+	return (coio_now() - when);
 }
 
 void
@@ -252,21 +253,19 @@ coio_now()
 {
 #if defined(__APPLE__)
 	clock_serv_t cclock;
-	mach_timespec_t mts;
+	mach_timespec_t ts;
 
 	host_get_clock_service(mach_host_self(), SYSTEM_CLOCK, &cclock);
-	clock_get_time(cclock, &mts);
-	mach_port_deallocate(mach_task_self(), cclock);
-
-	return (uvlong) mts.tv_sec * 1000 * 1000 * 1000 + mts.tv_nsec;
+	clock_get_time(cclock, &ts);
+	mach_port_deallocate(mach_task_self(), cclock);	
 #else
 	struct timespec ts;
 
 	if (clock_gettime(CLOCK_MONOTONIC, &ts) < 0)
 		return -1;
-
-	return (uvlong) ts.tv_sec * 1000 * 1000 * 1000 + ts.tv_nsec;
 #endif
+
+	return (uvlong) ts.tv_sec * 1000 + round(ts.tv_nsec / 1000000.0);
 }
 
 void

coio.h

@@ -33,6 +33,64 @@ uvlong 		coio_now();
 int 		coio_delay(int ms);
 void 		coio_ready(CoioTask* task);
 
+/* channel communication from libtask */
+typedef struct CoioAlt CoioAlt;
+typedef struct Altarray Altarray;
+typedef struct CoioChannel CoioChannel;
+
+enum
+{
+	COIO_CHANEND,
+	COIO_CHANSND,
+	COIO_CHANRCV,
+	COIO_CHANNOP,
+	COIO_CHANNOBLK,
+};
+
+struct CoioAlt
+{
+	CoioChannel*	c;
+	void*		v;
+	unsigned int	op;
+	CoioTask*	task;
+	CoioAlt*		xalt;
+};
+
+struct Altarray
+{
+	CoioAlt**		a;
+	unsigned int	n;
+	unsigned int	m;
+};
+
+struct CoioChannel
+{
+	unsigned int	bufsize;
+	unsigned int	elemsize;
+	unsigned char*	buf;
+	unsigned int	nbuf;
+	unsigned int	off;
+	Altarray	asend;
+	Altarray	arecv;
+	char*		name;
+};
+
+int		coio_chanalt(CoioAlt* alts);
+CoioChannel*	coio_chancreate(int elemsize, int elemcnt);
+void		coio_chanfree(CoioChannel* c);
+int		coio_channbrecv(CoioChannel* c, void* v);
+void*		coio_channbrecvp(CoioChannel* c);
+unsigned long	coio_channbrecvul(CoioChannel* c);
+int		coio_channbsend(CoioChannel* c, void* v);
+int		coio_channbsendp(CoioChannel* c, void* v);
+int		coio_channbsendul(CoioChannel* c, unsigned long v);
+int		coio_chanrecv(CoioChannel* c, void* v);
+void*		coio_chanrecvp(CoioChannel* c);
+unsigned long	coio_chanrecvul(CoioChannel* c);
+int		coio_chansend(CoioChannel* c, void* v);
+int		coio_chansendp(CoioChannel* c, void* v);
+int		coio_chansendul(CoioChannel* c, unsigned long v);
+
 #ifdef __cplusplus
 }
 #endif

coio_glib.c

@@ -19,6 +19,7 @@
 #include "coioimpl.h"
 
 #include <glib.h>
+#include <gio/gio.h>
 
 struct coio_source {
 	GSource base;
@@ -47,7 +48,7 @@ static gboolean coio_source_prepare(GSource* source, gint* timeout_)
 		if (now >= t->timeout) {
 			ms = 0;
 		} else {
-			ms = (t->timeout - now) / 1000000;
+			ms = (t->timeout - now);
 		}
 	}
 
@@ -68,7 +69,10 @@ static gboolean coio_source_check(GSource* source)
 		}
 	}
 
-	return TRUE;
+	if (coio_ready_list.head)
+		return TRUE;
+
+	return FALSE;
 }
 
 static gboolean coio_source_dispatch(GSource* source, GSourceFunc callback, gpointer user_data)
@@ -77,10 +81,6 @@ static gboolean coio_source_dispatch(GSource* source, GSourceFunc callback, gpoi
 	CoioTask* last;
 	gboolean result = G_SOURCE_CONTINUE;
 
-	/* error condition */
-	if (!coio_ready_list.head && !coio_sleeping.head)
-		return G_SOURCE_REMOVE;
-
 	if (!coio_ready_list.head)
 		return G_SOURCE_CONTINUE;
 
@@ -99,6 +99,8 @@ static gboolean coio_source_dispatch(GSource* source, GSourceFunc callback, gpoi
 		}
 	} while (coio_current != last);
 
+	coio_current = NULL;
+
 	if (callback) {
 		result = callback(user_data);
 	}
@@ -132,3 +134,11 @@ gboolean coio_task_wakeup_helper(gpointer task)
 	coio_ready((CoioTask*)task);
 	return G_SOURCE_REMOVE;
 }
+
+void coio_gasyncresult_wakeup_helper(GObject *source_object, GAsyncResult *res, gpointer user_data)
+{
+	(void)source_object;
+	CoioChannel* ch = (CoioChannel*)user_data;
+
+	coio_channbsendp(ch, g_object_ref(res));
+}

coio_glib.h

@@ -18,8 +18,11 @@
 #define COIO_GLIB_H
 
 #include <glib.h>
+#include <gio/gio.h>
+#include "coio.h"
 
 GSource* coio_gsource_create();
 gboolean coio_task_wakeup_helper(gpointer task);
+void coio_gasyncresult_wakeup_helper(GObject *source_object, GAsyncResult *res, gpointer user_data);
 
 #endif

coro.h

@@ -305,7 +305,7 @@ void coro_stack_free (struct coro_stack *stack);
 #  define CORO_LOSER 1 /* you don't win with windoze */
 # elif (__linux || __OpenBSD__ || __APPLE__) && (__i386 || (__x86_64 && !__ILP32))
 #  define CORO_ASM 1
-# elif defined HAVE_UCONTEXT_H
+# elif defined HAVE_UCONTEXT_H || (__linux)
 #  define CORO_UCONTEXT 1
 # elif defined HAVE_SETJMP_H && defined HAVE_SIGALTSTACK
 #  define CORO_SJLJ 1

testdelay.c

@@ -30,7 +30,7 @@ main(int argc, char** argv)
 	(void) argc;
 	(void) argv;
 
-	coio_create(_t1, NULL, 0x8000);
+	coio_create("t1", _t1, NULL, 0x8000);
 
 	if (coio_main() < 0) {
 		printf("Deadlocked\n");

testyield.c

@@ -38,8 +38,8 @@ main(int argc, char** argv)
 	(void) argc;
 	(void) argv;
 
-	coio_create(_t1, NULL, 0x8000);
-	coio_create(_t2, NULL, 0x8000);
+	coio_create("t1", _t1, NULL, 0x8000);
+	coio_create("t2", _t2, NULL, 0x8000);
 
 	if (coio_main() < 0) {
 		printf("Deadlocked\n");