[PATCH V2] libxenstore: filter watch events in libxenstore when we unwatch

[Julien Grall <julien.grall@citrix.com>]

While on entry to xs_unwatch, there may be an event on its way from
xenstored (eg in the ring or in the local kernel), all such events
will definitely come before the reply to the unwatch command.  So at
the point where the unwatch reply has been processed (after xs_talkv),
any such now-deleted watch events will definitely have made it to
libxenstore's queue where we can remove them.

As for other threads in the same process: if two threads call
xs_read_watch and xs_unwatch, it is acceptable for the xs_read_watch
to return the event being deleted.  What is not allowed is for an
xs_read_watch entered after xs_unwatch returns to return the deleted
event, and this code does indeed ensure that.

Signed-off-by: Ian Jackson <ian.jackson@eu.citrix.com>
Signed-off-by: Julien Grall <julien.grall@citrix.com>
---
 tools/xenstore/xs.c |   73 +++++++++++++++++++++++++++++++++++++++++++++-----
 1 files changed, 65 insertions(+), 8 deletions(-)

diff --git a/tools/xenstore/xs.c b/tools/xenstore/xs.c
index b951015..df89e37 100644
--- a/tools/xenstore/xs.c
+++ b/tools/xenstore/xs.c
@@ -753,6 +753,19 @@ bool xs_watch(struct xs_handle *h, const char *path, const char *token)
 				ARRAY_SIZE(iov), NULL));
 }
 
+
+/* Clear the pipe token if there are no more pending watchs.
+ * We suppose the watch_mutex is already taken.
+ */
+static void xs_clear_watch_pipe(struct xs_handle *h)
+{
+	char c;
+
+	if (list_empty(&h->watch_list) && (h->watch_pipe[0] != -1))
+		while (read(h->watch_pipe[0], &c, 1) != 1)
+			continue;
+}
+
 /* Find out what node change was on (will block if nothing pending).
  * Returns array of two pointers: path and token, or NULL.
  * Call free() after use.
@@ -761,7 +774,7 @@ static char **read_watch_internal(struct xs_handle *h, unsigned int *num,
 				  int nonblocking)
 {
 	struct xs_stored_msg *msg;
-	char **ret, *strings, c = 0;
+	char **ret, *strings;
 	unsigned int num_strings, i;
 
 	mutex_lock(&h->watch_mutex);
@@ -798,11 +811,7 @@ static char **read_watch_internal(struct xs_handle *h, unsigned int *num,
 	msg = list_top(&h->watch_list, struct xs_stored_msg, list);
 	list_del(&msg->list);
 
-	/* Clear the pipe token if there are no more pending watches. */
-	if (list_empty(&h->watch_list) && (h->watch_pipe[0] != -1))
-		while (read(h->watch_pipe[0], &c, 1) != 1)
-			continue;
-
+	xs_clear_watch_pipe(h);
 	mutex_unlock(&h->watch_mutex);
 
 	assert(msg->hdr.type == XS_WATCH_EVENT);
@@ -855,14 +864,62 @@ char **xs_read_watch(struct xs_handle *h, unsigned int *num)
 bool xs_unwatch(struct xs_handle *h, const char *path, const char *token)
 {
 	struct iovec iov[2];
+	struct xs_stored_msg *msg, *tmsg;
+	bool res;
+	char *s, *p;
+	unsigned int i;
+	char *l_token, *l_path;
 
 	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, XBT_NULL, XS_UNWATCH, iov,
-				ARRAY_SIZE(iov), NULL));
+	res = xs_bool(xs_talkv(h, XBT_NULL, XS_UNWATCH, iov,
+						   ARRAY_SIZE(iov), NULL));
+
+	/* Filter the watch list to remove potential message */
+	mutex_lock(&h->watch_mutex);
+
+	if (list_empty(&h->watch_list)) {
+		mutex_unlock(&h->watch_mutex);
+		return res;
+	}
+
+	list_for_each_entry_safe(msg, tmsg, &h->watch_list, list) {
+		assert(msg->hdr.type == XS_WATCH_EVENT);
+
+		s = msg->body;
+
+		l_token = NULL;
+		l_path = NULL;
+
+		for (p = s, i = 0; p < msg->body + msg->hdr.len; p++) {
+			if (*p == '\0')
+			{
+				if (i == XS_WATCH_TOKEN)
+					l_token = s;
+				else if (i == XS_WATCH_PATH)
+					l_path = s;
+				i++;
+				s = p + 1;
+			}
+		}
+
+		if (l_token && !strcmp(token, l_token)
+			/* Use strncmp because we can have a watch fired on sub-directory */
+			&& l_path && !strncmp(path, l_path, strlen(path))) {
+			fprintf (stderr, "DELETE\n");
+			list_del(&msg->list);
+			free(msg);
+		}
+	}
+
+	xs_clear_watch_pipe(h);
+
+	mutex_unlock(&h->watch_mutex);
+
+	return res;
 }
 
 /* Start a transaction: changes by others will not be seen during this
-- 
Julien Grall