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

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

XenStore puts in queued watch events via a thread and notifies the user.
Sometimes xs_unwatch is called before all related message is read. The use
case is non-threaded libevent, we have two event A and B:
    - Event A will destroy something and call xs_unwatch;
    - Event B is used to notify that a node has changed in XenStore.
As the event is called one by one, event A can be handled before event B.
So on next xs_watch_read the user could retrieve an unwatch token and
a segfault occured if the token store the pointer of the structure
(ie: "backend:0xcafe").

Signed-off-by: Julien Grall <julien.grall@citrix.com>
---
 tools/xenstore/xs.c |   41 ++++++++++++++++++++++++++++++++++++++++-
 1 files changed, 40 insertions(+), 1 deletions(-)

diff --git a/tools/xenstore/xs.c b/tools/xenstore/xs.c
index b951015..31aad14 100644
--- a/tools/xenstore/xs.c
+++ b/tools/xenstore/xs.c
@@ -855,14 +855,53 @@ 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 *strings;
+	unsigned int num_strings, i;
+	char c;
 
 	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,
+	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);
+
+		strings = msg->body;
+		num_strings = xs_count_strings(strings, msg->hdr.len);
+
+		for (i = 0; i < num_strings; i++) {
+			if (i == XS_WATCH_TOKEN && !strcmp (token, strings)) {
+				list_del(&msg->list);
+				free(msg);
+				break;
+			}
+			strings = strings + strlen (strings) + 1;
+		}
+	}
+
+	/* 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;
+	}
+
+	mutex_unlock(&h->watch_mutex);
+
+	return res;
 }
 
 /* Start a transaction: changes by others will not be seen during this
-- 
Julien Grall