minios: fix thread safety of xenbus watches

diff -r 076fb0d4b60c extras/mini-os/blkfront.c
@@ -49,6 +49,8 @@ struct blkfront_dev {
     char *nodename;
     char *backend;
     struct blkfront_info info;
+
+    xenbus_event_queue events;
 
 #ifdef HAVE_LIBC
     int fd;

@@ -101,6 +103,8 @@ struct blkfront_dev *init_blkfront(char 
 
     dev->ring_ref = gnttab_grant_access(dev->dom,virt_to_mfn(s),0);
 
+    dev->events = NULL;
+
     // FIXME: proper frees on failures
 again:
     err = xenbus_transaction_start(&xbt);

@@ -166,11 +170,9 @@ done:
 
         snprintf(path, sizeof(path), "%s/state", dev->backend);
 
-        xenbus_watch_path(XBT_NIL, path);
+        xenbus_watch_path_token(XBT_NIL, path, path, &dev->events);
 
-        xenbus_wait_for_value(path,"4");
-
-        xenbus_unwatch_path(XBT_NIL, path);
+        xenbus_wait_for_value(path, "4", &dev->events);
 
         snprintf(path, sizeof(path), "%s/info", dev->backend);
         dev->info.info = xenbus_read_integer(path);

@@ -211,10 +213,12 @@ void shutdown_blkfront(struct blkfront_d
 
     snprintf(path, sizeof(path), "%s/state", dev->backend);
     err = xenbus_printf(XBT_NIL, nodename, "state", "%u", 5); /* closing */
-    xenbus_wait_for_value(path,"5");
+    xenbus_wait_for_value(path, "5", &dev->events);
 
     err = xenbus_printf(XBT_NIL, nodename, "state", "%u", 6);
-    xenbus_wait_for_value(path,"6");
+    xenbus_wait_for_value(path, "6", &dev->events);
+
+    xenbus_unwatch_path(XBT_NIL, path);
 
     unbind_evtchn(dev->evtchn);
 

diff -r 076fb0d4b60c extras/mini-os/fbfront.c
@@ -30,6 +30,8 @@ struct kbdfront_dev {
 
     char *nodename;
     char *backend;
+
+    xenbus_event_queue events;
 
 #ifdef HAVE_LIBC
     int fd;

@@ -75,6 +77,8 @@ struct kbdfront_dev *init_kbdfront(char 
     dev->page = s = (struct xenkbd_page*) alloc_page();
     memset(s,0,PAGE_SIZE);
 
+    dev->events = NULL;
+
     s->in_cons = s->in_prod = 0;
     s->out_cons = s->out_prod = 0;
 

@@ -136,11 +140,9 @@ done:
 
         snprintf(path, sizeof(path), "%s/state", dev->backend);
 
-        xenbus_watch_path(XBT_NIL, path);
+        xenbus_watch_path_token(XBT_NIL, path, path, &dev->events);
 
-        xenbus_wait_for_value(path,"4");
-
-        xenbus_unwatch_path(XBT_NIL, path);
+        xenbus_wait_for_value(path, "4", &dev->events);
 
         printk("%s connected\n", dev->backend);
 

@@ -199,10 +201,12 @@ void shutdown_kbdfront(struct kbdfront_d
 
     snprintf(path, sizeof(path), "%s/state", dev->backend);
     err = xenbus_printf(XBT_NIL, nodename, "state", "%u", 5); /* closing */
-    xenbus_wait_for_value(path,"5");
+    xenbus_wait_for_value(path, "5", &dev->events);
 
     err = xenbus_printf(XBT_NIL, nodename, "state", "%u", 6);
-    xenbus_wait_for_value(path,"6");
+    xenbus_wait_for_value(path, "6", &dev->events);
+
+    xenbus_unwatch_path(XBT_NIL, path);
 
     unbind_evtchn(dev->evtchn);
 

@@ -251,6 +255,8 @@ struct fbfront_dev {
     int offset;
 
     int status;
+
+    xenbus_event_queue events;
 };
 
 void fbfront_handler(evtchn_port_t port, struct pt_regs *regs, void *data)

@@ -323,6 +329,7 @@ struct fbfront_dev *init_fbfront(char *n
     dev->mem_length = s->mem_length = n * PAGE_SIZE;
     dev->offset = 0;
     dev->status = XENFB_BACKEND_STATUS_ACTIVE;
+    dev->events = NULL;
 
     const int max_pd = sizeof(s->pd) / sizeof(s->pd[0]);
     unsigned long mapped = 0;

@@ -399,13 +406,11 @@ done:
 
         snprintf(path, sizeof(path), "%s/state", dev->backend);
 
-        xenbus_watch_path(XBT_NIL, path);
+        xenbus_watch_path_token(XBT_NIL, path, path, &dev->events);
 
-        xenbus_wait_for_value(path,"4");
+        xenbus_wait_for_value(path, "4", &dev->events);
 
         printk("%s connected\n", dev->backend);
-
-        xenbus_unwatch_path(XBT_NIL, path);
 
         snprintf(path, sizeof(path), "%s/request-update", dev->backend);
         dev->request_update = xenbus_read_integer(path);

@@ -499,10 +504,12 @@ void shutdown_fbfront(struct fbfront_dev
 
     snprintf(path, sizeof(path), "%s/state", dev->backend);
     err = xenbus_printf(XBT_NIL, nodename, "state", "%u", 5); /* closing */
-    xenbus_wait_for_value(path,"5");
+    xenbus_wait_for_value(path, "5", &dev->events);
 
     err = xenbus_printf(XBT_NIL, nodename, "state", "%u", 6);
-    xenbus_wait_for_value(path,"6");
+    xenbus_wait_for_value(path, "6", &dev->events);
+
+    xenbus_unwatch_path(XBT_NIL, path);
 
     unbind_evtchn(dev->evtchn);
 

diff -r 076fb0d4b60c extras/mini-os/fs-front.c
@@ -917,6 +917,7 @@ static int init_fs_import(struct fs_impo
     struct fsif_sring *sring;
     int retry = 0;
     domid_t self_id;
+    xenbus_event_queue events = NULL;
 
     printk("Initialising FS fortend to backend dom %d\n", import->dom_id);
     /* Allocate page for the shared ring */

@@ -1026,8 +1027,8 @@ done:
     sprintf(r_nodename, "%s/state", import->backend);
     sprintf(token, "fs-front-%d", import->import_id);
     /* The token will not be unique if multiple imports are inited */
-    xenbus_watch_path(XBT_NIL, r_nodename/*, token*/);
-    xenbus_wait_for_value(/*token,*/ r_nodename, STATE_READY);
+    xenbus_watch_path_token(XBT_NIL, r_nodename, r_nodename, &events);
+    xenbus_wait_for_value(r_nodename, STATE_READY, &events);
     xenbus_unwatch_path(XBT_NIL, r_nodename);
     printk("Backend ready.\n");
    

diff -r 076fb0d4b60c extras/mini-os/include/lib.h
@@ -178,7 +178,7 @@ extern struct file {
         struct {
             /* To each xenbus FD is associated a queue of watch events for this
              * FD.  */
-            struct xenbus_event *volatile events;
+            xenbus_event_queue events;
         } xenbus;
     };
     volatile int read;	/* maybe available for read */

diff -r 076fb0d4b60c extras/mini-os/include/xenbus.h
@@ -19,17 +19,18 @@ struct xenbus_event {
     char *token;
     struct xenbus_event *next;
 };
+typedef struct xenbus_event *xenbus_event_queue;
 
-char *xenbus_watch_path_token(xenbus_transaction_t xbt, const char *path, const char *token, struct xenbus_event *volatile *events);
+char *xenbus_watch_path_token(xenbus_transaction_t xbt, const char *path, const char *token, xenbus_event_queue *events);
 char *xenbus_unwatch_path_token(xenbus_transaction_t xbt, const char *path, const char *token);
 extern struct wait_queue_head xenbus_watch_queue;
-void xenbus_wait_for_watch(void);
-char **xenbus_wait_for_watch_return(void);
-char* xenbus_wait_for_value(const char *path, const char *value);
+void xenbus_wait_for_watch(xenbus_event_queue *queue);
+char **xenbus_wait_for_watch_return(xenbus_event_queue *queue);
+char* xenbus_wait_for_value(const char *path, const char *value, xenbus_event_queue *queue);
 
 /* When no token is provided, use a global queue. */
 #define XENBUS_WATCH_PATH_TOKEN "xenbus_watch_path"
-extern struct xenbus_event * volatile xenbus_events;
+extern xenbus_event_queue xenbus_events;
 #define xenbus_watch_path(xbt, path) xenbus_watch_path_token(xbt, path, XENBUS_WATCH_PATH_TOKEN, NULL)
 #define xenbus_unwatch_path(xbt, path) xenbus_unwatch_path_token(xbt, path, XENBUS_WATCH_PATH_TOKEN)
 

diff -r 076fb0d4b60c extras/mini-os/netfront.c
@@ -52,6 +52,8 @@ struct netfront_dev {
 
     char *nodename;
     char *backend;
+
+    xenbus_event_queue events;
 
 #ifdef HAVE_LIBC
     int fd;

@@ -328,6 +330,8 @@ struct netfront_dev *init_netfront(char 
 
     dev->netif_rx = thenetif_rx;
 
+    dev->events = NULL;
+
     // FIXME: proper frees on failures
 again:
     err = xenbus_transaction_start(&xbt);

@@ -399,11 +403,9 @@ done:
         char path[strlen(dev->backend) + 1 + 5 + 1];
         snprintf(path, sizeof(path), "%s/state", dev->backend);
 
-        xenbus_watch_path(XBT_NIL, path);
+        xenbus_watch_path_token(XBT_NIL, path, path, &dev->events);
 
-        xenbus_wait_for_value(path,"4");
-
-        xenbus_unwatch_path(XBT_NIL, path);
+        xenbus_wait_for_value(path, "4", &dev->events);
 
         if (ip) {
             snprintf(path, sizeof(path), "%s/ip", dev->backend);

@@ -458,10 +460,12 @@ void shutdown_netfront(struct netfront_d
 
     snprintf(path, sizeof(path), "%s/state", dev->backend);
     err = xenbus_printf(XBT_NIL, nodename, "state", "%u", 5); /* closing */
-    xenbus_wait_for_value(path,"5");
+    xenbus_wait_for_value(path, "5", &dev->events);
 
     err = xenbus_printf(XBT_NIL, nodename, "state", "%u", 6);
-    xenbus_wait_for_value(path,"6");
+    xenbus_wait_for_value(path, "6", &dev->events);
+
+    xenbus_unwatch_path(XBT_NIL, path);
 
     unbind_evtchn(dev->evtchn);
 

diff -r 076fb0d4b60c extras/mini-os/xenbus/xenbus.c
@@ -45,10 +45,10 @@ static DECLARE_WAIT_QUEUE_HEAD(xb_waitq)
 static DECLARE_WAIT_QUEUE_HEAD(xb_waitq);
 DECLARE_WAIT_QUEUE_HEAD(xenbus_watch_queue);
 
-struct xenbus_event *volatile xenbus_events;
+xenbus_event_queue xenbus_events;
 static struct watch {
     char *token;
-    struct xenbus_event *volatile *events;
+    xenbus_event_queue *events;
     struct watch *next;
 } *watches;
 struct xenbus_req_info 

@@ -75,28 +75,34 @@ static void memcpy_from_ring(const void 
     memcpy(dest + c1, ring, c2);
 }
 
-char **xenbus_wait_for_watch_return()
+char **xenbus_wait_for_watch_return(xenbus_event_queue *queue)
 {
     struct xenbus_event *event;
+    if (!queue)
+        queue = &xenbus_events;
     DEFINE_WAIT(w);
-    while (!(event = xenbus_events)) {
+    while (!(event = *queue)) {
         add_waiter(w, xenbus_watch_queue);
         schedule();
     }
     remove_waiter(w);
-    xenbus_events = event->next;
+    *queue = event->next;
     return &event->path;
 }
 
-void xenbus_wait_for_watch(void)
+void xenbus_wait_for_watch(xenbus_event_queue *queue)
 {
     char **ret;
-    ret = xenbus_wait_for_watch_return();
+    if (!queue)
+        queue = &xenbus_events;
+    ret = xenbus_wait_for_watch_return(queue);
     free(ret);
 }
 
-char* xenbus_wait_for_value(const char* path, const char* value)
+char* xenbus_wait_for_value(const char* path, const char* value, xenbus_event_queue *queue)
 {
+    if (!queue)
+        queue = &xenbus_events;
     for(;;)
     {
         char *res, *msg;

@@ -109,7 +115,7 @@ char* xenbus_wait_for_value(const char* 
         free(res);
 
         if(r==0) break;
-        else xenbus_wait_for_watch();
+        else xenbus_wait_for_watch(queue);
     }
     return NULL;
 }

@@ -147,8 +153,8 @@ static void xenbus_thread_func(void *ign
 
             if(msg.type == XS_WATCH_EVENT)
             {
-		struct xenbus_event *event = malloc(sizeof(*event) + msg.len),
-                                    *volatile *events = NULL;
+		struct xenbus_event *event = malloc(sizeof(*event) + msg.len);
+                xenbus_event_queue *events = NULL;
 		char *data = (char*)event + sizeof(*event);
                 struct watch *watch;
 

@@ -167,8 +173,6 @@ static void xenbus_thread_func(void *ign
                         events = watch->events;
                         break;
                     }
-                if (!events)
-                    events = &xenbus_events;
 
 		event->next = *events;
 		*events = event;

@@ -463,7 +467,7 @@ char *xenbus_write(xenbus_transaction_t 
     return NULL;
 }
 
-char* xenbus_watch_path_token( xenbus_transaction_t xbt, const char *path, const char *token, struct xenbus_event *volatile *events)
+char* xenbus_watch_path_token( xenbus_transaction_t xbt, const char *path, const char *token, xenbus_event_queue *events)
 {
     struct xsd_sockmsg *rep;
 

@@ -473,6 +477,9 @@ char* xenbus_watch_path_token( xenbus_tr
     };
 
     struct watch *watch = malloc(sizeof(*watch));
+
+    if (!events)
+        events = &xenbus_events;
 
     watch->token = strdup(token);
     watch->events = events;