From e0ca281c324d8d9f4a81a1ec1d6c2f7ae0cdbe67 Mon Sep 17 00:00:00 2001
From: Tony Garnock-Jones <tonygarnockjones@gmail.com>
Date: Mon, 10 Jan 2011 14:33:42 -0500
Subject: [PATCH] Make nap() interruptable. Use nap() in shoveller and exit if
 idle.

---
 server/harness.c | 25 +++++++++++++++++++------
 server/harness.h |  2 +-
 server/queue.c   | 14 ++++++++++++--
 3 files changed, 32 insertions(+), 9 deletions(-)

diff --git a/server/harness.c b/server/harness.c
index df18020..5e41527 100644
--- a/server/harness.c
+++ b/server/harness.c
@@ -117,24 +117,37 @@ Process *spawn(void (*f)(void *), void *arg) {
   return p;
 }
 
+typedef struct nap_context_t_ {
+  Process *p;
+  int timeout_fired;
+} nap_context_t;
+
 void nap_isr(int fd, short what, void *arg) {
-  Process *p = (Process *) arg;
+  nap_context_t *context = arg;
   //info("nap_isr %p\n", p);
-  assert((p->state == PROCESS_WAITING) && (p->wait_flags & EV_TIMEOUT));
-  p->wait_flags &= ~EV_TIMEOUT;
-  enqueue_runlist(p);
+  assert((context->p->state == PROCESS_WAITING) && (context->p->wait_flags & EV_TIMEOUT));
+  context->timeout_fired = 1;
+  enqueue_runlist(context->p);
 }
 
-void nap(long millis) {
+int nap(long millis) {
+  struct event ev;
   struct timeval tv;
+  nap_context_t context;
   assert(current_process != NULL);
   assert(current_process->state == PROCESS_RUNNING);
+  context.p = current_process;
+  context.timeout_fired = 0;
   tv.tv_sec = millis / 1000;
   tv.tv_usec = (millis % 1000) * 1000;
-  ICHECK(event_once(-1, EV_TIMEOUT, nap_isr, current_process, &tv), "event_once");
+  evtimer_set(&ev, nap_isr, &context);
+  ICHECK(evtimer_add(&ev, &tv), "evtimer_add");
   current_process->state = PROCESS_WAITING;
   current_process->wait_flags |= EV_TIMEOUT;
   schedule();
+  current_process->wait_flags &= ~EV_TIMEOUT;
+  evtimer_del(&ev);
+  return context.timeout_fired;
 }
 
 static void awaken_waiters(IOHandle *h, short mask) {
diff --git a/server/harness.h b/server/harness.h
index 9134842..bd54ef0 100644
--- a/server/harness.h
+++ b/server/harness.h
@@ -37,7 +37,7 @@ extern Process *current_process;
 
 extern void yield(void);
 extern Process *spawn(process_main_t f, void *arg);
-extern void nap(long millis);
+extern int nap(long millis); /* 1 for timeout expired; 0 for resumed early */
 
 extern void suspend(void);
 extern void resume(Process *p);
diff --git a/server/queue.c b/server/queue.c
index 6ac9488..3035ebe 100644
--- a/server/queue.c
+++ b/server/queue.c
@@ -91,6 +91,11 @@ static void shoveller(void *qv) {
   sexp_t *body = NULL; /* held */
   subscription_t *sub = NULL;
 
+  {
+    cmsg_bytes_t n = sexp_data(q->name);
+    info("Queue <<%.*s>> busy. Shoveller entering\n", n.len, n.bytes);
+  }
+
  check_for_work:
   //info("Checking for work\n");
 
@@ -139,19 +144,24 @@ static void shoveller(void *qv) {
   end_burst(q, &burst_count, total_count);
   //info("Waiting for throck\n");
   q->shovel_awake = 0;
-  suspend();
+  if (nap(100)) {
+    cmsg_bytes_t n = sexp_data(q->name);
+    info("Queue <<%.*s>> idle. Shoveller exiting\n", n.len, n.bytes);
+    q->shovel = NULL;
+    return;
+  }
   //info("Throck received!\n");
   goto check_for_work;
 }
 
 static void throck_shovel(queue_extension_t *q) {
   if (!q->shovel_awake) {
+    q->shovel_awake = 1;
     if (!q->shovel) {
       q->shovel = spawn(shoveller, q);
     } else {
       resume(q->shovel);
     }
-    q->shovel_awake = 1;
   }
 }