[PATCH/WIP v2 03/14] read-cache: connect to file watcher

["Nguyễn Thái Ngọc Duy" <pclouds@gmail.com>]

This patch establishes a connection between a new file watcher daemon
and git. Each index file may have at most one file watcher attached to
it. The file watcher maintains a UNIX socket at
$GIT_DIR/index.watcher. Any process that has write access to $GIT_DIR
can talk to the file watcher.

A validation is performed after git connects to the file watcher to
make sure both sides have the same view. This is done by exchanging
the index signature (*) The file watcher keeps a copy of the signature
locally while git computes the signature from the index. If the
signatures do not match, something has gone wrong so both sides
reinitialize wrt. to file watching: the file watcher clears all
watches while git clears CE_WATCHED flags.

If the signatures match, we can trust the file watcher and git can
start asking questions that are not important to this patch.

This file watching thing is all about speed. So if the daemon is not
responding within 20ms (or even hanging), git moves on without it.

A note about per-repo vs global (or per-user) daemon approach. While I
implement per-repo daemon, this is actually implementation
details. Nothing can stop you from writing a global daemon that opens
unix sockets to many repos, e.g. to avoid hitting inotify's 128 user
instances limit.

If env variable GIT_NO_FILE_WATCHER is set, the file watcher is
ignored. 'WATC' extension is kept, but if the index is updated
(likely), it'll become invalid at the next connection.

(*) for current index versions, the signature is the index SHA-1
trailer. But it could be something else (e.g. v5 does not have SHA-1
trailer)

Signed-off-by: Nguyễn Thái Ngọc Duy <pclouds@gmail.com>
---
 .gitignore               |   1 +
 Makefile                 |   2 +
 cache.h                  |   3 +
 file-watcher-lib.c (new) |  97 ++++++++++++++++++++++++++++++++
 file-watcher-lib.h (new) |   9 +++
 file-watcher.c (new)     | 142 +++++++++++++++++++++++++++++++++++++++++++++++
 read-cache.c             |  37 ++++++++++++
 trace.c                  |   3 +-
 8 files changed, 292 insertions(+), 2 deletions(-)
 create mode 100644 file-watcher-lib.c
 create mode 100644 file-watcher-lib.h
 create mode 100644 file-watcher.c

diff --git a/.gitignore b/.gitignore
index dc600f9..dc870cc 100644
--- a/.gitignore
+++ b/.gitignore
@@ -56,6 +56,7 @@
 /git-fast-import
 /git-fetch
 /git-fetch-pack
+/git-file-watcher
 /git-filter-branch
 /git-fmt-merge-msg
 /git-for-each-ref
diff --git a/Makefile b/Makefile
index 287e6f8..4369b3b 100644
--- a/Makefile
+++ b/Makefile
@@ -536,6 +536,7 @@ PROGRAMS += $(EXTRA_PROGRAMS)
 PROGRAM_OBJS += credential-store.o
 PROGRAM_OBJS += daemon.o
 PROGRAM_OBJS += fast-import.o
+PROGRAM_OBJS += file-watcher.o
 PROGRAM_OBJS += http-backend.o
 PROGRAM_OBJS += imap-send.o
 PROGRAM_OBJS += sh-i18n--envsubst.o
@@ -798,6 +799,7 @@ LIB_OBJS += entry.o
 LIB_OBJS += environment.o
 LIB_OBJS += exec_cmd.o
 LIB_OBJS += fetch-pack.o
+LIB_OBJS += file-watcher-lib.o
 LIB_OBJS += fsck.o
 LIB_OBJS += gettext.o
 LIB_OBJS += gpg-interface.o
diff --git a/cache.h b/cache.h
index 069dce7..0d55551 100644
--- a/cache.h
+++ b/cache.h
@@ -282,6 +282,7 @@ struct index_state {
 	struct hashmap name_hash;
 	struct hashmap dir_hash;
 	unsigned char sha1[20];
+	int watcher;
 };
 
 extern struct index_state the_index;
@@ -1241,6 +1242,8 @@ extern void alloc_report(void);
 __attribute__((format (printf, 1, 2)))
 extern void trace_printf(const char *format, ...);
 __attribute__((format (printf, 2, 3)))
