[PATCH v3 4/5] wal: add event_mask to wal_watcher

[Vladimir Davydov <vdavydov.dev@gmail.com>]

In order to implement WAL auto-deletion, we need a notification channel
through which the WAL thread could notify TX that a WAL file was deleted
so that the latter can shoot off stale replicas. We will reuse existing
wal_watcher API for this. Currently, wal_watcher invokes the registered
callback on each WAL write so using it as is would be inefficient. To
avoid that, let's allow the caller to specify events of interest when
registering a wal_watcher.

Needed for #3397
---
 src/box/relay.cc |  5 ++++-
 src/box/wal.c    | 10 +++++++++-
 src/box/wal.h    |  9 ++++++++-
 3 files changed, 21 insertions(+), 3 deletions(-)

diff --git a/src/box/relay.cc b/src/box/relay.cc
index 2a4cb412..0034f99a 100644
--- a/src/box/relay.cc
+++ b/src/box/relay.cc
@@ -408,6 +408,8 @@ relay_schedule_pending_gc(struct relay *relay, const struct vclock *vclock)
 static void
 relay_process_wal_event(struct wal_watcher_msg *msg)
 {
+	assert((msg->events & (WAL_EVENT_WRITE | WAL_EVENT_ROTATE)) != 0);
+
 	struct relay *relay = container_of(msg->watcher, struct relay,
 					   wal_watcher);
 	if (relay->state != RELAY_FOLLOW) {
@@ -505,7 +507,8 @@ relay_subscribe_f(va_list ap)
 	};
 	trigger_add(&r->on_close_log, &on_close_log);
 	wal_set_watcher(&relay->wal_watcher, cord_name(cord()),
-			relay_process_wal_event, cbus_process);
+			relay_process_wal_event, cbus_process,
+			WAL_EVENT_WRITE | WAL_EVENT_ROTATE);
 
 	relay_set_cord_name(relay->io.fd);
 
diff --git a/src/box/wal.c b/src/box/wal.c
index b239acd4..09927f6f 100644
--- a/src/box/wal.c
+++ b/src/box/wal.c
@@ -1027,6 +1027,12 @@ wal_watcher_notify(struct wal_watcher *watcher, unsigned events)
 {
 	assert(!rlist_empty(&watcher->next));
 
+	events &= watcher->event_mask;
+	if (events == 0) {
+		/* The watcher isn't interested in this event. */
+		return;
+	}
+
 	if (watcher->msg.cmsg.route != NULL) {
 		/*
 		 * If the notification message is still en route,
@@ -1100,7 +1106,8 @@ wal_watcher_detach(void *arg)
 void
 wal_set_watcher(struct wal_watcher *watcher, const char *name,
 		void (*watcher_cb)(struct wal_watcher_msg *),
-		void (*process_cb)(struct cbus_endpoint *))
+		void (*process_cb)(struct cbus_endpoint *),
+		unsigned event_mask)
 {
 	assert(journal_is_initialized(&wal_writer_singleton.base));
 
@@ -1110,6 +1117,7 @@ wal_set_watcher(struct wal_watcher *watcher, const char *name,
 	watcher->msg.events = 0;
 	watcher->msg.cmsg.route = NULL;
 	watcher->pending_events = 0;
+	watcher->event_mask = event_mask;
 
 	assert(lengthof(watcher->route) == 2);
 	watcher->route[0] = (struct cmsg_hop)
diff --git a/src/box/wal.h b/src/box/wal.h
index b7edcd43..e8a4299c 100644
--- a/src/box/wal.h
+++ b/src/box/wal.h
@@ -96,6 +96,11 @@ struct wal_watcher {
 	/** Message sent to notify the watcher. */
 	struct wal_watcher_msg msg;
 	/**
+	 * Bit mask of WAL events that this watcher is
+	 * interested in.
+	 */
+	unsigned event_mask;
+	/**
 	 * Bit mask of WAL events that happened while
 	 * the notification message was en route.
 	 * It indicates that the message must be resend
@@ -126,11 +131,13 @@ struct wal_watcher {
  * @param process_cb  Function called to process cbus messages
  *                    while the watcher is being attached or NULL
  *                    if the cbus loop is running elsewhere.
+ * @param event_mask  Bit mask of events the watcher is interested in.
  */
 void
 wal_set_watcher(struct wal_watcher *watcher, const char *name,
 		void (*watcher_cb)(struct wal_watcher_msg *),
-		void (*process_cb)(struct cbus_endpoint *));
+		void (*process_cb)(struct cbus_endpoint *),
+		unsigned event_mask);
 
 /**
  * Unsubscribe from WAL events.
-- 
2.11.0