Initial attempt at queue implementation

Makefile

@@ -1,5 +1,5 @@
 TARGET = cmsg
-OBJECTS = main.o harness.o net.o util.o relay.o hashtable.o dataq.o sexp.o sexpio.o node.o
+OBJECTS = main.o harness.o net.o util.o relay.o hashtable.o dataq.o sexp.o sexpio.o node.o queue.o
 
 CFLAGS = -D_XOPEN_SOURCE=600 -Wall -O0 -g
 #CFLAGS = -D_XOPEN_SOURCE=600 -Wall -O3

main.c

@@ -19,6 +19,7 @@ typedef unsigned char u_char;
 #include "sexp.h"
 #include "hashtable.h"
 #include "node.h"
+#include "queue.h"
 
 #define WANT_CONSOLE_LISTENER 1
 
@@ -99,6 +100,7 @@ int main(int argc, char *argv[]) {
   info("Using libevent version %s\n", event_get_version());
   init_node();
   init_factory();
+  init_queue();
 #if WANT_CONSOLE_LISTENER
   spawn(console_listener, NULL);
 #endif

queue.c

@@ -0,0 +1,201 @@
+/* Copyright (C) 2010 Tony Garnock-Jones. All rights reserved. */
+
+#include <stdlib.h>
+#include <string.h>
+#include <stdio.h>
+#include <errno.h>
+#include <stdint.h>
+#include <stddef.h>
+
+#include <assert.h>
+
+/* OSSP UUID */
+#include <uuid.h>
+
+#include "cmsg_private.h"
+#include "harness.h"
+#include "ref.h"
+#include "sexp.h"
+#include "hashtable.h"
+#include "node.h"
+#include "queue.h"
+#include "dataq.h"
+
+typedef struct queue_node_t_ {
+  node_t node;
+  sexp_t *backlog_q;
+  queue_t waiter_q;
+  hashtable_t subscriptions;
+  Process *shovel;
+  int shovel_awake;
+} queue_node_t;
+
+typedef struct subscription_t_ {
+  sexp_t *uuid;
+  sexp_t *sink;
+  sexp_t *name;
+  struct subscription_t_ *link;
+} subscription_t;
+
+static void free_subscription(subscription_t *sub) {
+  DECREF(sub->uuid, sexp_destructor);
+  DECREF(sub->sink, sexp_destructor);
+  DECREF(sub->name, sexp_destructor);
+  free(sub);
+}
+
+static node_t *queue_construct(node_class_t *nc, sexp_t *args) {
+  queue_node_t *q = calloc(1, sizeof(*q));
+  q->backlog_q = INCREF(sexp_new_queue());
+  q->waiter_q = EMPTY_QUEUE(subscription_t, link);
+  init_hashtable(&q->subscriptions, 5, NULL, NULL);
+  q->shovel = NULL;
+  q->shovel_awake = 0;
+  return (node_t *) q;
+}
+
+static void queue_destructor(node_t *n) {
+  queue_node_t *q = (queue_node_t *) n;
+  DECREF(q->backlog_q, sexp_destructor);
+  /* q->waiter_q will be automatically destroyed as part of the destruction of q->subscriptions */
+  destroy_hashtable(&q->subscriptions);
+  /* TODO: um, take down the shovel too! */
+  free(q);
+}
+
+static int send_to_waiter(subscription_t *sub, sexp_t *body) {
+  return post_node(sexp_data(sub->sink), sexp_data(sub->name), body);
+}
+
+static void shoveller(void *qv) {
+  queue_node_t *q = (queue_node_t *) qv;
+  sexp_t *body = NULL; /* held */
+  queue_t examined;
+  subscription_t *sub = NULL;
+
+ check_for_work:
+
+  if (!sexp_queue_emptyp(q->backlog_q)) {
+    goto wait_and_shovel;
+  }
+
+  body = INCREF(sexp_dequeue(q->backlog_q)); /* held */
+  examined = EMPTY_QUEUE(subscription_t, link);
+
+ find_valid_waiter:
+  if (q->waiter_q.count == 0) {
+    sexp_queue_pushback(q->backlog_q, body);
+    DECREF(body, sexp_destructor);
+    q->waiter_q = examined;
+    goto wait_and_shovel;
+  }
+
+  sub = dequeue(&q->waiter_q);
+
+  if ((sub->uuid == NULL) /* It has been unsubscribed. */
+      || !send_to_waiter(sub, body)) { /* Destination no longer exists. */
+    free_subscription(sub);
+    goto find_valid_waiter;
+  }
+
+  DECREF(body, sexp_destructor);
+  queue_append(&q->waiter_q, &examined);
+  enqueue(&q->waiter_q, sub);
+  goto check_for_work;
+
+ wait_and_shovel:
+  q->shovel_awake = 0;
+  suspend();
+  goto check_for_work;
+}
+
+static void throck_shovel(queue_node_t *q) {
+  if (!q->shovel_awake) {
+    if (!q->shovel) {
+      q->shovel = spawn(shoveller, q);
+    } else {
+      resume(q->shovel);
+    }
+    q->shovel_awake = 1;
+  }
+}
+
+static void queue_handle_message(node_t *n, sexp_t *m) {
+  queue_node_t *q = (queue_node_t *) n;
+
+  size_t msglen = sexp_length(m);
+  sexp_t *args;
+  cmsg_bytes_t selector;
+
+  if (msglen == 0 || !sexp_stringp(sexp_head(m))) {
+    warn("Invalid message in factory\n");
+    return;
+  }
+
+  selector = sexp_data(sexp_head(m));
+  args = sexp_tail(m);
+
+  if ((msglen == 4) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("post"))) {
+    sexp_enqueue(q->backlog_q, sexp_listref(args, 1));
+    throck_shovel(q);
+  } else if ((msglen == 6) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("subscribe"))) {
+    unsigned char uuid[UUID_LEN_BIN];
+    if (gen_uuid(uuid) != 0) {
+      warn("Could not generate UUID\n");
+    } else {
+      sexp_t *reply_sink = sexp_listref(args, 3);
+      sexp_t *reply_name = sexp_listref(args, 4);
+      subscription_t *sub = malloc(sizeof(*sub));
+      sub->uuid = INCREF(sexp_bytes(CMSG_BYTES(sizeof(uuid), uuid)));
+      sub->sink = sexp_listref(args, 1);
+      sub->name = sexp_listref(args, 2);
+      sub->link = NULL;
+      if (!sexp_stringp(sub->sink) || !sexp_stringp(sub->name)
+	  || !sexp_stringp(reply_sink) || !sexp_stringp(reply_name)) {
+	DECREF(sub->uuid, sexp_destructor);
+	warn("Bad sink/name/reply_sink/replay_name in subscribe");
+      } else {
+	INCREF(sub->sink);
+	INCREF(sub->name);
+	hashtable_put(&q->subscriptions, sexp_data(sub->uuid), sub);
+	enqueue(&q->waiter_q, sub);
+	throck_shovel(q);
+	{
+	  sexp_t *subok = sexp_cons(sexp_bytes(cmsg_cstring_bytes("subscribe-ok")),
+				    sexp_cons(sub->uuid, NULL));
+	  INCREF(subok);
+	  post_node(sexp_data(reply_sink), sexp_data(reply_name), subok);
+	  DECREF(subok, sexp_destructor);
+	}
+      }
+    }
+  } else if ((msglen == 2) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("unsubscribe"))) {
+    if (sexp_stringp(sexp_head(args))) {
+      cmsg_bytes_t uuid = sexp_data(sexp_head(args));
+      subscription_t *sub;
+      if (hashtable_get(&q->subscriptions, uuid, (void **) &sub)) {
+	/* TODO: clean up more eagerly perhaps? */
+	hashtable_erase(&q->subscriptions, uuid);
+	DECREF(sub->uuid, sexp_destructor);
+	sub->uuid = NULL;
+      }
+    } else {
+      warn("Invalid unsubscription\n");
+    }
+  } else {
+    warn("Message not understood in queue; selector <<%.*s>>, length %u\n",
+	 selector.len, selector.bytes,
+	 msglen);
+  }
+}
+
+static node_class_t queue_class = {
+  .name = "queue",
+  .construct = queue_construct,
+  .destroy = queue_destructor,
+  .handle_message = queue_handle_message
+};
+
+void init_queue(void) {
+  register_node_class(&queue_class);
+}

queue.h

@@ -0,0 +1,6 @@
+#ifndef cmsg_queue_h
+#define cmsg_queue_h
+
+extern void init_queue(void);
+
+#endif