sys/src/9/xen/devxenstore.c

   1 /*
   2  * Driver for xenstore - database shared between domains, used by xenbus to
   3  * communicate configuration info.
   4  */
   5
   6 #include "u.h"
   7 #include "../port/lib.h"
   8 #include "mem.h"
   9 #include "dat.h"
  10 #include "fns.h"
  11 #include "../port/error.h"
  12 #include "../pc/io.h"
  13
  14 #define LOG(a)
  15
  16 typedef struct Aux Aux;
  17
  18 enum {
  19         Qtopdir,
  20         Qctl,
  21         Qwatch,
  22         WRITING = 0,
  23         READING,
  24         WATCHING,
  25         MAXIO = 8*1024,
  26 };
  27
  28 Dirtab xsdir[] = {
  29         ".",    {Qtopdir, 0, QTDIR},    0,      0555,
  30         "xenstore",     {Qctl, 0},      0,      0660,
  31         "xenwatch", {Qwatch, 0}, 0, 0440,
  32 };
  33
  34 struct {
  35         struct xenstore_domain_interface        *intf;
  36         struct xsd_sockmsg      hdr;
  37         int     hdrvalid;
  38         int     evtchn;
  39         int     nextreqid;
  40         Aux *rhead;
  41         Aux *kernelaux;
  42         Queue *evq;
  43         Rendez wr;
  44         Rendez rr;
  45         QLock;
  46         Lock rlock;
  47 } xenstore;
  48
  49 struct Aux {
  50         QLock;
  51         Rendez qr;
  52         Queue *ioq;
  53         Aux     *next;
  54         int state;
  55         int     reqid;
  56 };
  57
  58 static char Ephase[] = "phase error";
  59 static char Eproto[] = "protocol error";
  60 static char NodeShutdown[] = "control/shutdown";
  61
  62 static void xenbusproc(void*);
  63
  64 static int
  65 notfull(void*)
  66 {
  67         struct xenstore_domain_interface *xs = xenstore.intf;
  68
  69         return (xs->req_prod-xs->req_cons) < XENSTORE_RING_SIZE;
  70 }
  71
  72 static int
  73 notempty(void*)
  74 {
  75         struct xenstore_domain_interface *xs = xenstore.intf;
  76
  77         return xs->rsp_prod > xs->rsp_cons;
  78 }
  79
  80 static int
  81 ishead(void* a)
  82 {
  83         return xenstore.rhead == a;
  84 }
  85
  86 static void
  87 xsintr(Ureg*, void*)
  88 {
  89         LOG(dprint("xsintr\n");)
  90         wakeup(&xenstore.rr);
  91         wakeup(&xenstore.wr);
  92 }
  93
  94 static void
  95 xwrite(Queue *q, char *buf, int len)
  96 {
  97         struct xenstore_domain_interface *xs;
  98         int m, n;
  99         XENSTORE_RING_IDX idx;
 100
 101         xs = xenstore.intf;
 102         while (len > 0) {
 103                 n = XENSTORE_RING_SIZE - (xs->req_prod - xs->req_cons);
 104                 if (n == 0) {
 105                         xenchannotify(xenstore.evtchn);
 106                         sleep(&xenstore.wr, notfull, 0);
 107                         continue;
 108                 }
 109                 if (n > len)
 110                         n = len;
 111                 idx = MASK_XENSTORE_IDX(xs->req_prod);
 112                 m = XENSTORE_RING_SIZE - idx;
 113                 if (m > n)
 114                         m = n;
 115                 if (q)
 116                         qread(q, xs->req+idx, m);
 117                 else
 118                         memmove(xs->req+idx, buf, m);
 119                 if (m < n) {
 120                         if (q)
 121                                 qread(q, xs->req, n-m);
 122                         else
 123                                 memmove(xs->req, buf+m, n-m);
 124                 }
 125                 coherence();
 126                 xs->req_prod += n;
 127                 xenchannotify(xenstore.evtchn);
 128                 if (buf)
 129                         buf += n;
 130                 len -= n;
 131         }
 132 }
 133
 134 static void
 135 xread(Queue *q, char *buf, int len)
 136 {
 137         struct xenstore_domain_interface *xs = xenstore.intf;
 138         int n, m;
 139         XENSTORE_RING_IDX idx;
 140
 141         for (n = len; n > 0; n -= m) {
 142                 while (xs->rsp_prod == xs->rsp_cons) {
 143                         xenchannotify(xenstore.evtchn);
 144                         if (up == 0)
 145                                 HYPERVISOR_yield();
 146                         else
 147                                 sleep(&xenstore.rr, notempty, 0);
 148                 }
 149                 idx = MASK_XENSTORE_IDX(xs->rsp_cons);
 150                 m = xs->rsp_prod - xs->rsp_cons;
 151                 if (m > n)
 152                         m = n;
 153                 if (m > XENSTORE_RING_SIZE - idx)
 154                         m = XENSTORE_RING_SIZE - idx;
 155                 if (q)
 156                         qwrite(q, xs->rsp+idx, m);
 157                 else if (buf) {
 158                         memmove(buf, xs->rsp+idx, m);
 159                         buf += m;
 160                 }
 161                 coherence();
 162                 xs->rsp_cons += m;
 163         }
 164         xenchannotify(xenstore.evtchn);
 165 }
 166
 167 static void
 168 xsrpc(Aux *aux)
 169 {
 170         Queue *q;
 171         Aux *l, *r, **lp;
 172         struct xsd_sockmsg hdr;
 173         long n;
 174
 175         q = aux->ioq;
 176
 177         if (aux->state == WATCHING)
 178                 aux->reqid = 0;
 179         else {
 180                 /* get the request header and check validity */
 181                 if (qlen(q) < sizeof hdr)
 182                         error(Eproto);
 183                 qread(q, &hdr, sizeof hdr);
 184                 n = hdr.len;
 185                 if (qlen(q) != n)
 186                         error(Eproto);
 187                 qlock(&xenstore);
 188                 /* generate a unique request id */
 189                 aux->reqid = ++xenstore.nextreqid;
 190                 hdr.req_id = aux->reqid;
 191                 hdr.tx_id = 0;
 192                 /* send the request */
 193                 xwrite(0, (char*)&hdr, sizeof hdr);
 194                 xwrite(q, 0, n);
 195                 qunlock(&xenstore);
 196         }
 197
 198         /* join list of requests awaiting response */
 199         ilock(&xenstore.rlock);
 200         if (xenstore.rhead == 0) {
 201                 aux->next = 0;
 202                 xenstore.rhead = aux;
 203         } else {
 204                 aux->next = xenstore.rhead->next;
 205                 xenstore.rhead->next = aux;
 206         }
 207         iunlock(&xenstore.rlock);
 208
 209         /* loop until matching response header has been received */
 210         if (waserror()) {
 211                 ilock(&xenstore.rlock);
 212                 for (lp = &xenstore.rhead; *lp && *lp != aux; lp = &(*lp)->next)
 213                         ;
 214                 if (*lp != 0) {
 215                         *lp = (*lp)->next;
 216                         if (lp == &xenstore.rhead && *lp)
 217                                 wakeup(&(*lp)->qr);
 218                 }
 219                 iunlock(&xenstore.rlock);
 220                 nexterror();
 221         }
 222         for (;;) {
 223                 /* wait until this request reaches head of queue */
 224                 if (xenstore.rhead != aux)
 225                         sleep(&aux->qr, ishead, aux);
 226                 /* wait until a response header (maybe for another request) has been read */
 227                 if (!xenstore.hdrvalid) {
 228                         xread(0, (char*)&xenstore.hdr, sizeof xenstore.hdr);
 229                         xenstore.hdrvalid = 1;
 230                 }
 231                 if (xenstore.hdr.req_id == aux->reqid)
 232                         break;
 233                 /* response was for a different request: move matching request to head of queue */
 234                 ilock(&xenstore.rlock);
 235                 for (l = xenstore.rhead; r = l->next; l = r)
 236                         if (xenstore.hdr.req_id == r->reqid) {
 237                                 l->next = r->next;
 238                                 r->next = xenstore.rhead;
 239                                 xenstore.rhead = r;
 240                                 break;
 241                         }
 242                 iunlock(&xenstore.rlock);
 243                 if (r) {
 244                         /* wake the matching request */
 245                         wakeup(&r->qr);
 246                 } else {
 247                         /* response without a request: should be a watch event */
 248                         xenstore.hdrvalid = 0;
 249                         xread(0, 0, xenstore.hdr.len);
 250                         continue;
 251                 }
 252         }
 253
 254         /* queue the response header, and data if any, for the caller to read */
 255         qwrite(q, &xenstore.hdr, sizeof xenstore.hdr);
 256         xenstore.hdrvalid = 0;
 257         /* read the data, if any */
 258         if (xenstore.hdr.len > 0)
 259                 xread(q, 0, xenstore.hdr.len);
 260
 261         /* remove finished request and wake the next request on the queue */
 262         ilock(&xenstore.rlock);
 263         xenstore.rhead = aux->next;
 264         iunlock(&xenstore.rlock);
 265         poperror();
 266         if (xenstore.rhead != 0)
 267                 wakeup(&xenstore.rhead->qr);
 268 }
 269
 270 static void
 271 xsreset()
 272 {
 273         LOG(dprint("xsreset\n");)
 274 }
 275
 276 static void
 277 xsinit()
 278 {
 279         intrenable(xenstore.evtchn, xsintr, 0, BUSUNKNOWN, "Xen store");
 280         kproc("xenbus", xenbusproc, 0);
 281 }
 282
 283 static Chan*
 284 xsattach(char *spec)
 285 {
 286         return devattach('x', spec);
 287 }
 288
 289 static Walkqid*
 290 xswalk(Chan *c, Chan *nc, char **name, int nname)
 291 {
 292         return devwalk(c, nc, name, nname, xsdir, nelem(xsdir), devgen);
 293 }
 294
 295 static int
 296 xsstat(Chan *c, uchar *dp, int n)
 297 {
 298         return devstat(c, dp, n, xsdir, nelem(xsdir), devgen);
 299 }
 300
 301 static Aux*
 302 auxalloc(int initstate)
 303 {
 304         Aux *aux;
 305         Queue *q;
 306
 307         aux = mallocz(sizeof(Aux), 1);
 308         if (aux == 0)
 309                 return 0;
 310         q = qopen(MAXIO, 0, 0, 0);
 311         if (q == 0) {
 312                 free(aux);
 313                 return 0;
 314         }
 315         qnoblock(q, 1);
 316         aux->state = initstate;
 317         aux->ioq = q;
 318         return aux;
 319 }
 320
 321 static Chan*
 322 xsopen(Chan *c, int omode)
 323 {
 324         Aux *aux;
 325         int state;
 326
 327         c = devopen(c, omode, xsdir, nelem(xsdir), devgen);
 328         state = WRITING;
 329         switch ((ulong)c->qid.path) {
 330         case Qwatch:
 331                 state = WATCHING;
 332         /* fall through */
 333         case Qctl:
 334                 aux = auxalloc(state);
 335                 if (aux == 0) {
 336                         c->flag &= ~COPEN;
 337                         error(Enomem);
 338                 }
 339                 c->aux = aux;
 340                 break;
 341         }
 342         return c;
 343 }
 344
 345 static void
 346 xsclose(Chan* c)
 347 {
 348         Aux *aux;
 349
 350         if ((c->flag&COPEN) == 0)
 351                 return;
 352
 353         switch ((ulong)c->qid.path) {
 354         case Qwatch:
 355         case Qctl:
 356                 if ((aux = (Aux*)c->aux) != 0) {
 357                         qfree(aux->ioq);
 358                         free(aux);
 359                         c->aux = 0;
 360                 }
 361                 break;
 362         }
 363 }
 364
 365 static long
 366 xsread(Chan *c, void *a, long n, vlong off)
 367 {
 368         Aux *aux;
 369         Queue *q;
 370         long nr;
 371
 372         USED(off);
 373         if (c->qid.type == QTDIR)
 374                 return devdirread(c, a, n, xsdir, nelem(xsdir), devgen);
 375
 376         aux = (Aux*)c->aux;
 377         qlock(aux);
 378         if (waserror()) {
 379                 qunlock(aux);
 380                 nexterror();
 381         }
 382         q = aux->ioq;
 383         switch (aux->state) {
 384         case WRITING:
 385                 if (qlen(q) == 0)
 386                         error(Ephase);
 387                 xsrpc(aux);
 388                 aux->state = READING;
 389                 break;
 390         case WATCHING:
 391                 if (qlen(q) == 0)
 392                         xsrpc(aux);
 393                 break;
 394         }
 395         if (!qcanread(q))
 396                 nr = 0;
 397         else
 398                 nr = qread(q, a, n);
 399         qunlock(aux);
 400         poperror();
 401         return nr;
 402 }
 403
 404 static long
 405 xswrite(Chan *c, void *a, long n, vlong off)
 406 {
 407         Aux *aux;
 408         Queue *q;
 409         long nr;
 410
 411         if (c->qid.type == QTDIR)
 412                 error(Eperm);
 413         if ((ulong)c->qid.path == Qwatch)
 414                 error(Ebadusefd);
 415
 416         aux = (Aux*)c->aux;
 417         qlock(aux);
 418         if (waserror()) {
 419                 qunlock(aux);
 420                 nexterror();
 421         }
 422         q = aux->ioq;
 423         if ((off == 0 || aux->state == READING) && qlen(q) > 0)
 424                 qflush(q);
 425         aux->state = WRITING;
 426         nr = qwrite(aux->ioq, a, n);
 427         qunlock(aux);
 428         poperror();
 429         return nr;
 430 }
 431
 432 Dev xenstoredevtab = {
 433         'x',
 434         "xenstore",
 435
 436         xsreset,
 437         xsinit,
 438         devshutdown,
 439         xsattach,
 440         xswalk,
 441         xsstat,
 442         xsopen,
 443         devcreate,
 444         xsclose,
 445         xsread,
 446         devbread,
 447         xswrite,
 448         devbwrite,
 449         devremove,
 450         devwstat,
 451 };
 452
 453 static char*
 454 xscmd(Aux *aux, char *buf, int cmd, char *s, char *val)
 455 {
 456         struct xsd_sockmsg *msg;
 457         char *arg;
 458         long n;
 459
 460         msg = (struct xsd_sockmsg*)buf;
 461         arg = buf + sizeof(*msg);
 462         msg->type = cmd;
 463         msg->len = strlen(s)+1;
 464         if (val) {
 465                 msg->len += strlen(val);
 466                 if (cmd == XS_WATCH)
 467                         msg->len++;             /* stupid special case */
 468         }
 469         strcpy(arg, s);
 470         if (val)
 471                 strcpy(arg+strlen(s)+1, val);
 472         n = sizeof(*msg)+msg->len;
 473         if (up == 0) {
 474                 msg->req_id = 1;
 475                 msg->tx_id = 0;
 476                 xwrite(0, buf, n);
 477                 xread(0, buf, sizeof(*msg));
 478                 xread(0, arg, msg->len);
 479         } else {
 480                 qlock(aux);
 481                 if (qlen(aux->ioq) > 0)
 482                         qflush(aux->ioq);
 483                 qwrite(aux->ioq, buf, n);
 484                 xsrpc(aux);
 485                 qread(aux->ioq, buf, sizeof(*msg));
 486                 LOG(dprint("xs: type %d req_id %d len %d\n", msg->type, msg->req_id, msg->len);)
 487                 // XXX buffer overflow
 488                 qread(aux->ioq, arg, msg->len);
 489                 qunlock(aux);
 490         }
 491         arg[msg->len] = 0;
 492         if (msg->type == XS_ERROR) {
 493                 return 0;
 494         }
 495         return arg;
 496 }
 497
 498 static void
 499 intfinit(void)
 500 {
 501         if (xenstore.intf == 0) {
 502                 xenstore.intf = (struct xenstore_domain_interface*)mmumapframe(XENBUS, xenstart->store_mfn);
 503                 xenstore.evtchn = xenstart->store_evtchn;
 504                 xenstore.kernelaux = auxalloc(WRITING);
 505         }
 506 }
 507
 508 void
 509 xenstore_write(char *s, char *val)
 510 {
 511         char buf[512];
 512
 513         intfinit();
 514         xscmd(xenstore.kernelaux, buf, XS_WRITE, s, val);
 515 }
 516
 517 int
 518 xenstore_read(char *s, char *val, int len)
 519 {
 520         char buf[512];
 521         char *p;
 522
 523         intfinit();
 524         p = xscmd(xenstore.kernelaux, buf, XS_READ, s, nil);
 525         if (p == 0)
 526                 return -1;
 527         strecpy(val, val+len, p);
 528         return 1;
 529 }
 530
 531 void
 532 xenstore_setd(char *dir, char *node, int value)
 533 {
 534         int off;
 535         char buf[12];
 536
 537         off = strlen(dir);
 538         sprint(dir+off, "%s", node);
 539         sprint(buf, "%ud", value);
 540         xenstore_write(dir, buf);
 541         dir[off] = 0;
 542 }
 543
 544 int
 545 xenstore_gets(char *dir, char *node, char *buf, int buflen)
 546 {
 547         int off;
 548         int n;
 549
 550         off = strlen(dir);
 551         sprint(dir+off, "%s", node);
 552         n = xenstore_read(dir, buf, buflen);
 553         dir[off] = 0;
 554         return n;
 555 }
 556
 557 static void
 558 xenbusproc(void*)
 559 {
 560         Chan *c;
 561         Aux *aux;
 562         char *p;
 563         struct xsd_sockmsg msg;
 564         char buf[512];
 565         int n, m;
 566
 567         c = namec("#x/xenstore", Aopen, ORDWR, 0);
 568         aux = (Aux*)c->aux;
 569         c = namec("#x/xenwatch", Aopen, OREAD, 0);
 570         xscmd(aux, buf, XS_WATCH, NodeShutdown, "$");
 571         for (;;) {
 572                 xsread(c, &msg, sizeof(msg), 0);
 573                 for (n = msg.len; n > 0; n -= m)
 574                         m = xsread(c, buf, msg.len, sizeof(msg));
 575                 buf[msg.len] = 0;
 576                 if (strcmp(buf, NodeShutdown) != 0)
 577                         continue;
 578                 p = xscmd(aux, buf, XS_READ, NodeShutdown, nil);
 579                 if (p == nil)
 580                         continue;
 581                 if (strcmp(p, "poweroff") == 0)
 582                         reboot(nil, nil, 0);
 583                 else if (strcmp(p, "reboot") == 0)
 584                         exit(0);
 585                 else {
 586                         print("xenbus: %s=%s\n", NodeShutdown, p);
 587                         xscmd(aux, buf, XS_WRITE, NodeShutdown, "");
 588                 }
 589         }
 590 }