1.1 --- a/tools/xenstore/TODO	Tue Jul 26 15:13:56 2005 +0000
     1.2 +++ b/tools/xenstore/TODO	Tue Jul 26 15:20:09 2005 +0000
     1.3 @@ -2,8 +2,9 @@ TODO in no particular order.  Some of th
     1.4  are omissions of important but necessary things.  It is up to the
     1.5  reader to fill in the blanks.
     1.6  
     1.7 -- Remove calls to system() from daemon
     1.8  - Timeout failed watch responses
     1.9 -- Dynamic nodes
    1.10 +- Dynamic/supply nodes
    1.11  - Persistant storage of introductions, watches and transactions, so daemon can restart
    1.12  - Remove assumption that rename doesn't fail
    1.13 +- Multi-root transactions, for setting up front and back ends at same time.
    1.14 +

     2.1 --- a/tools/xenstore/testsuite/07watch.sh	Tue Jul 26 15:13:56 2005 +0000
     2.2 +++ b/tools/xenstore/testsuite/07watch.sh	Tue Jul 26 15:20:09 2005 +0000
     2.3 @@ -3,45 +3,52 @@
     2.4  # Watch something, write to it, check watch has fired.
     2.5  [ "`echo -e 'write /test create contents' | ./xs_test 2>&1`" = "" ]
     2.6  
     2.7 -[ "`echo -e '1 watch /test token 100
     2.8 -2 write /test create contents2
     2.9 +[ "`echo -e '1 watch /test token
    2.10 +2 async write /test create contents2
    2.11  1 waitwatch
    2.12  1 ackwatch token' | ./xs_test 2>&1`" = "1:/test:token" ]
    2.13  
    2.14  # Check that reads don't set it off.
    2.15 -[ "`echo -e '1 watch /test token 100
    2.16 +[ "`echo -e '1 watch /test token
    2.17  2 read /test
    2.18  1 waitwatch' | ./xs_test 2>&1`" = "2:contents2
    2.19  1:waitwatch timeout" ]
    2.20  
    2.21  # mkdir, setperm and rm should (also tests watching dirs)
    2.22  [ "`echo -e 'mkdir /dir' | ./xs_test 2>&1`" = "" ]
    2.23 -[ "`echo -e '1 watch /dir token 100
    2.24 -2 mkdir /dir/newdir
    2.25 +[ "`echo -e '1 watch /dir token
    2.26 +2 async mkdir /dir/newdir
    2.27  1 waitwatch
    2.28  1 ackwatch token
    2.29 -2 setperm /dir/newdir 0 READ
    2.30 +asyncwait
    2.31 +2 async setperm /dir/newdir 0 READ
    2.32  1 waitwatch
    2.33  1 ackwatch token
    2.34 -2 rm /dir/newdir
    2.35 +asyncwait
    2.36 +2 async rm /dir/newdir
    2.37  1 waitwatch
    2.38  1 ackwatch token' | ./xs_test 2>&1`" = "1:/dir/newdir:token
    2.39  1:/dir/newdir:token
    2.40  1:/dir/newdir:token" ]
    2.41  
    2.42 +# We don't get a watch from our own commands.
    2.43 +[ "`echo -e 'watch /dir token
    2.44 +mkdir /dir/newdir
    2.45 +waitwatch' | ./xs_test 2>&1`" = "waitwatch timeout" ]
    2.46 +
    2.47  # ignore watches while doing commands, should work.
    2.48 -[ "`echo -e 'watch /dir token 100
    2.49 -write /dir/test create contents
    2.50 +[ "`echo -e 'watch /dir token
    2.51 +1 async write /dir/test create contents
    2.52  read /dir/test
    2.53  waitwatch
    2.54  ackwatch token' | ./xs_test 2>&1`" = "contents
    2.55  /dir/test:token" ]
    2.56  
    2.57 -# watch priority /test.
    2.58 -[ "`echo -e '1 watch /dir token1 1
    2.59 -3 watch /dir token3 3
    2.60 -2 watch /dir token2 2
    2.61 -write /dir/test create contents
    2.62 +# watch priority test: all simultaneous
    2.63 +[ "`echo -e '1 watch /dir token1
    2.64 +3 watch /dir token3
    2.65 +2 watch /dir token2
    2.66 +async write /dir/test create contents
    2.67  3 waitwatch
    2.68  3 ackwatch token3
    2.69  2 waitwatch
    2.70 @@ -52,9 +59,9 @@ 2:/dir/test:token2
    2.71  1:/dir/test:token1" ]
    2.72  
    2.73  # If one dies (without acking), the other should still get ack.
    2.74 -[ "`echo -e '1 watch /dir token1 0
    2.75 -2 watch /dir token2 1
    2.76 -write /dir/test create contents
    2.77 +[ "`echo -e '1 watch /dir token1
    2.78 +2 watch /dir token2
    2.79 +async write /dir/test create contents
    2.80  2 waitwatch
    2.81  2 close
    2.82  1 waitwatch
    2.83 @@ -62,51 +69,52 @@ 1 ackwatch token1' | ./xs_test 2>&1`" = 
    2.84  1:/dir/test:token1" ]
    2.85  
    2.86  # If one dies (without reading at all), the other should still get ack.
    2.87 -[ "`echo -e '1 watch /dir token1 0
    2.88 -2 watch /dir token2 1
    2.89 -write /dir/test create contents
    2.90 +[ "`echo -e '1 watch /dir token1
    2.91 +2 watch /dir token2
    2.92 +async write /dir/test create contents
    2.93  2 close
    2.94  1 waitwatch
    2.95  1 ackwatch token1' | ./xs_test 2>&1`" = "1:/dir/test:token1" ]
    2.96  
    2.97  # unwatch
    2.98 -[ "`echo -e '1 watch /dir token1 0
    2.99 +[ "`echo -e '1 watch /dir token1
   2.100  1 unwatch /dir token1
   2.101 -1 watch /dir token2 0
   2.102 -2 write /dir/test2 create contents
   2.103 +1 watch /dir token2
   2.104 +2 async write /dir/test2 create contents
   2.105  1 waitwatch
   2.106  1 unwatch /dir token2' | ./xs_test 2>&1`" = "1:/dir/test2:token2" ]
   2.107  
   2.108  # unwatch while watch pending.  Next watcher gets the event.
   2.109 -[ "`echo -e '1 watch /dir token1 0
   2.110 -2 watch /dir token2 1
   2.111 -write /dir/test create contents
   2.112 +[ "`echo -e '1 watch /dir token1
   2.113 +2 watch /dir token2
   2.114 +async write /dir/test create contents
   2.115  2 unwatch /dir token2
   2.116  1 waitwatch
   2.117  1 ackwatch token1' | ./xs_test 2>&1`" = "1:/dir/test:token1" ]
   2.118  
   2.119  # unwatch while watch pending.  Should clear this so we get next event.
   2.120 -[ "`echo -e '1 watch /dir token1 0
   2.121 -write /dir/test create contents
   2.122 +[ "`echo -e '1 watch /dir token1
   2.123 +async write /dir/test create contents
   2.124  1 unwatch /dir token1
   2.125 -1 watch /dir/test token2 0
   2.126 -write /dir/test none contents2
   2.127 +1 watch /dir/test token2
   2.128 +asyncwait
   2.129 +async write /dir/test none contents2
   2.130  1 waitwatch
   2.131  1 ackwatch token2' | ./xs_test 2>&1`" = "1:/dir/test:token2" ]
   2.132  
   2.133  # check we only get notified once.
   2.134 -[ "`echo -e '1 watch /test token 100
   2.135 -2 write /test create contents2
   2.136 +[ "`echo -e '1 watch /test token
   2.137 +2 async write /test create contents2
   2.138  1 waitwatch
   2.139  1 ackwatch token
   2.140  1 waitwatch' | ./xs_test 2>&1`" = "1:/test:token
   2.141  1:waitwatch timeout" ]
   2.142  
   2.143  # watches are queued in order.
   2.144 -[ "`echo -e '1 watch / token 100
   2.145 -2 write /test1 create contents
   2.146 -2 write /test2 create contents
   2.147 -2 write /test3 create contents
   2.148 +[ "`echo -e '1 watch / token
   2.149 +async 2 write /test1 create contents
   2.150 +async 2 write /test2 create contents
   2.151 +async 2 write /test3 create contents
   2.152  1 waitwatch
   2.153  1 ackwatch token
   2.154  1 waitwatch
   2.155 @@ -117,9 +125,9 @@ 1:/test2:token
   2.156  1:/test3:token" ]
   2.157  
   2.158  # Creation of subpaths should be covered correctly.
   2.159 -[ "`echo -e '1 watch / token 100
   2.160 -2 write /test/subnode create contents2
   2.161 -2 write /test/subnode/subnode create contents2
   2.162 +[ "`echo -e '1 watch / token
   2.163 +2 async write /test/subnode create contents2
   2.164 +2 async write /test/subnode/subnode create contents2
   2.165  1 waitwatch
   2.166  1 ackwatch token
   2.167  1 waitwatch
   2.168 @@ -129,23 +137,23 @@ 1:/test/subnode/subnode:token
   2.169  1:waitwatch timeout" ]
   2.170  
   2.171  # Watch event must have happened before we registered interest.
   2.172 -[ "`echo -e '1 watch / token 100
   2.173 -2 write /test/subnode create contents2
   2.174 -2 watch / token2 0
   2.175 +[ "`echo -e '1 watch / token
   2.176 +2 async write /test/subnode create contents2
   2.177 +1 watch / token2 0
   2.178  1 waitwatch
   2.179  1 ackwatch token
   2.180 -2 waitwatch' | ./xs_test 2>&1`" = "1:/test/subnode:token
   2.181 -2:waitwatch timeout" ]
   2.182 +1 waitwatch' | ./xs_test 2>&1`" = "1:/test/subnode:token
   2.183 +1:waitwatch timeout" ]
   2.184  
   2.185  # Rm fires notification on child.
   2.186 -[ "`echo -e '1 watch /test/subnode token 100
   2.187 -2 rm /test
   2.188 +[ "`echo -e '1 watch /test/subnode token
   2.189 +2 async rm /test
   2.190  1 waitwatch
   2.191  1 ackwatch token' | ./xs_test 2>&1`" = "1:/test/subnode:token" ]
   2.192  
   2.193  # Watch should not double-send after we ack, even if we did something in between.
   2.194 -[ "`echo -e '1 watch /test2 token 100
   2.195 -2 write /test2/foo create contents2
   2.196 +[ "`echo -e '1 watch /test2 token
   2.197 +2 async write /test2/foo create contents2
   2.198  1 waitwatch
   2.199  1 read /test2/foo
   2.200  1 ackwatch token

     3.1 --- a/tools/xenstore/testsuite/08transaction.sh	Tue Jul 26 15:13:56 2005 +0000
     3.2 +++ b/tools/xenstore/testsuite/08transaction.sh	Tue Jul 26 15:20:09 2005 +0000
     3.3 @@ -45,37 +45,37 @@ 2:entry1" ]
     3.4  sleep 1
     3.5  rm /test/entry1
     3.6  commit
     3.7 -dir /test' | ./xs_test`" = "" ]
     3.8 +dir /test' | ./xs_test --no-timeout`" = "" ]
     3.9  
    3.10  # ... as long as noone is waiting.
    3.11  [ "`echo -e '1 start /test
    3.12  2 mkdir /test/dir
    3.13  1 mkdir /test/dir
    3.14  1 dir /test
    3.15 -1 commit' | ./xs_test 2>&1`" = "1:dir
    3.16 +1 commit' | ./xs_test --no-timeout 2>&1`" = "1:dir
    3.17  FATAL: 1: commit: Connection timed out" ]
    3.18  
    3.19  # Events inside transactions don't trigger watches until (successful) commit.
    3.20 -[ "`echo -e '1 watch /test token 100
    3.21 +[ "`echo -e '1 watch /test token
    3.22  2 start /test
    3.23  2 mkdir /test/dir/sub
    3.24  1 waitwatch' | ./xs_test 2>&1`" = "1:waitwatch timeout" ]
    3.25 -[ "`echo -e '1 watch /test token 100
    3.26 +[ "`echo -e '1 watch /test token
    3.27  2 start /test
    3.28  2 mkdir /test/dir/sub
    3.29  2 abort
    3.30  1 waitwatch' | ./xs_test 2>&1`" = "1:waitwatch timeout" ]
    3.31 -[ "`echo -e '1 watch /test token 100
    3.32 +[ "`echo -e '1 watch /test token
    3.33  2 start /test
    3.34  2 mkdir /test/dir/sub
    3.35 -2 commit
    3.36 +2 async commit
    3.37  1 waitwatch
    3.38  1 ackwatch token' | ./xs_test 2>&1`" = "1:/test/dir/sub:token" ]
    3.39  
    3.40  # Rm inside transaction works like rm outside: children get notified.
    3.41 -[ "`echo -e '1 watch /test/dir/sub token 100
    3.42 +[ "`echo -e '1 watch /test/dir/sub token
    3.43  2 start /test
    3.44  2 rm /test/dir
    3.45 -2 commit
    3.46 +2 async commit
    3.47  1 waitwatch
    3.48  1 ackwatch token' | ./xs_test 2>&1`" = "1:/test/dir/sub:token" ]

     4.1 --- a/tools/xenstore/testsuite/10domain-homedir.sh	Tue Jul 26 15:13:56 2005 +0000
     4.2 +++ b/tools/xenstore/testsuite/10domain-homedir.sh	Tue Jul 26 15:20:09 2005 +0000
     4.3 @@ -13,8 +13,8 @@ entry1" ]
     4.4  # Place a watch using a relative path: expect relative answer.
     4.5  [ "`echo 'introduce 1 100 7 /home
     4.6  1 mkdir foo
     4.7 -1 watch foo token 0
     4.8 -write /home/foo/bar create contents
     4.9 +1 watch foo token
    4.10 +async write /home/foo/bar create contents
    4.11  1 waitwatch
    4.12  1 ackwatch token' | ./xs_test 2>&1`" = "handle is 1
    4.13  1:foo/bar:token" ]

     5.1 --- a/tools/xenstore/testsuite/11domain-watch.sh	Tue Jul 26 15:13:56 2005 +0000
     5.2 +++ b/tools/xenstore/testsuite/11domain-watch.sh	Tue Jul 26 15:20:09 2005 +0000
     5.3 @@ -6,42 +6,46 @@
     5.4  [ "`echo -e 'mkdir /dir' | ./xs_test 2>&1`" = "" ]
     5.5  
     5.6  [ "`echo -e 'introduce 1 100 7 /my/home
     5.7 -1 watch /test token 100
     5.8 -write /test create contents2
     5.9 +1 watch /test token
    5.10 +async write /test create contents2
    5.11  1 waitwatch
    5.12  1 ackwatch token
    5.13  1 unwatch /test token
    5.14 +asyncwait
    5.15  release 1' | ./xs_test 2>&1`" = "handle is 1
    5.16  1:/test:token" ]
    5.17  
    5.18  # ignore watches while doing commands, should work.
    5.19  [ "`echo -e 'introduce 1 100 7 /my/home
    5.20 -1 watch /dir token 100
    5.21 -1 write /dir/test create contents
    5.22 -1 read /dir/test
    5.23 +1 watch /dir token
    5.24 +async write /dir/test create contents
    5.25 +1 write /dir/test2 create contents2
    5.26 +1 write /dir/test3 create contents3
    5.27 +1 write /dir/test4 create contents4
    5.28  1 waitwatch
    5.29  1 ackwatch token
    5.30 +asyncwait
    5.31  release 1' | ./xs_test 2>&1`" = "handle is 1
    5.32 -1:contents
    5.33  1:/dir/test:token" ]
    5.34  
    5.35  # unwatch
    5.36  [ "`echo -e 'introduce 1 100 7 /my/home
    5.37 -1 watch /dir token1 0
    5.38 +1 watch /dir token1
    5.39  1 unwatch /dir token1
    5.40 -1 watch /dir token2 0
    5.41 -2 write /dir/test2 create contents
    5.42 +1 watch /dir token2
    5.43 +async 2 write /dir/test2 create contents
    5.44  1 waitwatch
    5.45  1 unwatch /dir token2
    5.46 +asyncwait
    5.47  release 1' | ./xs_test 2>&1`" = "handle is 1
    5.48  1:/dir/test2:token2" ]
    5.49  
    5.50  # unwatch while watch pending.
    5.51  [ "`echo -e 'introduce 1 100 7 /my/home
    5.52  introduce 2 101 8 /my/secondhome
    5.53 -1 watch /dir token1 0
    5.54 -2 watch /dir token2 1
    5.55 -write /dir/test create contents
    5.56 +1 watch /dir token1
    5.57 +2 watch /dir token2
    5.58 +3 async write /dir/test create contents
    5.59  2 unwatch /dir token2
    5.60  1 waitwatch
    5.61  1 ackwatch token1

     6.1 --- a/tools/xenstore/testsuite/12readonly.sh	Tue Jul 26 15:13:56 2005 +0000
     6.2 +++ b/tools/xenstore/testsuite/12readonly.sh	Tue Jul 26 15:20:09 2005 +0000
     6.3 @@ -9,7 +9,7 @@ tool" ]
     6.4  
     6.5  [ "`echo 'read /test
     6.6  getperm /test
     6.7 -watch /test token 0
     6.8 +watch /test token
     6.9  unwatch /test token 
    6.10  start /
    6.11  commit
    6.12 @@ -27,7 +27,7 @@ 0 READ" ]
    6.13  
    6.14  # Check that watches work like normal.
    6.15  set -m
    6.16 -[ "`echo 'watch / token 0
    6.17 +[ "`echo 'watch / token
    6.18  waitwatch
    6.19  ackwatch token' | ./xs_test --readonly 2>&1`" = "/test:token" ] &
    6.20  
    6.21 @@ -36,6 +36,3 @@ if wait; then :; else
    6.22      echo Readonly wait test failed: $?
    6.23      exit 1
    6.24  fi
    6.25 -    
    6.26 -    
    6.27 -

     7.1 --- a/tools/xenstore/testsuite/13watch-ack.sh	Tue Jul 26 15:13:56 2005 +0000
     7.2 +++ b/tools/xenstore/testsuite/13watch-ack.sh	Tue Jul 26 15:20:09 2005 +0000
     7.3 @@ -15,8 +15,9 @@ echo mkdir /test/3 | ./xs_test
     7.4  [ "`echo '1 watch /test/1 token1 0
     7.5  1 watch /test/2 token2 0
     7.6  1 watch /test/3 token3 0
     7.7 -2 write /test/2 create contents2
     7.8 +2 async write /test/2 create contents2
     7.9  1 waitwatch
    7.10 -2 write /test/1 create contents1
    7.11 -2 write /test/3 create contents3
    7.12 -1 ackwatch token2' | ./xs_test 2>&1`" = "1:/test/2:token2" ]
    7.13 +3 async write /test/1 create contents1
    7.14 +4 async write /test/3 create contents3
    7.15 +1 ackwatch token2
    7.16 +1 close' | ./xs_test 2>&1`" = "1:/test/2:token2" ]

     8.1 --- a/tools/xenstore/testsuite/test.sh	Tue Jul 26 15:13:56 2005 +0000
     8.2 +++ b/tools/xenstore/testsuite/test.sh	Tue Jul 26 15:20:09 2005 +0000
     8.3 @@ -9,7 +9,7 @@ run_test()
     8.4      mkdir $XENSTORED_ROOTDIR
     8.5  # Weird failures with this.
     8.6      if type valgrind >/dev/null 2>&1; then
     8.7 -	valgrind -q --logfile-fd=3 ./xenstored_test --output-pid --no-fork 3>testsuite/tmp/vgout > /tmp/pid 2> testsuite/tmp/xenstored_errors &
     8.8 +	valgrind -q --logfile-fd=3 ./xenstored_test --output-pid --trace-file=testsuite/tmp/trace --no-fork 3>testsuite/tmp/vgout > /tmp/pid 2> testsuite/tmp/xenstored_errors &
     8.9  	while [ ! -s /tmp/pid ]; do sleep 0; done
    8.10  	PID=`cat /tmp/pid`
    8.11  	rm /tmp/pid
    8.12 @@ -38,7 +38,9 @@ for f in testsuite/[0-9]*.sh; do
    8.13  	echo Test $f passed...
    8.14      else
    8.15  	echo Test $f failed, running verbosely...
    8.16 -	run_test $f -x
    8.17 +	run_test $f -x || true
    8.18 +	# That will have filled the screen, repeat message.
    8.19 +	echo Test $f failed
    8.20  	exit 1
    8.21      fi
    8.22  done

     9.1 --- a/tools/xenstore/xenstored_core.c	Tue Jul 26 15:13:56 2005 +0000
     9.2 +++ b/tools/xenstore/xenstored_core.c	Tue Jul 26 15:20:09 2005 +0000
     9.3 @@ -51,7 +51,7 @@
     9.4  #include "xenstored_domain.h"
     9.5  
     9.6  static bool verbose;
     9.7 -static LIST_HEAD(connections);
     9.8 +LIST_HEAD(connections);
     9.9  static int tracefd = -1;
    9.10  
    9.11  #ifdef TESTING
    9.12 @@ -959,8 +959,11 @@ static void do_write(struct connection *
    9.13  	}
    9.14  
    9.15  	add_change_node(conn->transaction, node, false);
    9.16 +	if (fire_watches(conn, node, false)) {
    9.17 +		conn->watch_ack = XS_WRITE;
    9.18 +		return;
    9.19 +	}
    9.20  	send_ack(conn, XS_WRITE);
    9.21 -	fire_watches(conn->transaction, node, false);
    9.22  }
    9.23  
    9.24  static void do_mkdir(struct connection *conn, const char *node)
    9.25 @@ -985,8 +988,11 @@ static void do_mkdir(struct connection *
    9.26  	}
    9.27  
    9.28  	add_change_node(conn->transaction, node, false);
    9.29 +	if (fire_watches(conn, node, false)) {
    9.30 +		conn->watch_ack = XS_MKDIR;
    9.31 +		return;
    9.32 +	}
    9.33  	send_ack(conn, XS_MKDIR);
    9.34 -	fire_watches(conn->transaction, node, false);
    9.35  }
    9.36  
    9.37  static void do_rm(struct connection *conn, const char *node)
    9.38 @@ -1023,8 +1029,11 @@ static void do_rm(struct connection *con
    9.39  	}
    9.40  
    9.41  	add_change_node(conn->transaction, node, true);
    9.42 +	if (fire_watches(conn, node, true)) {
    9.43 +		conn->watch_ack = XS_RM;
    9.44 +		return;
    9.45 +	}
    9.46  	send_ack(conn, XS_RM);
    9.47 -	fire_watches(conn->transaction, node, true);
    9.48  }
    9.49  
    9.50  static void do_get_perms(struct connection *conn, const char *node)
    9.51 @@ -1095,8 +1104,11 @@ static void do_set_perms(struct connecti
    9.52  	}
    9.53  
    9.54  	add_change_node(conn->transaction, node, false);
    9.55 +	if (fire_watches(conn, node, false)) {
    9.56 +		conn->watch_ack = XS_SET_PERMS;
    9.57 +		return;
    9.58 +	}
    9.59  	send_ack(conn, XS_SET_PERMS);
    9.60 -	fire_watches(conn->transaction, node, false);
    9.61  }
    9.62  
    9.63  /* Process "in" for conn: "in" will vanish after this conversation, so
    9.64 @@ -1321,14 +1333,23 @@ static void unblock_connections(void)
    9.65  	struct connection *i, *tmp;
    9.66  
    9.67  	list_for_each_entry_safe(i, tmp, &connections, list) {
    9.68 -		if (i->state == OK)
    9.69 -			continue;
    9.70 -
    9.71 -		if (!transaction_covering_node(i->blocked_by)) {
    9.72 -			talloc_free(i->blocked_by);
    9.73 -			i->blocked_by = NULL;
    9.74 -			i->state = OK;
    9.75 -			consider_message(i);
    9.76 +		switch (i->state) {
    9.77 +		case BLOCKED:
    9.78 +			if (!transaction_covering_node(i->blocked_by)) {
    9.79 +				talloc_free(i->blocked_by);
    9.80 +				i->blocked_by = NULL;
    9.81 +				i->state = OK;
    9.82 +				consider_message(i);
    9.83 +			}
    9.84 +			break;
    9.85 +		case WATCHED:
    9.86 +			if (i->watches_unacked == 0) {
    9.87 +				i->state = OK;
    9.88 +				send_ack(i, i->watch_ack);
    9.89 +			}
    9.90 +			break;
    9.91 +		case OK:
    9.92 +			break;
    9.93  		}
    9.94  	}
    9.95  
    9.96 @@ -1351,6 +1372,8 @@ struct connection *new_connection(connwr
    9.97  
    9.98  	new->state = OK;
    9.99  	new->blocked_by = NULL;
   9.100 +	new->watch_ack = XS_ERROR;
   9.101 +	new->watches_unacked = 0;
   9.102  	new->out = new->waiting_reply = NULL;
   9.103  	new->fd = -1;
   9.104  	new->id = 0;
   9.105 @@ -1359,6 +1382,7 @@ struct connection *new_connection(connwr
   9.106  	new->write = write;
   9.107  	new->read = read;
   9.108  	new->can_write = true;
   9.109 +	INIT_LIST_HEAD(&new->watches);
   9.110  
   9.111  	talloc_set_fail_handler(out_of_mem, &talloc_fail);
   9.112  	if (setjmp(talloc_fail)) {
   9.113 @@ -1430,13 +1454,12 @@ void dump_connection(void)
   9.114  		printf("    state = %s\n",
   9.115  		       i->state == OK ? "OK"
   9.116  		       : i->state == BLOCKED ? "BLOCKED"
   9.117 +		       : i->state == WATCHED ? "WATCHED"
   9.118  		       : "INVALID");
   9.119  		if (i->id)
   9.120  			printf("    id = %i\n", i->id);
   9.121  		if (i->blocked_by)
   9.122  			printf("    blocked on = %s\n", i->blocked_by);
   9.123 -		if (i->waiting_for_ack)
   9.124 -			printf("    waiting_for_ack TRUE\n");
   9.125  		if (!i->in->inhdr || i->in->used)
   9.126  			printf("    got %i bytes of %s\n",
   9.127  			       i->in->used, i->in->inhdr ? "header" : "data");

    10.1 --- a/tools/xenstore/xenstored_core.h	Tue Jul 26 15:13:56 2005 +0000
    10.2 +++ b/tools/xenstore/xenstored_core.h	Tue Jul 26 15:20:09 2005 +0000
    10.3 @@ -51,6 +51,8 @@ enum state
    10.4  {
    10.5  	/* Blocked by transaction. */
    10.6  	BLOCKED,
    10.7 +	/* Waiting for watchers to ack event we caused */
    10.8 +	WATCHED,
    10.9  	/* Completed */
   10.10  	OK,
   10.11  };
   10.12 @@ -71,6 +73,12 @@ struct connection
   10.13  	/* Node we are waiting for (if state == BLOCKED) */
   10.14  	char *blocked_by;
   10.15  
   10.16 +	/* Are we waiting for watches to be acked from an event we caused? */
   10.17 +	unsigned int watches_unacked;
   10.18 +
   10.19 +	/* Type of ack to send once watches fired. */
   10.20 +	enum xsd_sockmsg_type watch_ack;
   10.21 +
   10.22  	/* Is this a read-only connection? */
   10.23  	bool can_write;
   10.24  
   10.25 @@ -92,10 +100,14 @@ struct connection
   10.26  	/* The domain I'm associated with, if any. */
   10.27  	struct domain *domain;
   10.28  
   10.29 +	/* My watches. */
   10.30 +	struct list_head watches;
   10.31 +
   10.32  	/* Methods for communicating over this connection: write can be NULL */
   10.33  	connwritefn_t *write;
   10.34  	connreadfn_t *read;
   10.35  };
   10.36 +extern struct list_head connections;
   10.37  
   10.38  /* Return length of string (including nul) at this offset. */
   10.39  unsigned int get_string(const struct buffered_data *data,

    11.1 --- a/tools/xenstore/xenstored_transaction.c	Tue Jul 26 15:13:56 2005 +0000
    11.2 +++ b/tools/xenstore/xenstored_transaction.c	Tue Jul 26 15:20:09 2005 +0000
    11.3 @@ -288,7 +288,6 @@ void do_transaction_start(struct connect
    11.4  static bool commit_transaction(struct transaction *trans)
    11.5  {
    11.6  	char *tmp, *dir;
    11.7 -	struct changed_node *i;
    11.8  
    11.9  	/* Move: orig -> .old, repl -> orig.  Cleanup deletes .old. */
   11.10  	dir = node_dir_outside_transaction(trans->node);
   11.11 @@ -301,15 +300,15 @@ static bool commit_transaction(struct tr
   11.12  			trans->divert, dir);
   11.13  
   11.14  	trans->divert = tmp;
   11.15 -
   11.16 -	/* Fire off the watches for everything that changed. */
   11.17 -	list_for_each_entry(i, &trans->changes, list)
   11.18 -		fire_watches(NULL, i->node, i->recurse);
   11.19  	return true;
   11.20  }
   11.21  
   11.22  void do_transaction_end(struct connection *conn, const char *arg)
   11.23  {
   11.24 +	struct changed_node *i;
   11.25 +	struct transaction *trans;
   11.26 +	bool fired = false;
   11.27 +
   11.28  	if (!arg || (!streq(arg, "T") && !streq(arg, "F"))) {
   11.29  		send_error(conn, EINVAL);
   11.30  		return;
   11.31 @@ -320,24 +319,30 @@ void do_transaction_end(struct connectio
   11.32  		return;
   11.33  	}
   11.34  
   11.35 +	/* Set to NULL so fire_watches sends events. */
   11.36 +	trans = conn->transaction;
   11.37 +	conn->transaction = NULL;
   11.38 +	/* Attach transaction to arg for auto-cleanup */
   11.39 +	talloc_steal(arg, trans);
   11.40 +
   11.41  	if (streq(arg, "T")) {
   11.42 -		if (conn->transaction->destined_to_fail) {
   11.43 +		if (trans->destined_to_fail) {
   11.44  			send_error(conn, ETIMEDOUT);
   11.45 -			goto failed;
   11.46 +			return;
   11.47  		}
   11.48 -		if (!commit_transaction(conn->transaction)) {
   11.49 +		if (!commit_transaction(trans)) {
   11.50  			send_error(conn, errno);
   11.51 -			goto failed;
   11.52 +			return;
   11.53  		}
   11.54 +
   11.55 +		/* Fire off the watches for everything that changed. */
   11.56 +		list_for_each_entry(i, &trans->changes, list)
   11.57 +			fired |= fire_watches(conn, i->node, i->recurse);
   11.58  	}
   11.59  
   11.60 -	talloc_free(conn->transaction);
   11.61 -	conn->transaction = NULL;
   11.62 -	send_ack(conn, XS_TRANSACTION_END);
   11.63 -	return;
   11.64 -
   11.65 -failed:
   11.66 -	talloc_free(conn->transaction);
   11.67 -	conn->transaction = NULL;
   11.68 +	if (fired)
   11.69 +		conn->watch_ack = XS_TRANSACTION_END;
   11.70 +	else
   11.71 +		send_ack(conn, XS_TRANSACTION_END);
   11.72  }
   11.73  

    12.1 --- a/tools/xenstore/xenstored_watch.c	Tue Jul 26 15:13:56 2005 +0000
    12.2 +++ b/tools/xenstore/xenstored_watch.c	Tue Jul 26 15:20:09 2005 +0000
    12.3 @@ -33,69 +33,39 @@
    12.4  #include "xenstored_domain.h"
    12.5  
    12.6  /* FIXME: time out unacked watches. */
    12.7 -
    12.8 -/* We create this if anyone is interested "node", then we pass it from
    12.9 - * watch to watch as each connection acks it.
   12.10 - */
   12.11  struct watch_event
   12.12  {
   12.13 -	/* The watch we are firing for (watch->events) */
   12.14 +	/* The events on this watch. */
   12.15  	struct list_head list;
   12.16  
   12.17 -	/* Watches we need to fire for (watches[0]->events == this). */
   12.18 -	struct watch **watches;
   12.19 -	unsigned int num_watches;
   12.20 -
   12.21 -	struct timeval timeout;
   12.22 +	/* Data to send (node\0token\0). */
   12.23 +	unsigned int len;
   12.24 +	char *data;
   12.25  
   12.26 -	/* Name of node which changed. */
   12.27 -	char *node;
   12.28 -
   12.29 -	/* For remove, we trigger on all the children of this node too. */
   12.30 -	bool recurse;
   12.31 +	/* Connection which caused watch event (which we are blocking) */
   12.32 +	struct connection *cause;
   12.33  };
   12.34  
   12.35  struct watch
   12.36  {
   12.37 +	/* Watches on this connection */
   12.38  	struct list_head list;
   12.39 -	unsigned int priority;
   12.40  
   12.41  	/* Current outstanding events applying to this watch. */
   12.42  	struct list_head events;
   12.43  
   12.44  	/* Is this relative to connnection's implicit path? */
   12.45 -	bool relative;
   12.46 +	const char *relative_path;
   12.47  
   12.48  	char *token;
   12.49  	char *node;
   12.50 -	struct connection *conn;
   12.51  };
   12.52 -static LIST_HEAD(watches);
   12.53 -
   12.54 -static struct watch_event *get_first_event(struct connection *conn)
   12.55 -{
   12.56 -	struct watch *watch;
   12.57 -	struct watch_event *event;
   12.58 -
   12.59 -	/* Find first watch with an event. */
   12.60 -	list_for_each_entry(watch, &watches, list) {
   12.61 -		if (watch->conn != conn)
   12.62 -			continue;
   12.63 -
   12.64 -		event = list_top(&watch->events, struct watch_event, list);
   12.65 -		if (event)
   12.66 -			return event;
   12.67 -	}
   12.68 -	return NULL;
   12.69 -}
   12.70  
   12.71  /* Look through our watches: if any of them have an event, queue it. */
   12.72  void queue_next_event(struct connection *conn)
   12.73  {
   12.74  	struct watch_event *event;
   12.75 -	const char *node;
   12.76 -	char *buffer;
   12.77 -	unsigned int len;
   12.78 +	struct watch *watch;
   12.79  
   12.80  	/* We had a reply queued already?  Send it: other end will
   12.81  	 * discard watch. */
   12.82 @@ -110,170 +80,93 @@ void queue_next_event(struct connection 
   12.83  	if (conn->waiting_for_ack)
   12.84  		return;
   12.85  
   12.86 -	event = get_first_event(conn);
   12.87 -	if (!event)
   12.88 -		return;
   12.89 -
   12.90 -	/* If we decide to cancel, we will reset this. */
   12.91 -	conn->waiting_for_ack = event->watches[0];
   12.92 -
   12.93 -	/* If we deleted /foo and they're watching /foo/bar, that's what we
   12.94 -	 * tell them has changed. */
   12.95 -	if (!is_child(event->node, event->watches[0]->node)) {
   12.96 -		assert(event->recurse);
   12.97 -		node = event->watches[0]->node;
   12.98 -	} else
   12.99 -		node = event->node;
  12.100 -
  12.101 -	/* If watch placed using relative path, give them relative answer. */
  12.102 -	if (event->watches[0]->relative) {
  12.103 -		node += strlen(get_implicit_path(conn));
  12.104 -		if (node[0] == '/') /* Could be "". */
  12.105 -			node++;
  12.106 -	}
  12.107 -
  12.108 -	/* Create reply from path and token */
  12.109 -	len = strlen(node) + 1 + strlen(event->watches[0]->token) + 1;
  12.110 -	buffer = talloc_array(conn, char, len);
  12.111 -	strcpy(buffer, node);
  12.112 -	strcpy(buffer+strlen(node)+1, event->watches[0]->token);
  12.113 -	send_reply(conn, XS_WATCH_EVENT, buffer, len);
  12.114 -	talloc_free(buffer);
  12.115 -}
  12.116 -
  12.117 -static struct watch **find_watches(const char *node, bool recurse,
  12.118 -				   unsigned int *num)
  12.119 -{
  12.120 -	struct watch *i;
  12.121 -	struct watch **ret = NULL;
  12.122 -
  12.123 -	*num = 0;
  12.124 -
  12.125 -	/* We include children too if this is an rm. */
  12.126 -	list_for_each_entry(i, &watches, list) {
  12.127 -		if (is_child(node, i->node) ||
  12.128 -		    (recurse && is_child(i->node, node))) {
  12.129 -			(*num)++;
  12.130 -			ret = talloc_realloc(node, ret, struct watch *, *num);
  12.131 -			ret[*num - 1] = i;
  12.132 -		}
  12.133 -	}
  12.134 -	return ret;
  12.135 -}
  12.136 -
  12.137 -/* FIXME: we fail to fire on out of memory.  Should drop connections. */
  12.138 -void fire_watches(struct transaction *trans, const char *node, bool recurse)
  12.139 -{
  12.140 -	struct watch **watches;
  12.141 -	struct watch_event *event;
  12.142 -	unsigned int num_watches;
  12.143 -
  12.144 -	/* During transactions, don't fire watches. */
  12.145 -	if (trans)
  12.146 -		return;
  12.147 -
  12.148 -	watches = find_watches(node, recurse, &num_watches);
  12.149 -	if (!watches)
  12.150 -		return;
  12.151 -
  12.152 -	/* Create and fill in info about event. */
  12.153 -	event = talloc(talloc_autofree_context(), struct watch_event);
  12.154 -	event->node = talloc_strdup(event, node);
  12.155 -
  12.156 -	/* Tie event to this watch. */
  12.157 -	event->watches = watches;
  12.158 -	talloc_steal(event, watches);
  12.159 -	event->num_watches = num_watches;
  12.160 -	event->recurse = recurse;
  12.161 -	list_add_tail(&event->list, &watches[0]->events);
  12.162 -
  12.163 -	/* Warn if not finished after thirty seconds. */
  12.164 -	gettimeofday(&event->timeout, NULL);
  12.165 -	event->timeout.tv_sec += 30;
  12.166 -
  12.167 -	/* If connection not doing anything, queue this. */
  12.168 -	if (!watches[0]->conn->out)
  12.169 -		queue_next_event(watches[0]->conn);
  12.170 -}
  12.171 -
  12.172 -/* We're done with this event: see if anyone else wants it. */
  12.173 -static void move_event_onwards(struct watch_event *event)
  12.174 -{
  12.175 -	list_del(&event->list);
  12.176 -
  12.177 -	event->num_watches--;
  12.178 -	event->watches++;
  12.179 -	if (!event->num_watches) {
  12.180 -		talloc_free(event);
  12.181 -		return;
  12.182 -	}
  12.183 -
  12.184 -	list_add_tail(&event->list, &event->watches[0]->events);
  12.185 -
  12.186 -	/* If connection not doing anything, queue this. */
  12.187 -	if (!event->watches[0]->conn->out)
  12.188 -		queue_next_event(event->watches[0]->conn);
  12.189 -}
  12.190 -
  12.191 -static void remove_watch_from_events(struct watch *dying_watch)
  12.192 -{
  12.193 -	struct watch *watch;
  12.194 -	struct watch_event *event;
  12.195 -	unsigned int i;
  12.196 -
  12.197 -	list_for_each_entry(watch, &watches, list) {
  12.198 -		list_for_each_entry(event, &watch->events, list) {
  12.199 -			for (i = 0; i < event->num_watches; i++) {
  12.200 -				if (event->watches[i] != dying_watch)
  12.201 -					continue;
  12.202 -
  12.203 -				assert(i != 0);
  12.204 -				memmove(event->watches+i,
  12.205 -					event->watches+i+1,
  12.206 -					(event->num_watches - (i+1))
  12.207 -					* sizeof(struct watch *));
  12.208 -				event->num_watches--;
  12.209 -			}
  12.210 +	list_for_each_entry(watch, &conn->watches, list) {
  12.211 +		event = list_top(&watch->events, struct watch_event, list);
  12.212 +		if (event) {
  12.213 +			conn->waiting_for_ack = watch;
  12.214 +			send_reply(conn,XS_WATCH_EVENT,event->data,event->len);
  12.215 +			break;
  12.216  		}
  12.217  	}
  12.218  }
  12.219  
  12.220 -static int destroy_watch(void *_watch)
  12.221 +static int destroy_watch_event(void *_event)
  12.222  {
  12.223 -	struct watch *watch = _watch;
  12.224 -	struct watch_event *event;
  12.225 +	struct watch_event *event = _event;
  12.226  
  12.227 -	/* If we have pending events, pass them on to others. */
  12.228 -	while ((event = list_top(&watch->events, struct watch_event, list)))
  12.229 -		move_event_onwards(event);
  12.230 -
  12.231 -	/* Remove from global list. */
  12.232 -	list_del(&watch->list);
  12.233 -
  12.234 -	/* Other events which match this watch must be cleared. */
  12.235 -	remove_watch_from_events(watch);
  12.236 -
  12.237 -	trace_destroy(watch, "watch");
  12.238 +	trace_destroy(event, "watch_event");
  12.239 +	assert(event->cause->watches_unacked != 0);
  12.240 +	/* If it hits zero, will unblock in unblock_connections. */
  12.241 +	event->cause->watches_unacked--;
  12.242  	return 0;
  12.243  }
  12.244  
  12.245 -/* We keep watches in priority order. */
  12.246 -static void insert_watch(struct watch *watch)
  12.247 +static void add_event(struct connection *cause, struct watch *watch,
  12.248 +		      const char *node)
  12.249  {
  12.250 -	struct watch *i;
  12.251 +	struct watch_event *event;
  12.252 +
  12.253 +	if (watch->relative_path) {
  12.254 +		node += strlen(watch->relative_path);
  12.255 +		if (*node == '/') /* Could be "" */
  12.256 +			node++;
  12.257 +	}
  12.258 +
  12.259 +	event = talloc(watch, struct watch_event);
  12.260 +	event->len = strlen(node) + 1 + strlen(watch->token) + 1;
  12.261 +	event->data = talloc_array(event, char, event->len);
  12.262 +	strcpy(event->data, node);
  12.263 +	strcpy(event->data + strlen(node) + 1, watch->token);
  12.264 +	event->cause = cause;
  12.265 +	cause->watches_unacked++;
  12.266 +	talloc_set_destructor(event, destroy_watch_event);
  12.267 +	list_add_tail(&event->list, &watch->events);
  12.268 +	trace_create(event, "watch_event");
  12.269 +}
  12.270  
  12.271 -	list_for_each_entry(i, &watches, list) {
  12.272 -		if (i->priority <= watch->priority) {
  12.273 -			list_add_tail(&watch->list, &i->list);
  12.274 -			return;
  12.275 +/* FIXME: we fail to fire on out of memory.  Should drop connections. */
  12.276 +bool fire_watches(struct connection *conn, const char *node, bool recurse)
  12.277 +{
  12.278 +	struct connection *i;
  12.279 +	struct watch *watch;
  12.280 +
  12.281 +	/* During transactions, don't fire watches. */
  12.282 +	if (conn->transaction)
  12.283 +		return false;
  12.284 +
  12.285 +	assert(conn->state == OK);
  12.286 +
  12.287 +	/* Create an event for each watch.  Don't send to self. */
  12.288 +	list_for_each_entry(i, &connections, list) {
  12.289 +		if (i == conn)
  12.290 +			continue;
  12.291 +
  12.292 +		list_for_each_entry(watch, &i->watches, list) {
  12.293 +			if (is_child(node, watch->node))
  12.294 +				add_event(conn, watch, node);
  12.295 +			else if (recurse && is_child(watch->node, node))
  12.296 +				add_event(conn, watch, watch->node);
  12.297 +			else
  12.298 +				continue;
  12.299 +			conn->state = WATCHED;
  12.300 +			/* If connection not doing anything, queue this. */
  12.301 +			if (!i->out)
  12.302 +				queue_next_event(i);
  12.303  		}
  12.304  	}
  12.305 +	return conn->state == WATCHED;
  12.306 +}
  12.307  
  12.308 -	list_add_tail(&watch->list, &watches);
  12.309 +static int destroy_watch(void *_watch)
  12.310 +{
  12.311 +	trace_destroy(_watch, "watch");
  12.312 +	return 0;
  12.313  }
  12.314  
  12.315  void shortest_watch_ack_timeout(struct timeval *tv)
  12.316  {
  12.317 +	(void)tv;
  12.318 +#if 0 /* FIXME */
  12.319  	struct watch *watch;
  12.320  
  12.321  	list_for_each_entry(watch, &watches, list) {
  12.322 @@ -285,10 +178,12 @@ void shortest_watch_ack_timeout(struct t
  12.323  				*tv = i->timeout;
  12.324  		}
  12.325  	}
  12.326 +#endif
  12.327  }	
  12.328  
  12.329  void check_watch_ack_timeout(void)
  12.330  {
  12.331 +#if 0
  12.332  	struct watch *watch;
  12.333  	struct timeval now;
  12.334  
  12.335 @@ -308,12 +203,13 @@ void check_watch_ack_timeout(void)
  12.336  			}
  12.337  		}
  12.338  	}
  12.339 +#endif
  12.340  }
  12.341  
  12.342  void do_watch(struct connection *conn, struct buffered_data *in)
  12.343  {
  12.344  	struct watch *watch;
  12.345 -	char *vec[3];
  12.346 +	char *vec[2];
  12.347  	bool relative;
  12.348  
  12.349  	if (get_strings(in, vec, ARRAY_SIZE(vec)) != ARRAY_SIZE(vec)) {
  12.350 @@ -331,14 +227,16 @@ void do_watch(struct connection *conn, s
  12.351  	watch = talloc(conn, struct watch);
  12.352  	watch->node = talloc_strdup(watch, vec[0]);
  12.353  	watch->token = talloc_strdup(watch, vec[1]);
  12.354 -	watch->conn = conn;
  12.355 -	watch->priority = strtoul(vec[2], NULL, 0);
  12.356 -	watch->relative = relative;
  12.357 +	if (relative)
  12.358 +		watch->relative_path = get_implicit_path(conn);
  12.359 +	else
  12.360 +		watch->relative_path = NULL;
  12.361 +
  12.362  	INIT_LIST_HEAD(&watch->events);
  12.363  
  12.364 -	insert_watch(watch);
  12.365 +	list_add_tail(&watch->list, &conn->watches);
  12.366 +	trace_create(watch, "watch");
  12.367  	talloc_set_destructor(watch, destroy_watch);
  12.368 -	trace_create(watch, "watch");
  12.369  	send_ack(conn, XS_WATCH);
  12.370  }
  12.371  
  12.372 @@ -356,9 +254,6 @@ void do_watch_ack(struct connection *con
  12.373  		return;
  12.374  	}
  12.375  
  12.376 -	event = list_top(&conn->waiting_for_ack->events,
  12.377 -			 struct watch_event, list);
  12.378 -	assert(event->watches[0] == conn->waiting_for_ack);
  12.379  	if (!streq(conn->waiting_for_ack->token, token)) {
  12.380  		/* They're confused: this will cause us to send event again */
  12.381  		conn->waiting_for_ack = NULL;
  12.382 @@ -366,7 +261,12 @@ void do_watch_ack(struct connection *con
  12.383  		return;
  12.384  	}
  12.385  
  12.386 -	move_event_onwards(event);
  12.387 +	/* Remove event: after ack sent, core will call queue_next_event */
  12.388 +	event = list_top(&conn->waiting_for_ack->events, struct watch_event,
  12.389 +			 list);
  12.390 +	list_del(&event->list);
  12.391 +	talloc_free(event);
  12.392 +
  12.393  	conn->waiting_for_ack = NULL;
  12.394  	send_ack(conn, XS_WATCH_ACK);
  12.395  }
  12.396 @@ -385,11 +285,9 @@ void do_unwatch(struct connection *conn,
  12.397  	 * watch we're deleting: conn->waiting_for_ack was reset by
  12.398  	 * this command in consider_message anyway. */
  12.399  	node = canonicalize(conn, vec[0]);
  12.400 -	list_for_each_entry(watch, &watches, list) {
  12.401 -		if (watch->conn != conn)
  12.402 -			continue;
  12.403 -
  12.404 +	list_for_each_entry(watch, &conn->watches, list) {
  12.405  		if (streq(watch->node, node) && streq(watch->token, vec[1])) {
  12.406 +			list_del(&watch->list);
  12.407  			talloc_free(watch);
  12.408  			send_ack(conn, XS_UNWATCH);
  12.409  			return;
  12.410 @@ -404,15 +302,16 @@ void dump_watches(struct connection *con
  12.411  	struct watch *watch;
  12.412  	struct watch_event *event;
  12.413  
  12.414 -	/* Find first watch with an event. */
  12.415 -	list_for_each_entry(watch, &watches, list) {
  12.416 -		if (watch->conn != conn)
  12.417 -			continue;
  12.418 +	if (conn->waiting_for_ack)
  12.419 +		printf("    waiting_for_ack for watch on %s token %s\n",
  12.420 +		       conn->waiting_for_ack->node,
  12.421 +		       conn->waiting_for_ack->token);
  12.422  
  12.423 -		printf("    watch on %s token %s prio %i\n",
  12.424 -		       watch->node, watch->token, watch->priority);
  12.425 +	list_for_each_entry(watch, &conn->watches, list) {
  12.426 +		printf("    watch on %s token %s\n",
  12.427 +		       watch->node, watch->token);
  12.428  		list_for_each_entry(event, &watch->events, list)
  12.429 -			printf("        event: %s\n", event->node);
  12.430 +			printf("        event: %s\n", event->data);
  12.431  	}
  12.432  }
  12.433  #endif

    13.1 --- a/tools/xenstore/xenstored_watch.h	Tue Jul 26 15:13:56 2005 +0000
    13.2 +++ b/tools/xenstore/xenstored_watch.h	Tue Jul 26 15:20:09 2005 +0000
    13.3 @@ -32,8 +32,10 @@ bool is_watch_event(struct connection *c
    13.4  /* Look through our watches: if any of them have an event, queue it. */
    13.5  void queue_next_event(struct connection *conn);
    13.6  
    13.7 -/* Fire all watches: recurse means all the children are effected (ie. rm) */
    13.8 -void fire_watches(struct transaction *trans, const char *node, bool recurse);
    13.9 +/* Fire all watches: recurse means all the children are effected (ie. rm).
   13.10 + * Returns true if there were any, meaning connection has to wait.
   13.11 + */
   13.12 +bool fire_watches(struct connection *conn, const char *node, bool recurse);
   13.13  
   13.14  /* Find shortest timeout: if any, reduce tv (may already be set). */
   13.15  void shortest_watch_ack_timeout(struct timeval *tv);

    14.1 --- a/tools/xenstore/xs.c	Tue Jul 26 15:13:56 2005 +0000
    14.2 +++ b/tools/xenstore/xs.c	Tue Jul 26 15:20:09 2005 +0000
    14.3 @@ -401,22 +401,16 @@ unwind:
    14.4  /* Watch a node for changes (poll on fd to detect, or call read_watch()).
    14.5   * When the node (or any child) changes, fd will become readable.
    14.6   * Token is returned when watch is read, to allow matching.
    14.7 - * Priority indicates order if multiple watchers: higher is first.
    14.8   * Returns false on failure.
    14.9   */
   14.10 -bool xs_watch(struct xs_handle *h, const char *path, const char *token,
   14.11 -	      unsigned int priority)
   14.12 +bool xs_watch(struct xs_handle *h, const char *path, const char *token)
   14.13  {
   14.14 -	char prio[MAX_STRLEN(priority)];
   14.15 -	struct iovec iov[3];
   14.16 +	struct iovec iov[2];
   14.17  
   14.18 -	sprintf(prio, "%u", priority);
   14.19  	iov[0].iov_base = (void *)path;
   14.20  	iov[0].iov_len = strlen(path) + 1;
   14.21  	iov[1].iov_base = (void *)token;
   14.22  	iov[1].iov_len = strlen(token) + 1;
   14.23 -	iov[2].iov_base = prio;
   14.24 -	iov[2].iov_len = strlen(prio) + 1;
   14.25  
   14.26  	return xs_bool(xs_talkv(h, XS_WATCH, iov, ARRAY_SIZE(iov), NULL));
   14.27  }

    15.1 --- a/tools/xenstore/xs.h	Tue Jul 26 15:13:56 2005 +0000
    15.2 +++ b/tools/xenstore/xs.h	Tue Jul 26 15:20:09 2005 +0000
    15.3 @@ -82,11 +82,9 @@ bool xs_set_permissions(struct xs_handle
    15.4  /* Watch a node for changes (poll on fd to detect, or call read_watch()).
    15.5   * When the node (or any child) changes, fd will become readable.
    15.6   * Token is returned when watch is read, to allow matching.
    15.7 - * Priority indicates order if multiple watchers: higher is first.
    15.8   * Returns false on failure.
    15.9   */
   15.10 -bool xs_watch(struct xs_handle *h, const char *path, const char *token,
   15.11 -	      unsigned int priority);
   15.12 +bool xs_watch(struct xs_handle *h, const char *path, const char *token);
   15.13  
   15.14  /* Return the FD to poll on to see if a watch has fired. */
   15.15  int xs_fileno(struct xs_handle *h);

    16.1 --- a/tools/xenstore/xs_test.c	Tue Jul 26 15:13:56 2005 +0000
    16.2 +++ b/tools/xenstore/xs_test.c	Tue Jul 26 15:20:09 2005 +0000
    16.3 @@ -20,6 +20,7 @@
    16.4  #include <stdio.h>
    16.5  #include <stdlib.h>
    16.6  #include <sys/types.h>
    16.7 +#include <sys/wait.h>
    16.8  #include <sys/stat.h>
    16.9  #include <fcntl.h>
   16.10  #include <signal.h>
   16.11 @@ -33,6 +34,10 @@
   16.12  #define XSTEST
   16.13  
   16.14  static struct xs_handle *handles[10] = { NULL };
   16.15 +static unsigned int children;
   16.16 +
   16.17 +static bool timeout = true;
   16.18 +static bool readonly = false;
   16.19  
   16.20  struct ringbuf_head
   16.21  {
   16.22 @@ -173,7 +178,9 @@ static void __attribute__((noreturn)) us
   16.23  	     "  getperm <path>\n"
   16.24  	     "  setperm <path> <id> <flags> ...\n"
   16.25  	     "  shutdown\n"
   16.26 -	     "  watch <path> <token> <prio>\n"
   16.27 +	     "  watch <path> <token>\n"
   16.28 +	     "  async <command>...\n"
   16.29 +	     "  asyncwait\n"
   16.30  	     "  waitwatch\n"
   16.31  	     "  ackwatch <token>\n"
   16.32  	     "  unwatch <path> <token>\n"
   16.33 @@ -186,22 +193,34 @@ static void __attribute__((noreturn)) us
   16.34  	     "  dump\n");
   16.35  }
   16.36  
   16.37 +static int argpos(const char *line, unsigned int num)
   16.38 +{
   16.39 +	unsigned int i, len = 0, off = 0;
   16.40 +
   16.41 +	for (i = 0; i <= num; i++) {
   16.42 +		off += len;
   16.43 +		off += strspn(line + off, " \t\n");
   16.44 +		len = strcspn(line + off, " \t\n");
   16.45 +		if (!len)
   16.46 +			return off;
   16.47 +	}
   16.48 +	return off;
   16.49 +}
   16.50 +
   16.51  static char *arg(char *line, unsigned int num)
   16.52  {
   16.53  	static char *args[10];
   16.54 -	unsigned int i, len = 0;
   16.55 +	unsigned int off, len;
   16.56  
   16.57 -	for (i = 0; i <= num; i++) {
   16.58 -		line += len;
   16.59 -		line += strspn(line, " \t\n");
   16.60 -		len = strcspn(line, " \t\n");
   16.61 -		if (!len)
   16.62 -			barf("Can't get arg %u", num);
   16.63 -	}
   16.64 +	off = argpos(line, num);
   16.65 +	len = strcspn(line + off, " \t\n");
   16.66 +
   16.67 +	if (!len)
   16.68 +		barf("Can't get arg %u", num);
   16.69  
   16.70  	free(args[num]);
   16.71  	args[num] = malloc(len + 1);
   16.72 -	memcpy(args[num], line, len);
   16.73 +	memcpy(args[num], line+off, len);
   16.74  	args[num][len] = '\0';
   16.75  	return args[num];
   16.76  }
   16.77 @@ -360,10 +379,9 @@ static void do_shutdown(unsigned int han
   16.78  		failed(handle);
   16.79  }
   16.80  
   16.81 -static void do_watch(unsigned int handle, const char *node, const char *token,
   16.82 -		     const char *pri)
   16.83 +static void do_watch(unsigned int handle, const char *node, const char *token)
   16.84  {
   16.85 -	if (!xs_watch(handles[handle], node, token, atoi(pri)))
   16.86 +	if (!xs_watch(handles[handle], node, token))
   16.87  		failed(handle);
   16.88  }
   16.89  
   16.90 @@ -388,6 +406,47 @@ static void do_ackwatch(unsigned int han
   16.91  		failed(handle);
   16.92  }
   16.93  
   16.94 +/* Async wait for watch on handle */
   16.95 +static void do_command(unsigned int default_handle, char *line);
   16.96 +static void do_async(unsigned int handle, char *line)
   16.97 +{
   16.98 +	int child;
   16.99 +	unsigned int i;
  16.100 +	children++;
  16.101 +	if ((child = fork()) != 0)
  16.102 +		return;
  16.103 +
  16.104 +	/* Don't keep other handles open in parent. */
  16.105 +	for (i = 0; i < ARRAY_SIZE(handles); i++) {
  16.106 +		if (handles[i] && i != handle) {
  16.107 +			xs_daemon_close(handles[i]);
  16.108 +			handles[i] = NULL;
  16.109 +		}
  16.110 +	}
  16.111 +
  16.112 +	do_command(handle, line + argpos(line, 1));
  16.113 +	exit(0);
  16.114 +}
  16.115 +
  16.116 +static void do_asyncwait(unsigned int handle)
  16.117 +{
  16.118 +	int status;
  16.119 +
  16.120 +	if (handle)
  16.121 +		barf("handle has no meaning with asyncwait");
  16.122 +
  16.123 +	if (children == 0)
  16.124 +		barf("No children to wait for!");
  16.125 +
  16.126 +	if (waitpid(0, &status, 0) > 0) {
  16.127 +		if (!WIFEXITED(status))
  16.128 +			barf("async died");
  16.129 +		if (WEXITSTATUS(status))
  16.130 +			exit(WEXITSTATUS(status));
  16.131 +	}
  16.132 +	children--;
  16.133 +}
  16.134 +
  16.135  static void do_unwatch(unsigned int handle, const char *node, const char *token)
  16.136  {
  16.137  	if (!xs_unwatch(handles[handle], node, token))
  16.138 @@ -533,23 +592,106 @@ static void dump(int handle)
  16.139  	free(subdirs);
  16.140  }
  16.141  
  16.142 +static int handle;
  16.143 +
  16.144 +static void alarmed(int sig __attribute__((unused)))
  16.145 +{
  16.146 +	if (handle) {
  16.147 +		char handlename[10];
  16.148 +		sprintf(handlename, "%u:", handle);
  16.149 +		write(STDOUT_FILENO, handlename, strlen(handlename));
  16.150 +	}
  16.151 +	write(STDOUT_FILENO, command, strlen(command));
  16.152 +	write(STDOUT_FILENO, " timeout\n", strlen(" timeout\n"));
  16.153 +	exit(1);
  16.154 +}
  16.155 +
  16.156 +static void do_command(unsigned int default_handle, char *line)
  16.157 +{
  16.158 +	char *endp;
  16.159 +
  16.160 +	if (strspn(line, " \n") == strlen(line))
  16.161 +		return;
  16.162 +	if (strstarts(line, "#"))
  16.163 +		return;
  16.164 +
  16.165 +	handle = strtoul(line, &endp, 10);
  16.166 +	if (endp != line)
  16.167 +		memmove(line, endp+1, strlen(endp));
  16.168 +	else
  16.169 +		handle = default_handle;
  16.170 +
  16.171 +	if (!handles[handle]) {
  16.172 +		if (readonly)
  16.173 +			handles[handle] = xs_daemon_open_readonly();
  16.174 +		else
  16.175 +			handles[handle] = xs_daemon_open();
  16.176 +		if (!handles[handle])
  16.177 +			barf_perror("Opening connection to daemon");
  16.178 +	}
  16.179 +	command = arg(line, 0);
  16.180 +
  16.181 +	if (timeout)
  16.182 +		alarm(5);
  16.183 +
  16.184 +	if (streq(command, "dir"))
  16.185 +		do_dir(handle, arg(line, 1));
  16.186 +	else if (streq(command, "read"))
  16.187 +		do_read(handle, arg(line, 1));
  16.188 +	else if (streq(command, "write"))
  16.189 +		do_write(handle,
  16.190 +			 arg(line, 1), arg(line, 2), arg(line, 3));
  16.191 +	else if (streq(command, "setid"))
  16.192 +		do_setid(handle, arg(line, 1));
  16.193 +	else if (streq(command, "mkdir"))
  16.194 +		do_mkdir(handle, arg(line, 1));
  16.195 +	else if (streq(command, "rm"))
  16.196 +		do_rm(handle, arg(line, 1));
  16.197 +	else if (streq(command, "getperm"))
  16.198 +		do_getperm(handle, arg(line, 1));
  16.199 +	else if (streq(command, "setperm"))
  16.200 +		do_setperm(handle, arg(line, 1), line);
  16.201 +	else if (streq(command, "shutdown"))
  16.202 +		do_shutdown(handle);
  16.203 +	else if (streq(command, "watch"))
  16.204 +		do_watch(handle, arg(line, 1), arg(line, 2));
  16.205 +	else if (streq(command, "waitwatch"))
  16.206 +		do_waitwatch(handle);
  16.207 +	else if (streq(command, "async"))
  16.208 +		do_async(handle, line);
  16.209 +	else if (streq(command, "asyncwait"))
  16.210 +		do_asyncwait(handle);
  16.211 +	else if (streq(command, "ackwatch"))
  16.212 +		do_ackwatch(handle, arg(line, 1));
  16.213 +	else if (streq(command, "unwatch"))
  16.214 +		do_unwatch(handle, arg(line, 1), arg(line, 2));
  16.215 +	else if (streq(command, "close")) {
  16.216 +		xs_daemon_close(handles[handle]);
  16.217 +		handles[handle] = NULL;
  16.218 +	} else if (streq(command, "start"))
  16.219 +		do_start(handle, arg(line, 1));
  16.220 +	else if (streq(command, "commit"))
  16.221 +		do_end(handle, false);
  16.222 +	else if (streq(command, "abort"))
  16.223 +		do_end(handle, true);
  16.224 +	else if (streq(command, "introduce"))
  16.225 +		do_introduce(handle, arg(line, 1), arg(line, 2),
  16.226 +			     arg(line, 3), arg(line, 4));
  16.227 +	else if (streq(command, "release"))
  16.228 +		do_release(handle, arg(line, 1));
  16.229 +	else if (streq(command, "dump"))
  16.230 +		dump(handle);
  16.231 +	else if (streq(command, "sleep"))
  16.232 +		sleep(atoi(arg(line, 1)));
  16.233 +	else
  16.234 +		barf("Unknown command %s", command);
  16.235 +	fflush(stdout);
  16.236 +	alarm(0);
  16.237 +}
  16.238 +
  16.239  int main(int argc, char *argv[])
  16.240  {
  16.241  	char line[1024];
  16.242 -	bool readonly = false, timeout = true;
  16.243 -	int handle;
  16.244 -
  16.245 -	static void alarmed(int sig __attribute__((unused)))
  16.246 -	{
  16.247 -		if (handle) {
  16.248 -			char handlename[10];
  16.249 -			sprintf(handlename, "%u:", handle);
  16.250 -			write(STDOUT_FILENO, handlename, strlen(handlename));
  16.251 -		}
  16.252 -		write(STDOUT_FILENO, command, strlen(command));
  16.253 -		write(STDOUT_FILENO, " timeout\n", strlen(" timeout\n"));
  16.254 -		exit(1);
  16.255 -	}
  16.256  
  16.257  	if (argc > 1 && streq(argv[1], "--readonly")) {
  16.258  		readonly = true;
  16.259 @@ -557,7 +699,7 @@ int main(int argc, char *argv[])
  16.260  		argv++;
  16.261  	}
  16.262  
  16.263 -	if (argc > 1 && streq(argv[1], "--notimeout")) {
  16.264 +	if (argc > 1 && streq(argv[1], "--no-timeout")) {
  16.265  		timeout = false;
  16.266  		argc--;
  16.267  		argv++;
  16.268 @@ -570,81 +712,10 @@ int main(int argc, char *argv[])
  16.269  	ringbuf_datasize = getpagesize() / 2 - sizeof(struct ringbuf_head);
  16.270  
  16.271  	signal(SIGALRM, alarmed);
  16.272 -	while (fgets(line, sizeof(line), stdin)) {
  16.273 -		char *endp;
  16.274 -
  16.275 -		if (strspn(line, " \n") == strlen(line))
  16.276 -			continue;
  16.277 -		if (strstarts(line, "#"))
  16.278 -			continue;
  16.279 -
  16.280 -		handle = strtoul(line, &endp, 10);
  16.281 -		if (endp != line)
  16.282 -			memmove(line, endp+1, strlen(endp));
  16.283 -		else
  16.284 -			handle = 0;
  16.285 -
  16.286 -		if (!handles[handle]) {
  16.287 -			if (readonly)
  16.288 -				handles[handle] = xs_daemon_open_readonly();
  16.289 -			else
  16.290 -				handles[handle] = xs_daemon_open();
  16.291 -			if (!handles[handle])
  16.292 -				barf_perror("Opening connection to daemon");
  16.293 -		}
  16.294 -		command = arg(line, 0);
  16.295 +	while (fgets(line, sizeof(line), stdin))
  16.296 +		do_command(0, line);
  16.297  
  16.298 -		if (timeout)
  16.299 -			alarm(5);
  16.300 -		if (streq(command, "dir"))
  16.301 -			do_dir(handle, arg(line, 1));
  16.302 -		else if (streq(command, "read"))
  16.303 -			do_read(handle, arg(line, 1));
  16.304 -		else if (streq(command, "write"))
  16.305 -			do_write(handle,
  16.306 -				 arg(line, 1), arg(line, 2), arg(line, 3));
  16.307 -		else if (streq(command, "setid"))
  16.308 -			do_setid(handle, arg(line, 1));
  16.309 -		else if (streq(command, "mkdir"))
  16.310 -			do_mkdir(handle, arg(line, 1));
  16.311 -		else if (streq(command, "rm"))
  16.312 -			do_rm(handle, arg(line, 1));
  16.313 -		else if (streq(command, "getperm"))
  16.314 -			do_getperm(handle, arg(line, 1));
  16.315 -		else if (streq(command, "setperm"))
  16.316 -			do_setperm(handle, arg(line, 1), line);
  16.317 -		else if (streq(command, "shutdown"))
  16.318 -			do_shutdown(handle);
  16.319 -		else if (streq(command, "watch"))
  16.320 -			do_watch(handle, arg(line, 1), arg(line, 2), arg(line, 3));
  16.321 -		else if (streq(command, "waitwatch"))
  16.322 -			do_waitwatch(handle);
  16.323 -		else if (streq(command, "ackwatch"))
  16.324 -			do_ackwatch(handle, arg(line, 1));
  16.325 -		else if (streq(command, "unwatch"))
  16.326 -			do_unwatch(handle, arg(line, 1), arg(line, 2));
  16.327 -		else if (streq(command, "close")) {
  16.328 -			xs_daemon_close(handles[handle]);
  16.329 -			handles[handle] = NULL;
  16.330 -		} else if (streq(command, "start"))
  16.331 -			do_start(handle, arg(line, 1));
  16.332 -		else if (streq(command, "commit"))
  16.333 -			do_end(handle, false);
  16.334 -		else if (streq(command, "abort"))
  16.335 -			do_end(handle, true);
  16.336 -		else if (streq(command, "introduce"))
  16.337 -			do_introduce(handle, arg(line, 1), arg(line, 2),
  16.338 -				     arg(line, 3), arg(line, 4));
  16.339 -		else if (streq(command, "release"))
  16.340 -			do_release(handle, arg(line, 1));
  16.341 -		else if (streq(command, "dump"))
  16.342 -			dump(handle);
  16.343 -		else if (streq(command, "sleep"))
  16.344 -			sleep(atoi(arg(line, 1)));
  16.345 -		else
  16.346 -			barf("Unknown command %s", command);
  16.347 -		fflush(stdout);
  16.348 -		alarm(0);
  16.349 -	}
  16.350 +	while (children)
  16.351 +		do_asyncwait(0);
  16.352  	return 0;
  16.353  }