implementing the blocking channel communitation between routines

Makefile

@@ -1,4 +1,4 @@
-SRCS = ctx.c
+SRCS = chan.c ctx.c
 OBJS = $(SRCS:.c=.o)
 
 CFLAGS := -Wall -g

chan.c

@@ -0,0 +1,129 @@
+#include "hip-prv.h"
+
+#include <stdlib.h>
+#include <string.h>
+
+struct hip_chan_t
+{
+	hip_t* hip;
+
+	int closed;
+
+	hip_oow_t unit_size;
+	hip_oow_t buf_size;
+
+	hip_uint8_t* buf;
+	hip_uint8_t* endptr;
+	hip_uint8_t* readptr;
+	hip_uint8_t* writeptr;
+
+	hip_uctx_link_t sendq; // list of senders waiting on this channel
+	hip_uctx_link_t recvq; // list of receivers waiting on this channel
+};
+
+hip_chan_t* hip_chan_open(hip_t* hip, hip_oow_t buf_size)
+{
+	hip_chan_t* c;
+
+	c = (hip_chan_t*)malloc(HIP_SIZEOF(*c) + buf_size);
+	if (HIP_UNLIKELY(!c)) return HIP_NULL;
+
+	memset(c, 0, HIP_SIZEOF(*c));
+	c->hip = hip;
+	c->closed = 0;
+	c->buf_size = buf_size;
+	c->buf = (hip_uint8_t*)(c + 1);
+	c->endptr = 0;
+	c->readptr = 0;
+	c->writeptr = 0;
+
+	HIP_LIST_INIT(&c->recvq);
+	HIP_LIST_INIT(&c->sendq);
+
+	return c;
+}
+
+void hip_chan_close(hip_chan_t* c)
+{
+	free(c);
+}
+
+int hip_chan_send(hip_chan_t* c, const void* ptr, hip_oow_t len)
+{
+	hip_uctx_t* self;
+	int ret;
+
+	self = UCTX_FROM_LINK(c->hip->running);
+
+// TODO: IS BUFFER EMPTY
+
+	if (!HIP_LIST_IS_EMPTY(&c->recvq)) /* there is a receving routine */
+	{
+		hip_uctx_link_t* x;
+		hip_uctx_t* xx;
+
+		/* remove the receiving routine from the queue and make it runnable */
+		x = HIP_LIST_HEAD(&c->recvq);
+		HIP_LIST_UNCHAIN(x);
+		HIP_LIST_ADD_BACK(x, &c->hip->runnables);
+
+		xx = UCTX_FROM_LINK(x);
+
+		if (len > xx->chan_data_len) len = xx->chan_data_len;
+		memcpy(xx->chan_data_ptr, ptr, len);
+		*(xx->chan_ret_ptr) = len; /* manipulate the return value of the receiver */
+
+		hip_yield(c->hip); /* let other routines go first */
+		return len;
+	}
+
+
+	/* remember the data holder and places the current running routine in the send queue */
+	self->chan_data_ptr = (hip_uint8_t*)ptr;
+	self->chan_data_len = len;
+	self->chan_ret_ptr = &ret; /* remember the pointer to the return value holder to be set by the receiver */
+	HIP_LIST_ADD_BACK(c->hip->running, &c->sendq);
+
+	hip_switch(c->hip);
+	return ret;
+}
+
+int hip_chan_recv(hip_chan_t* c, void* ptr, hip_oow_t len)
+{
+	hip_uctx_t* self;
+	int ret;
+
+	self = UCTX_FROM_LINK(c->hip->running);
+
+	/* TODO: IS THERE DATA IN BUFFER ? */
+
+	if (!HIP_LIST_IS_EMPTY(&c->sendq)) /* there is a sending routine */
+	{
+		hip_uctx_link_t* x;
+		hip_uctx_t* xx;
+
+		/* remove the sending routine from the queue and make it runnable */
+		x = HIP_LIST_HEAD(&c->sendq);
+		HIP_LIST_UNCHAIN(x);
+		HIP_LIST_ADD_BACK(x, &c->hip->runnables);
+
+		xx = UCTX_FROM_LINK(x);
+
+		/* copy the data from the sending routine's buffer */
+		if (len > xx->chan_data_len) len = xx->chan_data_len; /* TODO: do something else instead of simple truncation */
+		memcpy(ptr, xx->chan_data_ptr, len);
+		*(xx->chan_data_ptr) = len; /* manipulate the return value of the sender */
+
+		hip_yield(c->hip); /* let other routines go first */
+		return len;
+	}
+
+	/* remember the data holder and place the calling routine in the receive queue */
+	self->chan_data_ptr = ptr;
+	self->chan_data_len = len;
+	self->chan_ret_ptr = &ret; /* remember the pointer to the return value holder to be set by the sender */
+	HIP_LIST_ADD_BACK(c->hip->running, &c->recvq);
+
+	hip_switch(c->hip); /* switch to the scheduler */
+	return ret;
+}

