minios: fix thread safety of xenbus watches by requiring callers to

provide their own queue of events, because else we can not dispatch to
watchers running in parallel.

Signed-off-by: Samuel Thibault <samuel.thibault@eu.citrix.com>

extras/mini-os/blkfront.c

@@ -50,6 +50,8 @@ struct blkfront_dev {
     char *backend;
     struct blkfront_info info;
 
+    xenbus_event_queue events;
+
 #ifdef HAVE_LIBC
     int fd;
 #endif
@@ -101,6 +103,8 @@ struct blkfront_dev *init_blkfront(char *nodename, struct blkfront_info *info)
 
     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_dev *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);
 

extras/mini-os/fbfront.c

@@ -31,6 +31,8 @@ struct kbdfront_dev {
     char *nodename;
     char *backend;
 
+    xenbus_event_queue events;
+
 #ifdef HAVE_LIBC
     int fd;
 #endif
@@ -75,6 +77,8 @@ struct kbdfront_dev *init_kbdfront(char *nodename, int abs_pointer)
     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_dev *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);
 
@@ -249,6 +253,8 @@ struct fbfront_dev {
     int stride;
     int mem_length;
     int offset;
+
+    xenbus_event_queue events;
 };
 
 void fbfront_handler(evtchn_port_t port, struct pt_regs *regs, void *data)
@@ -292,6 +298,7 @@ struct fbfront_dev *init_fbfront(char *nodename, unsigned long *mfns, int width,
     dev->stride = s->line_length = stride;
     dev->mem_length = s->mem_length = n * PAGE_SIZE;
     dev->offset = 0;
+    dev->events = NULL;
 
     const int max_pd = sizeof(s->pd) / sizeof(s->pd[0]);
     unsigned long mapped = 0;
@@ -368,14 +375,12 @@ 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);
 
@@ -463,10 +468,12 @@ void shutdown_fbfront(struct fbfront_dev *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);
 

extras/mini-os/fs-front.c

@@ -917,6 +917,7 @@ static int init_fs_import(struct fs_import *import)
     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");
    

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 */

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)
 

extras/mini-os/netfront.c

@@ -53,6 +53,8 @@ struct netfront_dev {
     char *nodename;
     char *backend;
 
+    xenbus_event_queue events;
+
 #ifdef HAVE_LIBC
     int fd;
     unsigned char *data;
@@ -328,6 +330,8 @@ struct netfront_dev *init_netfront(char *nodename, void (*thenetif_rx)(unsigned
 
     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_dev *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);
 

extras/mini-os/xenbus/xenbus.c

@@ -45,10 +45,10 @@ static struct xenstore_domain_interface *xenstore_buf;
 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 *Ring,
     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* path, const char* value)
         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 xbt, const char *path, const char *value
     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;
 
@@ -474,6 +478,9 @@ char* xenbus_watch_path_token( xenbus_transaction_t xbt, const char *path, const
 
     struct watch *watch = malloc(sizeof(*watch));
 
+    if (!events)
+        events = &xenbus_events;
+
     watch->token = strdup(token);
     watch->events = events;
     watch->next = watches;