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