ctx.c

@@ -1,7 +1,5 @@
-#include "hip.h"
+#include "hip-prv.h"
 
-#include <signal.h>
-#include <time.h>
 #include <string.h>
 #include <stdlib.h>
 #include <stdio.h>
@@ -22,67 +20,6 @@
 #include <valgrind/valgrind.h>
 #endif
 
-#define CHAIN(_link, _prev, _next) do { \
-	hip_uctx_link_t* _p = _prev; \
-	hip_uctx_link_t* _n = _next; \
-	(_n)->prev = _link; \
-	(_link)->prev = _p; \
-	(_link)->next = _n; \
-	(_p)->next = _link; \
-} while(0)
-
-#define UNCHAIN(_link) do { \
-	(_link)->next->prev = (_link)->prev; \
-	(_link)->prev->next = (_link)->next; \
-} while(0)
-
-#define HEAD(ll) ((ll)->next)
-#define TAIL(ll) ((ll)->prev)
-
-#define ADD_FRONT(_link, ll) CHAIN(_link, (ll), HEAD(ll))
-#define ADD_BACK(_link, ll) CHAIN(_link, TAIL(ll), (ll))
-#define IS_EMPTY(ll) (HEAD(ll) == ll)
-
-#define UCTX_FROM_LINK(_link) HIP_CONTAINEROF(_link, hip_uctx_t, uctx)
-
-struct hip_uctx_t
-{
-	hip_t* hip;
-	ucontext_t uc;
-	int flags;
-	hip_ufun_t uf;
-	void* ctx;
-	hip_oow_t csi; /* number of context switches in */
-	hip_oow_t cso; /* number of context switches out */
-
-	hip_nsecdur_t wakeup_time;
-	int waiting_fd;
-	unsigned int stid;
-
-	hip_uctx_link_t uctx;
-	hip_uctx_link_t io_done;
-};
-
-struct hip_t
-{
-	int flags;
-
-	struct sigaction oldsa;
-	timer_t tmr_id;
-	int mux_id;
-
-	hip_uctx_link_t runnables;
-	hip_uctx_link_t terminated;
-	hip_uctx_link_t sleeping;
-	hip_uctx_link_t io_waiting;
-	hip_uctx_link_t io_done;
-
-	hip_uctx_link_t* running;
-
-	hip_uctx_t* uctx_sched;
-	hip_uctx_t* uctx_mux;
-	ucontext_t uc_main;
-};
 /* ---------------------------------------------------- */
 
 #if (MKCTX_NARGS <= 1)