+extern void trace_printf_key(const char *key, const char *fmt, ...);
+__attribute__((format (printf, 2, 3)))
 extern void trace_argv_printf(const char **argv, const char *format, ...);
 extern void trace_repo_setup(const char *prefix);
 extern int trace_want(const char *key);
diff --git a/file-watcher-lib.c b/file-watcher-lib.c
new file mode 100644
index 0000000..ed14ef9
--- /dev/null
+++ b/file-watcher-lib.c
@@ -0,0 +1,97 @@
+#include "cache.h"
+
+#define WAIT_TIME 20		/* in ms */
+#define TRACE_KEY "GIT_TRACE_WATCHER"
+
+int connect_watcher(const char *path)
+{
+	struct strbuf sb = STRBUF_INIT;
+	struct stat st;
+	int fd = -1;
+
+	strbuf_addf(&sb, "%s.watcher", path);
+	if (!stat(sb.buf, &st) && S_ISSOCK(st.st_mode)) {
+		struct sockaddr_un sun;
+		fd = socket(AF_UNIX, SOCK_DGRAM, 0);
+		sun.sun_family = AF_UNIX;
+		strlcpy(sun.sun_path, sb.buf, sizeof(sun.sun_path));
+		if (connect(fd, (struct sockaddr *)&sun, sizeof(sun))) {
+			error(_("unable to connect to file watcher: %s"),
+			      strerror(errno));
+			close(fd);
+			fd = -1;
+		} else {
+			sprintf(sun.sun_path, "%c%"PRIuMAX, 0, (uintmax_t)getpid());
+			if (bind(fd, (struct sockaddr *)&sun, sizeof(sun))) {
+				error(_("unable to bind socket: %s"),
+				      strerror(errno));
+				close(fd);
+				fd = -1;
+			}
+		}
+	}
+	strbuf_release(&sb);
+	return fd;
+}
+
+ssize_t send_watcher(int sockfd, struct sockaddr_un *dest,
+		     const char *fmt, ...)
+{
+	struct strbuf sb = STRBUF_INIT;
+	struct pollfd pfd;
+	int ret;
+
+	va_list ap;
+	va_start(ap, fmt);
+	strbuf_vaddf(&sb, fmt, ap);
+	va_end(ap);
+
+	pfd.fd = sockfd;
+	pfd.events = POLLOUT;
+	ret = poll(&pfd, 1, WAIT_TIME);
+	if (ret > 0 && pfd.revents & POLLOUT) {
+		trace_printf_key(TRACE_KEY, "< %s\n", sb.buf);
+		if (dest)
+			ret = sendto(sockfd, sb.buf, sb.len, 0,
+				     (struct sockaddr *)dest,
+				     sizeof(struct sockaddr_un));
+		else
+			ret = write(sockfd, sb.buf, sb.len);
+	}
+	strbuf_release(&sb);
+	return ret;
+}
+
+char *read_watcher(int fd, ssize_t *size, struct sockaddr_un *sun)
+{
+	static char *buf;
+	static int buf_size;
+	struct pollfd pfd;
+	ssize_t len;
+
+	if (!buf_size) {
+		socklen_t vallen = sizeof(buf_size);
+		if (getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &buf_size, &vallen))
+			die_errno("could not get SO_SNDBUF from socket %d", fd);
+		buf = xmalloc(buf_size + 1);
+	}
+
+	pfd.fd = fd;
+	pfd.events = POLLIN;
+	if (poll(&pfd, 1, WAIT_TIME) > 0 &&
+	    (pfd.revents & POLLIN)) {
+		if (sun) {
+			socklen_t socklen = sizeof(*sun);
+			len = recvfrom(fd, buf, buf_size, 0, sun, &socklen);
+		} else
+			len = read(fd, buf, buf_size);
+		if (len > 0)
+			buf[len] = '\0';
+		if (size)
+			*size = len;
+		trace_printf_key(TRACE_KEY, "> %s\n", buf);
+		return buf;
+	} else if (size)
+		*size = 0;
+	return NULL;
+}
diff --git a/file-watcher-lib.h b/file-watcher-lib.h
new file mode 100644
index 0000000..0fe9399
--- /dev/null
+++ b/file-watcher-lib.h
@@ -0,0 +1,9 @@
+#ifndef __FILE_WATCHER_LIB__
+#define __FILE_WATCHER_LIB__
+
+int connect_watcher(const char *path);
+ssize_t send_watcher(int sockfd, struct sockaddr_un *dest,
+		     const char *fmt, ...)
+	__attribute__((format (printf, 3, 4)));
+char *read_watcher(int fd, ssize_t *size, struct sockaddr_un *sun);
+#endif
diff --git a/file-watcher.c b/file-watcher.c
new file mode 100644
index 0000000..36a9a8d
--- /dev/null
+++ b/file-watcher.c
@@ -0,0 +1,142 @@
+#include "cache.h"
+#include "sigchain.h"
+#include "parse-options.h"
+#include "exec_cmd.h"
+#include "file-watcher-lib.h"
+
+static const char *const file_watcher_usage[] = {
+	N_("git file-watcher [options]"),
+	NULL
+};
+
+static char index_signature[41];
+
+static int handle_command(int fd)
+{
+	struct sockaddr_un sun;
+	ssize_t len;
+	const char *arg;
+	char *msg;
+
+	if (!(msg = read_watcher(fd, &len, &sun)))
+		die_errno("read from socket");
+
+	if ((arg = skip_prefix(msg, "hello "))) {
+		send_watcher(fd, &sun, "hello %s", index_signature);
+		if (strcmp(arg, index_signature))
+			/*
+			 * Index SHA-1 mismatch, something has gone
+			 * wrong. Clean up and start over.
+			 */
+			index_signature[0] = '\0';
+	} else if (!strcmp(msg, "die")) {
+		exit(0);
+	} else {
+		die("unrecognized command %s", msg);
+	}
+	return 0;
+}
+
+static const char *socket_path;
+static int do_not_clean_up;
+
+static void cleanup(void)
+{
+	if (do_not_clean_up)
+		return;
+	unlink(socket_path);
+}
+
+static void cleanup_on_signal(int signo)
+{
+	cleanup();
+	sigchain_pop(signo);
+	raise(signo);
+}
+
+static void daemonize(void)
+{
+#ifdef NO_POSIX_GOODIES
+	die("fork not supported on this platform");
+#else
+	switch (fork()) {
+		case 0:
+			break;
+		case -1:
+			die_errno("fork failed");
+		default:
+			do_not_clean_up = 1;
+			exit(0);
+	}
+	if (setsid() == -1)
+		die_errno("setsid failed");
+	close(0);
+	close(1);
+	close(2);
+	sanitize_stdfds();
+#endif
+}
+
+int main(int argc, const char **argv)
+{
+	struct strbuf sb = STRBUF_INIT;
+	struct sockaddr_un sun;
+	struct pollfd pfd[2];
+	int fd, err, nr;
+	const char *prefix;
+	int daemon = 0;
+	struct option options[] = {
+		OPT_BOOL(0, "daemon", &daemon,
+			 N_("run in background")),
+		OPT_END()
+	};
+
+	git_extract_argv0_path(argv[0]);
+	git_setup_gettext();
+	prefix = setup_git_directory();
+	argc = parse_options(argc, argv, prefix, options,
+			     file_watcher_usage, 0);
+	if (argc)
+		die("too many arguments");
+
+	strbuf_addf(&sb, "%s.watcher", get_index_file());
+	socket_path = strbuf_detach(&sb, NULL);
+	memset(index_signature, 0, sizeof(index_signature));
+	fd = socket(AF_UNIX, SOCK_DGRAM, 0);
+	sun.sun_family = AF_UNIX;
+	strlcpy(sun.sun_path, socket_path, sizeof(sun.sun_path));
+	if (bind(fd, (struct sockaddr *)&sun, sizeof(sun)))
+		die_errno("unable to bind to %s", socket_path);
+	atexit(cleanup);
+	sigchain_push_common(cleanup_on_signal);
+
+	if (daemon) {
+		strbuf_addf(&sb, "%s.log", socket_path);
+		err = open(sb.buf, O_CREAT | O_TRUNC | O_WRONLY, 0600);
+		adjust_shared_perm(sb.buf);
+		if (err == -1)
+			die_errno("unable to create %s", sb.buf);
+		daemonize();
+		dup2(err, 1);
+		dup2(err, 2);
+		close(err);
+	}
+
+	nr = 0;
+	pfd[nr].fd = fd;
+	pfd[nr++].events = POLLIN;
+
+	for (;;) {
+		if (poll(pfd, nr, -1) < 0) {
+			if (errno != EINTR) {
+				error("Poll failed, resuming: %s", strerror(errno));
+				sleep(1);
+			}
+			continue;
+		}
+
+		if ((pfd[0].revents & POLLIN) && handle_command(fd))
+			break;
+	}
+	return 0;
+}
diff --git a/read-cache.c b/read-cache.c
index 6f21e3f..76cf0e3 100644
--- a/read-cache.c
+++ b/read-cache.c
@@ -14,6 +14,7 @@
 #include "resolve-undo.h"
 #include "strbuf.h"
 #include "varint.h"
