bitkeeper revision 1.1713.3.5 (42b2a4e2r6SNlC_nq2hAkXEQjEFAmA)

Many files:
  - watch now takes a token, returned when reading watch
  - More tests
  - Fix domain shared page communication (flush output)
  - Add "home" path for domains
  - More permissions checks in various functions
  - Simplify watch acknowledgement code and fix occasional bug
xs_watch_stress.c, 12readonly.sh, 11domain-watch.sh, 10domain-homedir.sh:
  new file
xs_stress.c, xs_lib.h, xs_lib.c:
  Cleanup whitespace.
ignore:
  Add tools/xenstore/xs_watch_stress
Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>
Signed-off-by: Christian Limpach <Christian.Limpach@cl.cam.ac.uk>

23 files changed, 717 insertions, 216 deletions

diff --git a/.rootkeys b/.rootkeys
index 4d3b69442d..d2c3de4b04 100644
--- a/.rootkeys
+++ b/.rootkeys
@@ -1066,6 +1066,9 @@
 42a57d98fdO519YyATk4_Zwr1STNfQ tools/xenstore/testsuite/07watch.sh
 42a57d98zZUtvirUMjmHxFphJjmO7Q tools/xenstore/testsuite/08transaction.sh
 42a57d98sn9RbpBgHRv1D99Kt7LwYA tools/xenstore/testsuite/09domain.sh
+42b2a4bfxAwHlRgd31SJBgFnj8g3MA tools/xenstore/testsuite/10domain-homedir.sh
+42b2a4bfHbUp4IB8tfNIa8j37S27fw tools/xenstore/testsuite/11domain-watch.sh
+42b2a4bfhrB5v6uYKPj6jSO_Ng0PAA tools/xenstore/testsuite/12readonly.sh
 42a57d98tSuoFCHnnM2GgENXJrRQmw tools/xenstore/testsuite/test.sh
 42a57d98zxDP2Ti7dTznGROi66rUGw tools/xenstore/utils.c
 42a57d98SDvOYCEjmCjwHSk6390GLA tools/xenstore/utils.h
@@ -1087,6 +1090,7 @@
 42a57d99Kl6Ba8oCHv2fggl7QN9QZA tools/xenstore/xs_random.c
 42a57d99SHYR1lQOD0shuErPDg9NKQ tools/xenstore/xs_stress.c
 42a57d996aBawpkQNOWkNWXD6LrhPg tools/xenstore/xs_test.c
+42b2a4bfp-lhxBfenUyHlvw7bPcVgA tools/xenstore/xs_watch_stress.c
 403a3edbrr8RE34gkbR40zep98SXbg tools/xentrace/Makefile
 40a107afN60pFdURgBv9KwEzgRl5mQ tools/xentrace/formats
 420d52d2_znVbT4JAPIU36vQOme83g tools/xentrace/xenctx.c
diff --git a/BitKeeper/etc/ignore b/BitKeeper/etc/ignore
index 2bd05d6e4f..a6d5a390a0 100644
--- a/BitKeeper/etc/ignore
+++ b/BitKeeper/etc/ignore
@@ -148,6 +148,7 @@ tools/xenstore/xs_dom0_test
 tools/xenstore/xs_random
 tools/xenstore/xs_stress
 tools/xenstore/xs_test
+tools/xenstore/xs_watch_stress
 tools/xentrace/xentrace
 tools/xfrd/xfrd
 xen/BLOG
diff --git a/tools/python/xen/lowlevel/xs/xs.c b/tools/python/xen/lowlevel/xs/xs.c
index 0da0fbcb3e..9c26c0f51c 100644
--- a/tools/python/xen/lowlevel/xs/xs.c
+++ b/tools/python/xen/lowlevel/xs/xs.c
@@ -277,10 +277,11 @@ static PyObject *xspy_set_permissions(PyObject *self, PyObject *args,
 
 static PyObject *xspy_watch(PyObject *self, PyObject *args, PyObject *kwds)
 {
-    static char *kwd_spec[] = { "path", "priority", NULL };
-    static char *arg_spec = "s|i";
+    static char *kwd_spec[] = { "path", "priority", "token", NULL };
+    static char *arg_spec = "s|is";
     char *path = NULL;
     int priority = 0;
+    char *token;
 
     struct xs_handle *xh = xshandle(self);
     PyObject *val = NULL;
@@ -289,9 +290,9 @@ static PyObject *xspy_watch(PyObject *self, PyObject *args, PyObject *kwds)
     if (!xh)
 	goto exit;
     if (!PyArg_ParseTupleAndKeywords(args, kwds, arg_spec, kwd_spec, 
-                                     &path, &priority))
+                                     &path, &priority, &token))
         goto exit;
-    xsval = xs_watch(xh, path, priority);
+    xsval = xs_watch(xh, path, token, priority);
     val = pyvalue_int(xsval);
  exit:
     return val;