@@ -117,7 +54,7 @@ printf ("invoke_uf XXXXXXXXXXXXXXXXXXXXX...%p\n", uctx);
 	else
 	{
 		/* switch to the scheduler */
-		UNCHAIN(&uctx->uctx);
+		HIP_LIST_UNCHAIN(&uctx->uctx);
 
 		if (uctx->flags & HIP_RTN_FLAG_AUTO_DESTROY)
 		{
@@ -126,7 +63,7 @@ printf (">>>>>>>>>> *********************** CLOSED TERMINATED\n");
 		}
 		else
 		{
-			ADD_BACK(&uctx->uctx, &hip->terminated);
+			HIP_LIST_ADD_BACK(&uctx->uctx, &hip->terminated);
 printf (">>>>>>>>>> ************************ ADDED TO TERMINATED\n");
 		}
 
@@ -144,11 +81,11 @@ hip_uctx_t* hip_uctx_open(hip_t* hip, hip_oow_t stack_size, int flags, hip_ufun_
 	hip_uctx_t* uc;
 	void* sp;
 
-	uc = (hip_uctx_t*)malloc(sizeof(*uc) + stack_size);
+	uc = (hip_uctx_t*)malloc(HIP_SIZEOF(*uc) + stack_size);
 	if (!uc) return HIP_NULL;
 	sp = (void*)(uc + 1);
 
-	memset(uc, 0, sizeof(*uc) + stack_size);
+	memset(uc, 0, HIP_SIZEOF(*uc) + stack_size);
 	uc->hip = hip;
 	uc->flags = flags;
 	uc->uf = uf;
@@ -228,16 +165,16 @@ static void schedule(hip_uctx_t* uctx, void *ctx)
 		hip_uctx_link_t* h;
 		hip_uctx_t* u;
 
-		if (IS_EMPTY(&hip->runnables))
+		if (HIP_LIST_IS_EMPTY(&hip->runnables))
 		{
 /* TODO: check if there are sleeping ... */
 //	printf ("<scheduler> NO TASK\n");
 			swapcontext(&hip->uctx_sched->uc, &hip->uctx_mux->uc);
-			if (IS_EMPTY(&hip->runnables)) break; /* no task? */
+			if (HIP_LIST_IS_EMPTY(&hip->runnables)) break; /* no task? */
 		}
 
-		h = HEAD(&hip->runnables);
+		h = HIP_LIST_HEAD(&hip->runnables);
-		UNCHAIN(h);
+		HIP_LIST_UNCHAIN(h);
 		hip->running = h;
 		u = UCTX_FROM_LINK(h);
 
@@ -259,8 +196,8 @@ static void multiplex(hip_uctx_t* uctx, void *ctx)
 	hip = uctx->hip;
 	while (1)
 	{
-//printf ("sleeping empty[%d] io_waiting empty[%d]\n", IS_EMPTY(&hip->sleeping), IS_EMPTY(&hip->io_waiting));
+//printf ("sleeping empty[%d] io_waiting empty[%d]\n", HIP_LIST_IS_EMPTY(&hip->sleeping), HIP_LIST_IS_EMPTY(&hip->io_waiting));
-		if (IS_EMPTY(&hip->sleeping) && IS_EMPTY(&hip->io_waiting))
+		if (HIP_LIST_IS_EMPTY(&hip->sleeping) && HIP_LIST_IS_EMPTY(&hip->io_waiting))
 		{
 // STILL NEED TO HANDLE IO FILE DESCRIPTORS...
 			/* go back to the scheduler */
@@ -276,7 +213,7 @@ static void multiplex(hip_uctx_t* uctx, void *ctx)
 			int n;
 			hip_uctx_t* u;
 
-			l = HEAD(&hip->sleeping);
+			l = HIP_LIST_HEAD(&hip->sleeping);
 			u = UCTX_FROM_LINK(l);
 			now = monotime();
 			wait = now >= u->wakeup_time? 0: u->wakeup_time - now;
@@ -286,11 +223,11 @@ static void multiplex(hip_uctx_t* uctx, void *ctx)
  * 	epoll_ctl(DELETE here
  */
 			// lazy deletion of file descriptors
-			while (!IS_EMPTY(&hip->io_done))
+			while (!HIP_LIST_IS_EMPTY(&hip->io_done))
 			{
-				l = HEAD(&hip->io_done);
+				l = HIP_LIST_HEAD(&hip->io_done);
 				u = UCTX_FROM_LINK(l);
-				UNCHAIN(l);
+				HIP_LIST_UNCHAIN(l);
 				epoll_ctl(hip->mux_id, EPOLL_CTL_DEL, u->waiting_fd, HIP_NULL);
 			}
 
@@ -324,8 +261,8 @@ static void multiplex(hip_uctx_t* uctx, void *ctx)
 					l = ee[--n].data.ptr;
 					u = UCTX_FROM_LINK(l);
 
-					UNCHAIN(l); /* unchain it from io_waiting */
+					HIP_LIST_UNCHAIN(l); /* unchain it from io_waiting */
-					ADD_BACK(l, &hip->runnables);
+					HIP_LIST_ADD_BACK(l, &hip->runnables);
 
 					if (hip->flags & HIP_FLAG_LAZY)
 					{
@@ -334,12 +271,12 @@ static void multiplex(hip_uctx_t* uctx, void *ctx)
 						 * add this to the io_done list for clean-up
 						 * before waiting and keep waiting_fd untouched
 						 * for adjustment later. */
-						ADD_BACK(&u->io_done, &hip->io_done);
+						HIP_LIST_ADD_BACK(&u->io_done, &hip->io_done);
 					}
 					else
 					{
 
-//printf ("PWAIT link MULTIPLEX DEL %p fd[%d] io_waiting empty[%d]\n", l, u->waiting_fd, IS_EMPTY(&hip->io_waiting));
+//printf ("PWAIT link MULTIPLEX DEL %p fd[%d] io_waiting empty[%d]\n", l, u->waiting_fd, HIP_LIST_IS_EMPTY(&hip->io_waiting));
 
 						epoll_ctl(hip->mux_id, EPOLL_CTL_DEL, u->waiting_fd, HIP_NULL);
 						u->waiting_fd = HIP_INVALID_FD;
@@ -350,18 +287,18 @@ static void multiplex(hip_uctx_t* uctx, void *ctx)
 			}
 
 			now = monotime();
-			while (!IS_EMPTY(&hip->sleeping))
+			while (!HIP_LIST_IS_EMPTY(&hip->sleeping))
 			{
-				l = HEAD(&hip->sleeping);
+				l = HIP_LIST_HEAD(&hip->sleeping);
 				u = UCTX_FROM_LINK(l);
 				if (now < u->wakeup_time) break;
 
-				UNCHAIN(l);
+				HIP_LIST_UNCHAIN(l);
-				ADD_BACK(l, &hip->runnables);
+				HIP_LIST_ADD_BACK(l, &hip->runnables);
 			}
 
 			/* go back to the scheduler */
-			if (!IS_EMPTY(&hip->runnables))
+			if (!HIP_LIST_IS_EMPTY(&hip->runnables))
 			{
 //printf ("BACK TO SCHEDULER \n");
 				swapcontext(&hip->uctx_mux->uc, &hip->uctx_sched->uc);
@@ -379,40 +316,36 @@ hip_t* hip_open(int flags)
 	sigset_t sm, oldsm;
 	timer_t tid;
 
-	hip = (hip_t*)malloc(sizeof(*hip));
+	hip = (hip_t*)malloc(HIP_SIZEOF(*hip));
 	if (!hip) return HIP_NULL;
 
-	memset(hip, 0, sizeof(*hip));
+	memset(hip, 0, HIP_SIZEOF(*hip));
 	hip->flags = flags;
 
 	/* initialize to an empty list by making each pointer point to itself.*/
-	hip->runnables.next = &hip->runnables;
+	HIP_LIST_INIT(&hip->runnables);
-	hip->runnables.prev = &hip->runnables;
+	HIP_LIST_INIT(&hip->terminated);
-	hip->terminated.next = &hip->terminated;
+	HIP_LIST_INIT(&hip->sleeping);
-	hip->terminated.prev = &hip->terminated;
+	HIP_LIST_INIT(&hip->io_waiting);
-	hip->sleeping.next = &hip->sleeping;
+	HIP_LIST_INIT(&hip->io_done);
-	hip->sleeping.prev = &hip->sleeping;
+	HIP_LIST_INIT(&hip->suspended);
-	hip->io_waiting.next = &hip->io_waiting;
-	hip->io_waiting.prev = &hip->io_waiting;
-	hip->io_done.next = &hip->io_done;
-	hip->io_done.prev = &hip->io_done;
 
 	sigemptyset(&sm);
 	sigaddset(&sm, SIGRTMIN);
 	sigprocmask(SIG_BLOCK, &sm, &oldsm);
 
-	memset(&sa, 0, sizeof(sa));
+	memset(&sa, 0, HIP_SIZEOF(sa));
 	sa.sa_flags = SA_SIGINFO /*| SA_RESTART*/;
 	sa.sa_sigaction = on_timer_signal;
 	sigemptyset(&sa.sa_mask);
 	sigaction(SIGRTMIN, &sa, &hip->oldsa);
 
 	ss.ss_sp = signal_stack; /* TODO: allocate this dynamically */
-	ss.ss_size = sizeof(signal_stack);
+	ss.ss_size = HIP_SIZEOF(signal_stack);
 	ss.ss_flags = 0;
 	sigaltstack(&ss, HIP_NULL);
 
-	memset(&se, 0, sizeof(se));
+	memset(&se, 0, HIP_SIZEOF(se));
 	se.sigev_notify = SIGEV_SIGNAL;
 	se.sigev_signo = SIGRTMIN;
 	/*se.sigev_value.sival_ptr = hip;  TOOD: possible to use sigev_notify_function and use this field?
@@ -449,18 +382,52 @@ void hip_close(hip_t* hip)
 	close(hip->mux_id);
 	timer_delete(hip->tmr_id);
 
-	while (!IS_EMPTY(&hip->terminated))
+	while (!HIP_LIST_IS_EMPTY(&hip->terminated))
 	{
-		l = HEAD(&hip->terminated);
+		l = HIP_LIST_HEAD(&hip->terminated);
-		UNCHAIN(l);
+		HIP_LIST_UNCHAIN(l);
 		uctx = UCTX_FROM_LINK(l);
 		hip_uctx_close(uctx);
 	}
 
-	while (!IS_EMPTY(&hip->runnables))
+	while (!HIP_LIST_IS_EMPTY(&hip->runnables))
 	{
-		l = HEAD(&hip->runnables);
+		l = HIP_LIST_HEAD(&hip->runnables);
-		UNCHAIN(l);
+		HIP_LIST_UNCHAIN(l);
+		uctx = UCTX_FROM_LINK(l);
+		hip_uctx_close(uctx);
+	}
+
+/* TODO: minimize repeated code in the same pattern */
+/* TODO: or is it better to maintain a list of all contexts? */
+	while (!HIP_LIST_IS_EMPTY(&hip->sleeping))
+	{
+		l = HIP_LIST_HEAD(&hip->sleeping);
+		HIP_LIST_UNCHAIN(l);
+		uctx = UCTX_FROM_LINK(l);
+		hip_uctx_close(uctx);
+	}
+
+	while (!HIP_LIST_IS_EMPTY(&hip->io_waiting))
+	{
+		l = HIP_LIST_HEAD(&hip->io_waiting);
+		HIP_LIST_UNCHAIN(l);
+		uctx = UCTX_FROM_LINK(l);
+		hip_uctx_close(uctx);
+	}
+
+	while (!HIP_LIST_IS_EMPTY(&hip->io_done))
+	{
+		l = HIP_LIST_HEAD(&hip->io_done);
+		HIP_LIST_UNCHAIN(l);
+		uctx = UCTX_FROM_LINK(l);
+		hip_uctx_close(uctx);
+	}
+
+	while (!HIP_LIST_IS_EMPTY(&hip->suspended))
+	{
+		l = HIP_LIST_HEAD(&hip->suspended);
+		HIP_LIST_UNCHAIN(l);
 		uctx = UCTX_FROM_LINK(l);
 		hip_uctx_close(uctx);
 	}
@@ -478,7 +445,7 @@ hip_uctx_t* hip_newrtn(hip_t* hip, int flags, hip_ufun_t uf, void* ctx)
 	if (!uctx) return HIP_NULL;
 
 	/* append to the list */
-	ADD_BACK(&uctx->uctx, &hip->runnables);
+	HIP_LIST_ADD_BACK(&uctx->uctx, &hip->runnables);
 
 	return uctx;
 }
@@ -501,7 +468,7 @@ int hip_schedule(hip_t* hip, int preempt)
 	{
 		/* start the timer tick */
 		struct itimerspec its;
-		memset(&its, 0, sizeof(its));
+		memset(&its, 0, HIP_SIZEOF(its));
 		its.it_interval.tv_sec = 0;
 		//its.it_interval.tv_nsec = 10000000; /* 10 milliseconds */
 		its.it_interval.tv_nsec = 100000000; /* 100 milliseconds */
@@ -516,7 +483,7 @@ int hip_schedule(hip_t* hip, int preempt)
 	{
 		/* stop the timer tick */
 		struct itimerspec its;
-		memset(&its, 0, sizeof(its));
+		memset(&its, 0, HIP_SIZEOF(its));
 		its.it_interval.tv_sec = 0;
 		its.it_interval.tv_nsec = 0;
 		its.it_value = its.it_interval;
@@ -528,17 +495,17 @@ int hip_schedule(hip_t* hip, int preempt)
 
 void hip_yield(hip_t* hip)
 {
-	hip_uctx_t* r;
+	hip_uctx_t* rr;
 	assert (hip->running != HIP_NULL);
 
 	/* switch from the running context to the scheduler context */
-	ADD_BACK(hip->running, &hip->runnables);
+	HIP_LIST_ADD_BACK(hip->running, &hip->runnables);
-	r = UCTX_FROM_LINK(hip->running);
+	rr = UCTX_FROM_LINK(hip->running);
 
 /* TODO: extract these lines to a macro or something */
-r->cso++;
+rr->cso++;
 hip->uctx_sched->csi++;
-	swapcontext(&r->uc, &hip->uctx_sched->uc);
+	swapcontext(&rr->uc, &hip->uctx_sched->uc);
 }
 
 void hip_sleep(hip_t* hip, hip_nsecdur_t nsecs)
@@ -550,28 +517,43 @@ void hip_sleep(hip_t* hip, hip_nsecdur_t nsecs)
 	rr->wakeup_time = monotime() + nsecs;
 
 	/* TODO: switch to HEAP. for now simply linear search to keep this list sorted. */
-	if (!IS_EMPTY(&hip->sleeping))
+	if (!HIP_LIST_IS_EMPTY(&hip->sleeping))
 	{
 		hip_uctx_link_t* l;
 		hip_uctx_t* r;
-		for (l = HEAD(&hip->sleeping); l != &hip->sleeping; l = l->next)
+		for (l = HIP_LIST_HEAD(&hip->sleeping); l != &hip->sleeping; l = l->next)
 		{
 			r = UCTX_FROM_LINK(l);
 			if (rr->wakeup_time <= r->wakeup_time)
 			{
-				CHAIN(hip->running, l->prev, l);
+				HIP_LIST_CHAIN(hip->running, l->prev, l);
 				goto do_sched;
 			}
 		}
 	}
 	/* place the running rountine at the end of the sleeping routines list */
-	ADD_BACK(hip->running, &hip->sleeping);
+	HIP_LIST_ADD_BACK(hip->running, &hip->sleeping);
 
 do_sched:
 	hip->running = HIP_NULL;
 	swapcontext(&rr->uc, &hip->uctx_sched->uc); /* switch to the scheduler */
 }
 
+void hip_suspend(hip_t* hip)
+{
+	HIP_LIST_ADD_BACK(hip->running, &hip->suspended);
+	hip_switch(hip);
+}
+
+void hip_switch(hip_t* hip)
+{
+	hip_uctx_t* rr;
+	/* the caller must deal with hip->running before calling this function */
+	rr = UCTX_FROM_LINK(hip->running);
+	hip->running = HIP_NULL;
+	swapcontext(&rr->uc, &hip->uctx_sched->uc); /* switch to the scheduler */
+}
+
 void hip_awaitio(hip_t* hip, int fd, int flags)
 {
 	struct epoll_event ev;
@@ -582,12 +564,12 @@ void hip_awaitio(hip_t* hip, int fd, int flags)
 
 	if (rr->waiting_fd != HIP_INVALID_FD)
 	{
-		UNCHAIN(&rr->io_done);
+		HIP_LIST_UNCHAIN(&rr->io_done);
 
 		if (rr->waiting_fd == fd)
 		{
 			/* update */
-			memset(&ev, 0, sizeof(ev));
+			memset(&ev, 0, HIP_SIZEOF(ev));
 			if (flags & HIP_IO_READ) ev.events |= EPOLLIN;
 			if (flags & HIP_IO_WRITE) ev.events |= EPOLLOUT;
 
@@ -609,7 +591,7 @@ void hip_awaitio(hip_t* hip, int fd, int flags)
 	else
 	{
 	add:
-		memset(&ev, 0, sizeof(ev));
+		memset(&ev, 0, HIP_SIZEOF(ev));
 		if (flags & HIP_IO_READ) ev.events |= EPOLLIN;
 		if (flags & HIP_IO_WRITE) ev.events |= EPOLLOUT;
 		ev.data.ptr = hip->running; /* store running context link */
@@ -618,11 +600,11 @@ void hip_awaitio(hip_t* hip, int fd, int flags)
 		/* TODO: error handling - probably panic? */
 	}
 
-	ADD_BACK(hip->running, &hip->io_waiting);
+	HIP_LIST_ADD_BACK(hip->running, &hip->io_waiting);
 	hip->running = HIP_NULL;
 	rr->waiting_fd = fd; /* remember the file descriptor being waited on */
 
-	swapcontext(&rr->uc, &hip->uctx_sched->uc);
+	swapcontext(&rr->uc, &hip->uctx_sched->uc); /* switch to the scheduler */
 }
 
 /* ---------------------------------------------------- */
@@ -687,7 +669,7 @@ static int socket_connect(hip_t* hip, int proto, const struct sockaddr* addr, so
 
 		hip_awaitio(hip, fd, HIP_IO_WRITE);
 
-		sl = sizeof(ss);
+		sl = HIP_SIZEOF(ss);
 		if (getsockopt(fd, SOL_SOCKET, SO_ERROR, (void*)&ss, &sl) <= -1)
 		{
 			close(fd);
@@ -728,7 +710,12 @@ static hip_ooi_t socket_write(hip_t* hip, int fd, const void* buf, hip_oow_t len
 
 static void uf1(hip_uctx_t* uc, void* ctx)
 {
-int i;
+	hip_chan_t* chan;
+	int i;
+	char buf[32];
+
+	chan = (hip_chan_t*)ctx;
+
 for (i =0; i < 5; i++)
 {
 printf (">> UF1 TOP [%d]\n", i);
@@ -738,11 +725,16 @@ printf (">> UF1 SEC [%d]\n", i);
 	//hip_yield(uc->hip);
 hip_sleep(uc->hip, 1000000000);
 }
+
+	printf (">> UF1 ALMOST DONE. WAITING FOR UF3\n");
+	i = hip_chan_recv(chan, buf, sizeof(buf));
+	printf (">> UF1 RECEIVED [%.*s]\n", (int)i, buf);
 }
 
 static void uf2(hip_uctx_t* uc, void* ctx)
 {
 	int i;
+
 	for (i =0; i < 10; i++)
 	{
 printf (">> UF2 TOP [%d]\n", i);
@@ -757,16 +749,18 @@ hip_sleep(uc->hip, 500000000);
 static void uf3(hip_uctx_t* uc, void* ctx)
 {
 	hip_t* hip;
+	hip_chan_t* chan;
 	struct sockaddr_in sa;
 	int i, fd;
 
 	hip = uc->hip;
+	chan = (hip_chan_t*)ctx;
 
-	memset(&sa, 0, sizeof(sa));
+	memset(&sa, 0, HIP_SIZEOF(sa));
 	sa.sin_family = AF_INET;
 	sa.sin_addr.s_addr = inet_addr("142.250.206.196");
 	sa.sin_port = htons(80);
-	fd = socket_connect(hip, SOCK_STREAM, (struct sockaddr*)&sa, sizeof(sa));
+	fd = socket_connect(hip, SOCK_STREAM, (struct sockaddr*)&sa, HIP_SIZEOF(sa));
 printf ("*********CONNECTED>>>>>>>>>>>>>>>\n");
 	{
 		int n;
@@ -779,10 +773,10 @@ printf ("*********CONNECTED>>>>>>>>>>>>>>>\n");
 		//const char x[] = "GET / HTTP/1.1\r\nHost: www.google.com\r\n\r\n";
 		const char x[] = "GET / HTTP/1.1\r\nHost: www.google.com\r\nConnection: close\r\n\r\n";
 
-		socket_write(hip, fd, x, sizeof(x) - 1);
+		socket_write(hip, fd, x, HIP_SIZEOF(x) - 1);
 		do
 		{
-			n = socket_read(hip, fd, buf, sizeof(buf));
+			n = socket_read(hip, fd, buf, HIP_SIZEOF(buf));
 			if (n <= 0) break;
 			printf("GOT [%.*s]\n", n, buf);
 		}
@@ -802,22 +796,26 @@ hip_sleep(uc->hip, 1000000000);
 		if (i == 8) hip_newrtn(uc->hip, HIP_RTN_FLAG_AUTO_DESTROY, uf2, HIP_NULL);
 	}
 printf (">> UF3 DONE DONE\n");
+	hip_chan_send(chan, "hello", 5);
 }
 
 int main()
 {
 	hip_t* hip;
+	hip_chan_t* chan;
 
 	hip = hip_open(HIP_FLAG_LAZY);
+	chan = hip_chan_open(hip, 100);
 
-	hip_newrtn(hip, HIP_RTN_FLAG_AUTO_DESTROY, uf1, HIP_NULL);
+	hip_newrtn(hip, HIP_RTN_FLAG_AUTO_DESTROY, uf1, chan);
 	hip_newrtn(hip, HIP_RTN_FLAG_AUTO_DESTROY, uf2, HIP_NULL);
-	hip_newrtn(hip, HIP_RTN_FLAG_AUTO_DESTROY, uf3, HIP_NULL);
+	hip_newrtn(hip, HIP_RTN_FLAG_AUTO_DESTROY, uf3, chan);
 
 	/* jump to the scheduler */
 	hip_schedule(hip, 0);
 printf("XXXXXXXXXXXXXXXXX ABOUT TO CLOSE ********************\n");
 
+	hip_chan_close(chan);
 /* TODO: close all uctxes ? */
 	hip_close(hip);
 

hip-prv.h

@@ -0,0 +1,56 @@
+#ifndef _HIP_PRV_H_
+#define _HIP_PRV_H_
+
+#include "hip.h"
+#include <ucontext.h>
+#include <signal.h>
+#include <time.h>
+
+#define UCTX_FROM_LINK(_link) HIP_CONTAINEROF(_link, hip_uctx_t, uctx)
+
+struct hip_uctx_t
+{
+	hip_t* hip;
+/* TODO: add a small name field for debugging and identification */
+	ucontext_t uc;
+	int flags;
+	hip_ufun_t uf;
+	void* ctx;
+	hip_oow_t csi; /* number of context switches in */
+	hip_oow_t cso; /* number of context switches out */
+
+	hip_nsecdur_t wakeup_time;
+	int waiting_fd;
+	unsigned int stid;
+
+	hip_uint8_t* chan_data_ptr;
+	hip_oow_t chan_data_len;
+	int* chan_ret_ptr;
+
+	hip_uctx_link_t uctx;
+	hip_uctx_link_t io_done;
+};
+
+struct hip_t
+{
+	int flags;
+
+	struct sigaction oldsa;
+	timer_t tmr_id;
+	int mux_id;
+
+	hip_uctx_link_t runnables;
+	hip_uctx_link_t terminated;
+	hip_uctx_link_t sleeping;
+	hip_uctx_link_t io_waiting;
+	hip_uctx_link_t io_done;
+	hip_uctx_link_t suspended;
+
+	hip_uctx_link_t* running;
+
+	hip_uctx_t* uctx_sched;
+	hip_uctx_t* uctx_mux;
+	ucontext_t uc_main;
+};
+
+#endif

hip.h

@@ -3,6 +3,12 @@
 
 #include <ucontext.h>
 
+/* TODO: define these properly */
+#define HIP_LIKELY(x) x
+#define HIP_UNLIKELY(x) x
+
+#define HIP_SIZEOF(x) sizeof(x)
+
 typedef unsigned char hip_uint8_t;
 typedef unsigned short hip_uint16_t;
 typedef unsigned int hip_uint32_t;
@@ -33,6 +39,7 @@ typedef struct hip_t hip_t;
 typedef struct hip_uctx_t hip_uctx_t;
 typedef struct hip_uctx_link_t hip_uctx_link_t;
 typedef void (*hip_ufun_t) (hip_uctx_t* uc, void* ctx);
+typedef struct hip_chan_t hip_chan_t;
 
 enum hip_io_flag_t
 {
@@ -62,6 +69,32 @@ enum hip_rtn_flag_t
 };
 typedef enum hip_rtn_flag_t hip_rtn_flag_t;
 
+#define HIP_LIST_INIT(_link) do { \
+	(_link)->prev = _link; \
+	(_link)->next = _link; \
+} while(0)
+
+#define HIP_LIST_CHAIN(_link, _prev, _next) do { \
+	hip_uctx_link_t* _p = _prev; \
+	hip_uctx_link_t* _n = _next; \
+	(_n)->prev = _link; \
+	(_link)->prev = _p; \
+	(_link)->next = _n; \
+	(_p)->next = _link; \
+} while(0)
+
+#define HIP_LIST_UNCHAIN(_link) do { \
+	(_link)->next->prev = (_link)->prev; \
+	(_link)->prev->next = (_link)->next; \
+} while(0)
+
+#define HIP_LIST_HEAD(ll) ((ll)->next)
+#define HIP_LIST_TAIL(ll) ((ll)->prev)
+
+#define HIP_LIST_ADD_FRONT(_link, ll) HIP_LIST_CHAIN(_link, (ll), HIP_LIST_HEAD(ll))
+#define HIP_LIST_ADD_BACK(_link, ll) HIP_LIST_CHAIN(_link, HIP_LIST_TAIL(ll), (ll))
+#define HIP_LIST_IS_EMPTY(ll) (HIP_LIST_HEAD(ll) == ll)
+
 #if defined(__cplusplus)
 extern "C" {
 #endif
@@ -75,6 +108,21 @@ hip_uctx_t* hip_newrtn(hip_t* hip, int flags, hip_ufun_t uf, void* ctx);
 hip_uctx_t* hip_uctx_open(hip_t* hip, hip_oow_t stack_size, int flags, hip_ufun_t uf, void* ctx);
 void hip_uctx_close(hip_uctx_t* uctx);
 
+
+
+
+
+void hip_yield(hip_t* hip);
+void hip_sleep(hip_t* hip, hip_nsecdur_t nsecs);
+void hip_switch(hip_t* hip);
+void hip_suspend(hip_t* hip);
+
+
+hip_chan_t* hip_chan_open(hip_t* hip, hip_oow_t buf_size);
+void hip_chan_close(hip_chan_t* c);
+int hip_chan_send(hip_chan_t* c, const void* ptr, hip_oow_t len);
+int hip_chan_recv(hip_chan_t* c, void* ptr, hip_oow_t len);
+
 #if defined(__cplusplus)
 }
 #endif