drivers/xen/xenbus/xenbus_xs.c

   1 /******************************************************************************
   2  * xenbus_xs.c
   3  *
   4  * This is the kernel equivalent of the "xs" library.  We don't need everything
   5  * and we use xenbus_comms for communication.
   6  *
   7  * Copyright (C) 2005 Rusty Russell, IBM Corporation
   8  *
   9  * This program is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU General Public License version 2
  11  * as published by the Free Software Foundation; or, when distributed
  12  * separately from the Linux kernel or incorporated into other
  13  * software packages, subject to the following license:
  14  *
  15  * Permission is hereby granted, free of charge, to any person obtaining a copy
  16  * of this source file (the "Software"), to deal in the Software without
  17  * restriction, including without limitation the rights to use, copy, modify,
  18  * merge, publish, distribute, sublicense, and/or sell copies of the Software,
  19  * and to permit persons to whom the Software is furnished to do so, subject to
  20  * the following conditions:
  21  *
  22  * The above copyright notice and this permission notice shall be included in
  23  * all copies or substantial portions of the Software.
  24  *
  25  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  26  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  27  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  28  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  29  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  30  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
  31  * IN THE SOFTWARE.
  32  */
  33
  34 #include <linux/unistd.h>
  35 #include <linux/errno.h>
  36 #include <linux/types.h>
  37 #include <linux/uio.h>
  38 #include <linux/kernel.h>
  39 #include <linux/string.h>
  40 #include <linux/err.h>
  41 #include <linux/slab.h>
  42 #include <linux/fcntl.h>
  43 #include <linux/kthread.h>
  44 #include <linux/rwsem.h>
  45 #include <linux/module.h>
  46 #include <linux/mutex.h>
  47 #include <xen/xenbus.h>
  48 #include "xenbus_comms.h"
  49
  50 struct xs_stored_msg {
  51         struct list_head list;
  52
  53         struct xsd_sockmsg hdr;
  54
  55         union {
  56                 /* Queued replies. */
  57                 struct {
  58                         char *body;
  59                 } reply;
  60
  61                 /* Queued watch events. */
  62                 struct {
  63                         struct xenbus_watch *handle;
  64                         char **vec;
  65                         unsigned int vec_size;
  66                 } watch;
  67         } u;
  68 };
  69
  70 struct xs_handle {
  71         /* A list of replies. Currently only one will ever be outstanding. */
  72         struct list_head reply_list;
  73         spinlock_t reply_lock;
  74         wait_queue_head_t reply_waitq;
  75
  76         /*
  77          * Mutex ordering: transaction_mutex -> watch_mutex -> request_mutex.
  78          * response_mutex is never taken simultaneously with the other three.
  79          *
  80          * transaction_mutex must be held before incrementing
  81          * transaction_count. The mutex is held when a suspend is in
  82          * progress to prevent new transactions starting.
  83          *
  84          * When decrementing transaction_count to zero the wait queue
  85          * should be woken up, the suspend code waits for count to
  86          * reach zero.
  87          */
  88
  89         /* One request at a time. */
  90         struct mutex request_mutex;
  91
  92         /* Protect xenbus reader thread against save/restore. */
  93         struct mutex response_mutex;
  94
  95         /* Protect transactions against save/restore. */
  96         struct mutex transaction_mutex;
  97         atomic_t transaction_count;
  98         wait_queue_head_t transaction_wq;
  99
 100         /* Protect watch (de)register against save/restore. */
 101         struct rw_semaphore watch_mutex;
 102 };
 103
 104 static struct xs_handle xs_state;
 105
 106 /* List of registered watches, and a lock to protect it. */
 107 static LIST_HEAD(watches);
 108 static DEFINE_SPINLOCK(watches_lock);
 109
 110 /* List of pending watch callback events, and a lock to protect it. */
 111 static LIST_HEAD(watch_events);
 112 static DEFINE_SPINLOCK(watch_events_lock);
 113
 114 /*
 115  * Details of the xenwatch callback kernel thread. The thread waits on the
 116  * watch_events_waitq for work to do (queued on watch_events list). When it
 117  * wakes up it acquires the xenwatch_mutex before reading the list and
 118  * carrying out work.
 119  */
 120 static pid_t xenwatch_pid;
 121 static DEFINE_MUTEX(xenwatch_mutex);
 122 static DECLARE_WAIT_QUEUE_HEAD(watch_events_waitq);
 123
 124 static int get_error(const char *errorstring)
 125 {
 126         unsigned int i;
 127
 128         for (i = 0; strcmp(errorstring, xsd_errors[i].errstring) != 0; i++) {
 129                 if (i == ARRAY_SIZE(xsd_errors) - 1) {
 130                         printk(KERN_WARNING
 131                                "XENBUS xen store gave: unknown error %s",
 132                                errorstring);
 133                         return EINVAL;
 134                 }
 135         }
 136         return xsd_errors[i].errnum;
 137 }
 138
 139 static void *read_reply(enum xsd_sockmsg_type *type, unsigned int *len)
 140 {
 141         struct xs_stored_msg *msg;
 142         char *body;
 143
 144         spin_lock(&xs_state.reply_lock);
 145
 146         while (list_empty(&xs_state.reply_list)) {
 147                 spin_unlock(&xs_state.reply_lock);
 148                 /* XXX FIXME: Avoid synchronous wait for response here. */
 149                 wait_event(xs_state.reply_waitq,
 150                            !list_empty(&xs_state.reply_list));
 151                 spin_lock(&xs_state.reply_lock);
 152         }
 153
 154         msg = list_entry(xs_state.reply_list.next,
 155                          struct xs_stored_msg, list);
 156         list_del(&msg->list);
 157
 158         spin_unlock(&xs_state.reply_lock);
 159
 160         *type = msg->hdr.type;
 161         if (len)
 162                 *len = msg->hdr.len;
 163         body = msg->u.reply.body;
 164
 165         kfree(msg);
 166
 167         return body;
 168 }
 169
 170 static void transaction_start(void)
 171 {
 172         mutex_lock(&xs_state.transaction_mutex);
 173         atomic_inc(&xs_state.transaction_count);
 174         mutex_unlock(&xs_state.transaction_mutex);
 175 }
 176
 177 static void transaction_end(void)
 178 {
 179         if (atomic_dec_and_test(&xs_state.transaction_count))
 180                 wake_up(&xs_state.transaction_wq);
 181 }
 182
 183 static void transaction_suspend(void)
 184 {
 185         mutex_lock(&xs_state.transaction_mutex);
 186         wait_event(xs_state.transaction_wq,
 187                    atomic_read(&xs_state.transaction_count) == 0);
 188 }
 189
 190 static void transaction_resume(void)
 191 {
 192         mutex_unlock(&xs_state.transaction_mutex);
 193 }
 194
 195 void *xenbus_dev_request_and_reply(struct xsd_sockmsg *msg)
 196 {
 197         void *ret;
 198         struct xsd_sockmsg req_msg = *msg;
 199         int err;
 200
 201         if (req_msg.type == XS_TRANSACTION_START)
 202                 transaction_start();
 203
 204         mutex_lock(&xs_state.request_mutex);
 205
 206         err = xb_write(msg, sizeof(*msg) + msg->len);
 207         if (err) {
 208                 msg->type = XS_ERROR;
 209                 ret = ERR_PTR(err);
 210         } else
 211                 ret = read_reply(&msg->type, &msg->len);
 212
 213         mutex_unlock(&xs_state.request_mutex);
 214
 215         if ((msg->type == XS_TRANSACTION_END) ||
 216             ((req_msg.type == XS_TRANSACTION_START) &&
 217              (msg->type == XS_ERROR)))
 218                 transaction_end();
 219
 220         return ret;
 221 }
 222 EXPORT_SYMBOL(xenbus_dev_request_and_reply);
 223
 224 /* Send message to xs, get kmalloc'ed reply.  ERR_PTR() on error. */
 225 static void *xs_talkv(struct xenbus_transaction t,
 226                       enum xsd_sockmsg_type type,
 227                       const struct kvec *iovec,
 228                       unsigned int num_vecs,
 229                       unsigned int *len)
 230 {
 231         struct xsd_sockmsg msg;
 232         void *ret = NULL;
 233         unsigned int i;
 234         int err;
 235
 236         msg.tx_id = t.id;
 237         msg.req_id = 0;
 238         msg.type = type;
 239         msg.len = 0;
 240         for (i = 0; i < num_vecs; i++)
 241                 msg.len += iovec[i].iov_len;
 242
 243         mutex_lock(&xs_state.request_mutex);
 244
 245         err = xb_write(&msg, sizeof(msg));
 246         if (err) {
 247                 mutex_unlock(&xs_state.request_mutex);
 248                 return ERR_PTR(err);
 249         }
 250
 251         for (i = 0; i < num_vecs; i++) {
 252                 err = xb_write(iovec[i].iov_base, iovec[i].iov_len);
 253                 if (err) {
 254                         mutex_unlock(&xs_state.request_mutex);
 255                         return ERR_PTR(err);
 256                 }
 257         }
 258
 259         ret = read_reply(&msg.type, len);
 260
 261         mutex_unlock(&xs_state.request_mutex);
 262
 263         if (IS_ERR(ret))
 264                 return ret;
 265
 266         if (msg.type == XS_ERROR) {
 267                 err = get_error(ret);
 268                 kfree(ret);
 269                 return ERR_PTR(-err);
 270         }
 271
 272         if (msg.type != type) {
 273                 if (printk_ratelimit())
 274                         printk(KERN_WARNING
 275                                "XENBUS unexpected type [%d], expected [%d]\n",
 276                                msg.type, type);
 277                 kfree(ret);
 278                 return ERR_PTR(-EINVAL);
 279         }
 280         return ret;
 281 }
 282
 283 /* Simplified version of xs_talkv: single message. */
 284 static void *xs_single(struct xenbus_transaction t,
 285                        enum xsd_sockmsg_type type,
 286                        const char *string,
 287                        unsigned int *len)
 288 {
 289         struct kvec iovec;
 290
 291         iovec.iov_base = (void *)string;
 292         iovec.iov_len = strlen(string) + 1;
 293         return xs_talkv(t, type, &iovec, 1, len);
 294 }
 295
 296 /* Many commands only need an ack, don't care what it says. */
 297 static int xs_error(char *reply)
 298 {
 299         if (IS_ERR(reply))
 300                 return PTR_ERR(reply);
 301         kfree(reply);
 302         return 0;
 303 }
 304
 305 static unsigned int count_strings(const char *strings, unsigned int len)
 306 {
 307         unsigned int num;
 308         const char *p;
 309
 310         for (p = strings, num = 0; p < strings + len; p += strlen(p) + 1)
 311                 num++;
 312
 313         return num;
 314 }
 315
 316 /* Return the path to dir with /name appended. Buffer must be kfree()'ed. */
 317 static char *join(const char *dir, const char *name)
 318 {
 319         char *buffer;
 320
 321         if (strlen(name) == 0)
 322                 buffer = kasprintf(GFP_NOIO | __GFP_HIGH, "%s", dir);
 323         else
 324                 buffer = kasprintf(GFP_NOIO | __GFP_HIGH, "%s/%s", dir, name);
 325         return (!buffer) ? ERR_PTR(-ENOMEM) : buffer;
 326 }
 327
 328 static char **split(char *strings, unsigned int len, unsigned int *num)
 329 {
 330         char *p, **ret;
 331
 332         /* Count the strings. */
 333         *num = count_strings(strings, len);
 334
 335         /* Transfer to one big alloc for easy freeing. */
 336         ret = kmalloc(*num * sizeof(char *) + len, GFP_NOIO | __GFP_HIGH);
 337         if (!ret) {
 338                 kfree(strings);
 339                 return ERR_PTR(-ENOMEM);
 340         }
 341         memcpy(&ret[*num], strings, len);
 342         kfree(strings);
 343
 344         strings = (char *)&ret[*num];
 345         for (p = strings, *num = 0; p < strings + len; p += strlen(p) + 1)
 346                 ret[(*num)++] = p;
 347
 348         return ret;
 349 }
 350
 351 char **xenbus_directory(struct xenbus_transaction t,
 352                         const char *dir, const char *node, unsigned int *num)
 353 {
 354         char *strings, *path;
 355         unsigned int len;
 356
 357         path = join(dir, node);
 358         if (IS_ERR(path))
 359                 return (char **)path;
 360
 361         strings = xs_single(t, XS_DIRECTORY, path, &len);
 362         kfree(path);
 363         if (IS_ERR(strings))
 364                 return (char **)strings;
 365
 366         return split(strings, len, num);
 367 }
 368 EXPORT_SYMBOL_GPL(xenbus_directory);
 369
 370 /* Check if a path exists. Return 1 if it does. */
 371 int xenbus_exists(struct xenbus_transaction t,
 372                   const char *dir, const char *node)
 373 {
 374         char **d;
 375         int dir_n;
 376
 377         d = xenbus_directory(t, dir, node, &dir_n);
 378         if (IS_ERR(d))
 379                 return 0;
 380         kfree(d);
 381         return 1;
 382 }
 383 EXPORT_SYMBOL_GPL(xenbus_exists);
 384
 385 /* Get the value of a single file.
 386  * Returns a kmalloced value: call free() on it after use.
 387  * len indicates length in bytes.
 388  */
 389 void *xenbus_read(struct xenbus_transaction t,
 390                   const char *dir, const char *node, unsigned int *len)
 391 {
 392         char *path;
 393         void *ret;
 394
 395         path = join(dir, node);
 396         if (IS_ERR(path))
 397                 return (void *)path;
 398
 399         ret = xs_single(t, XS_READ, path, len);
 400         kfree(path);
 401         return ret;
 402 }
 403 EXPORT_SYMBOL_GPL(xenbus_read);
 404
 405 /* Write the value of a single file.
 406  * Returns -err on failure.
 407  */
 408 int xenbus_write(struct xenbus_transaction t,
 409                  const char *dir, const char *node, const char *string)
 410 {
 411         const char *path;
 412         struct kvec iovec[2];
 413         int ret;
 414
 415         path = join(dir, node);
 416         if (IS_ERR(path))
 417                 return PTR_ERR(path);
 418
 419         iovec[0].iov_base = (void *)path;
 420         iovec[0].iov_len = strlen(path) + 1;
 421         iovec[1].iov_base = (void *)string;
 422         iovec[1].iov_len = strlen(string);
 423
 424         ret = xs_error(xs_talkv(t, XS_WRITE, iovec, ARRAY_SIZE(iovec), NULL));
 425         kfree(path);
 426         return ret;
 427 }
 428 EXPORT_SYMBOL_GPL(xenbus_write);
 429
 430 /* Create a new directory. */
 431 int xenbus_mkdir(struct xenbus_transaction t,
 432                  const char *dir, const char *node)
 433 {
 434         char *path;
 435         int ret;
 436
 437         path = join(dir, node);
 438         if (IS_ERR(path))
 439                 return PTR_ERR(path);
 440
 441         ret = xs_error(xs_single(t, XS_MKDIR, path, NULL));
 442         kfree(path);
 443         return ret;
 444 }
 445 EXPORT_SYMBOL_GPL(xenbus_mkdir);
 446
 447 /* Destroy a file or directory (directories must be empty). */
 448 int xenbus_rm(struct xenbus_transaction t, const char *dir, const char *node)
 449 {
 450         char *path;
 451         int ret;
 452
 453         path = join(dir, node);
 454         if (IS_ERR(path))
 455                 return PTR_ERR(path);
 456
 457         ret = xs_error(xs_single(t, XS_RM, path, NULL));
 458         kfree(path);
 459         return ret;
 460 }
 461 EXPORT_SYMBOL_GPL(xenbus_rm);
 462
 463 /* Start a transaction: changes by others will not be seen during this
 464  * transaction, and changes will not be visible to others until end.
 465  */
 466 int xenbus_transaction_start(struct xenbus_transaction *t)
 467 {
 468         char *id_str;
 469
 470         transaction_start();
 471
 472         id_str = xs_single(XBT_NIL, XS_TRANSACTION_START, "", NULL);
 473         if (IS_ERR(id_str)) {
 474                 transaction_end();
 475                 return PTR_ERR(id_str);
 476         }
 477
 478         t->id = simple_strtoul(id_str, NULL, 0);
 479         kfree(id_str);
 480         return 0;
 481 }
 482 EXPORT_SYMBOL_GPL(xenbus_transaction_start);
 483
 484 /* End a transaction.
 485  * If abandon is true, transaction is discarded instead of committed.
 486  */
 487 int xenbus_transaction_end(struct xenbus_transaction t, int abort)
 488 {
 489         char abortstr[2];
 490         int err;
 491
 492         if (abort)
 493                 strcpy(abortstr, "F");
 494         else
 495                 strcpy(abortstr, "T");
 496
 497         err = xs_error(xs_single(t, XS_TRANSACTION_END, abortstr, NULL));
 498
 499         transaction_end();
 500
 501         return err;
 502 }
 503 EXPORT_SYMBOL_GPL(xenbus_transaction_end);
 504
 505 /* Single read and scanf: returns -errno or num scanned. */
 506 int xenbus_scanf(struct xenbus_transaction t,
 507                  const char *dir, const char *node, const char *fmt, ...)
 508 {
 509         va_list ap;
 510         int ret;
 511         char *val;
 512
 513         val = xenbus_read(t, dir, node, NULL);
 514         if (IS_ERR(val))
 515                 return PTR_ERR(val);
 516
 517         va_start(ap, fmt);
 518         ret = vsscanf(val, fmt, ap);
 519         va_end(ap);
 520         kfree(val);
 521         /* Distinctive errno. */
 522         if (ret == 0)
 523                 return -ERANGE;
 524         return ret;
 525 }
 526 EXPORT_SYMBOL_GPL(xenbus_scanf);
 527
 528 /* Single printf and write: returns -errno or 0. */
 529 int xenbus_printf(struct xenbus_transaction t,
 530                   const char *dir, const char *node, const char *fmt, ...)
 531 {
 532         va_list ap;
 533         int ret;
 534 #define PRINTF_BUFFER_SIZE 4096
 535         char *printf_buffer;
 536
 537         printf_buffer = kmalloc(PRINTF_BUFFER_SIZE, GFP_NOIO | __GFP_HIGH);
 538         if (printf_buffer == NULL)
 539                 return -ENOMEM;
 540
 541         va_start(ap, fmt);
 542         ret = vsnprintf(printf_buffer, PRINTF_BUFFER_SIZE, fmt, ap);
 543         va_end(ap);
 544
 545         BUG_ON(ret > PRINTF_BUFFER_SIZE-1);
 546         ret = xenbus_write(t, dir, node, printf_buffer);
 547
 548         kfree(printf_buffer);
 549
 550         return ret;
 551 }
 552 EXPORT_SYMBOL_GPL(xenbus_printf);
 553
 554 /* Takes tuples of names, scanf-style args, and void **, NULL terminated. */
 555 int xenbus_gather(struct xenbus_transaction t, const char *dir, ...)
 556 {
 557         va_list ap;
 558         const char *name;
 559         int ret = 0;
 560
 561         va_start(ap, dir);
 562         while (ret == 0 && (name = va_arg(ap, char *)) != NULL) {
 563                 const char *fmt = va_arg(ap, char *);
 564                 void *result = va_arg(ap, void *);
 565                 char *p;
 566
 567                 p = xenbus_read(t, dir, name, NULL);
 568                 if (IS_ERR(p)) {
 569                         ret = PTR_ERR(p);
 570                         break;
 571                 }
 572                 if (fmt) {
 573                         if (sscanf(p, fmt, result) == 0)
 574                                 ret = -EINVAL;
 575                         kfree(p);
 576                 } else
 577                         *(char **)result = p;
 578         }
 579         va_end(ap);
 580         return ret;
 581 }
 582 EXPORT_SYMBOL_GPL(xenbus_gather);
 583
 584 static int xs_watch(const char *path, const char *token)
 585 {
 586         struct kvec iov[2];
 587
 588         iov[0].iov_base = (void *)path;
 589         iov[0].iov_len = strlen(path) + 1;
 590         iov[1].iov_base = (void *)token;
 591         iov[1].iov_len = strlen(token) + 1;
 592
 593         return xs_error(xs_talkv(XBT_NIL, XS_WATCH, iov,
 594                                  ARRAY_SIZE(iov), NULL));
 595 }
 596
 597 static int xs_unwatch(const char *path, const char *token)
 598 {
 599         struct kvec iov[2];
 600
 601         iov[0].iov_base = (char *)path;
 602         iov[0].iov_len = strlen(path) + 1;
 603         iov[1].iov_base = (char *)token;
 604         iov[1].iov_len = strlen(token) + 1;
 605
 606         return xs_error(xs_talkv(XBT_NIL, XS_UNWATCH, iov,
 607                                  ARRAY_SIZE(iov), NULL));
 608 }
 609
 610 static struct xenbus_watch *find_watch(const char *token)
 611 {
 612         struct xenbus_watch *i, *cmp;
 613
 614         cmp = (void *)simple_strtoul(token, NULL, 16);
 615
 616         list_for_each_entry(i, &watches, list)
 617                 if (i == cmp)
 618                         return i;
 619
 620         return NULL;
 621 }
 622
 623 /* Register callback to watch this node. */
 624 int register_xenbus_watch(struct xenbus_watch *watch)
 625 {
 626         /* Pointer in ascii is the token. */
 627         char token[sizeof(watch) * 2 + 1];
 628         int err;
 629
 630         sprintf(token, "%lX", (long)watch);
 631
 632         down_read(&xs_state.watch_mutex);
 633
 634         spin_lock(&watches_lock);
 635         BUG_ON(find_watch(token));
 636         list_add(&watch->list, &watches);
 637         spin_unlock(&watches_lock);
 638
 639         err = xs_watch(watch->node, token);
 640
 641         /* Ignore errors due to multiple registration. */
 642         if ((err != 0) && (err != -EEXIST)) {
 643                 spin_lock(&watches_lock);
 644                 list_del(&watch->list);
 645                 spin_unlock(&watches_lock);
 646         }
 647
 648         up_read(&xs_state.watch_mutex);
 649
 650         return err;
 651 }
 652 EXPORT_SYMBOL_GPL(register_xenbus_watch);
 653
 654 void unregister_xenbus_watch(struct xenbus_watch *watch)
 655 {
 656         struct xs_stored_msg *msg, *tmp;
 657         char token[sizeof(watch) * 2 + 1];
 658         int err;
 659
 660         sprintf(token, "%lX", (long)watch);
 661
 662         down_read(&xs_state.watch_mutex);
 663
 664         spin_lock(&watches_lock);
 665         BUG_ON(!find_watch(token));
 666         list_del(&watch->list);
 667         spin_unlock(&watches_lock);
 668
 669         err = xs_unwatch(watch->node, token);
 670         if (err)
 671                 printk(KERN_WARNING
 672                        "XENBUS Failed to release watch %s: %i\n",
 673                        watch->node, err);
 674
 675         up_read(&xs_state.watch_mutex);
 676
 677         /* Make sure there are no callbacks running currently (unless
 678            its us) */
 679         if (current->pid != xenwatch_pid)
 680                 mutex_lock(&xenwatch_mutex);
 681
 682         /* Cancel pending watch events. */
 683         spin_lock(&watch_events_lock);
 684         list_for_each_entry_safe(msg, tmp, &watch_events, list) {
 685                 if (msg->u.watch.handle != watch)
 686                         continue;
 687                 list_del(&msg->list);
 688                 kfree(msg->u.watch.vec);
 689                 kfree(msg);
 690         }
 691         spin_unlock(&watch_events_lock);
 692
 693         if (current->pid != xenwatch_pid)
 694                 mutex_unlock(&xenwatch_mutex);
 695 }
 696 EXPORT_SYMBOL_GPL(unregister_xenbus_watch);
 697
 698 void xs_suspend(void)
 699 {
 700         transaction_suspend();
 701         down_write(&xs_state.watch_mutex);
 702         mutex_lock(&xs_state.request_mutex);
 703         mutex_lock(&xs_state.response_mutex);
 704 }
 705
 706 void xs_resume(void)
 707 {
 708         struct xenbus_watch *watch;
 709         char token[sizeof(watch) * 2 + 1];
 710
 711         xb_init_comms();
 712
 713         mutex_unlock(&xs_state.response_mutex);
 714         mutex_unlock(&xs_state.request_mutex);
 715         transaction_resume();
 716
 717         /* No need for watches_lock: the watch_mutex is sufficient. */
 718         list_for_each_entry(watch, &watches, list) {
 719                 sprintf(token, "%lX", (long)watch);
 720                 xs_watch(watch->node, token);
 721         }
 722
 723         up_write(&xs_state.watch_mutex);
 724 }
 725
 726 void xs_suspend_cancel(void)
 727 {
 728         mutex_unlock(&xs_state.response_mutex);
 729         mutex_unlock(&xs_state.request_mutex);
 730         up_write(&xs_state.watch_mutex);
 731         mutex_unlock(&xs_state.transaction_mutex);
 732 }
 733
 734 static int xenwatch_thread(void *unused)
 735 {
 736         struct list_head *ent;
 737         struct xs_stored_msg *msg;
 738
 739         for (;;) {
 740                 wait_event_interruptible(watch_events_waitq,
 741                                          !list_empty(&watch_events));
 742
 743                 if (kthread_should_stop())
 744                         break;
 745
 746                 mutex_lock(&xenwatch_mutex);
 747
 748                 spin_lock(&watch_events_lock);
 749                 ent = watch_events.next;
 750                 if (ent != &watch_events)
 751                         list_del(ent);
 752                 spin_unlock(&watch_events_lock);
 753
 754                 if (ent != &watch_events) {
 755                         msg = list_entry(ent, struct xs_stored_msg, list);
 756                         msg->u.watch.handle->callback(
 757                                 msg->u.watch.handle,
 758                                 (const char **)msg->u.watch.vec,
 759                                 msg->u.watch.vec_size);
 760                         kfree(msg->u.watch.vec);
 761                         kfree(msg);
 762                 }
 763
 764                 mutex_unlock(&xenwatch_mutex);
 765         }
 766
 767         return 0;
 768 }
 769
 770 static int process_msg(void)
 771 {
 772         struct xs_stored_msg *msg;
 773         char *body;
 774         int err;
 775
 776         /*
 777          * We must disallow save/restore while reading a xenstore message.
 778          * A partial read across s/r leaves us out of sync with xenstored.
 779          */
 780         for (;;) {
 781                 err = xb_wait_for_data_to_read();
 782                 if (err)
 783                         return err;
 784                 mutex_lock(&xs_state.response_mutex);
 785                 if (xb_data_to_read())
 786                         break;
 787                 /* We raced with save/restore: pending data 'disappeared'. */
 788                 mutex_unlock(&xs_state.response_mutex);
 789         }
 790
 791
 792         msg = kmalloc(sizeof(*msg), GFP_NOIO | __GFP_HIGH);
 793         if (msg == NULL) {
 794                 err = -ENOMEM;
 795                 goto out;
 796         }
 797
 798         err = xb_read(&msg->hdr, sizeof(msg->hdr));
 799         if (err) {
 800                 kfree(msg);
 801                 goto out;
 802         }
 803
 804         body = kmalloc(msg->hdr.len + 1, GFP_NOIO | __GFP_HIGH);
 805         if (body == NULL) {
 806                 kfree(msg);
 807                 err = -ENOMEM;
 808                 goto out;
 809         }
 810
 811         err = xb_read(body, msg->hdr.len);
 812         if (err) {
 813                 kfree(body);
 814                 kfree(msg);
 815                 goto out;
 816         }
 817         body[msg->hdr.len] = '\0';
 818
 819         if (msg->hdr.type == XS_WATCH_EVENT) {
 820                 msg->u.watch.vec = split(body, msg->hdr.len,
 821                                          &msg->u.watch.vec_size);
 822                 if (IS_ERR(msg->u.watch.vec)) {
 823                         err = PTR_ERR(msg->u.watch.vec);
 824                         kfree(msg);
 825                         goto out;
 826                 }
 827
 828                 spin_lock(&watches_lock);
 829                 msg->u.watch.handle = find_watch(
 830                         msg->u.watch.vec[XS_WATCH_TOKEN]);
 831                 if (msg->u.watch.handle != NULL) {
 832                         spin_lock(&watch_events_lock);
 833                         list_add_tail(&msg->list, &watch_events);
 834                         wake_up(&watch_events_waitq);
 835                         spin_unlock(&watch_events_lock);
 836                 } else {
 837                         kfree(msg->u.watch.vec);
 838                         kfree(msg);
 839                 }
 840                 spin_unlock(&watches_lock);
 841         } else {
 842                 msg->u.reply.body = body;
 843                 spin_lock(&xs_state.reply_lock);
 844                 list_add_tail(&msg->list, &xs_state.reply_list);
 845                 spin_unlock(&xs_state.reply_lock);
 846                 wake_up(&xs_state.reply_waitq);
 847         }
 848
 849  out:
 850         mutex_unlock(&xs_state.response_mutex);
 851         return err;
 852 }
 853
 854 static int xenbus_thread(void *unused)
 855 {
 856         int err;
 857
 858         for (;;) {
 859                 err = process_msg();
 860                 if (err)
 861                         printk(KERN_WARNING "XENBUS error %d while reading "
 862                                "message\n", err);
 863                 if (kthread_should_stop())
 864                         break;
 865         }
 866
 867         return 0;
 868 }
 869
 870 int xs_init(void)
 871 {
 872         int err;
 873         struct task_struct *task;
 874
 875         INIT_LIST_HEAD(&xs_state.reply_list);
 876         spin_lock_init(&xs_state.reply_lock);
 877         init_waitqueue_head(&xs_state.reply_waitq);
 878
 879         mutex_init(&xs_state.request_mutex);
 880         mutex_init(&xs_state.response_mutex);
 881         mutex_init(&xs_state.transaction_mutex);
 882         init_rwsem(&xs_state.watch_mutex);
 883         atomic_set(&xs_state.transaction_count, 0);
 884         init_waitqueue_head(&xs_state.transaction_wq);
 885
 886         /* Initialize the shared memory rings to talk to xenstored */
 887         err = xb_init_comms();
 888         if (err)
 889                 return err;
 890
 891         task = kthread_run(xenwatch_thread, NULL, "xenwatch");
 892         if (IS_ERR(task))
 893                 return PTR_ERR(task);
 894         xenwatch_pid = task->pid;
 895
 896         task = kthread_run(xenbus_thread, NULL, "xenbus");
 897         if (IS_ERR(task))
 898                 return PTR_ERR(task);
 899
 900         return 0;
 901 }