From fd85e4d24306ef6bc4e519ce766d0e66a4a27ce8 Mon Sep 17 00:00:00 2001 From: Tony Garnock-Jones Date: Sat, 1 Jan 2011 21:14:37 -0500 Subject: [PATCH] Initial attempt at queue implementation --- Makefile | 2 +- main.c | 2 + queue.c | 201 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ queue.h | 6 ++ 4 files changed, 210 insertions(+), 1 deletion(-) create mode 100644 queue.c create mode 100644 queue.h diff --git a/Makefile b/Makefile index af11407..a3483a6 100644 --- a/Makefile +++ b/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 diff --git a/main.c b/main.c index ec9c10a..217719d 100644 --- a/main.c +++ b/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 diff --git a/queue.c b/queue.c new file mode 100644 index 0000000..452e8e6 --- /dev/null +++ b/queue.c @@ -0,0 +1,201 @@ +/* Copyright (C) 2010 Tony Garnock-Jones. All rights reserved. */ + +#include +#include +#include +#include +#include +#include + +#include + +/* OSSP UUID */ +#include + +#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); +} diff --git a/queue.h b/queue.h new file mode 100644 index 0000000..f7a0e65 --- /dev/null +++ b/queue.h @@ -0,0 +1,6 @@ +#ifndef cmsg_queue_h +#define cmsg_queue_h + +extern void init_queue(void); + +#endif