Xenbus implementation ported from Linux to Mini-os, simple thread support introduced

to simplify the porting. 64 bit version of Mini-os now compiles, but does not work
because of the pagetables and some bits of scheduler not being written.

Signed-off-by: Grzegorz Milos <gm281@cam.ac.uk>

4 files changed, 834 insertions, 0 deletions

diff --git a/extras/mini-os/xenbus/Makefile b/extras/mini-os/xenbus/Makefile
new file mode 100644
index 0000000000..b61820a9cb
--- /dev/null
+++ b/extras/mini-os/xenbus/Makefile
@@ -0,0 +1,9 @@
+all: xenstore.h xenbus_comms.o xenbus_xs.o xenbus_probe.o
+
+xenstore.h:
+	[ -e xenstored.h ] || ln -sf ../../../tools/xenstore/xenstored.h xenstored.h
+
+clean:
+	#Taken care of by main Makefile
+	#rm xenstored.h
+	#rm *.o
diff --git a/extras/mini-os/xenbus/xenbus_comms.c b/extras/mini-os/xenbus/xenbus_comms.c
new file mode 100644
index 0000000000..b299c05683
--- /dev/null
+++ b/extras/mini-os/xenbus/xenbus_comms.c
@@ -0,0 +1,231 @@
+/******************************************************************************
+ * xenbus_comms.c
+ *
+ * Low level code to talks to Xen Store: ringbuffer and event channel.
+ *
+ * Copyright (C) 2005 Rusty Russell, IBM Corporation
+ * 
+ * This file may be distributed separately from the Linux kernel, or
+ * incorporated into other software packages, subject to the following license:
+ * 
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this source file (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use, copy, modify,
+ * merge, publish, distribute, sublicense, and/or sell copies of the Software,
+ * and to permit persons to whom the Software is furnished to do so, subject to
+ * the following conditions:
+ * 
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ * 
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+ * IN THE SOFTWARE.
+ */
+#include <types.h>
+#include <wait.h>
+#include <mm.h>
+#include <hypervisor.h>
+#include <events.h>
+#include <os.h>
+#include <lib.h>
+
+
+#ifdef XENBUS_COMMS_DEBUG
+#define DEBUG(_f, _a...) \
+    printk("MINI_OS(file=xenbus_comms.c, line=%d) " _f "\n", __LINE__, ## _a)
+#else
+#define DEBUG(_f, _a...)    ((void)0)
+#endif
+
+
+#define RINGBUF_DATASIZE ((PAGE_SIZE / 2) - sizeof(struct ringbuf_head))
+struct ringbuf_head
+{
+	u32 write; /* Next place to write to */
+	u32 read; /* Next place to read from */
+	u8 flags;
+	char buf[0];
+} __attribute__((packed));
+
+DECLARE_WAIT_QUEUE_HEAD(xb_waitq);
+
+static inline struct ringbuf_head *outbuf(void)
+{
+	return mfn_to_virt(start_info.store_mfn);
+}
+
+static inline struct ringbuf_head *inbuf(void)
+{
+	return (struct ringbuf_head *)((char *)mfn_to_virt(start_info.store_mfn) + PAGE_SIZE/2);
+}
+
+static void wake_waiting(int port, struct pt_regs *regs)
+{
+	wake_up(&xb_waitq);
+}
+
+static int check_buffer(const struct ringbuf_head *h)
+{
+	return (h->write < RINGBUF_DATASIZE && h->read < RINGBUF_DATASIZE);
+}
+
+/* We can't fill last byte: would look like empty buffer. */
+static void *get_output_chunk(const struct ringbuf_head *h,
+			      void *buf, u32 *len)
+{
+	u32 read_mark;
+
+	if (h->read == 0)
+		read_mark = RINGBUF_DATASIZE - 1;
+	else
+		read_mark = h->read - 1;
+
+	/* Here to the end of buffer, unless they haven't read some out. */
+	*len = RINGBUF_DATASIZE - h->write;
+	if (read_mark >= h->write)
+		*len = read_mark - h->write;
+	return (void *)((char *)buf + h->write);
+}
+
+static const void *get_input_chunk(const struct ringbuf_head *h,
+				   const void *buf, u32 *len)
+{
+	/* Here to the end of buffer, unless they haven't written some. */
+	*len = RINGBUF_DATASIZE - h->read;
+	if (h->write >= h->read)
+		*len = h->write - h->read;
+	return (void *)((char *)buf + h->read);
+}
+
+static void update_output_chunk(struct ringbuf_head *h, u32 len)
+{
+	h->write += len;
+	if (h->write == RINGBUF_DATASIZE)
+		h->write = 0;
+}
+
+static void update_input_chunk(struct ringbuf_head *h, u32 len)
+{
+	h->read += len;
+	if (h->read == RINGBUF_DATASIZE)
+		h->read = 0;
+}
+
+static int output_avail(struct ringbuf_head *out)
+{
+	unsigned int avail;
+
+	get_output_chunk(out, out->buf, &avail);
+	return avail != 0;
+}
+
+int xb_write(const void *data, unsigned len)
+{
+	struct ringbuf_head h;
+	struct ringbuf_head *out = outbuf();
+
+	do {
+		void *dst;
+		unsigned int avail;
+
+		wait_event(xb_waitq, output_avail(out));
+
+		/* Read, then check: not that we don't trust store.
+		 * Hell, some of my best friends are daemons.  But,
+		 * in this post-911 world... */
+		h = *out;
+		mb();
+		if (!check_buffer(&h)) {
+			return -1; /* ETERRORIST! */
+		}
+
+		dst = get_output_chunk(&h, out->buf, &avail);
+		if (avail > len)
+			avail = len;
+		memcpy(dst, data, avail);
+		data = (void *)((char *)data + avail);
+		len -= avail;
+		update_output_chunk(out, avail);
+		notify_via_evtchn(start_info.store_evtchn);
+	} while (len != 0);
+
+	return 0;
+}
+
+int xs_input_avail(void)
+{
+	unsigned int avail;
+	struct ringbuf_head *in = inbuf();
+
+	get_input_chunk(in, in->buf, &avail);
+	return avail != 0;
+}
+
+int xb_read(void *data, unsigned len)
+{
+	struct ringbuf_head h;
+	struct ringbuf_head *in = inbuf();
+	int was_full;
+
+	while (len != 0) {
+		unsigned int avail;
+		const char *src;
+
+		wait_event(xb_waitq, xs_input_avail());
+		h = *in;
+		mb();
+		if (!check_buffer(&h)) {
+			return -1;
+		}
+
+		src = get_input_chunk(&h, in->buf, &avail);
+		if (avail > len)
+			avail = len;
+		was_full = !output_avail(&h);
+
+		memcpy(data, src, avail);
+		data = (void *)((char *)data + avail);
+		len -= avail;
+		update_input_chunk(in, avail);
+		DEBUG("Finished read of %i bytes (%i to go)\n", avail, len);
+		/* If it was full, tell them we've taken some. */
+		if (was_full)
+			notify_via_evtchn(start_info.store_evtchn);
+	}
+
+	/* If we left something, wake watch thread to deal with it. */
+	if (xs_input_avail())
+		wake_up(&xb_waitq);
+
+	return 0;
+}
+
+/* Set up interrupt handler off store event channel. */
+int xb_init_comms(void)
+{
+    printk("Init xenbus comms, store event channel %d\n", start_info.store_evtchn);
+	if (!start_info.store_evtchn)
+		return 0;
+    printk("Binding virq\n");
+	bind_evtchn(start_info.store_evtchn, &wake_waiting);
+
+	/* FIXME zero out page -- domain builder should probably do this*/
+	memset(mfn_to_virt(start_info.store_mfn), 0, PAGE_SIZE);
+    notify_via_evtchn(start_info.store_evtchn);
+	return 0;
+}
+
+void xb_suspend_comms(void)
+{
+
+	if (!start_info.store_evtchn)
+		return;
+
+    // TODO
+	//unbind_evtchn_from_irqhandler(xen_start_info.store_evtchn, &xb_waitq);
+}
diff --git a/extras/mini-os/xenbus/xenbus_comms.h b/extras/mini-os/xenbus/xenbus_comms.h
new file mode 100644
index 0000000000..def0ddfb4a
--- /dev/null
+++ b/extras/mini-os/xenbus/xenbus_comms.h
@@ -0,0 +1,40 @@
+/*
+ * Private include for xenbus communications.
+ * 
+ * Copyright (C) 2005 Rusty Russell, IBM Corporation
+ *
+ * This file may be distributed separately from the Linux kernel, or
+ * incorporated into other software packages, subject to the following license:
+ * 
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this source file (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use, copy, modify,
+ * merge, publish, distribute, sublicense, and/or sell copies of the Software,
+ * and to permit persons to whom the Software is furnished to do so, subject to
+ * the following conditions:
+ * 
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ * 
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+ * IN THE SOFTWARE.
+ */
+
+#ifndef _XENBUS_COMMS_H
+#define _XENBUS_COMMS_H
+
+int xb_init_comms(void);
+void xb_suspend_comms(void);
+
+/* Low level routines. */
+int xb_write(const void *data, unsigned len);
+int xb_read(void *data, unsigned len);
+int xs_input_avail(void);
+extern struct wait_queue_head xb_waitq;
+
+#endif /* _XENBUS_COMMS_H */
diff --git a/extras/mini-os/xenbus/xenbus_xs.c b/extras/mini-os/xenbus/xenbus_xs.c
new file mode 100644
index 0000000000..89b424055c
--- /dev/null
+++ b/extras/mini-os/xenbus/xenbus_xs.c
@@ -0,0 +1,554 @@
+/******************************************************************************
+ * xenbus_xs.c
+ *
+ * This is the kernel equivalent of the "xs" library.  We don't need everything
+ * and we use xenbus_comms for communication.
+ *
+ * Copyright (C) 2005 Rusty Russell, IBM Corporation
+ * 
+ * This file may be distributed separately from the Linux kernel, or
+ * incorporated into other software packages, subject to the following license:
+ * 
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this source file (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use, copy, modify,
+ * merge, publish, distribute, sublicense, and/or sell copies of the Software,
+ * and to permit persons to whom the Software is furnished to do so, subject to
+ * the following conditions:
+ * 
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ * 
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+ * IN THE SOFTWARE.
+ */
+#include <errno.h>
+#include <types.h>
+#include <list.h>
+#include <lib.h>
+#include <err.h>
+#include <os.h>
+#include <xmalloc.h>
+#include <fcntl.h>
+#include <xenbus.h>
+#include <wait.h>
+#include <sched.h>
+#include <semaphore.h>
+#include "xenstored.h"
+#include "xenbus_comms.h"
+
+#define streq(a, b) (strcmp((a), (b)) == 0)
+
+static char printf_buffer[4096];
+static LIST_HEAD(watches);
+//TODO
+DECLARE_MUTEX(xenbus_lock);
+
+static int get_error(const char *errorstring)
+{
+	unsigned int i;
+
+	for (i = 0; !streq(errorstring, xsd_errors[i].errstring); i++) {
+		if (i == ARRAY_SIZE(xsd_errors) - 1) {
+			printk("XENBUS xen store gave: unknown error %s",
+			       errorstring);
+			return EINVAL;
+		}
+	}
+	return xsd_errors[i].errnum;
+}
+
+static void *read_reply(enum xsd_sockmsg_type *type, unsigned int *len)
+{
+	struct xsd_sockmsg msg;
+	void *ret;
+	int err;
+
+	err = xb_read(&msg, sizeof(msg));
+	if (err)
+		return ERR_PTR(err);
+
+	ret = xmalloc_array(char, msg.len + 1);
+	if (!ret)
+		return ERR_PTR(-ENOMEM);
+
+	err = xb_read(ret, msg.len);
+	if (err) {
+		xfree(ret);
+		return ERR_PTR(err);
+	}
+	((char*)ret)[msg.len] = '\0';
+
+	*type = msg.type;
+	if (len)
+		*len = msg.len;
+	return ret;
+}
+
+/* Emergency write. */
+void xenbus_debug_write(const char *str, unsigned int count)
+{
+	struct xsd_sockmsg msg;
+
+	msg.type = XS_DEBUG;
+	msg.len = sizeof("print") + count + 1;
+
+	xb_write(&msg, sizeof(msg));
+	xb_write("print", sizeof("print"));
+	xb_write(str, count);
+	xb_write("", 1);
+}
+
+/* Send message to xs, get kmalloc'ed reply.  ERR_PTR() on error. */
+static void *xs_talkv(enum xsd_sockmsg_type type,
+		      const struct kvec *iovec,
+		      unsigned int num_vecs,
+		      unsigned int *len)
+{
+	struct xsd_sockmsg msg;
+	void *ret = NULL;
+	unsigned int i;
+	int err;
+
+	//WARN_ON(down_trylock(&xenbus_lock) == 0);
+
+	msg.type = type;
+	msg.len = 0;
+	for (i = 0; i < num_vecs; i++)
+		msg.len += iovec[i].iov_len;
+
+	err = xb_write(&msg, sizeof(msg));
+	if (err)
+		return ERR_PTR(err);
+
+	for (i = 0; i < num_vecs; i++) {
+		err = xb_write(iovec[i].iov_base, iovec[i].iov_len);;
+		if (err)
+			return ERR_PTR(err);
+	}
+
+	/* Watches can have fired before reply comes: daemon detects
+	 * and re-transmits, so we can ignore this. */
+	do {
+		xfree(ret);
+		ret = read_reply(&msg.type, len);
+		if (IS_ERR(ret))
+			return ret;
+	} while (msg.type == XS_WATCH_EVENT);
+
+	if (msg.type == XS_ERROR) {
+		err = get_error(ret);
+		xfree(ret);
+		return ERR_PTR(-err);
+	}
+
+	//BUG_ON(msg.type != type);
+	return ret;
+}
+
+/* Simplified version of xs_talkv: single message. */
+static void *xs_single(enum xsd_sockmsg_type type,
+		       const char *string, unsigned int *len)
+{
+	struct kvec iovec;
+
+	iovec.iov_base = (void *)string;
+	iovec.iov_len = strlen(string) + 1;
+	return xs_talkv(type, &iovec, 1, len);
+}
+
+/* Many commands only need an ack, don't care what it says. */
+static int xs_error(char *reply)
+{
+	if (IS_ERR(reply))
+		return PTR_ERR(reply);
+	xfree(reply);
+	return 0;
+}
+
+static unsigned int count_strings(const char *strings, unsigned int len)
+{
+	unsigned int num;
+	const char *p;
+
+	for (p = strings, num = 0; p < strings + len; p += strlen(p) + 1)
+		num++;
+
+	return num;
+}
+
+/* Return the path to dir with /name appended. */ 
+static char *join(const char *dir, const char *name)
+{
+	static char buffer[4096];
+
+	//BUG_ON(down_trylock(&xenbus_lock) == 0);
+	/* XXX FIXME: might not be correct if name == "" */
+	//BUG_ON(strlen(dir) + strlen("/") + strlen(name) + 1 > sizeof(buffer));
+
+	strcpy(buffer, dir);
+	if (!streq(name, "")) {
+		strcat(buffer, "/");
+		strcat(buffer, name);
+	}
+	return buffer;
+}
+
+char **xenbus_directory(const char *dir, const char *node, unsigned int *num)
+{
+	char *strings, *p, **ret;
+	unsigned int len;
+
+	strings = xs_single(XS_DIRECTORY, join(dir, node), &len);
+	if (IS_ERR(strings))
+		return (char **)strings;
+
+	/* Count the strings. */
+	*num = count_strings(strings, len);
+
+	/* Transfer to one big alloc for easy freeing. */
+	ret = (char **)xmalloc_array(char, *num * sizeof(char *) + len);
+	if (!ret) {
+		xfree(strings);
+		return ERR_PTR(-ENOMEM);
+	}
+	memcpy(&ret[*num], strings, len);
+	xfree(strings);
+
+	strings = (char *)&ret[*num];
+	for (p = strings, *num = 0; p < strings + len; p += strlen(p) + 1)
+		ret[(*num)++] = p;
+	return ret;
+}
+
+/* Check if a path exists. Return 1 if it does. */
+int xenbus_exists(const char *dir, const char *node)
+{
+	char **d;
+	int dir_n;
+
+	d = xenbus_directory(dir, node, &dir_n);
+	if (IS_ERR(d))
+		return 0;
+	xfree(d);
+	return 1;
+}
+
+/* Get the value of a single file.
+ * Returns a kmalloced value: call free() on it after use.
+ * len indicates length in bytes.
+ */
+void *xenbus_read(const char *dir, const char *node, unsigned int *len)
+{
+	return xs_single(XS_READ, join(dir, node), len);
+}
+
+/* Write the value of a single file.
+ * Returns -err on failure.  createflags can be 0, O_CREAT, or O_CREAT|O_EXCL.
+ */
+int xenbus_write(const char *dir, const char *node,
+		 const char *string, int createflags)
+{
+	const char *flags, *path;
+	struct kvec iovec[3];
+
+	path = join(dir, node);
+	/* Format: Flags (as string), path, data. */
+	if (createflags == 0)
+		flags = XS_WRITE_NONE;
+	else if (createflags == O_CREAT)
+		flags = XS_WRITE_CREATE;
+	else if (createflags == (O_CREAT|O_EXCL))
+		flags = XS_WRITE_CREATE_EXCL;
+	else
+		return -EINVAL;
+
+	iovec[0].iov_base = (void *)path;
+	iovec[0].iov_len = strlen(path) + 1;
+	iovec[1].iov_base = (void *)flags;
+	iovec[1].iov_len = strlen(flags) + 1;
+	iovec[2].iov_base = (void *)string;
+	iovec[2].iov_len = strlen(string);
+
+	return xs_error(xs_talkv(XS_WRITE, iovec, ARRAY_SIZE(iovec), NULL));
+}
+
+/* Create a new directory. */
+int xenbus_mkdir(const char *dir, const char *node)
+{
+	return xs_error(xs_single(XS_MKDIR, join(dir, node), NULL));
+}
+
+/* Destroy a file or directory (directories must be empty). */
+int xenbus_rm(const char *dir, const char *node)
+{
+	return xs_error(xs_single(XS_RM, join(dir, node), NULL));
+}
+
+/* Start a transaction: changes by others will not be seen during this
+ * transaction, and changes will not be visible to others until end.
+ * Transaction only applies to the given subtree.
+ * You can only have one transaction at any time.
+ */
+int xenbus_transaction_start(const char *subtree)
+{
+	return xs_error(xs_single(XS_TRANSACTION_START, subtree, NULL));
+}
+
+/* End a transaction.
+ * If abandon is true, transaction is discarded instead of committed.
+ */
+int xenbus_transaction_end(int abort)
+{
+	char abortstr[2];
+
+	if (abort)
+		strcpy(abortstr, "F");
+	else
+		strcpy(abortstr, "T");
+	return xs_error(xs_single(XS_TRANSACTION_END, abortstr, NULL));
+}
+
+/* Single read and scanf: returns -errno or num scanned. */
+int xenbus_scanf(const char *dir, const char *node, const char *fmt, ...)
+{
+	va_list ap;
+	int ret;
+	char *val;
+
+	val = xenbus_read(dir, node, NULL);
+	if (IS_ERR(val))
+		return PTR_ERR(val);
+
+	va_start(ap, fmt);
+	ret = vsscanf(val, fmt, ap);
+	va_end(ap);
+	xfree(val);
+	/* Distinctive errno. */
+	if (ret == 0)
+		return -ERANGE;
+	return ret;
+}
+
+/* Single printf and write: returns -errno or 0. */
+int xenbus_printf(const char *dir, const char *node, const char *fmt, ...)
+{
+	va_list ap;
+	int ret;
+
+	//BUG_ON(down_trylock(&xenbus_lock) == 0);
+	va_start(ap, fmt);
+	ret = vsnprintf(printf_buffer, sizeof(printf_buffer), fmt, ap);
+	va_end(ap);
+
+	//BUG_ON(ret > sizeof(printf_buffer)-1);
+	return xenbus_write(dir, node, printf_buffer, O_CREAT);
+}
+
+	
+/* Takes tuples of names, scanf-style args, and void **, NULL terminated. */
+int xenbus_gather(const char *dir, ...)
+{
+	va_list ap;
+	const char *name;
+	int ret = 0;
+
+	va_start(ap, dir);
+	while (ret == 0 && (name = va_arg(ap, char *)) != NULL) {
+		const char *fmt = va_arg(ap, char *);
+		void *result = va_arg(ap, void *);
+		char *p;
+
+		p = xenbus_read(dir, name, NULL);
+		if (IS_ERR(p)) {
+			ret = PTR_ERR(p);
+			break;
+		}
+		if (fmt) {
+			if (sscanf(p, fmt, result) == 0)
+				ret = -EINVAL;
+			xfree(p);
+		} else
+			*(char **)result = p;
+	}
+	va_end(ap);
+	return ret;
+}
+
+static int xs_watch(const char *path, const char *token)
+{
+	struct kvec iov[2];
+
+	iov[0].iov_base = (void *)path;
+	iov[0].iov_len = strlen(path) + 1;
+	iov[1].iov_base = (void *)token;
+	iov[1].iov_len = strlen(token) + 1;
+
+	return xs_error(xs_talkv(XS_WATCH, iov, ARRAY_SIZE(iov), NULL));
+}
+
+static char *xs_read_watch(char **token)
+{
+	enum xsd_sockmsg_type type;
+	char *ret;
+
+	ret = read_reply(&type, NULL);
+	if (IS_ERR(ret))
+		return ret;
+
+	//BUG_ON(type != XS_WATCH_EVENT);
+	*token = ret + strlen(ret) + 1;
+	return ret;
+}
+
+static int xs_acknowledge_watch(const char *token)
+{
+	return xs_error(xs_single(XS_WATCH_ACK, token, NULL));
+}
+
+static int xs_unwatch(const char *path, const char *token)
+{
+	struct kvec iov[2];
+
+	iov[0].iov_base = (char *)path;
+	iov[0].iov_len = strlen(path) + 1;
+	iov[1].iov_base = (char *)token;
+	iov[1].iov_len = strlen(token) + 1;
+
+	return xs_error(xs_talkv(XS_UNWATCH, iov, ARRAY_SIZE(iov), NULL));
+}
+
+/* A little paranoia: we don't just trust token. */
+static struct xenbus_watch *find_watch(const char *token)
+{
+	struct xenbus_watch *i, *cmp;
+
+	cmp = (void *)simple_strtoul(token, NULL, 16);
+
+	list_for_each_entry(i, &watches, list)
+		if (i == cmp)
+			return i;
+	return NULL;
+}
+
+/* Register callback to watch this node. */
+int register_xenbus_watch(struct xenbus_watch *watch)
+{
+	/* Pointer in ascii is the token. */
+	char token[sizeof(watch) * 2 + 1];
+	int err;
+
+	sprintf(token, "%lX", (long)watch);
+	//BUG_ON(find_watch(token));
+printk("Registered watch for: %s\n", token);
+	err = xs_watch(watch->node, token);
+	if (!err)
+		list_add(&watch->list, &watches);
+	return err;
+}
+
+void unregister_xenbus_watch(struct xenbus_watch *watch)
+{
+	char token[sizeof(watch) * 2 + 1];
+	int err;
+
+	sprintf(token, "%lX", (long)watch);
+	//BUG_ON(!find_watch(token));
+
+	err = xs_unwatch(watch->node, token);
+	list_del(&watch->list);
+
+	if (err)
+		printk("XENBUS Failed to release watch %s: %i\n",
+		       watch->node, err);
+}
+
+/* Re-register callbacks to all watches. */
+void reregister_xenbus_watches(void)
+{
+	struct xenbus_watch *watch;
+	char token[sizeof(watch) * 2 + 1];
+
+	list_for_each_entry(watch, &watches, list) {
+		sprintf(token, "%lX", (long)watch);
+		xs_watch(watch->node, token);
+	}
+}
+
+void watch_thread(void *unused)
+{
+	for (;;) {
+		char *token;
+		char *node = NULL;
+
+		wait_event(xb_waitq, xs_input_avail());
+
+		/* If this is a spurious wakeup caused by someone
+		 * doing an op, they'll hold the lock and the buffer
+		 * will be empty by the time we get there.		 
+		 */
+		down(&xenbus_lock);
+		if (xs_input_avail())
+			node = xs_read_watch(&token);
+
+		if (node && !IS_ERR(node)) {
+			struct xenbus_watch *w;
+			int err;
+
+			err = xs_acknowledge_watch(token);
+			if (err)
+				printk("XENBUS ack %s fail %i\n", node, err);
+			w = find_watch(token);
+			//BUG_ON(!w);
+			w->callback(w, node);
+			xfree(node);
+		} else
+			printk("XENBUS xs_read_watch: %li\n", PTR_ERR(node));
+		up(&xenbus_lock);
+	}
+}
+
+
+static void ballon_changed(struct xenbus_watch *watch, const char *node)
+{
+    unsigned long new_target;
+    int err;
+    err = xenbus_scanf("memory", "target", "%lu", &new_target);
+
+    if(err != 1)
+    {
+        printk("Unable to read memory/target\n");
+        return;
+    }
+
+    printk("Memory target changed to: %ld bytes, ignoring.\n", new_target);
+}
+
+
+static struct xenbus_watch ballon_watch = {
+    .node = "memory/target",
+    .callback = ballon_changed,
+};
+
+
+
+int xs_init(void)
+{
+	int err;
+	struct thread *watcher;
+    printk("xb_init_comms\n");
+	err = xb_init_comms();
+	if (err)
+		return err;
+	
+	watcher = create_thread("kxwatch", watch_thread, NULL);
+    down(&xenbus_lock);
+    register_xenbus_watch(&ballon_watch);
+    up(&xenbus_lock);
+	return 0;
+}