From 86b465cfccef5552aa111941fb74ec622e2e7c03 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roman=20Smr=C5=BE?= Date: Thu, 17 Aug 2023 21:54:45 +0200 Subject: Network: use digest instead of partial ref in protocol header --- src/network.cpp | 156 ++++++++++++++++++++++++----------------------- src/network/protocol.cpp | 39 ++++++------ src/network/protocol.h | 4 +- 3 files changed, 100 insertions(+), 99 deletions(-) diff --git a/src/network.cpp b/src/network.cpp index ef06d6e..725ac50 100644 --- a/src/network.cpp +++ b/src/network.cpp @@ -115,7 +115,7 @@ void Server::addPeer(const string & node, const string & service) const { shared_lock lock(p->selfMutex); header.push_back(NetworkProtocol::Header::Item { - NetworkProtocol::Header::Type::AnnounceSelf, *p->self.ref() }); + NetworkProtocol::Header::Type::AnnounceSelf, p->self.ref()->digest() }); } peer.send(header, {}, false); return; @@ -234,7 +234,7 @@ bool Peer::send(UUID uuid, const Ref & ref, const Object & obj) const if (auto speer = p->speer.lock()) { NetworkProtocol::Header header({ { NetworkProtocol::Header::Type::ServiceType, uuid }, - { NetworkProtocol::Header::Type::ServiceRef, ref }, + { NetworkProtocol::Header::Type::ServiceRef, ref.digest() }, }); speer->send(header, { obj }, true); return true; @@ -442,6 +442,8 @@ void Server::Priv::doListen() void Server::Priv::doAnnounce() { + auto pst = self.ref()->storage().derivePartialStorage(); + unique_lock lock(dataMutex); auto lastAnnounce = steady_clock::now() - announceInterval; @@ -451,10 +453,10 @@ void Server::Priv::doAnnounce() if (lastAnnounce + announceInterval < now) { shared_lock slock(selfMutex); NetworkProtocol::Header header({ - { NetworkProtocol::Header::Type::AnnounceSelf, *self.ref() } + { NetworkProtocol::Header::Type::AnnounceSelf, self.ref()->digest() } }); - vector bytes = header.toObject().encode(); + vector bytes = header.toObject(pst).encode(); for (const auto & in : bcastAddresses) { sockaddr_in sin = {}; @@ -544,56 +546,55 @@ void Server::Priv::handlePacket(Server::Peer & peer, const NetworkProtocol::Head for (auto & item : header.items) { switch (item.type) { - case NetworkProtocol::Header::Type::Acknowledged: - if (auto pref = std::get(item.value)) { - if (holds_alternative>(peer.channel) && - std::get>(peer.channel).ref().digest() == pref.digest()) - peer.finalizeChannel(reply, - std::get>(peer.channel)->data->channel()); - } + case NetworkProtocol::Header::Type::Acknowledged: { + auto dgst = std::get(item.value); + if (holds_alternative>(peer.channel) && + std::get>(peer.channel).ref().digest() == dgst) + peer.finalizeChannel(reply, + std::get>(peer.channel)->data->channel()); break; + } case NetworkProtocol::Header::Type::DataRequest: { - auto pref = std::get(item.value); + auto dgst = std::get(item.value); if (holds_alternative>(peer.channel) || - plaintextRefs.find(pref.digest()) != plaintextRefs.end()) { - if (auto ref = peer.tempStorage.ref(pref.digest())) { - NetworkProtocol::Header::Item hitem { NetworkProtocol::Header::Type::DataResponse, *ref }; - reply.header({ NetworkProtocol::Header::Type::DataResponse, *ref }); + plaintextRefs.find(dgst) != plaintextRefs.end()) { + if (auto ref = peer.tempStorage.ref(dgst)) { + reply.header({ NetworkProtocol::Header::Type::DataResponse, ref->digest() }); reply.body(*ref); } } break; } - case NetworkProtocol::Header::Type::DataResponse: - if (auto pref = std::get(item.value)) { - reply.header({ NetworkProtocol::Header::Type::Acknowledged, pref }); - for (auto & pwref : waiting) { - if (auto wref = pwref.lock()) { - if (std::find(wref->missing.begin(), wref->missing.end(), pref.digest()) != - wref->missing.end()) { - if (wref->check(reply)) - pwref.reset(); - } + case NetworkProtocol::Header::Type::DataResponse: { + auto dgst = std::get(item.value); + reply.header({ NetworkProtocol::Header::Type::Acknowledged, dgst }); + for (auto & pwref : waiting) { + if (auto wref = pwref.lock()) { + if (std::find(wref->missing.begin(), wref->missing.end(), dgst) != + wref->missing.end()) { + if (wref->check(reply)) + pwref.reset(); } } - waiting.erase(std::remove_if(waiting.begin(), waiting.end(), - [](auto & wref) { return wref.expired(); }), waiting.end()); } + waiting.erase(std::remove_if(waiting.begin(), waiting.end(), + [](auto & wref) { return wref.expired(); }), waiting.end()); break; + } case NetworkProtocol::Header::Type::AnnounceSelf: { - auto pref = std::get(item.value); - if (pref.digest() == self.ref()->digest()) + auto dgst = std::get(item.value); + if (dgst == self.ref()->digest()) break; if (holds_alternative(peer.identity)) { - reply.header({ NetworkProtocol::Header::Type::AnnounceSelf, *self.ref()}); + reply.header({ NetworkProtocol::Header::Type::AnnounceSelf, self.ref()->digest()}); shared_ptr wref(new WaitingRef { .storage = peer.tempStorage, - .ref = pref, + .ref = peer.partStorage.ref(dgst), .peer = peer, .missing = {}, }); @@ -606,12 +607,12 @@ void Server::Priv::handlePacket(Server::Peer & peer, const NetworkProtocol::Head case NetworkProtocol::Header::Type::AnnounceUpdate: if (holds_alternative(peer.identity)) { - auto pref = std::get(item.value); - reply.header({ NetworkProtocol::Header::Type::Acknowledged, pref }); + auto dgst = std::get(item.value); + reply.header({ NetworkProtocol::Header::Type::Acknowledged, dgst }); shared_ptr wref(new WaitingRef { .storage = peer.tempStorage, - .ref = pref, + .ref = peer.partStorage.ref(dgst), .peer = peer, .missing = {}, }); @@ -621,46 +622,46 @@ void Server::Priv::handlePacket(Server::Peer & peer, const NetworkProtocol::Head } break; - case NetworkProtocol::Header::Type::ChannelRequest: - if (auto pref = std::get(item.value)) { - reply.header({ NetworkProtocol::Header::Type::Acknowledged, pref }); + case NetworkProtocol::Header::Type::ChannelRequest: { + auto dgst = std::get(item.value); + reply.header({ NetworkProtocol::Header::Type::Acknowledged, dgst }); - if (holds_alternative>(peer.channel) && - std::get>(peer.channel).ref().digest() < pref.digest()) - break; + if (holds_alternative>(peer.channel) && + std::get>(peer.channel).ref().digest() < dgst) + break; - if (holds_alternative>(peer.channel)) - break; + if (holds_alternative>(peer.channel)) + break; - shared_ptr wref(new WaitingRef { - .storage = peer.tempStorage, - .ref = pref, - .peer = peer, - .missing = {}, - }); - waiting.push_back(wref); - peer.channel = wref; - wref->check(reply); - } + shared_ptr wref(new WaitingRef { + .storage = peer.tempStorage, + .ref = peer.partStorage.ref(dgst), + .peer = peer, + .missing = {}, + }); + waiting.push_back(wref); + peer.channel = wref; + wref->check(reply); break; + } - case NetworkProtocol::Header::Type::ChannelAccept: - if (auto pref = std::get(item.value)) { - if (holds_alternative>(peer.channel) && - std::get>(peer.channel).ref().digest() < pref.digest()) - break; - - auto cres = peer.tempStorage.copy(pref); - if (auto r = std::get_if(&cres)) { - auto acc = ChannelAccept::load(*r); - if (holds_alternative(peer.identity) && - acc.isSignedBy(std::get(peer.identity).keyMessage())) { - reply.header({ NetworkProtocol::Header::Type::Acknowledged, pref }); - peer.finalizeChannel(reply, acc.data->channel()); - } + case NetworkProtocol::Header::Type::ChannelAccept: { + auto dgst = std::get(item.value); + if (holds_alternative>(peer.channel) && + std::get>(peer.channel).ref().digest() < dgst) + break; + + auto cres = peer.tempStorage.copy(peer.partStorage.ref(dgst)); + if (auto r = std::get_if(&cres)) { + auto acc = ChannelAccept::load(*r); + if (holds_alternative(peer.identity) && + acc.isSignedBy(std::get(peer.identity).keyMessage())) { + reply.header({ NetworkProtocol::Header::Type::Acknowledged, dgst }); + peer.finalizeChannel(reply, acc.data->channel()); } } break; + } case NetworkProtocol::Header::Type::ServiceType: if (!serviceType) @@ -677,9 +678,10 @@ void Server::Priv::handlePacket(Server::Peer & peer, const NetworkProtocol::Head if (!serviceType) break; - auto pref = std::get(item.value); + auto dgst = std::get(item.value); + auto pref = peer.partStorage.ref(dgst); if (pref) - reply.header({ NetworkProtocol::Header::Type::Acknowledged, pref }); + reply.header({ NetworkProtocol::Header::Type::Acknowledged, dgst }); shared_ptr wref(new WaitingRef { .storage = peer.tempStorage, @@ -706,10 +708,10 @@ void Server::Priv::handleLocalHeadChange(const Head & head) vector hitems; for (const auto & r : self.refs()) hitems.push_back(NetworkProtocol::Header::Item { - NetworkProtocol::Header::Type::AnnounceUpdate, r }); + NetworkProtocol::Header::Type::AnnounceUpdate, r.digest() }); for (const auto & r : self.updates()) hitems.push_back(NetworkProtocol::Header::Item { - NetworkProtocol::Header::Type::AnnounceUpdate, r }); + NetworkProtocol::Header::Type::AnnounceUpdate, r.digest() }); NetworkProtocol::Header header(hitems); @@ -723,7 +725,7 @@ void Server::Peer::send(const NetworkProtocol::Header & header, const vector data, part, out; - part = header.toObject().encode(); + part = header.toObject(partStorage).encode(); data.insert(data.end(), part.begin(), part.end()); for (const auto & obj : objs) { part = obj.encode(); @@ -786,7 +788,7 @@ void Server::Peer::updateChannel(ReplyBuilder & reply) auto req = Channel::generateRequest(tempStorage, server.self, std::get(identity)); channel.emplace>(req); - reply.header({ NetworkProtocol::Header::Type::ChannelRequest, req.ref() }); + reply.header({ NetworkProtocol::Header::Type::ChannelRequest, req.ref().digest() }); reply.body(req.ref()); reply.body(req->data.ref()); reply.body(req->data->key.ref()); @@ -801,7 +803,7 @@ void Server::Peer::updateChannel(ReplyBuilder & reply) req->isSignedBy(std::get(identity).keyMessage())) { if (auto acc = Channel::acceptRequest(server.self, std::get(identity), req)) { channel.emplace>(*acc); - reply.header({ NetworkProtocol::Header::Type::ChannelAccept, acc->ref() }); + reply.header({ NetworkProtocol::Header::Type::ChannelAccept, acc->ref().digest() }); reply.body(acc->ref()); reply.body(acc.value()->data.ref()); reply.body(acc.value()->data->key.ref()); @@ -824,10 +826,10 @@ void Server::Peer::finalizeChannel(ReplyBuilder & reply, unique_ptr ch) vector hitems; for (const auto & r : server.self.refs()) reply.header(NetworkProtocol::Header::Item { - NetworkProtocol::Header::Type::AnnounceUpdate, r }); + NetworkProtocol::Header::Type::AnnounceUpdate, r.digest() }); for (const auto & r : server.self.updates()) reply.header(NetworkProtocol::Header::Item { - NetworkProtocol::Header::Type::AnnounceUpdate, r }); + NetworkProtocol::Header::Type::AnnounceUpdate, r.digest() }); } void Server::Peer::updateService(ReplyBuilder & reply) @@ -926,7 +928,7 @@ optional WaitingRef::check(ReplyBuilder & reply) return r; for (const auto & d : missing) - reply.header({ NetworkProtocol::Header::Type::DataRequest, peer.partStorage.ref(d) }); + reply.header({ NetworkProtocol::Header::Type::DataRequest, d }); return nullopt; } diff --git a/src/network/protocol.cpp b/src/network/protocol.cpp index c2c6c5d..fb3a5ea 100644 --- a/src/network/protocol.cpp +++ b/src/network/protocol.cpp @@ -215,9 +215,8 @@ bool NetworkProtocol::Header::Item::operator==(const Item & other) const if (value.index() != other.value.index()) return false; - if (holds_alternative(value)) - return std::get(value).digest() == - std::get(other.value).digest(); + if (holds_alternative(value)) + return std::get(value) == std::get(other.value); if (holds_alternative(value)) return std::get(value) == std::get(other.value); @@ -242,43 +241,43 @@ optional NetworkProtocol::Header::load(const PartialObj if (auto ref = item.asRef()) items.emplace_back(Item { .type = Type::Acknowledged, - .value = *ref, + .value = ref->digest(), }); } else if (item.name == "REQ") { if (auto ref = item.asRef()) items.emplace_back(Item { .type = Type::DataRequest, - .value = *ref, + .value = ref->digest(), }); } else if (item.name == "RSP") { if (auto ref = item.asRef()) items.emplace_back(Item { .type = Type::DataResponse, - .value = *ref, + .value = ref->digest(), }); } else if (item.name == "ANN") { if (auto ref = item.asRef()) items.emplace_back(Item { .type = Type::AnnounceSelf, - .value = *ref, + .value = ref->digest(), }); } else if (item.name == "ANU") { if (auto ref = item.asRef()) items.emplace_back(Item { .type = Type::AnnounceUpdate, - .value = *ref, + .value = ref->digest(), }); } else if (item.name == "CRQ") { if (auto ref = item.asRef()) items.emplace_back(Item { .type = Type::ChannelRequest, - .value = *ref, + .value = ref->digest(), }); } else if (item.name == "CAC") { if (auto ref = item.asRef()) items.emplace_back(Item { .type = Type::ChannelAccept, - .value = *ref, + .value = ref->digest(), }); } else if (item.name == "STP") { if (auto val = item.asUUID()) @@ -290,7 +289,7 @@ optional NetworkProtocol::Header::load(const PartialObj if (auto ref = item.asRef()) items.emplace_back(Item { .type = Type::ServiceRef, - .value = *ref, + .value = ref->digest(), }); } } @@ -298,38 +297,38 @@ optional NetworkProtocol::Header::load(const PartialObj return NetworkProtocol::Header(items); } -PartialObject NetworkProtocol::Header::toObject() const +PartialObject NetworkProtocol::Header::toObject(const PartialStorage & st) const { vector ritems; for (const auto & item : items) { switch (item.type) { case Type::Acknowledged: - ritems.emplace_back("ACK", std::get(item.value)); + ritems.emplace_back("ACK", st.ref(std::get(item.value))); break; case Type::DataRequest: - ritems.emplace_back("REQ", std::get(item.value)); + ritems.emplace_back("REQ", st.ref(std::get(item.value))); break; case Type::DataResponse: - ritems.emplace_back("RSP", std::get(item.value)); + ritems.emplace_back("RSP", st.ref(std::get(item.value))); break; case Type::AnnounceSelf: - ritems.emplace_back("ANN", std::get(item.value)); + ritems.emplace_back("ANN", st.ref(std::get(item.value))); break; case Type::AnnounceUpdate: - ritems.emplace_back("ANU", std::get(item.value)); + ritems.emplace_back("ANU", st.ref(std::get(item.value))); break; case Type::ChannelRequest: - ritems.emplace_back("CRQ", std::get(item.value)); + ritems.emplace_back("CRQ", st.ref(std::get(item.value))); break; case Type::ChannelAccept: - ritems.emplace_back("CAC", std::get(item.value)); + ritems.emplace_back("CAC", st.ref(std::get(item.value))); break; case Type::ServiceType: @@ -337,7 +336,7 @@ PartialObject NetworkProtocol::Header::toObject() const break; case Type::ServiceRef: - ritems.emplace_back("SRF", std::get(item.value)); + ritems.emplace_back("SRF", st.ref(std::get(item.value))); break; } } diff --git a/src/network/protocol.h b/src/network/protocol.h index 8aa22a2..4794ba6 100644 --- a/src/network/protocol.h +++ b/src/network/protocol.h @@ -107,7 +107,7 @@ struct NetworkProtocol::Header struct Item { const Type type; - const variant value; + const variant value; bool operator==(const Item &) const; bool operator!=(const Item & other) const { return !(*this == other); } @@ -116,7 +116,7 @@ struct NetworkProtocol::Header Header(const vector & items): items(items) {} static optional
load(const PartialRef &); static optional
load(const PartialObject &); - PartialObject toObject() const; + PartialObject toObject(const PartialStorage &) const; const vector items; }; -- cgit v1.2.3