+#include "file-watcher-lib.h"
 
 static struct cache_entry *refresh_cache_entry(struct cache_entry *ce, int really);
 
@@ -1447,6 +1448,37 @@ static struct cache_entry *create_from_disk(struct ondisk_cache_entry *ondisk,
 	return ce;
 }
 
+static void validate_watcher(struct index_state *istate, const char *path)
+{
+	int i;
+
+	if (getenv("GIT_NO_FILE_WATCHER")) {
+		istate->watcher = -1;
+		return;
+	}
+
+	istate->watcher = connect_watcher(path);
+	if (istate->watcher != -1) {
+		struct strbuf sb = STRBUF_INIT;
+		char *msg;
+		strbuf_addf(&sb, "hello %s", sha1_to_hex(istate->sha1));
+		if (send_watcher(istate->watcher, NULL, "%s", sb.buf) > 0 &&
+		    (msg = read_watcher(istate->watcher, NULL, NULL)) != NULL &&
+		    !strcmp(msg, sb.buf)) { /* good */
+			strbuf_release(&sb);
+			return;
+		}
+		strbuf_release(&sb);
+	}
+
+	/* No the file watcher is out of date, clear everything */
+	for (i = 0; i < istate->cache_nr; i++)
+		if (istate->cache[i]->ce_flags & CE_WATCHED) {
+			istate->cache[i]->ce_flags &= ~CE_WATCHED;
+			istate->cache_changed = 1;
+		}
+}
+
 /* remember to discard_cache() before reading a different cache! */
 int read_index_from(struct index_state *istate, const char *path)
 {
@@ -1532,6 +1564,7 @@ int read_index_from(struct index_state *istate, const char *path)
 		src_offset += extsize;
 	}
 	munmap(mmap, mmap_size);
+	validate_watcher(istate, path);
 	return istate->cache_nr;
 
 unmap:
@@ -1557,6 +1590,10 @@ int discard_index(struct index_state *istate)
 	istate->timestamp.nsec = 0;
 	free_name_hash(istate);
 	cache_tree_free(&(istate->cache_tree));
+	if (istate->watcher > 0) {
+		close(istate->watcher);
+		istate->watcher = -1;
+	}
 	istate->initialized = 0;
 	free(istate->cache);
 	istate->cache = NULL;
diff --git a/trace.c b/trace.c
index 3d744d1..0b8ebe0 100644
--- a/trace.c
+++ b/trace.c
@@ -75,8 +75,7 @@ static void trace_vprintf(const char *key, const char *fmt, va_list ap)
 	strbuf_release(&buf);
 }
 
-__attribute__((format (printf, 2, 3)))
-static void trace_printf_key(const char *key, const char *fmt, ...)
+void trace_printf_key(const char *key, const char *fmt, ...)
 {
 	va_list ap;
 	va_start(ap, fmt);
-- 
1.8.5.1.208.g05b12ea