Server watching local state head

1 files changed, 124 insertions, 4 deletions

diff --git a/src/storage.cpp b/src/storage.cpp
index 45caadb..4f48c67 100644
--- a/src/storage.cpp
+++ b/src/storage.cpp
@@ -9,7 +9,10 @@
 #include <stdexcept>
 #include <thread>
 
+#include <poll.h>
 #include <stdio.h>
+#include <sys/eventfd.h>
+#include <sys/inotify.h>
 
 #include <blake2.h>
 #include <zlib.h>
@@ -28,8 +31,10 @@ using std::monostate;
 using std::nullopt;
 using std::ofstream;
 using std::runtime_error;
+using std::scoped_lock;
 using std::shared_ptr;
 using std::string;
+using std::system_error;
 using std::to_string;
 
 FilesystemStorage::FilesystemStorage(const fs::path & path):
@@ -45,6 +50,24 @@ FilesystemStorage::FilesystemStorage(const fs::path & path):
 		fs::create_directory(path/"heads");
 }
 
+FilesystemStorage::~FilesystemStorage()
+{
+	if (inotifyWakeup >= 0) {
+		uint64_t x = 1;
+		write(inotifyWakeup, &x, sizeof(x));
+	}
+
+	if (watcherThread.joinable())
+		watcherThread.join();
+
+	if (inotify >= 0)
+		close(inotify);
+
+	if (inotifyWakeup >= 0)
+		close(inotifyWakeup);
+
+}
+
 bool FilesystemStorage::contains(const Digest & digest) const
 {
 	return fs::exists(objectPath(digest));
@@ -237,6 +260,29 @@ bool FilesystemStorage::replaceHead(UUID type, UUID id, const Digest & old, cons
 	return true;
 }
 
+void FilesystemStorage::watchHead(UUID type, const function<void(UUID id, const Digest &)> & watcher)
+{
+	scoped_lock lock(watcherLock);
+	watchers.emplace(type, watcher);
+
+	if (inotify < 0) {
+		inotify = inotify_init();
+		if (inotify < 0)
+			throw system_error(errno, std::generic_category());
+
+		inotifyWakeup = eventfd(0, 0);
+		if (inotifyWakeup < 0)
+			throw system_error(errno, std::generic_category());
+
+		watcherThread = std::thread(&FilesystemStorage::inotifyWatch, this);
+	}
+
+	int fd = inotify_add_watch(inotify, headPath(type).c_str(), IN_MOVED_TO);
+	if (fd < 0)
+		throw system_error(errno, std::generic_category());
+	watchMap[fd] = type;
+}
+
 optional<vector<uint8_t>> FilesystemStorage::loadKey(const Digest & pubref) const
 {
 	fs::path path = keyPath(pubref);
@@ -259,6 +305,55 @@ void FilesystemStorage::storeKey(const Digest & pubref, const vector<uint8_t> &
 	file.write((const char *) key.data(), key.size());
 }
 
+void FilesystemStorage::inotifyWatch()
+{
+	char buf[4096]
+		__attribute__ ((aligned(__alignof__(struct inotify_event))));
+	const struct inotify_event * event;
+
+	array pfds {
+		pollfd { inotify, POLLIN, 0 },
+		pollfd { inotifyWakeup, POLLIN, 0 },
+	};
+
+	while (true) {
+		int ret = poll(pfds.data(), pfds.size(), -1);
+		if (ret < 0)
+			throw system_error(errno, std::generic_category());
+
+		if (!(pfds[0].revents & POLLIN))
+			break;
+
+		ssize_t len = read(inotify, buf, sizeof buf);
+		if (len < 0) {
+			if (errno == EAGAIN)
+				continue;
+
+			throw system_error(errno, std::generic_category());
+		}
+
+		if (len == 0)
+			break;
+
+		for (char * ptr = buf; ptr < buf + len;
+				ptr += sizeof(struct inotify_event) + event->len) {
+			event = (const struct inotify_event *) ptr;
+
+			if (event->mask & IN_MOVED_TO) {
+				UUID type = watchMap[event->wd];
+				if (auto mbid = UUID::fromString(event->name)) {
+					if (auto mbref = headRef(type, *mbid)) {
+						scoped_lock lock(watcherLock);
+						auto range = watchers.equal_range(type);
+						for (auto it = range.first; it != range.second; it++)
+							it->second(*mbid, *mbref);
+					}
+				}
+			}
+		}
+	}
+}
+
 fs::path FilesystemStorage::objectPath(const Digest & digest) const
 {
 	string name(digest);
@@ -267,12 +362,16 @@ fs::path FilesystemStorage::objectPath(const Digest & digest) const
 		fs::path(name.begin() + 2, name.end());
 }
 
+fs::path FilesystemStorage::headPath(UUID type) const
+{
+	string stype(type);
+	return root/"heads"/fs::path(stype.begin(), stype.end());
+}
+
 fs::path FilesystemStorage::headPath(UUID type, UUID id) const
 {
-	string stype(type), sid(id);
-	return root/"heads"/
-		fs::path(stype.begin(), stype.end())/
-		fs::path(sid.begin(), sid.end());
+	string sid(id);
+	return headPath(type) / fs::path(sid.begin(), sid.end());
 }
 
 fs::path FilesystemStorage::keyPath(const Digest & digest) const
@@ -367,6 +466,12 @@ bool MemoryStorage::replaceHead(UUID type, UUID id, const Digest & old, const Di
 	return false;
 }
 
+void MemoryStorage::watchHead(UUID type, const function<void(UUID id, const Digest &)> & watcher)
+{
+	scoped_lock lock(watcherLock);
+	watchers.emplace(type, watcher);
+}
+
 optional<vector<uint8_t>> MemoryStorage::loadKey(const Digest & digest) const
 {
 	auto it = keys.find(digest);
@@ -436,6 +541,12 @@ bool ChainStorage::replaceHead(UUID type, UUID id, const Digest & old, const Dig
 	return storage->replaceHead(type, id, old, dgst);
 }
 
+void ChainStorage::watchHead(UUID type, const function<void(UUID id, const Digest &)> & watcher)
+{
+	parent->watchHead(type, watcher);
+	storage->watchHead(type, watcher);
+}
+
 optional<vector<uint8_t>> ChainStorage::loadKey(const Digest & digest) const
 {
 	if (auto res = storage->loadKey(digest))
@@ -669,6 +780,15 @@ optional<Ref> Storage::updateHead(UUID type, UUID id, const Ref & old, const std
 		return nullopt;
 }
 
+void Storage::watchHead(UUID type, UUID wid, const std::function<void(const Ref &)> watcher) const
+{
+	p->backend->watchHead(type, [this, wid, watcher] (UUID id, const Digest & dgst) {
+		if (id == wid)
+			if (auto r = ref(dgst))
+				watcher(*r);
+	});
+}
+
 
 Digest::Digest(const string & str)
 {