From 2abd6593c8b047d3fd579aa6cc0058bbebe266f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roman=20Smr=C5=BE?= Date: Sun, 17 Jan 2021 21:59:46 +0100 Subject: Server watching local state head --- include/erebos/network.h | 4 +- include/erebos/storage.h | 10 ++++ src/network.cpp | 27 ++++++++-- src/network.h | 6 ++- src/storage.cpp | 128 +++++++++++++++++++++++++++++++++++++++++++++-- src/storage.h | 24 ++++++++- 6 files changed, 188 insertions(+), 11 deletions(-) diff --git a/include/erebos/network.h b/include/erebos/network.h index a2f989e..8f3debe 100644 --- a/include/erebos/network.h +++ b/include/erebos/network.h @@ -1,7 +1,7 @@ #pragma once -#include #include +#include #include #include @@ -11,7 +11,7 @@ namespace erebos { class Server { public: - Server(const Identity &, std::vector> &&); + Server(const Head &, std::vector> &&); ~Server(); template S & svc(); diff --git a/include/erebos/storage.h b/include/erebos/storage.h index 34ed9df..29eaa8f 100644 --- a/include/erebos/storage.h +++ b/include/erebos/storage.h @@ -107,6 +107,7 @@ protected: static UUID storeHead(UUID type, const Ref & ref); static bool replaceHead(UUID type, UUID id, const Ref & old, const Ref & ref); static std::optional updateHead(UUID type, UUID id, const Ref & old, const std::function &); + void watchHead(UUID type, UUID id, const std::function) const; }; class Digest @@ -482,6 +483,7 @@ public: const Ref & ref() const { return mstored.ref(); } std::optional> update(const std::function(const Stored &)> &) const; + void watch(const std::function &)> &) const; private: UUID mid; @@ -536,6 +538,14 @@ std::optional> Head::update(const std::function(const Store return Head(mid, *res); } +template +void Head::watch(const std::function &)> & watcher) const +{ + stored().ref().storage().watchHead(T::headTypeId, id(), [id = id(), watcher] (const Ref & ref) { + watcher(Head(id, ref)); + }); +} + } namespace std diff --git a/src/network.cpp b/src/network.cpp index 4b79dcb..bd64e07 100644 --- a/src/network.cpp +++ b/src/network.cpp @@ -21,8 +21,8 @@ using std::unique_lock; using namespace erebos; -Server::Server(const Identity & self, vector> && svcs): - p(new Priv(self, std::move(svcs))) +Server::Server(const Head & head, vector> && svcs): + p(new Priv(head, *head->identity(), std::move(svcs))) { } @@ -139,7 +139,9 @@ void PeerList::onUpdate(function w) } -Server::Priv::Priv(const Identity & self, vector> && svcs): +Server::Priv::Priv(const Head & local, const Identity & self, + vector> && svcs): + localHead(local), self(self), services(std::move(svcs)) { @@ -176,6 +178,8 @@ Server::Priv::Priv(const Identity & self, vector> && svcs): threadListen = thread([this] { doListen(); }); threadAnnounce = thread([this] { doAnnounce(); }); + + local.watch(std::bind(&Priv::handleLocalHeadChange, this, std::placeholders::_1)); } Server::Priv::~Priv() @@ -447,6 +451,23 @@ void Server::Priv::handlePacket(Server::Peer & peer, const TransportHeader & hea } } +void Server::Priv::handleLocalHeadChange(const Head & head) +{ + scoped_lock lock(dataMutex); + if (auto id = head->identity()) { + if (id->ref()->digest() != self.ref()->digest()) { + self = *id; + + TransportHeader header({ + { TransportHeader::Type::AnnounceSelf, *self.ref() } + }); + + for (const auto & peer : peers) + peer->send(header, { **self.ref() }); + } + } +} + void Server::Peer::send(const TransportHeader & header, const vector & objs) const { vector data, part, out; diff --git a/src/network.h b/src/network.h index 9b146a9..c02dbc3 100644 --- a/src/network.h +++ b/src/network.h @@ -136,7 +136,8 @@ struct WaitingRef struct Server::Priv { - Priv(const Identity & self, vector> && svcs); + Priv(const Head & local, const Identity & self, + vector> && svcs); ~Priv(); void doListen(); void doAnnounce(); @@ -144,6 +145,8 @@ struct Server::Priv Peer & getPeer(const sockaddr_in & paddr); void handlePacket(Peer &, const TransportHeader &, ReplyBuilder &); + void handleLocalHeadChange(const Head &); + constexpr static uint16_t discoveryPort { 29665 }; constexpr static chrono::seconds announceInterval { 60 }; @@ -151,6 +154,7 @@ struct Server::Priv condition_variable announceCondvar; bool finish = false; + Head localHead; Identity self; vector> services; 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 #include +#include #include +#include +#include #include #include @@ -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 & 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> FilesystemStorage::loadKey(const Digest & pubref) const { fs::path path = keyPath(pubref); @@ -259,6 +305,55 @@ void FilesystemStorage::storeKey(const Digest & pubref, const vector & 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 & watcher) +{ + scoped_lock lock(watcherLock); + watchers.emplace(type, watcher); +} + optional> 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 & watcher) +{ + parent->watchHead(type, watcher); + storage->watchHead(type, watcher); +} + optional> ChainStorage::loadKey(const Digest & digest) const { if (auto res = storage->loadKey(digest)) @@ -669,6 +780,15 @@ optional Storage::updateHead(UUID type, UUID id, const Ref & old, const std return nullopt; } +void Storage::watchHead(UUID type, UUID wid, const std::function 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) { diff --git a/src/storage.h b/src/storage.h index 18ac1ad..b8d769c 100644 --- a/src/storage.h +++ b/src/storage.h @@ -2,16 +2,21 @@ #include "erebos/storage.h" +#include +#include #include #include namespace fs = std::filesystem; +using std::function; +using std::mutex; using std::optional; using std::shared_future; using std::shared_ptr; using std::unique_ptr; using std::unordered_map; +using std::unordered_multimap; using std::unordered_set; using std::tuple; using std::variant; @@ -34,6 +39,7 @@ public: virtual vector> headRefs(UUID type) const = 0; virtual UUID storeHead(UUID type, const Digest & dgst) = 0; virtual bool replaceHead(UUID type, UUID id, const Digest & old, const Digest & dgst) = 0; + virtual void watchHead(UUID type, const function &) = 0; virtual optional> loadKey(const Digest &) const = 0; virtual void storeKey(const Digest &, const vector &) = 0; @@ -43,7 +49,7 @@ class FilesystemStorage : public StorageBackend { public: FilesystemStorage(const fs::path &); - virtual ~FilesystemStorage() = default; + virtual ~FilesystemStorage(); virtual bool contains(const Digest &) const override; @@ -54,20 +60,31 @@ public: virtual vector> headRefs(UUID type) const override; virtual UUID storeHead(UUID type, const Digest & dgst) override; virtual bool replaceHead(UUID type, UUID id, const Digest & old, const Digest & dgst) override; + virtual void watchHead(UUID type, const function &) override; virtual optional> loadKey(const Digest &) const override; virtual void storeKey(const Digest &, const vector &) override; private: + void inotifyWatch(); + static constexpr size_t CHUNK = 16384; fs::path objectPath(const Digest &) const; + fs::path headPath(UUID id) const; fs::path headPath(UUID id, UUID type) const; fs::path keyPath(const Digest &) const; FILE * openLockFile(const fs::path & path) const; fs::path root; + + mutex watcherLock; + std::thread watcherThread; + int inotify = -1; + int inotifyWakeup = -1; + unordered_multimap> watchers; + unordered_map watchMap; }; class MemoryStorage : public StorageBackend @@ -85,6 +102,7 @@ public: virtual vector> headRefs(UUID type) const override; virtual UUID storeHead(UUID type, const Digest & dgst) override; virtual bool replaceHead(UUID type, UUID id, const Digest & old, const Digest & dgst) override; + virtual void watchHead(UUID type, const function &) override; virtual optional> loadKey(const Digest &) const override; virtual void storeKey(const Digest &, const vector &) override; @@ -93,6 +111,9 @@ private: unordered_map> storage; unordered_map>> heads; unordered_map> keys; + + mutex watcherLock; + unordered_multimap> watchers; }; class ChainStorage : public StorageBackend @@ -113,6 +134,7 @@ public: virtual vector> headRefs(UUID type) const override; virtual UUID storeHead(UUID type, const Digest & dgst) override; virtual bool replaceHead(UUID type, UUID id, const Digest & old, const Digest & dgst) override; + virtual void watchHead(UUID type, const function &) override; virtual optional> loadKey(const Digest &) const override; virtual void storeKey(const Digest &, const vector &) override; -- cgit v1.2.3