From e7687e020f6d31a22aadb08aa73ae000796ae139 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roman=20Smr=C5=BE?= Date: Tue, 13 Aug 2024 19:42:02 +0200 Subject: Server: avoid executing callbacks while holding locks --- src/network.cpp | 197 +++++++++++++++++++++++++++++++++++--------------------- src/network.h | 5 +- 2 files changed, 126 insertions(+), 76 deletions(-) diff --git a/src/network.cpp b/src/network.cpp index 7f37cf7..e63949f 100644 --- a/src/network.cpp +++ b/src/network.cpp @@ -158,9 +158,11 @@ string Peer::name() const optional Peer::identity() const { - if (auto speer = p->speer.lock()) + if (auto speer = p->speer.lock()) { + scoped_lock lock(speer->server.dataMutex); if (holds_alternative(speer->identity)) return std::get(speer->identity); + } return nullopt; } @@ -192,10 +194,37 @@ uint16_t Peer::port() const void Peer::Priv::notifyWatchers() { - if (auto slist = list.lock()) { - Peer p(shared_from_this()); - for (const auto & w : slist->watchers) - w(listIndex, &p); + vector> callbacks; + + { + if (auto slist = list.lock()) { + scoped_lock lock(slist->dataMutex); + callbacks = slist->watchers; + } + } + + Peer p(shared_from_this()); + for (const auto & w : callbacks) + w(listIndex, &p); +} + +void Peer::Priv::runServicesHandler(Service & service, Ref ref) +{ + if (auto sptr = speer.lock()) { + Service::Context ctx { nullptr }; + + sptr->server.localHead.update([&] (const Stored & local) { + ctx = Service::Context(new Service::Context::Priv { + .ref = ref, + .peer = erebos::Peer(shared_from_this()), + .local = local, + }); + + service.handle(ctx); + return ctx.local(); + }); + + ctx.runAfterCommitHooks(); } } @@ -242,41 +271,53 @@ PeerList::~PeerList() = default; void PeerList::Priv::push(const shared_ptr & speer) { - scoped_lock lock(dataMutex); - size_t s = peers.size(); + vector> callbacks; + size_t s; - speer->lpeer.reset(new Peer::Priv); - speer->lpeer->speer = speer; - speer->lpeer->list = shared_from_this(); - speer->lpeer->listIndex = s; + { + scoped_lock lock(dataMutex); + s = peers.size(); - Peer p(speer->lpeer); + speer->lpeer.reset(new Peer::Priv); + speer->lpeer->speer = speer; + speer->lpeer->list = shared_from_this(); + speer->lpeer->listIndex = s; - peers.push_back(speer->lpeer); - for (const auto & w : watchers) + peers.push_back(speer->lpeer); + callbacks = watchers; + } + + Peer p(speer->lpeer); + for (const auto & w : callbacks) w(s, &p); } size_t PeerList::size() const { + scoped_lock lock(p->dataMutex); return p->peers.size(); } Peer PeerList::at(size_t i) const { + scoped_lock lock(p->dataMutex); return Peer(p->peers.at(i)); } void PeerList::onUpdate(function w) { - scoped_lock lock(p->dataMutex); - for (size_t i = 0; i < p->peers.size(); i++) { - if (auto speer = p->peers[i]->speer.lock()) { - Peer peer(speer->lpeer); - w(i, &peer); - } + vector peers; + + { + scoped_lock lock(p->dataMutex); + for (size_t i = 0; i < p->peers.size(); i++) + if (auto speer = p->peers[i]->speer.lock()) + peers.emplace_back(speer->lpeer); + p->watchers.push_back(w); } - p->watchers.push_back(w); + + for (size_t i = 0; i < peers.size(); i++) + w(i, &peers[i]); } @@ -381,6 +422,9 @@ void Server::Priv::doListen() if (!peer) continue; + vector> notifyPeers; + vector, Service &, Ref>> readyServices; + if (auto header = peer->connection.receive(peer->partStorage)) { ReplyBuilder reply; @@ -388,9 +432,9 @@ void Server::Priv::doListen() shared_lock slock(selfMutex); handlePacket(*peer, *header, reply); - peer->updateIdentity(reply); + peer->updateIdentity(reply, notifyPeers); peer->updateChannel(reply); - peer->updateService(reply); + peer->updateService(reply, readyServices); if (!reply.header().empty()) peer->connection.send(peer->partStorage, @@ -398,6 +442,11 @@ void Server::Priv::doListen() peer->connection.trySendOutQueue(); } + + for (const auto & p : notifyPeers) + p->notifyWatchers(); + for (const auto & [ lpeer, service, ref ] : readyServices) + lpeer->runServicesHandler(service, ref); } } @@ -452,41 +501,55 @@ Server::Peer * Server::Priv::findPeer(NetworkProtocol::Connection::Id cid) const Server::Peer & Server::Priv::getPeer(const sockaddr_in6 & paddr) { - scoped_lock lock(dataMutex); + shared_ptr peer; + shared_ptr sptr; - for (auto & peer : peers) - if (memcmp(&peer->connection.peerAddress(), &paddr, sizeof paddr) == 0) - return *peer; - - auto st = self.ref()->storage().deriveEphemeralStorage(); - shared_ptr peer(new Peer { - .server = *this, - .connection = protocol.connect(paddr), - .identity = monostate(), - .identityUpdates = {}, - .tempStorage = st, - .partStorage = st.derivePartialStorage(), - }); - peers.push_back(peer); - plist.p->push(peer); + { + scoped_lock lock(dataMutex); + + for (auto & peer : peers) + if (memcmp(&peer->connection.peerAddress(), &paddr, sizeof paddr) == 0) + return *peer; + + auto st = self.ref()->storage().deriveEphemeralStorage(); + peer.reset(new Peer { + .server = *this, + .connection = protocol.connect(paddr), + .identity = monostate(), + .identityUpdates = {}, + .tempStorage = st, + .partStorage = st.derivePartialStorage(), + }); + peers.push_back(peer); + sptr = plist.p; + } + + sptr->push(peer); return *peer; } Server::Peer & Server::Priv::addPeer(NetworkProtocol::Connection conn) { - scoped_lock lock(dataMutex); + shared_ptr peer; + shared_ptr sptr; - auto st = self.ref()->storage().deriveEphemeralStorage(); - shared_ptr peer(new Peer { - .server = *this, - .connection = move(conn), - .identity = monostate(), - .identityUpdates = {}, - .tempStorage = st, - .partStorage = st.derivePartialStorage(), - }); - peers.push_back(peer); - plist.p->push(peer); + { + scoped_lock lock(dataMutex); + + auto st = self.ref()->storage().deriveEphemeralStorage(); + peer.reset(new Peer { + .server = *this, + .connection = move(conn), + .identity = monostate(), + .identityUpdates = {}, + .tempStorage = st, + .partStorage = st.derivePartialStorage(), + }); + peers.push_back(peer); + sptr = plist.p; + } + + sptr->push(peer); return *peer; } @@ -655,14 +718,14 @@ void Server::Priv::handleLocalHeadChange(const Head & head) } } -void Server::Peer::updateIdentity(ReplyBuilder &) +void Server::Peer::updateIdentity(ReplyBuilder &, vector> & notifyPeers) { if (holds_alternative>(identity)) { if (auto ref = std::get>(identity)->check()) if (auto id = Identity::load(*ref)) { identity.emplace(*id); if (lpeer) - lpeer->notifyWatchers(); + notifyPeers.push_back(lpeer); } } else if (holds_alternative(identity)) { @@ -684,7 +747,7 @@ void Server::Peer::updateIdentity(ReplyBuilder &) if (nid != get(identity)) { identity = move(nid); if (lpeer) - lpeer->notifyWatchers(); + notifyPeers.push_back(lpeer); } } } @@ -742,32 +805,18 @@ void Server::Peer::finalizeChannel(ReplyBuilder & reply, unique_ptr ch) reply.header(NetworkProtocol::Header::AnnounceUpdate { r.digest() }); } -void Server::Peer::updateService(ReplyBuilder & reply) +void Server::Peer::updateService(ReplyBuilder & reply, vector, Service &, Ref>> & readyServices) { decltype(serviceQueue) next; for (auto & x : serviceQueue) { if (auto ref = std::get<1>(x)->check(reply)) { if (lpeer) { - Service::Context ctx { nullptr }; - - server.localHead.update([&] (const Stored & local) { - ctx = Service::Context(new Service::Context::Priv { - .ref = *ref, - .peer = erebos::Peer(lpeer), - .local = local, - }); - - for (auto & svc : server.services) { - if (svc->uuid() == std::get(x)) { - svc->handle(ctx); - break; - } + for (auto & svc : server.services) { + if (svc->uuid() == std::get(x)) { + readyServices.emplace_back(lpeer, *svc, *ref); + break; } - - return ctx.local(); - }); - - ctx.runAfterCommitHooks(); + } } } else { next.push_back(std::move(x)); diff --git a/src/network.h b/src/network.h index d1fae15..12ec4e1 100644 --- a/src/network.h +++ b/src/network.h @@ -57,10 +57,10 @@ struct Server::Peer shared_ptr lpeer = nullptr; - void updateIdentity(ReplyBuilder &); + void updateIdentity(ReplyBuilder &, vector> & notifyPeers); void updateChannel(ReplyBuilder &); void finalizeChannel(ReplyBuilder &, unique_ptr); - void updateService(ReplyBuilder &); + void updateService(ReplyBuilder &, vector, Service &, Ref>> & readyServices); }; struct Peer::Priv : enable_shared_from_this @@ -70,6 +70,7 @@ struct Peer::Priv : enable_shared_from_this size_t listIndex; void notifyWatchers(); + void runServicesHandler(Service & service, Ref ref); }; struct PeerList::Priv : enable_shared_from_this -- cgit v1.2.3