ref: 956ff38121ca10af2fc33dd1700f29bb80bf716c
dir: /sys/src/9/xen/devxenstore.c/
/*
* Driver for xenstore - database shared between domains, used by xenbus to
* communicate configuration info.
*/
#include "u.h"
#include "../port/lib.h"
#include "mem.h"
#include "dat.h"
#include "fns.h"
#include "../port/error.h"
#include "../pc/io.h"
#define LOG(a)
typedef struct Aux Aux;
enum {
Qtopdir,
Qctl,
Qwatch,
WRITING = 0,
READING,
WATCHING,
MAXIO = 8*1024,
};
Dirtab xsdir[] = {
".", {Qtopdir, 0, QTDIR}, 0, 0555,
"xenstore", {Qctl, 0}, 0, 0660,
"xenwatch", {Qwatch, 0}, 0, 0440,
};
struct {
struct xenstore_domain_interface *intf;
struct xsd_sockmsg hdr;
int hdrvalid;
int evtchn;
int nextreqid;
Aux *rhead;
Aux *kernelaux;
Queue *evq;
Rendez wr;
Rendez rr;
QLock;
Lock rlock;
} xenstore;
struct Aux {
QLock;
Rendez qr;
Queue *ioq;
Aux *next;
int state;
int reqid;
};
static char Ephase[] = "phase error";
static char Eproto[] = "protocol error";
static char NodeShutdown[] = "control/shutdown";
static void xenbusproc(void*);
static int
notfull(void*)
{
struct xenstore_domain_interface *xs = xenstore.intf;
return (xs->req_prod-xs->req_cons) < XENSTORE_RING_SIZE;
}
static int
notempty(void*)
{
struct xenstore_domain_interface *xs = xenstore.intf;
return xs->rsp_prod > xs->rsp_cons;
}
static int
ishead(void* a)
{
return xenstore.rhead == a;
}
static void
xsintr(Ureg*, void*)
{
LOG(dprint("xsintr\n");)
wakeup(&xenstore.rr);
wakeup(&xenstore.wr);
}
static void
xwrite(Queue *q, char *buf, int len)
{
struct xenstore_domain_interface *xs;
int m, n;
XENSTORE_RING_IDX idx;
xs = xenstore.intf;
while (len > 0) {
n = XENSTORE_RING_SIZE - (xs->req_prod - xs->req_cons);
if (n == 0) {
xenchannotify(xenstore.evtchn);
sleep(&xenstore.wr, notfull, 0);
continue;
}
if (n > len)
n = len;
idx = MASK_XENSTORE_IDX(xs->req_prod);
m = XENSTORE_RING_SIZE - idx;
if (m > n)
m = n;
if (q)
qread(q, xs->req+idx, m);
else
memmove(xs->req+idx, buf, m);
if (m < n) {
if (q)
qread(q, xs->req, n-m);
else
memmove(xs->req, buf+m, n-m);
}
coherence();
xs->req_prod += n;
xenchannotify(xenstore.evtchn);
if (buf)
buf += n;
len -= n;
}
}
static void
xread(Queue *q, char *buf, int len)
{
struct xenstore_domain_interface *xs = xenstore.intf;
int n, m;
XENSTORE_RING_IDX idx;
for (n = len; n > 0; n -= m) {
while (xs->rsp_prod == xs->rsp_cons) {
xenchannotify(xenstore.evtchn);
if (up == 0)
HYPERVISOR_yield();
else
sleep(&xenstore.rr, notempty, 0);
}
idx = MASK_XENSTORE_IDX(xs->rsp_cons);
m = xs->rsp_prod - xs->rsp_cons;
if (m > n)
m = n;
if (m > XENSTORE_RING_SIZE - idx)
m = XENSTORE_RING_SIZE - idx;
if (q)
qwrite(q, xs->rsp+idx, m);
else if (buf) {
memmove(buf, xs->rsp+idx, m);
buf += m;
}
coherence();
xs->rsp_cons += m;
}
xenchannotify(xenstore.evtchn);
}
static void
xsrpc(Aux *aux)
{
Queue *q;
Aux *l, *r, **lp;
struct xsd_sockmsg hdr;
long n;
q = aux->ioq;
if (aux->state == WATCHING)
aux->reqid = 0;
else {
/* get the request header and check validity */
if (qlen(q) < sizeof hdr)
error(Eproto);
qread(q, &hdr, sizeof hdr);
n = hdr.len;
if (qlen(q) != n)
error(Eproto);
qlock(&xenstore);
/* generate a unique request id */
aux->reqid = ++xenstore.nextreqid;
hdr.req_id = aux->reqid;
hdr.tx_id = 0;
/* send the request */
xwrite(0, (char*)&hdr, sizeof hdr);
xwrite(q, 0, n);
qunlock(&xenstore);
}
/* join list of requests awaiting response */
ilock(&xenstore.rlock);
if (xenstore.rhead == 0) {
aux->next = 0;
xenstore.rhead = aux;
} else {
aux->next = xenstore.rhead->next;
xenstore.rhead->next = aux;
}
iunlock(&xenstore.rlock);
/* loop until matching response header has been received */
if (waserror()) {
ilock(&xenstore.rlock);
for (lp = &xenstore.rhead; *lp && *lp != aux; lp = &(*lp)->next)
;
if (*lp != 0) {
*lp = (*lp)->next;
if (lp == &xenstore.rhead && *lp)
wakeup(&(*lp)->qr);
}
iunlock(&xenstore.rlock);
nexterror();
}
for (;;) {
/* wait until this request reaches head of queue */
if (xenstore.rhead != aux)
sleep(&aux->qr, ishead, aux);
/* wait until a response header (maybe for another request) has been read */
if (!xenstore.hdrvalid) {
xread(0, (char*)&xenstore.hdr, sizeof xenstore.hdr);
xenstore.hdrvalid = 1;
}
if (xenstore.hdr.req_id == aux->reqid)
break;
/* response was for a different request: move matching request to head of queue */
ilock(&xenstore.rlock);
for (l = xenstore.rhead; r = l->next; l = r)
if (xenstore.hdr.req_id == r->reqid) {
l->next = r->next;
r->next = xenstore.rhead;
xenstore.rhead = r;
break;
}
iunlock(&xenstore.rlock);
if (r) {
/* wake the matching request */
wakeup(&r->qr);
} else {
/* response without a request: should be a watch event */
xenstore.hdrvalid = 0;
xread(0, 0, xenstore.hdr.len);
continue;
}
}
/* queue the response header, and data if any, for the caller to read */
qwrite(q, &xenstore.hdr, sizeof xenstore.hdr);
xenstore.hdrvalid = 0;
/* read the data, if any */
if (xenstore.hdr.len > 0)
xread(q, 0, xenstore.hdr.len);
/* remove finished request and wake the next request on the queue */
ilock(&xenstore.rlock);
xenstore.rhead = aux->next;
iunlock(&xenstore.rlock);
poperror();
if (xenstore.rhead != 0)
wakeup(&xenstore.rhead->qr);
}
static void
xsreset()
{
LOG(dprint("xsreset\n");)
}
static void
xsinit()
{
intrenable(xenstore.evtchn, xsintr, 0, BUSUNKNOWN, "Xen store");
kproc("xenbus", xenbusproc, 0);
}
static Chan*
xsattach(char *spec)
{
return devattach('x', spec);
}
static Walkqid*
xswalk(Chan *c, Chan *nc, char **name, int nname)
{
return devwalk(c, nc, name, nname, xsdir, nelem(xsdir), devgen);
}
static int
xsstat(Chan *c, uchar *dp, int n)
{
return devstat(c, dp, n, xsdir, nelem(xsdir), devgen);
}
static Aux*
auxalloc(int initstate)
{
Aux *aux;
Queue *q;
aux = mallocz(sizeof(Aux), 1);
if (aux == 0)
return 0;
q = qopen(MAXIO, 0, 0, 0);
if (q == 0) {
free(aux);
return 0;
}
qnoblock(q, 1);
aux->state = initstate;
aux->ioq = q;
return aux;
}
static Chan*
xsopen(Chan *c, int omode)
{
Aux *aux;
int state;
c = devopen(c, omode, xsdir, nelem(xsdir), devgen);
state = WRITING;
switch ((ulong)c->qid.path) {
case Qwatch:
state = WATCHING;
/* fall through */
case Qctl:
aux = auxalloc(state);
if (aux == 0) {
c->flag &= ~COPEN;
error(Enomem);
}
c->aux = aux;
break;
}
return c;
}
static void
xsclose(Chan* c)
{
Aux *aux;
if ((c->flag&COPEN) == 0)
return;
switch ((ulong)c->qid.path) {
case Qwatch:
case Qctl:
if ((aux = (Aux*)c->aux) != 0) {
qfree(aux->ioq);
free(aux);
c->aux = 0;
}
break;
}
}
static long
xsread(Chan *c, void *a, long n, vlong off)
{
Aux *aux;
Queue *q;
long nr;
USED(off);
if (c->qid.type == QTDIR)
return devdirread(c, a, n, xsdir, nelem(xsdir), devgen);
aux = (Aux*)c->aux;
qlock(aux);
if (waserror()) {
qunlock(aux);
nexterror();
}
q = aux->ioq;
switch (aux->state) {
case WRITING:
if (qlen(q) == 0)
error(Ephase);
xsrpc(aux);
aux->state = READING;
break;
case WATCHING:
if (qlen(q) == 0)
xsrpc(aux);
break;
}
if (!qcanread(q))
nr = 0;
else
nr = qread(q, a, n);
qunlock(aux);
poperror();
return nr;
}
static long
xswrite(Chan *c, void *a, long n, vlong off)
{
Aux *aux;
Queue *q;
long nr;
if (c->qid.type == QTDIR)
error(Eperm);
if ((ulong)c->qid.path == Qwatch)
error(Ebadusefd);
aux = (Aux*)c->aux;
qlock(aux);
if (waserror()) {
qunlock(aux);
nexterror();
}
q = aux->ioq;
if ((off == 0 || aux->state == READING) && qlen(q) > 0)
qflush(q);
aux->state = WRITING;
nr = qwrite(aux->ioq, a, n);
qunlock(aux);
poperror();
return nr;
}
Dev xenstoredevtab = {
'x',
"xenstore",
xsreset,
xsinit,
devshutdown,
xsattach,
xswalk,
xsstat,
xsopen,
devcreate,
xsclose,
xsread,
devbread,
xswrite,
devbwrite,
devremove,
devwstat,
};
static char*
xscmd(Aux *aux, char *buf, int cmd, char *s, char *val)
{
struct xsd_sockmsg *msg;
char *arg;
long n;
msg = (struct xsd_sockmsg*)buf;
arg = buf + sizeof(*msg);
msg->type = cmd;
msg->len = strlen(s)+1;
if (val) {
msg->len += strlen(val);
if (cmd == XS_WATCH)
msg->len++; /* stupid special case */
}
strcpy(arg, s);
if (val)
strcpy(arg+strlen(s)+1, val);
n = sizeof(*msg)+msg->len;
if (up == 0) {
msg->req_id = 1;
msg->tx_id = 0;
xwrite(0, buf, n);
xread(0, buf, sizeof(*msg));
xread(0, arg, msg->len);
} else {
qlock(aux);
if (qlen(aux->ioq) > 0)
qflush(aux->ioq);
qwrite(aux->ioq, buf, n);
xsrpc(aux);
qread(aux->ioq, buf, sizeof(*msg));
LOG(dprint("xs: type %d req_id %d len %d\n", msg->type, msg->req_id, msg->len);)
// XXX buffer overflow
qread(aux->ioq, arg, msg->len);
qunlock(aux);
}
arg[msg->len] = 0;
if (msg->type == XS_ERROR) {
return 0;
}
return arg;
}
static void
intfinit(void)
{
if (xenstore.intf == 0) {
xenstore.intf = (struct xenstore_domain_interface*)mmumapframe(XENBUS, xenstart->store_mfn);
xenstore.evtchn = xenstart->store_evtchn;
xenstore.kernelaux = auxalloc(WRITING);
}
}
void
xenstore_write(char *s, char *val)
{
char buf[512];
intfinit();
xscmd(xenstore.kernelaux, buf, XS_WRITE, s, val);
}
int
xenstore_read(char *s, char *val, int len)
{
char buf[512];
char *p;
intfinit();
p = xscmd(xenstore.kernelaux, buf, XS_READ, s, nil);
if (p == 0)
return -1;
strecpy(val, val+len, p);
return 1;
}
void
xenstore_setd(char *dir, char *node, int value)
{
int off;
char buf[12];
off = strlen(dir);
sprint(dir+off, "%s", node);
sprint(buf, "%ud", value);
xenstore_write(dir, buf);
dir[off] = 0;
}
int
xenstore_gets(char *dir, char *node, char *buf, int buflen)
{
int off;
int n;
off = strlen(dir);
sprint(dir+off, "%s", node);
n = xenstore_read(dir, buf, buflen);
dir[off] = 0;
return n;
}
static void
xenbusproc(void*)
{
Chan *c;
Aux *aux;
char *p;
struct xsd_sockmsg msg;
char buf[512];
int n, m;
c = namec("#x/xenstore", Aopen, ORDWR, 0);
aux = (Aux*)c->aux;
c = namec("#x/xenwatch", Aopen, OREAD, 0);
xscmd(aux, buf, XS_WATCH, NodeShutdown, "$");
for (;;) {
xsread(c, &msg, sizeof(msg), 0);
for (n = msg.len; n > 0; n -= m)
m = xsread(c, buf, msg.len, sizeof(msg));
buf[msg.len] = 0;
if (strcmp(buf, NodeShutdown) != 0)
continue;
p = xscmd(aux, buf, XS_READ, NodeShutdown, nil);
if (p == nil)
continue;
if (strcmp(p, "poweroff") == 0)
reboot(nil, nil, 0);
else if (strcmp(p, "reboot") == 0)
exit(0);
else {
print("xenbus: %s=%s\n", NodeShutdown, p);
xscmd(aux, buf, XS_WRITE, NodeShutdown, "");
}
}
}