@@ -305,14 +306,19 @@ static PyObject *xspy_read_watch(PyObject *self, PyObject *args,
 
     struct xs_handle *xh = xshandle(self);
     PyObject *val = NULL;
-    char *xsval = NULL;
+    char **xsval = NULL;
 
     if (!xh)
 	goto exit;
     if (!PyArg_ParseTupleAndKeywords(args, kwds, arg_spec, kwd_spec))
         goto exit;
     xsval = xs_read_watch(xh);
-    val = pyvalue_str(xsval);
+    if(!xsval){
+            val = PyErr_SetFromErrno(PyExc_RuntimeError);
+            goto exit;
+    }
+    /* Create tuple (path, token). */
+    val = Py_BuildValue("(ss)", xsval[0], xsval[1]);
  exit:
     if (xsval)
 	free(xsval);
@@ -323,7 +329,8 @@ static PyObject *xspy_acknowledge_watch(PyObject *self, PyObject *args,
 					PyObject *kwds)
 {
     static char *kwd_spec[] = { NULL };
-    static char *arg_spec = "";
+    static char *arg_spec = "s";
+    char *token = "";
 
     struct xs_handle *xh = xshandle(self);
     PyObject *val = NULL;
@@ -331,9 +338,9 @@ static PyObject *xspy_acknowledge_watch(PyObject *self, PyObject *args,
 
     if (!xh)
 	goto exit;
-    if (!PyArg_ParseTupleAndKeywords(args, kwds, arg_spec, kwd_spec))
+    if (!PyArg_ParseTupleAndKeywords(args, kwds, arg_spec, kwd_spec, &token))
         goto exit;
-    xsval = xs_acknowledge_watch(xh);
+    xsval = xs_acknowledge_watch(xh, token);
     val = pyvalue_int(xsval);
  exit:
     return val;
@@ -341,9 +348,10 @@ static PyObject *xspy_acknowledge_watch(PyObject *self, PyObject *args,
 
 static PyObject *xspy_unwatch(PyObject *self, PyObject *args, PyObject *kwds)
 {
-    static char *kwd_spec[] = { "path", NULL };
-    static char *arg_spec = "s|";
+    static char *kwd_spec[] = { "path", "token", NULL };
+    static char *arg_spec = "s|s";
     char *path = NULL;
+    char *token = "";
 
     struct xs_handle *xh = xshandle(self);
     PyObject *val = NULL;
@@ -351,9 +359,10 @@ static PyObject *xspy_unwatch(PyObject *self, PyObject *args, PyObject *kwds)
 
     if (!xh)
 	goto exit;
-    if (!PyArg_ParseTupleAndKeywords(args, kwds, arg_spec, kwd_spec, &path))
+    if (!PyArg_ParseTupleAndKeywords(args, kwds, arg_spec, kwd_spec, &path,
+				     &token))
         goto exit;
-    xsval = xs_unwatch(xh, path);
+    xsval = xs_unwatch(xh, path, token);
     val = pyvalue_int(xsval);
  exit:
     return val;
diff --git a/tools/xenstore/Makefile b/tools/xenstore/Makefile
index 0a78cfbad7..494a7f20d3 100644
--- a/tools/xenstore/Makefile
+++ b/tools/xenstore/Makefile
@@ -41,8 +41,9 @@ xenstored_test: xenstored_core_test.o xenstored_watch_test.o xenstored_domain_te
 xs_test: xs_test.o xs_lib.o utils.o
 xs_random: xs_random.o xs_test_lib.o xs_lib.o talloc.o utils.o
 xs_stress: xs_stress.o xs_test_lib.o xs_lib.o talloc.o utils.o
+xs_watch_stress: xs_watch_stress.o xs_test_lib.o xs_lib.o talloc.o utils.o
 
-xs_test.o xs_stress.o xenstored_core_test.o xenstored_watch_test.o xenstored_transaction_test.o xenstored_domain_test.o xs_random.o xs_test_lib.o talloc_test.o fake_libxc.o: CFLAGS=$(BASECFLAGS) $(TESTFLAGS)
+xs_test.o xs_stress.o xs_watch_stress.o xenstored_core_test.o xenstored_watch_test.o xenstored_transaction_test.o xenstored_domain_test.o xs_random.o xs_test_lib.o talloc_test.o fake_libxc.o: CFLAGS=$(BASECFLAGS) $(TESTFLAGS)
 
 xenstored_%_test.o: xenstored_%.c
 	$(COMPILE.c) -o $@ $<
@@ -63,8 +64,9 @@ libxenstore.a: $(LIB_OBJS_A)
 libxenstore-pic.a: $(LIB_OBJS_PIC)
 
 clean: testsuite-clean
-	rm -f *.o *.opic *.a xen
-	rm -f xs_test xenstored xenstored_test xs_random xs_stress xs_dom0_test
+	rm -f *.o *.opic *.a
+	rm -f xen xenstored xs_random xs_stress xs_watch_stress
+	rm -f xs_test xenstored_test xs_dom0_test
 	-$(RM) $(PROG_DEP)
 
 check: testsuite-run randomcheck stresstest
@@ -83,9 +85,11 @@ randomcheck: xs_random xenstored_test
 	$(TESTENV) ./xs_random --fast /tmp/xs_random 100000 $(RANDSEED)
 	$(TESTENV) ./xs_random --fail /tmp/xs_random 10000 $(RANDSEED)
 
-stresstest: xs_stress xenstored_test
+stresstest: xs_stress xs_watch_stress xenstored_test
 	rm -rf $(TESTDIR)/store
-	export $(TESTENV); PID=`./xenstored_test --output-pid`; ./xs_stress 10000; ret=$$?; kill $$PID; exit $$ret
+	export $(TESTENV); PID=`./xenstored_test --output-pid`; ./xs_stress 5000; ret=$$?; kill $$PID; exit $$ret
+	rm -rf $(TESTDIR)/store
+	export $(TESTENV); PID=`./xenstored_test --output-pid`; ./xs_watch_stress; ret=$$?; kill $$PID; exit $$ret
 
 xs_dom0_test: xs_dom0_test.o utils.o
 	$(LINK.o) $^ $(LOADLIBES) $(LDLIBS) -lxc -o $@
diff --git a/tools/xenstore/testsuite/07watch.sh b/tools/xenstore/testsuite/07watch.sh
index bedce6ad5b..e6156ea7f8 100644
--- a/tools/xenstore/testsuite/07watch.sh
+++ b/tools/xenstore/testsuite/07watch.sh
@@ -3,30 +3,118 @@
 # Watch something, write to it, check watch has fired.
 [ "`echo -e 'write /test create contents' | ./xs_test 2>&1`" = "" ]
 
-[ "`echo -e '1 watch /test 100\n2 write /test create contents2\n1 waitwatch\n1 ackwatch' | ./xs_test 2>&1`" = "1:/test" ]
+[ "`echo -e '1 watch /test token 100
+2 write /test create contents2
+1 waitwatch
+1 ackwatch token' | ./xs_test 2>&1`" = "1:/test:token" ]
 
 # Check that reads don't set it off.
-[ "`echo -e '1 watch /test 100\n2 read /test\n1 waitwatch' | ./xs_test 2>&1`" = "2:contents2
+[ "`echo -e '1 watch /test token 100
+2 read /test
+1 waitwatch' | ./xs_test 2>&1`" = "2:contents2
 1:waitwatch timeout" ]
 
-# mkdir, setperm and rm should (also /tests watching dirs)
+# mkdir, setperm and rm should (also tests watching dirs)
 [ "`echo -e 'mkdir /dir' | ./xs_test 2>&1`" = "" ]
-[ "`echo -e '1 watch /dir 100\n2 mkdir /dir/newdir\n1 waitwatch\n1 ackwatch\n2 setperm /dir/newdir 0 READ\n1 waitwatch\n1 ackwatch\n2 rm /dir/newdir\n1 waitwatch\n1 ackwatch' | ./xs_test 2>&1`" = "1:/dir/newdir
-1:/dir/newdir
-1:/dir/newdir" ]
+[ "`echo -e '1 watch /dir token 100
+2 mkdir /dir/newdir
+1 waitwatch
+1 ackwatch token
+2 setperm /dir/newdir 0 READ
+1 waitwatch
+1 ackwatch token
+2 rm /dir/newdir
+1 waitwatch
+1 ackwatch token' | ./xs_test 2>&1`" = "1:/dir/newdir:token
+1:/dir/newdir:token
+1:/dir/newdir:token" ]
 
 # ignore watches while doing commands, should work.
-[ "`echo -e 'watch /dir 100\nwrite /dir/test create contents\nread /dir/test\nwaitwatch\nackwatch' | ./xs_test 2>&1`" = "contents
-/dir/test" ]
+[ "`echo -e 'watch /dir token 100
+write /dir/test create contents
+read /dir/test
+waitwatch
+ackwatch token' | ./xs_test 2>&1`" = "contents
+/dir/test:token" ]
 
 # watch priority /test.
-[ "`echo -e '1 watch /dir 1\n3 watch /dir 3\n2 watch /dir 2\nwrite /dir/test create contents\n3 waitwatch\n3 ackwatch\n2 waitwatch\n2 ackwatch\n1 waitwatch\n1 ackwatch' | ./xs_test 2>&1`" = "3:/dir/test
-2:/dir/test
-1:/dir/test" ]
+[ "`echo -e '1 watch /dir token1 1
+3 watch /dir token3 3
+2 watch /dir token2 2
+write /dir/test create contents
+3 waitwatch
+3 ackwatch token3
+2 waitwatch
+2 ackwatch token2
+1 waitwatch
+1 ackwatch token1' | ./xs_test 2>&1`" = "3:/dir/test:token3
+2:/dir/test:token2
+1:/dir/test:token1" ]
 
 # If one dies (without acking), the other should still get ack.
-[ "`echo -e '1 watch /dir 0\n2 watch /dir 1\nwrite /dir/test create contents\n2 waitwatch\n2 close\n1 waitwatch\n1 ackwatch' | ./xs_test 2>&1`" = "2:/dir/test
-1:/dir/test" ]
+[ "`echo -e '1 watch /dir token1 0
+2 watch /dir token2 1
+write /dir/test create contents
+2 waitwatch
+2 close
+1 waitwatch
+1 ackwatch token1' | ./xs_test 2>&1`" = "2:/dir/test:token2
+1:/dir/test:token1" ]
 
 # If one dies (without reading at all), the other should still get ack.
-[ "`echo -e '1 watch /dir 0\n2 watch /dir 1\nwrite /dir/test create contents\n2 close\n1 waitwatch\n1 ackwatch' | ./xs_test 2>&1`" = "1:/dir/test" ]
+[ "`echo -e '1 watch /dir token1 0
+2 watch /dir token2 1
+write /dir/test create contents
+2 close
+1 waitwatch
+1 ackwatch token1' | ./xs_test 2>&1`" = "1:/dir/test:token1" ]
+
+# unwatch
+[ "`echo -e '1 watch /dir token1 0
+1 unwatch /dir token1
+1 watch /dir token2 0
+2 write /dir/test2 create contents
+1 waitwatch
+1 unwatch /dir token2' | ./xs_test 2>&1`" = "1:/dir/test2:token2" ]
+
+# unwatch while watch pending.
+[ "`echo -e '1 watch /dir token1 0
+2 watch /dir token2 1
+write /dir/test create contents
+2 unwatch /dir token2
+1 waitwatch
+1 ackwatch token1' | ./xs_test 2>&1`" = "1:/dir/test:token1" ]
+
+# check we only get notified once.
+[ "`echo -e '1 watch /test token 100
+2 write /test create contents2
+1 waitwatch
+1 ackwatch token
+1 waitwatch' | ./xs_test 2>&1`" = "1:/test:token
+1:waitwatch timeout" ]
+
+# watches are queued in order.
+[ "`echo -e '1 watch / token 100
+2 write /test1 create contents
+2 write /test2 create contents
+2 write /test3 create contents
+1 waitwatch
+1 ackwatch token
+1 waitwatch
+1 ackwatch token
+1 waitwatch
+1 ackwatch token' | ./xs_test 2>&1`" = "1:/test1:token
+1:/test2:token
+1:/test3:token" ]
+
+# Creation of subpaths should be covered correctly.
+[ "`echo -e '1 watch / token 100
+2 write /test/subnode create contents2
+2 write /test/subnode/subnode create contents2
+1 waitwatch
+1 ackwatch token
+1 waitwatch
+1 ackwatch token
+1 waitwatch' | ./xs_test 2>&1`" = "1:/test/subnode:token
+1:/test/subnode/subnode:token
+1:waitwatch timeout" ]
diff --git a/tools/xenstore/testsuite/10domain-homedir.sh b/tools/xenstore/testsuite/10domain-homedir.sh
new file mode 100644
index 0000000000..a3587d2a5e
--- /dev/null
+++ b/tools/xenstore/testsuite/10domain-homedir.sh
@@ -0,0 +1,12 @@
+#! /bin/sh
+# Test domain "implicit" paths.
+
+# Create a domain, write an entry using implicit path, read using implicit
+[ "`echo -e 'mkdir /home
+introduce 1 100 7 /home
+1 write entry1 create contents
+read /home/entry1
+dir /home' | ./xs_test 2>&1`" = "handle is 1
+contents
+entry1" ]
+
diff --git a/tools/xenstore/testsuite/11domain-watch.sh b/tools/xenstore/testsuite/11domain-watch.sh
new file mode 100644
index 0000000000..f42fb5f8c6
--- /dev/null
+++ b/tools/xenstore/testsuite/11domain-watch.sh
@@ -0,0 +1,51 @@
+#! /bin/sh
+# Test watching from a domain.
+
+# Watch something, write to it, check watch has fired.
+[ "`echo -e 'write /test create contents' | ./xs_test 2>&1`" = "" ]
+[ "`echo -e 'mkdir /dir' | ./xs_test 2>&1`" = "" ]
+
+[ "`echo -e 'introduce 1 100 7 /my/home
+1 watch /test token 100
+write /test create contents2
+1 waitwatch
+1 ackwatch token
+1 unwatch /test token
+release 1' | ./xs_test 2>&1`" = "handle is 1
+1:/test:token" ]
+
+# ignore watches while doing commands, should work.
+[ "`echo -e 'introduce 1 100 7 /my/home
+1 watch /dir token 100
+1 write /dir/test create contents
+1 read /dir/test
+1 waitwatch
+1 ackwatch token
+release 1' | ./xs_test 2>&1`" = "handle is 1
+1:contents
+1:/dir/test:token" ]
+
+# unwatch
+[ "`echo -e 'introduce 1 100 7 /my/home
+1 watch /dir token1 0
+1 unwatch /dir token1
+1 watch /dir token2 0
+2 write /dir/test2 create contents
+1 waitwatch
+1 unwatch /dir token2
+release 1' | ./xs_test 2>&1`" = "handle is 1
+1:/dir/test2:token2" ]
+
+# unwatch while watch pending.
+[ "`echo -e 'introduce 1 100 7 /my/home
+introduce 2 101 8 /my/secondhome
+1 watch /dir token1 0
+2 watch /dir token2 1
+write /dir/test create contents
+2 unwatch /dir token2
+1 waitwatch
+1 ackwatch token1
+release 1
+release 2' | ./xs_test 2>&1`" = "handle is 1
+handle is 2
+1:/dir/test:token1" ]
diff --git a/tools/xenstore/testsuite/12readonly.sh b/tools/xenstore/testsuite/12readonly.sh
new file mode 100644
index 0000000000..5e7501a2e6
--- /dev/null
+++ b/tools/xenstore/testsuite/12readonly.sh
@@ -0,0 +1,40 @@
+#! /bin/sh
+# Test that read only connection can't alter store.
+
+[ "`echo 'write /test create contents' | ./xs_test 2>&1`" = "" ]
+
+# These are all valid.
+[ "`echo 'dir /
+read /test
+getperm /test
+watch /test token 0
+unwatch /test token 
+start /
+commit
+start /
+abort' | ./xs_test --readonly 2>&1`" = "test
+contents
+0 NONE" ]
+
+# These don't work
+[ "`echo 'write /test2 create contents' | ./xs_test --readonly 2>&1`" = "FATAL: write: Read-only file system" ]
+[ "`echo 'write /test create contents' | ./xs_test --readonly 2>&1`" = "FATAL: write: Read-only file system" ]
+[ "`echo 'setperm /test 100 NONE' | ./xs_test --readonly 2>&1`" = "FATAL: setperm: Read-only file system" ]
+[ "`echo 'setperm /test 100 NONE' | ./xs_test --readonly 2>&1`" = "FATAL: setperm: Read-only file system" ]
+[ "`echo 'shutdown' | ./xs_test --readonly 2>&1`" = "FATAL: shutdown: Read-only file system" ]
+[ "`echo 'introduce 1 100 7 /home' | ./xs_test --readonly 2>&1`" = "FATAL: introduce: Read-only file system" ]
+
+# Check that watches work like normal.
+set -m
+[ "`echo 'watch / token 0
+waitwatch
+ackwatch token' | ./xs_test --readonly 2>&1`" = "/test:token" ] &
+
+[ "`echo 'write /test create contents' | ./xs_test 2>&1`" = "" ]
+if wait; then :; else
+    echo Readonly wait test failed: $?
+    exit 1
+fi
+    
+    
+
diff --git a/tools/xenstore/testsuite/test.sh b/tools/xenstore/testsuite/test.sh
index 5718e84a15..3f0055842d 100755
--- a/tools/xenstore/testsuite/test.sh
+++ b/tools/xenstore/testsuite/test.sh
@@ -9,7 +9,7 @@ run_test()
     mkdir $XENSTORED_ROOTDIR
 # Weird failures with this.
     if type valgrind >/dev/null 2>&1; then
-	valgrind -q --logfile-fd=3 ./xenstored_test --output-pid --no-fork 3>testsuite/tmp/vgout > /tmp/pid &
+	valgrind -q --logfile-fd=3 ./xenstored_test --output-pid --no-fork 3>testsuite/tmp/vgout > /tmp/pid 2> testsuite/tmp/xenstored_errors &
 	while [ ! -s /tmp/pid ]; do sleep 0; done
 	PID=`cat /tmp/pid`
 	rm /tmp/pid
diff --git a/tools/xenstore/xenstored_core.c b/tools/xenstore/xenstored_core.c
index 1df00f37b4..9f7ff2e1ce 100644
--- a/tools/xenstore/xenstored_core.c
+++ b/tools/xenstore/xenstored_core.c
@@ -122,6 +122,33 @@ void __attribute__((noreturn)) corrupt(struct connection *conn,
 	_exit(2);
 }
 
+static char *sockmsg_string(enum xsd_sockmsg_type type)
+{
+	switch (type) {
+	case XS_DEBUG: return "DEBUG";
+	case XS_SHUTDOWN: return "SHUTDOWN";
+	case XS_DIRECTORY: return "DIRECTORY";
+	case XS_READ: return "READ";
+	case XS_GET_PERMS: return "GET_PERMS";
+	case XS_WATCH: return "WATCH";
+	case XS_WATCH_ACK: return "WATCH_ACK";
+	case XS_UNWATCH: return "UNWATCH";
+	case XS_TRANSACTION_START: return "TRANSACTION_START";
+	case XS_TRANSACTION_END: return "TRANSACTION_END";
+	case XS_INTRODUCE: return "INTRODUCE";
+	case XS_RELEASE: return "RELEASE";
+	case XS_GETDOMAINPATH: return "GETDOMAINPATH";
+	case XS_WRITE: return "WRITE";
+	case XS_MKDIR: return "MKDIR";
+	case XS_RM: return "RM";
+	case XS_SET_PERMS: return "SET_PERMS";
+	case XS_WATCH_EVENT: return "WATCH_EVENT";
+	case XS_ERROR: return "ERROR";
+	default:
+		return "**UNKNOWN**";
+	}
+}
+
 static bool write_message(struct connection *conn)
 {
 	int ret;
@@ -129,8 +156,9 @@ static bool write_message(struct connection *conn)
 
 	if (out->inhdr) {
 		if (verbose)
-			xprintf("Writing msg %i out to %p\n",
-				out->hdr.msg.type, conn);
+			xprintf("Writing msg %s (%s) out to %p\n",
+				sockmsg_string(out->hdr.msg.type),
+				out->buffer, conn);
 		ret = conn->write(conn, out->hdr.raw + out->used,
 				  sizeof(out->hdr) - out->used);
 		if (ret < 0)
@@ -148,9 +176,6 @@ static bool write_message(struct connection *conn)
 			return true;
 	}
 
-	if (verbose)
-		xprintf("Writing data len %i out to %p\n",
-			out->hdr.msg.len, conn);
 	ret = conn->write(conn, out->buffer + out->used,
 			  out->hdr.msg.len - out->used);
 
@@ -162,10 +187,7 @@ static bool write_message(struct connection *conn)
 		return true;
 
 	conn->out = NULL;
-
-	/* If this was an event, we wait for ack, otherwise we're done. */
-	if (!is_watch_event(conn, out))
-		talloc_free(out);
+	talloc_free(out);
 
 	queue_next_event(conn);
 	return true;
@@ -402,7 +424,7 @@ static bool valid_chars(const char *node)
 		       "0123456789-/_@") == strlen(node));
 }
 
-static bool is_valid_nodename(const char *node)
+bool is_valid_nodename(const char *node)
 {
 	/* Must start in /. */
 	if (!strstarts(node, "/"))
@@ -601,17 +623,24 @@ static int check_with_parents(struct connection *conn, const char *node,
 	return errnum;
 }
 
+char *canonicalize(struct connection *conn, const char *node)
+{
+	const char *prefix;
+
+	if (!node || strstarts(node, "/"))
+		return (char *)node;
+	prefix = get_implicit_path(conn);
+	if (prefix)
+		return talloc_asprintf(node, "%s/%s", prefix, node);
+	return (char *)node;
+}
+
 bool check_node_perms(struct connection *conn, const char *node,
 		      enum xs_perm_type perm)
 {
 	struct xs_permissions *perms;
 	unsigned int num;
 
-	if (!node) {
-		errno = EINVAL;
-		return false;
-	}
-
 	if (!node || !is_valid_nodename(node)) {
 		errno = EINVAL;
 		return false;
@@ -651,6 +680,7 @@ static bool send_directory(struct connection *conn, const char *node)
 	DIR *dir;
 	struct dirent *dirent;
 
+	node = canonicalize(conn, node);
 	if (!check_node_perms(conn, node, XS_PERM_READ))
 		return send_error(conn, errno);
 
@@ -680,6 +710,7 @@ static bool do_read(struct connection *conn, const char *node)
 	unsigned int size;
 	int *fd;
 
+	node = canonicalize(conn, node);
 	if (!check_node_perms(conn, node, XS_PERM_READ))
 		return send_error(conn, errno);
 
@@ -750,7 +781,7 @@ static bool do_write(struct connection *conn, struct buffered_data *in)
 	if (get_strings(in, vec, ARRAY_SIZE(vec)) < ARRAY_SIZE(vec))
 		return send_error(conn, EINVAL);
 
-	node = vec[0];
+	node = canonicalize(conn, vec[0]);
 	if (!within_transaction(conn->transaction, node))
 		return send_error(conn, EROFS);
 
@@ -804,6 +835,7 @@ static bool do_write(struct connection *conn, struct buffered_data *in)
 
 static bool do_mkdir(struct connection *conn, const char *node)
 {
+	node = canonicalize(conn, node);
 	if (!check_node_perms(conn, node, XS_PERM_WRITE|XS_PERM_CREATE))
 		return send_error(conn, errno);
 
@@ -826,6 +858,7 @@ static bool do_rm(struct connection *conn, const char *node)
 {
 	char *tmppath, *path;
 
+	node = canonicalize(conn, node);
 	if (!check_node_perms(conn, node, XS_PERM_WRITE))
 		return send_error(conn, errno);
 
@@ -848,6 +881,7 @@ static bool do_rm(struct connection *conn, const char *node)
 
 	add_change_node(conn->transaction, node);
 	send_ack(conn, XS_RM);
+	/* FIXME: traverse and fire watches for ALL of them! */
 	fire_watches(conn->transaction, node);
 	return false;
 }
@@ -858,6 +892,7 @@ static bool do_get_perms(struct connection *conn, const char *node)
 	char *strings;
 	unsigned int len, num;
 
+	node = canonicalize(conn, node);
 	if (!check_node_perms(conn, node, XS_PERM_READ))
 		return send_error(conn, errno);
 
@@ -883,7 +918,7 @@ static bool do_set_perms(struct connection *conn, struct buffered_data *in)
 		return send_error(conn, EINVAL);
 
 	/* First arg is node name. */
-	node = in->buffer;
+	node = canonicalize(conn, in->buffer);
 	in->buffer += strlen(in->buffer) + 1;
 	num--;
 
@@ -968,10 +1003,10 @@ static bool process_message(struct connection *conn, struct buffered_data *in)
 		return do_watch(conn, in);
 
 	case XS_WATCH_ACK:
-		return do_watch_ack(conn);
+		return do_watch_ack(conn, onearg(in));
 
 	case XS_UNWATCH:
-		return do_unwatch(conn, onearg(in));
+		return do_unwatch(conn, in);
 
 	case XS_TRANSACTION_START:
 		return do_transaction_start(conn, onearg(in));
@@ -1015,13 +1050,13 @@ static void consider_message(struct connection *conn)
 	}
 
 	if (verbose)
-		xprintf("Got message %i len %i from %p\n",
-			type, conn->in->hdr.msg.len, conn);
+		xprintf("Got message %s len %i from %p\n",
+			sockmsg_string(type), conn->in->hdr.msg.len, conn);
 
 	/* We might get a command while waiting for an ack: this means
 	 * the other end discarded it: we will re-transmit. */
 	if (type != XS_WATCH_ACK)
-		reset_watch_event(conn);
+		conn->waiting_for_ack = false;
 
 	/* Careful: process_message may free connection.  We detach
 	 * "in" beforehand and allocate the new buffer to avoid
@@ -1136,7 +1171,6 @@ struct connection *new_connection(connwritefn_t *write, connreadfn_t *read)
 
 	new->blocked = false;
 	new->out = new->waiting_reply = NULL;
-	new->event = NULL;
 	new->fd = -1;
 	new->id = 0;
 	new->domain = NULL;
@@ -1203,6 +1237,42 @@ static void time_relative_to_now(struct timeval *tv)
 	}
 }
 
+#ifdef TESTING
+/* Useful for running under debugger. */
+void dump_connection(void)
+{
+	struct connection *i;
+
+	list_for_each_entry(i, &connections, list) {
+		printf("Connection %p:\n", i);
+		if (i->id)
+			printf("    id = %i\n", i->id);
+		if (i->blocked)
+			printf("    blocked on = %s\n", i->blocked);
+		if (i->waiting_for_ack)
+			printf("    waiting_for_ack TRUE\n");
+		if (!i->in->inhdr || i->in->used)
+			printf("    got %i bytes of %s\n",
+			       i->in->used, i->in->inhdr ? "header" : "data");
+		if (i->out)
+			printf("    sending message %s (%s) out\n",
+			       sockmsg_string(i->out->hdr.msg.type),
+			       i->out->buffer);
+		if (i->waiting_reply)
+			printf("    ... and behind is queued %s (%s)\n",
+			       sockmsg_string(i->waiting_reply->hdr.msg.type),
+			       i->waiting_reply->buffer);
+#if 0
+		if (i->transaction)
+			dump_transaction(i);
+		if (i->domain)
+			dump_domain(i);
+#endif
+		dump_watches(i);
+	}
+}
+#endif
+
 static struct option options[] = { { "no-fork", 0, NULL, 'N' },
 				   { "verbose", 0, NULL, 'V' },
 				   { "output-pid", 0, NULL, 'P' },
@@ -1314,6 +1384,7 @@ int main(int argc, char *argv[])
 
 		timerclear(&tv);
 		shortest_transaction_timeout(&tv);
+		shortest_watch_ack_timeout(&tv);
 		if (timerisset(&tv)) {
 			time_relative_to_now(&tv);
 			tvp = &tv;
@@ -1351,8 +1422,15 @@ int main(int argc, char *argv[])
 			}
 		}
 
-		if (tvp)
+		/* Flush output for domain connections,  */
+		list_for_each_entry(i, &connections, list)
+			if (i->domain && i->out)
+				handle_output(i);
+
+		if (tvp) {
 			check_transaction_timeout();
+			check_watch_ack_timeout();
+		}
 
 		/* If transactions ended, we might be able to do more work. */
 		unblock_connections();
diff --git a/tools/xenstore/xenstored_core.h b/tools/xenstore/xenstored_core.h
index 0d0ebcaae0..a82ae8b22e 100644
--- a/tools/xenstore/xenstored_core.h
+++ b/tools/xenstore/xenstored_core.h
@@ -16,8 +16,10 @@
     along with this program; if not, write to the Free Software
     Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
-#ifndef _XENSTORED_INTERNAL_H
-#define _XENSTORED_INTERNAL_H
+
+#ifndef _XENSTORED_CORE_H
+#define _XENSTORED_CORE_H
+
 #include <stdbool.h>
 #include <stdint.h>
 #include <errno.h>
@@ -59,8 +61,8 @@ struct connection
 	/* Is this a read-only connection? */
 	bool can_write;
 
-	/* Our current event.  If all used, we're waiting for ack. */
-	struct watch_event *event;
+	/* Are we waiting for a watch event ack? */
+	bool waiting_for_ack;
 
 	/* Buffered incoming data. */
 	struct buffered_data *in;
@@ -105,6 +107,9 @@ bool send_ack(struct connection *conn, enum xsd_sockmsg_type type);
 /* Send an error: error is usually "errno". */
 bool send_error(struct connection *conn, int error);
 
+/* Canonicalize this path if possible. */
+char *canonicalize(struct connection *conn, const char *node);
+
 /* Check permissions on this node. */
 bool check_node_perms(struct connection *conn, const char *node,
 		      enum xs_perm_type perm);
@@ -121,6 +126,10 @@ struct connection *new_connection(connwritefn_t *write, connreadfn_t *read);
 void handle_input(struct connection *conn);
 void handle_output(struct connection *conn);
 
+/* Is this a valid node name? */
+bool is_valid_nodename(const char *node);
+
 /* Convenient talloc-style destructor for paths. */
 int destroy_path(void *path);
-#endif /* _XENSTORED_INTERNAL_H */
+
+#endif /* _XENSTORED_CORE_H */
diff --git a/tools/xenstore/xenstored_domain.c b/tools/xenstore/xenstored_domain.c
index a6f69ddf5b..8b154c49e4 100644
--- a/tools/xenstore/xenstored_domain.c
+++ b/tools/xenstore/xenstored_domain.c
@@ -65,11 +65,6 @@ struct domain
 
 static LIST_HEAD(domains);
 
-void domain_set_conn(struct domain *domain, struct connection *conn)
-{
-	domain->conn = conn;
-}
-
 struct ringbuf_head
 {
 	u32 write; /* Next place to write to */
@@ -268,6 +263,9 @@ bool do_introduce(struct connection *conn, struct buffered_data *in)
 	if (get_strings(in, vec, ARRAY_SIZE(vec)) < ARRAY_SIZE(vec))
 		return send_error(conn, EINVAL);
 
+	if (conn->id != 0)
+		return send_error(conn, EACCES);
+
 	if (!conn->can_write)
 		return send_error(conn, EROFS);
 
@@ -275,10 +273,9 @@ bool do_introduce(struct connection *conn, struct buffered_data *in)
 	domain = talloc(in, struct domain);
 	domain->domid = atoi(vec[0]);
 	domain->port = atoi(vec[2]);
-	domain->path = talloc_strdup(domain, vec[3]);
-	talloc_set_destructor(domain, destroy_domain);
-	if (!domain->port || !domain->domid)
+	if (!domain->port || !domain->domid || !is_valid_nodename(vec[3]))
 		return send_error(conn, EINVAL);
+	domain->path = talloc_strdup(domain, vec[3]);
 	domain->page = xc_map_foreign_range(*xc_handle, domain->domid,
 					    getpagesize(),
 					    PROT_READ|PROT_WRITE,
@@ -286,6 +283,9 @@ bool do_introduce(struct connection *conn, struct buffered_data *in)
 	if (!domain->page)
 		return send_error(conn, errno);
 
+	list_add(&domain->list, &domains);
+	talloc_set_destructor(domain, destroy_domain);
+
 	/* One in each half of page. */
 	domain->input = domain->page;
 	domain->output = domain->page + getpagesize()/2;
@@ -298,7 +298,6 @@ bool do_introduce(struct connection *conn, struct buffered_data *in)
 	domain->conn->domain = domain;
 
 	talloc_steal(domain->conn, domain);
-	list_add(&domain->list, &domains);
 
 	return send_ack(conn, XS_INTRODUCE);
 }
@@ -327,6 +326,9 @@ bool do_release(struct connection *conn, const char *domid_str)
 	if (!domid)
 		return send_error(conn, EINVAL);
 
+	if (conn->id != 0)
+		return send_error(conn, EACCES);
+
 	domain = find_domain_by_domid(domid);
 	if (!domain)
 		return send_error(conn, ENOENT);
@@ -365,6 +367,14 @@ static int close_xc_handle(void *_handle)
 	return 0;
 }
 
+/* Returns the implicit path of a connection (only domains have this) */
+const char *get_implicit_path(const struct connection *conn)
+{
+	if (!conn->domain)
+		return NULL;
+	return conn->domain->path;
+}
+
 /* Returns the event channel handle. */
 int domain_init(void)
 {
diff --git a/tools/xenstore/xenstored_domain.h b/tools/xenstore/xenstored_domain.h
index 20e85a54b5..74dc34e8e5 100644
--- a/tools/xenstore/xenstored_domain.h
+++ b/tools/xenstore/xenstored_domain.h
@@ -16,6 +16,7 @@
     along with this program; if not, write to the Free Software
     Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
+
 #ifndef _XENSTORED_DOMAIN_H
 #define _XENSTORED_DOMAIN_H
 
@@ -33,6 +34,7 @@ bool do_get_domain_path(struct connection *conn, const char *domid_str);
 /* Returns the event channel handle */
 int domain_init(void);
 
-void domain_set_conn(struct domain *domain, struct connection *conn);
+/* Returns the implicit path of a connection (only domains have this) */
+const char *get_implicit_path(const struct connection *conn);
 
 #endif /* _XENSTORED_DOMAIN_H */
diff --git a/tools/xenstore/xenstored_transaction.c b/tools/xenstore/xenstored_transaction.c
index ca37307f8c..dd70579109 100644
--- a/tools/xenstore/xenstored_transaction.c
+++ b/tools/xenstore/xenstored_transaction.c
@@ -201,6 +201,7 @@ bool do_transaction_start(struct connection *conn, const char *node)
 	if (conn->transaction)
 		return send_error(conn, EBUSY);
 
+	node = canonicalize(conn, node);
 	if (!check_node_perms(conn, node, XS_PERM_READ))
 		return send_error(conn, errno);
 
diff --git a/tools/xenstore/xenstored_watch.c b/tools/xenstore/xenstored_watch.c
index 2df83e1a54..d0e00f53c2 100644
--- a/tools/xenstore/xenstored_watch.c
+++ b/tools/xenstore/xenstored_watch.c
@@ -21,6 +21,8 @@
 #include <sys/types.h>
 #include <stdarg.h>
 #include <stdlib.h>
+#include <sys/time.h>
+#include <time.h>
 #include "talloc.h"
 #include "list.h"
 #include "xenstored_watch.h"
@@ -28,6 +30,8 @@
 #include "utils.h"
 #include "xenstored_test.h"
 
+/* FIXME: time out unacked watches. */
+
 /* We create this if anyone is interested "node", then we pass it from
  * watch to watch as each connection acks it.
  */
@@ -39,7 +43,10 @@ struct watch_event
 	/* Watch we are currently attached to. */
 	struct watch *watch;
 
-	struct buffered_data *data;
+	struct timeval timeout;
+
+	/* Name of node which changed. */
+	char *node;
 };
 
 struct watch
@@ -50,72 +57,63 @@ struct watch
 	/* Current outstanding events applying to this watch. */
 	struct list_head events;
 
+	char *token;
 	char *node;
 	struct connection *conn;
 };
 static LIST_HEAD(watches);
 
-static void reset_event(struct watch_event *event)
-{
-	event->data->inhdr = true;
-	event->data->used = 0;
-}
-
-/* We received a non-ACK response: re-queue any watch we just sent. */
-void reset_watch_event(struct connection *conn)
-{
-	if (waiting_for_ack(conn))
-		reset_event(conn->event);
-}
-
-/* We're waiting if we have an event and we sent it all. */
-bool waiting_for_ack(struct connection *conn)
+static struct watch_event *get_first_event(struct connection *conn)
 {
-	if (!conn->event)
-		return false;
+	struct watch *watch;
+	struct watch_event *event;
 
-	if (conn->event->data->inhdr)
-		return false;
-	return conn->event->data->used == conn->event->data->hdr.msg.len;
-}
+	/* Find first watch with an event. */
+	list_for_each_entry(watch, &watches, list) {
+		if (watch->conn != conn)
+			continue;
 
-bool is_watch_event(struct connection *conn, struct buffered_data *out)
-{
-	return (conn->event && out == conn->event->data);
+		event = list_top(&watch->events, struct watch_event, list);
+		if (event)
+			return event;
+	}
+	return NULL;
 }
 
 /* Look through our watches: if any of them have an event, queue it. */
 void queue_next_event(struct connection *conn)
 {
-	struct watch *watch;
+	struct watch_event *event;
+	char *buffer;
+	unsigned int len;
 
-	/* We had a reply queued already?  Send it. */
+	/* We had a reply queued already?  Send it: other end will
+	 * discard watch. */
 	if (conn->waiting_reply) {
 		conn->out = conn->waiting_reply;
 		conn->waiting_reply = NULL;
+		conn->waiting_for_ack = false;
 		return;
 	}
 
-	/* If we're waiting for ack, don't queue more. */
-	if (waiting_for_ack(conn))
+	/* If we're already waiting for ack, don't queue more. */
+	if (conn->waiting_for_ack)
 		return;
 
-	/* Find a good event to send. */
-	if (!conn->event) {
-		list_for_each_entry(watch, &watches, list) {
-			if (watch->conn != conn)
-				continue;
+	event = get_first_event(conn);
+	if (!event)
+		return;
 
-			conn->event = list_top(&watch->events,
-					       struct watch_event, list);
-			if (conn->event)
-				break;
-		}
-		if (!conn->event)
-			return;
-	}
+	/* If we decide to cancel, we will reset this. */
+	conn->waiting_for_ack = true;
 
-	conn->out = conn->event->data;
+	/* Create reply from path and token */
+	len = strlen(event->node) + 1 + strlen(event->watch->token) + 1;
+	buffer = talloc_array(conn, char, len);
+	strcpy(buffer, event->node);
+	strcpy(buffer+strlen(event->node)+1, event->watch->token);
+	send_reply(conn, XS_WATCH_EVENT, buffer, len);
+	talloc_free(buffer);
 }
 
 /* Watch on DIR applies to DIR, DIR/FILE, but not DIRLONG. */
@@ -160,14 +158,15 @@ void fire_watches(struct transaction *trans, const char *node)
 
 	/* Create and fill in info about event. */
 	event = talloc(talloc_autofree_context(), struct watch_event);
-	event->data = new_buffer(event);
-	event->data->hdr.msg.type = XS_WATCH_EVENT;
-	event->data->hdr.msg.len = strlen(node) + 1;
-	event->data->buffer = talloc_strdup(event->data, node);
+	event->node = talloc_strdup(event, node);
 
 	/* Tie event to this watch. */
 	event->watch = watch;
-	list_add(&event->list, &watch->events);
+	list_add_tail(&event->list, &watch->events);
+
+	/* Warn if not finished after thirty seconds. */
+	gettimeofday(&event->timeout, NULL);
+	event->timeout.tv_sec += 30;
 
 	/* If connection not doing anything, queue this. */
 	if (!watch->conn->out)
@@ -178,16 +177,15 @@ void fire_watches(struct transaction *trans, const char *node)
 static void move_event_onwards(struct watch_event *event)
 {
 	list_del(&event->list);
-	reset_event(event);
 
 	/* Remove from this watch, and find next watch to put this on. */
-	event->watch = find_next_watch(event->watch, event->data->buffer);
+	event->watch = find_next_watch(event->watch, event->node);
 	if (!event->watch) {
 		talloc_free(event);
 		return;
 	}
 
-	list_add(&event->list, &event->watch->events);
+	list_add_tail(&event->list, &event->watch->events);
 
 	/* If connection not doing anything, queue this. */
 	if (!event->watch->conn->out)
@@ -199,10 +197,6 @@ static int destroy_watch(void *_watch)
 	struct watch *watch = _watch;
 	struct watch_event *event;
 
-	/* Forget about sending out or waiting for acks for this watch.  */
-	if (watch->conn->event && watch->conn->event->watch == watch)
-		watch->conn->event = NULL;
-
 	/* If we have pending events, pass them on to others. */
 	while ((event = list_top(&watch->events, struct watch_event, list)))
 		move_event_onwards(event);
@@ -227,21 +221,59 @@ static void insert_watch(struct watch *watch)
 	list_add_tail(&watch->list, &watches);
 }
 
+void shortest_watch_ack_timeout(struct timeval *tv)
+{
+	struct watch *watch;
+
+	list_for_each_entry(watch, &watches, list) {
+		struct watch_event *i;
+		list_for_each_entry(i, &watch->events, list) {
+			if (!timerisset(&i->timeout))
+				continue;
+			if (!timerisset(tv) || timercmp(&i->timeout, tv, <))
+				*tv = i->timeout;
+		}
+	}
+}	
+
+void check_watch_ack_timeout(void)
+{
+	struct watch *watch;
+	struct timeval now;
+
+	gettimeofday(&now, NULL);
+	list_for_each_entry(watch, &watches, list) {
+		struct watch_event *i, *tmp;
+		list_for_each_entry_safe(i, tmp, &watch->events, list) {
+			if (!timerisset(&i->timeout))
+				continue;
+			if (timercmp(&i->timeout, &now, <)) {
+				xprintf("Warning: timeout on watch event %s"
+					" token %s\n",
+					i->node, watch->token);
+				timerclear(&i->timeout);
+			}
+		}
+	}
+}
+
 bool do_watch(struct connection *conn, struct buffered_data *in)
 {
 	struct watch *watch;
-	char *vec[2];
+	char *vec[3];
 
 	if (get_strings(in, vec, ARRAY_SIZE(vec)) != ARRAY_SIZE(vec))
 		return send_error(conn, EINVAL);
 
+	vec[0] = canonicalize(conn, vec[0]);
 	if (!check_node_perms(conn, vec[0], XS_PERM_READ))
 		return send_error(conn, errno);
 
 	watch = talloc(conn, struct watch);
 	watch->node = talloc_strdup(watch, vec[0]);
+	watch->token = talloc_strdup(watch, vec[1]);
 	watch->conn = conn;
-	watch->priority = strtoul(vec[1], NULL, 0);
+	watch->priority = strtoul(vec[2], NULL, 0);
 	INIT_LIST_HEAD(&watch->events);
 
 	insert_watch(watch);
@@ -249,31 +281,58 @@ bool do_watch(struct connection *conn, struct buffered_data *in)
 	return send_ack(conn, XS_WATCH);
 }
 
-bool do_watch_ack(struct connection *conn)
+bool do_watch_ack(struct connection *conn, const char *token)
 {
 	struct watch_event *event;
 
-	if (!waiting_for_ack(conn))
+	if (!conn->waiting_for_ack)
 		return send_error(conn, ENOENT);
 
-	/* Remove this watch event. */
-	event = conn->event;
-	conn->event = NULL;
+	event = get_first_event(conn);
+	if (!streq(event->watch->token, token))
+		return send_error(conn, EINVAL);
 
 	move_event_onwards(event);
+	conn->waiting_for_ack = false;
 	return send_ack(conn, XS_WATCH_ACK);
 }
 
-bool do_unwatch(struct connection *conn, const char *node)
+bool do_unwatch(struct connection *conn, struct buffered_data *in)
 {
 	struct watch *watch;
+	char *node, *vec[2];
+
+	if (get_strings(in, vec, ARRAY_SIZE(vec)) != ARRAY_SIZE(vec))
+		return send_error(conn, EINVAL);
 
+	node = canonicalize(conn, vec[0]);
 	list_for_each_entry(watch, &watches, list) {
-		if (watch->conn == conn
-		    && streq(watch->node, node)) {
+		if (watch->conn != conn)
+			continue;
+
+		if (streq(watch->node, node) && streq(watch->token, vec[1])) {
 			talloc_free(watch);
 			return send_ack(conn, XS_UNWATCH);
 		}
 	}
 	return send_error(conn, ENOENT);
 }
+
+#ifdef TESTING
+void dump_watches(struct connection *conn)
+{
+	struct watch *watch;
+	struct watch_event *event;
+
+	/* Find first watch with an event. */
+	list_for_each_entry(watch, &watches, list) {
+		if (watch->conn != conn)
+			continue;
+
+		printf("    watch on %s token %s prio %i\n",
+		       watch->node, watch->token, watch->priority);
+		list_for_each_entry(event, &watch->events, list)
+			printf("        event: %s\n", event->node);
+	}
+}
+#endif
diff --git a/tools/xenstore/xenstored_watch.h b/tools/xenstore/xenstored_watch.h
index 656ce4c36b..e9c0ad8f0b 100644
--- a/tools/xenstore/xenstored_watch.h
+++ b/tools/xenstore/xenstored_watch.h
@@ -16,13 +16,15 @@
     along with this program; if not, write to the Free Software
     Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
+
 #ifndef _XENSTORED_WATCH_H
 #define _XENSTORED_WATCH_H
+
 #include "xenstored_core.h"
 
 bool do_watch(struct connection *conn, struct buffered_data *in);
-bool do_watch_ack(struct connection *conn);
-bool do_unwatch(struct connection *conn, const char *node);
+bool do_watch_ack(struct connection *conn, const char *token);
+bool do_unwatch(struct connection *conn, struct buffered_data *in);
 
 /* Is this a watch event message for this connection? */
 bool is_watch_event(struct connection *conn, struct buffered_data *out);
@@ -30,13 +32,15 @@ bool is_watch_event(struct connection *conn, struct buffered_data *out);
 /* Look through our watches: if any of them have an event, queue it. */
 void queue_next_event(struct connection *conn);
 
-/* Is this connection waiting for a watch acknowledgement? */
-bool waiting_for_ack(struct connection *conn);
-
-/* Reset event if we were sending one */
-void reset_watch_event(struct connection *conn);
-
 /* Fire all watches. */
 void fire_watches(struct transaction *trans, const char *node);
 
+/* Find shortest timeout: if any, reduce tv (may already be set). */
+void shortest_watch_ack_timeout(struct timeval *tv);
+
+/* Check for watches which may have timed out. */
+void check_watch_ack_timeout(void);
+
+void dump_watches(struct connection *conn);
+
 #endif /* _XENSTORED_WATCH_H */
diff --git a/tools/xenstore/xs.c b/tools/xenstore/xs.c
index d6e41380f9..e41ca652bd 100644
--- a/tools/xenstore/xs.c
+++ b/tools/xenstore/xs.c
@@ -159,8 +159,7 @@ static void *read_reply(int fd, enum xsd_sockmsg_type *type, unsigned int *len)
 
 /* Send message to xs, get malloc'ed reply.  NULL and set errno on error. */
 static void *xs_talkv(struct xs_handle *h, enum xsd_sockmsg_type type,
-		      const struct iovec *iovec,
-		      unsigned int num_vecs,
+		      const struct iovec *iovec, unsigned int num_vecs,
 		      unsigned int *len)
 {
 	struct xsd_sockmsg msg;
@@ -330,8 +329,7 @@ bool xs_rm(struct xs_handle *h, const char *path)
  * Returns malloced array, or NULL: call free() after use.
  */
 struct xs_permissions *xs_get_permissions(struct xs_handle *h,
-					  const char *path,
-					  unsigned int *num)
+					  const char *path, unsigned int *num)
 {
 	char *strings;
 	unsigned int len;
@@ -400,61 +398,75 @@ unwind:
 
 /* Watch a node for changes (poll on fd to detect, or call read_watch()).
  * When the node (or any child) changes, fd will become readable.
+ * Token is returned when watch is read, to allow matching.
  * Priority indicates order if multiple watchers: higher is first.
  * Returns false on failure.
  */
-bool xs_watch(struct xs_handle *h, const char *path, unsigned int priority)
+bool xs_watch(struct xs_handle *h, const char *path, const char *token,
+	      unsigned int priority)
 {
 	char prio[MAX_STRLEN(priority)];
-	struct iovec iov[2];
+	struct iovec iov[3];
 
 	sprintf(prio, "%u", priority);
 	iov[0].iov_base = (void *)path;
 	iov[0].iov_len = strlen(path) + 1;
-	iov[1].iov_base = prio;
-	iov[1].iov_len = strlen(prio) + 1;
+	iov[1].iov_base = (void *)token;
+	iov[1].iov_len = strlen(token) + 1;
+	iov[2].iov_base = prio;
+	iov[2].iov_len = strlen(prio) + 1;
 
 	return xs_bool(xs_talkv(h, XS_WATCH, iov, ARRAY_SIZE(iov), NULL));
 }
 
 /* Find out what node change was on (will block if nothing pending).
- * Returns malloced path, or NULL: call free() after use.
+ * Returns array of two pointers: path and token, or NULL.
+ * Call free() after use.
  */
-char *xs_read_watch(struct xs_handle *h)
+char **xs_read_watch(struct xs_handle *h)
 {
 	struct xsd_sockmsg msg;
-	char *path;
+	char **ret;
 
 	if (!read_all(h->fd, &msg, sizeof(msg)))
 		return NULL;
 
 	assert(msg.type == XS_WATCH_EVENT);
-	path = malloc(msg.len);
-	if (!path)
+	ret = malloc(sizeof(char *)*2 + msg.len);
+	if (!ret)
 		return NULL;
 
-	if (!read_all(h->fd, path, msg.len)) {
-		free_no_errno(path);
+	ret[0] = (char *)(ret + 2);
+	if (!read_all(h->fd, ret[0], msg.len)) {
+		free_no_errno(ret);
 		return NULL;
 	}
-	return path;
+	ret[1] = ret[0] + strlen(ret[0]) + 1;
+	return ret;
 }
 
 /* Acknowledge watch on node.  Watches must be acknowledged before
  * any other watches can be read.
  * Returns false on failure.
  */
-bool xs_acknowledge_watch(struct xs_handle *h)
+bool xs_acknowledge_watch(struct xs_handle *h, const char *token)
 {
-	return xs_bool(xs_single(h, XS_WATCH_ACK, "OK", NULL));
+	return xs_bool(xs_single(h, XS_WATCH_ACK, token, NULL));
 }
 
 /* Remove a watch on a node.
  * Returns false on failure (no watch on that node).
  */
-bool xs_unwatch(struct xs_handle *h, const char *path)
+bool xs_unwatch(struct xs_handle *h, const char *path, const char *token)
 {
-	return xs_bool(xs_single(h, XS_UNWATCH, path, NULL));
+	struct iovec 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_bool(xs_talkv(h, XS_UNWATCH, iov, ARRAY_SIZE(iov), NULL));
 }
 
 /* Start a transaction: changes by others will not be seen during this
@@ -488,11 +500,8 @@ bool xs_transaction_end(struct xs_handle *h, bool abort)
  * This tells the store daemon about a shared memory page and event channel
  * associated with a domain: the domain uses these to communicate.
  */
-bool xs_introduce_domain(struct xs_handle *h,
-			 domid_t domid,
-			 unsigned long mfn,
-			 unsigned int eventchn,
-			 const char *path)
+bool xs_introduce_domain(struct xs_handle *h, domid_t domid, unsigned long mfn,
+			 unsigned int eventchn, const char *path)
 {
 	char domid_str[MAX_STRLEN(domid)];
 	char mfn_str[MAX_STRLEN(mfn)];
@@ -515,8 +524,7 @@ bool xs_introduce_domain(struct xs_handle *h,
 	return xs_bool(xs_talkv(h, XS_INTRODUCE, iov, ARRAY_SIZE(iov), NULL));
 }
 
-bool xs_release_domain(struct xs_handle *h,
-		       domid_t domid)
+bool xs_release_domain(struct xs_handle *h, domid_t domid)
 {
 	char domid_str[MAX_STRLEN(domid)];
 
diff --git a/tools/xenstore/xs.h b/tools/xenstore/xs.h
index ff9481c3a6..b778cedd65 100644
--- a/tools/xenstore/xs.h
+++ b/tools/xenstore/xs.h
@@ -1,5 +1,3 @@
-#ifndef _XS_H
-#define _XS_H
 /* 
     Xen Store Daemon providing simple tree-like database.
     Copyright (C) 2005 Rusty Russell IBM Corporation
@@ -19,11 +17,15 @@
     Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
 
-/* On failure, these routines set errno. */
+#ifndef _XS_H
+#define _XS_H
+
 #include "xs_lib.h"
 
 struct xs_handle;
 
+/* On failure, these routines set errno. */
+
 /* Connect to the xs daemon.
  * Returns a handle or NULL.
  */
@@ -52,8 +54,8 @@ void *xs_read(struct xs_handle *h, const char *path, unsigned int *len);
 /* Write the value of a single file.
  * Returns false on failure.  createflags can be 0, O_CREAT, or O_CREAT|O_EXCL.
  */
-bool xs_write(struct xs_handle *h, const char *path, const void *data, unsigned int len,
-	      int createflags);
+bool xs_write(struct xs_handle *h, const char *path, const void *data,
+	      unsigned int len, int createflags);
 
 /* Create a new directory.
  * Returns false on failure.
@@ -69,42 +71,42 @@ bool xs_rm(struct xs_handle *h, const char *path);
  * Returns malloced array, or NULL: call free() after use.
  */
 struct xs_permissions *xs_get_permissions(struct xs_handle *h,
-					  const char *path,
-					  unsigned int *num);
+					  const char *path, unsigned int *num);
 
 /* Set permissions of node (must be owner).
  * Returns false on failure.
  */
-bool xs_set_permissions(struct xs_handle *h,
-			const char *path,
-			struct xs_permissions *perms,
-			unsigned int num_perms);
+bool xs_set_permissions(struct xs_handle *h, const char *path,
+			struct xs_permissions *perms, unsigned int num_perms);
 
 /* Watch a node for changes (poll on fd to detect, or call read_watch()).
  * When the node (or any child) changes, fd will become readable.
+ * Token is returned when watch is read, to allow matching.
  * Priority indicates order if multiple watchers: higher is first.
  * Returns false on failure.
  */
-bool xs_watch(struct xs_handle *h, const char *path, unsigned int priority);
+bool xs_watch(struct xs_handle *h, const char *path, const char *token,
+	      unsigned int priority);
 
 /* Return the FD to poll on to see if a watch has fired. */
 int xs_fileno(struct xs_handle *h);
 
 /* Find out what node change was on (will block if nothing pending).
- * Returns malloced path, or NULL: call free() after use.
+ * Returns array of two pointers: path and token, or NULL.
+ * Call free() after use.
  */
-char *xs_read_watch(struct xs_handle *h);
+char **xs_read_watch(struct xs_handle *h);
 
 /* Acknowledge watch on node.  Watches must be acknowledged before
  * any other watches can be read.
  * Returns false on failure.
  */
-bool xs_acknowledge_watch(struct xs_handle *h);
+bool xs_acknowledge_watch(struct xs_handle *h, const char *token);
 
 /* Remove a watch on a node.
  * Returns false on failure (no watch on that node).
  */
-bool xs_unwatch(struct xs_handle *h, const char *path);
+bool xs_unwatch(struct xs_handle *h, const char *path, const char *token);
 
 /* Start a transaction: changes by others will not be seen during this
  * transaction, and changes will not be visible to others until end.
@@ -125,11 +127,8 @@ bool xs_transaction_end(struct xs_handle *h, bool abort);
  * This tells the store daemon about a shared memory page, event channel
  * and store path associated with a domain: the domain uses these to communicate.
  */
-bool xs_introduce_domain(struct xs_handle *h,
-                         domid_t domid,
-                         unsigned long mfn,
-                         unsigned int eventchn,
-                         const char *path);
+bool xs_introduce_domain(struct xs_handle *h, domid_t domid, unsigned long mfn,
+                         unsigned int eventchn, const char *path);
 
 /* Release a domain.
  * Tells the store domain to release the memory page to the domain.
diff --git a/tools/xenstore/xs_lib.c b/tools/xenstore/xs_lib.c
index 3f4f4b0899..cc9f9e1706 100644
--- a/tools/xenstore/xs_lib.c
+++ b/tools/xenstore/xs_lib.c
@@ -67,7 +67,7 @@ bool xs_write_all(int fd, const void *data, unsigned int len)
 
 /* Convert strings to permissions.  False if a problem. */
 bool xs_strings_to_perms(struct xs_permissions *perms, unsigned int num,
-		      const char *strings)
+			 const char *strings)
 {
 	const char *p;
 	char *end;
@@ -138,4 +138,3 @@ unsigned int xs_count_strings(const char *strings, unsigned int len)
 
 	return num;
 }
-	
diff --git a/tools/xenstore/xs_lib.h b/tools/xenstore/xs_lib.h
index 76ea9b67fe..97b72c8c7e 100644
--- a/tools/xenstore/xs_lib.h
+++ b/tools/xenstore/xs_lib.h
@@ -1,5 +1,3 @@
-#ifndef _XR_LIB_H
-#define _XR_LIB_H
 /* 
     Common routines between Xen store user library and daemon.
     Copyright (C) 2005 Rusty Russell IBM Corporation
@@ -18,6 +16,10 @@
     along with this program; if not, write to the Free Software
     Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
+
+#ifndef _XS_LIB_H
+#define _XS_LIB_H
+
 #include <stdbool.h>
 #include <limits.h>
 #include <xc.h>
@@ -52,7 +54,7 @@ bool xs_write_all(int fd, const void *data, unsigned int len);
 
 /* Convert strings to permissions.  False if a problem. */
 bool xs_strings_to_perms(struct xs_permissions *perms, unsigned int num,
-		      const char *strings);
+			 const char *strings);
 
 /* Convert permissions to a string (up to len MAX_STRLEN(domid_t)+1). */
 bool xs_perm_to_string(const struct xs_permissions *perm, char *buffer);
diff --git a/tools/xenstore/xs_stress.c b/tools/xenstore/xs_stress.c
index 9c480b1553..0c257e465b 100644
--- a/tools/xenstore/xs_stress.c
+++ b/tools/xenstore/xs_stress.c
@@ -50,7 +50,7 @@ static void work(unsigned int cycles, unsigned int childnum)
 		}
 		if (streq(lockdir, ""))
 			strcpy(lockdir, "/");
-		
+
 		if (!xs_transaction_start(h, lockdir))
 			barf_perror("%i: starting transaction %i on %s",
 				    childnum, i, lockdir);
diff --git a/tools/xenstore/xs_test.c b/tools/xenstore/xs_test.c
index 4d769e220d..29929b7693 100644
--- a/tools/xenstore/xs_test.c
+++ b/tools/xenstore/xs_test.c
@@ -173,9 +173,9 @@ static void __attribute__((noreturn)) usage(void)
 	     "  getperm <path>\n"
 	     "  setperm <path> <id> <flags> ...\n"
 	     "  shutdown\n"
-	     "  watch <path> <prio>\n"
+	     "  watch <path> <token> <prio>\n"
 	     "  waitwatch\n"
-	     "  ackwatch\n"
+	     "  ackwatch <token>\n"
 	     "  unwatch <path> <token>\n"
 	     "  close\n"
 	     "  start <node>\n"
@@ -358,36 +358,37 @@ static void do_shutdown(unsigned int handle)
 		failed(handle);
 }
 
-static void do_watch(unsigned int handle, const char *node, const char *pri)
+static void do_watch(unsigned int handle, const char *node, const char *token,
+		     const char *pri)
 {
-	if (!xs_watch(handles[handle], node, atoi(pri)))
+	if (!xs_watch(handles[handle], node, token, atoi(pri)))
 		failed(handle);
 }
 
 static void do_waitwatch(unsigned int handle)
 {
-	char *node;
+	char **vec;
 
-	node = xs_read_watch(handles[handle]);
-	if (!node)
+	vec = xs_read_watch(handles[handle]);
+	if (!vec)
 		failed(handle);
 
 	if (handle)
-		printf("%i:%s\n", handle, node);
+		printf("%i:%s:%s\n", handle, vec[0], vec[1]);
 	else
-		printf("%s\n", node);
-	free(node);
+		printf("%s:%s\n", vec[0], vec[1]);
+	free(vec);
 }
 
-static void do_ackwatch(unsigned int handle)
+static void do_ackwatch(unsigned int handle, const char *token)
 {
-	if (!xs_acknowledge_watch(handles[handle]))
+	if (!xs_acknowledge_watch(handles[handle], token))
 		failed(handle);
 }
 
-static void do_unwatch(unsigned int handle, const char *node)
+static void do_unwatch(unsigned int handle, const char *node, const char *token)
 {
-	if (!xs_unwatch(handles[handle], node))
+	if (!xs_unwatch(handles[handle], node, token))
 		failed(handle);
 }
 
@@ -613,13 +614,13 @@ int main(int argc, char *argv[])
 		else if (streq(command, "shutdown"))
 			do_shutdown(handle);
 		else if (streq(command, "watch"))
-			do_watch(handle, arg(line, 1), arg(line, 2));
+			do_watch(handle, arg(line, 1), arg(line, 2), arg(line, 3));
 		else if (streq(command, "waitwatch"))
 			do_waitwatch(handle);
 		else if (streq(command, "ackwatch"))
-			do_ackwatch(handle);
+			do_ackwatch(handle, arg(line, 1));
 		else if (streq(command, "unwatch"))
-			do_unwatch(handle, arg(line, 1));
+			do_unwatch(handle, arg(line, 1), arg(line, 2));
 		else if (streq(command, "close")) {
 			xs_daemon_close(handles[handle]);
 			handles[handle] = NULL;
diff --git a/tools/xenstore/xs_watch_stress.c b/tools/xenstore/xs_watch_stress.c
new file mode 100644
index 0000000000..91431e2376
--- /dev/null
+++ b/tools/xenstore/xs_watch_stress.c
@@ -0,0 +1,120 @@
+/* Stress test for watch code: two processes communicating by watches */
+#include "xs.h"
+#include "utils.h"
+#include <stdlib.h>
+#include <stdio.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <unistd.h>
+
+int main(int argc __attribute__((unused)), char *argv[])
+{
+	int childpid, status, fds[2];
+	bool parent;
+	unsigned int i, acks = 0;
+	struct xs_handle *h;
+	char *data;
+	unsigned int len;
+	const char *path, *otherpath;
+
+	pipe(fds);
+	childpid = fork();
+	if (childpid == -1)
+		barf_perror("Failed fork");
+	parent = (childpid != 0);
+
+	h = xs_daemon_open();
+	if (!h)
+		barf_perror("Could not connect to daemon");
+
+	if (!xs_watch(h, "/", "token", 0))
+		barf_perror("Could not set watch");
+
+	if (parent) {
+		char c;
+
+		if (read(fds[0], &c, 1) != 1)
+			barf("Child exited");
+
+		path = "/parent";
+		otherpath = "/child";
+		/* Create initial node. */
+		if (!xs_write(h, path, "0", 2, O_CREAT))
+			barf_perror("Write to %s failed", path);
+	} else {
+		path = "/child";
+		otherpath = "/parent";
+
+		if (write(fds[1], "", 1) != 1)
+			barf_perror("Write to parent failed");
+	}
+
+	for (i = 0; i < (argv[1] ? (unsigned)atoi(argv[1]) : 100);) {
+		char **vec;
+
+		vec = xs_read_watch(h);
+		if (!vec)
+			barf_perror("Read watch failed");
+
+		if (!streq(vec[1], "token"))
+			barf("Watch token %s bad", vec[1]);
+		if (streq(vec[0], otherpath)) {
+			char number[32];
+
+			data = xs_read(h, otherpath, &len);
+			if (!data)
+				barf_perror("reading %s", otherpath);
+			sprintf(number, "%i", atoi(data) + 1);
+			free(data);
+			if (!xs_write(h, path, number, strlen(number) + 1,
+				      O_CREAT))
+				barf_perror("writing %s", path);
+			i++;
+		} else if (!streq(vec[0], path))
+			barf_perror("Watch fired on unknown path %s", vec[0]);
+		xs_acknowledge_watch(h, vec[1]);
+		acks++;
+		free(vec);
+	}
+
+	if (!parent) {
+		while (acks != 2 * i - 1) {
+			char **vec;
+			vec = xs_read_watch(h);
+			if (!vec)
+				barf_perror("Watch failed");
+			if (!streq(vec[0], path))
+				barf_perror("Watch fired path %s", vec[0]);
+			if (!streq(vec[1], "token"))
+				barf("Watch token %s bad", vec[1]);
+			free(vec);
+
+			printf("Expect %i events, only got %i\n",
+			       2 * i - 1, acks);
+			acks++;
+		}
+		exit(0);
+	}
+
+	if (acks != 2 * i)
+		barf("Parent got %i watch events\n", acks);
+
+	printf("Waiting for %i\n", childpid);
+	if (waitpid(childpid, &status, 0) != childpid)
+		barf_perror("Child wait failed");
+	if (!WIFEXITED(status) || WEXITSTATUS(status) != 0)
+		barf_perror("Child status %i", status);
+
+	data = xs_read(h, path, &len);
+	if (atoi(data) != 2 * (int)i)
+		barf("%s count is %s\n", path, data);
+	free(data);
+	data = xs_read(h, otherpath, &len);
+	if (atoi(data) != 2 * (int)i - 1)
+		barf("%s count is %s\n", otherpath, data);
+	free(data);
+	printf("Success!\n");
+	exit(0);
+}