mirror of
https://github.com/oxen-io/lokinet.git
synced 2024-11-17 15:25:35 +00:00
430 lines
9.8 KiB
C++
430 lines
9.8 KiB
C++
#include <link/link_manager.hpp>
|
|
|
|
#include <router/i_outbound_session_maker.hpp>
|
|
#include <crypto/crypto.hpp>
|
|
|
|
#include <algorithm>
|
|
#include <set>
|
|
|
|
namespace llarp
|
|
{
|
|
LinkLayer_ptr
|
|
LinkManager::GetCompatibleLink(const RouterContact& rc) const
|
|
{
|
|
if (stopping)
|
|
return nullptr;
|
|
|
|
for (auto& link : outboundLinks)
|
|
{
|
|
// TODO: may want to add some memory of session failures for a given
|
|
// router on a given link and not return that link here for a
|
|
// duration
|
|
if (!link->IsCompatable(rc))
|
|
continue;
|
|
|
|
return link;
|
|
}
|
|
|
|
return nullptr;
|
|
}
|
|
|
|
IOutboundSessionMaker*
|
|
LinkManager::GetSessionMaker() const
|
|
{
|
|
return _sessionMaker;
|
|
}
|
|
|
|
bool
|
|
LinkManager::SendTo(
|
|
const RouterID& remote, const llarp_buffer_t& buf, ILinkSession::CompletionHandler completed)
|
|
{
|
|
if (stopping)
|
|
return false;
|
|
|
|
auto link = GetLinkWithSessionTo(remote);
|
|
if (link == nullptr)
|
|
{
|
|
if (completed)
|
|
{
|
|
completed(ILinkSession::DeliveryStatus::eDeliveryDropped);
|
|
}
|
|
return false;
|
|
}
|
|
|
|
return link->SendTo(remote, buf, completed);
|
|
}
|
|
|
|
bool
|
|
LinkManager::HasSessionTo(const RouterID& remote) const
|
|
{
|
|
return GetLinkWithSessionTo(remote) != nullptr;
|
|
}
|
|
|
|
void
|
|
LinkManager::PumpLinks()
|
|
{
|
|
for (const auto& link : inboundLinks)
|
|
{
|
|
link->Pump();
|
|
}
|
|
for (const auto& link : outboundLinks)
|
|
{
|
|
link->Pump();
|
|
}
|
|
}
|
|
|
|
void
|
|
LinkManager::AddLink(LinkLayer_ptr link, bool inbound)
|
|
{
|
|
util::Lock l(_mutex);
|
|
|
|
if (inbound)
|
|
{
|
|
inboundLinks.emplace(link);
|
|
}
|
|
else
|
|
{
|
|
outboundLinks.emplace(link);
|
|
}
|
|
}
|
|
|
|
bool
|
|
LinkManager::StartLinks(Logic_ptr logic)
|
|
{
|
|
LogInfo("starting ", outboundLinks.size(), " outbound links");
|
|
for (const auto& link : outboundLinks)
|
|
{
|
|
if (!link->Start(logic))
|
|
{
|
|
LogWarn("outbound link '", link->Name(), "' failed to start");
|
|
return false;
|
|
}
|
|
LogDebug("Outbound Link ", link->Name(), " started");
|
|
}
|
|
|
|
if (inboundLinks.size())
|
|
{
|
|
LogInfo("starting ", inboundLinks.size(), " inbound links");
|
|
for (const auto& link : inboundLinks)
|
|
{
|
|
if (!link->Start(logic))
|
|
{
|
|
LogWarn("Link ", link->Name(), " failed to start");
|
|
return false;
|
|
}
|
|
LogDebug("Inbound Link ", link->Name(), " started");
|
|
}
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
void
|
|
LinkManager::Stop()
|
|
{
|
|
if (stopping)
|
|
{
|
|
return;
|
|
}
|
|
|
|
util::Lock l(_mutex);
|
|
|
|
LogInfo("stopping links");
|
|
stopping = true;
|
|
|
|
for (const auto& link : outboundLinks)
|
|
link->Stop();
|
|
for (const auto& link : inboundLinks)
|
|
link->Stop();
|
|
}
|
|
|
|
void
|
|
LinkManager::PersistSessionUntil(const RouterID& remote, llarp_time_t until)
|
|
{
|
|
if (stopping)
|
|
return;
|
|
|
|
util::Lock l(_mutex);
|
|
auto& curr = m_PersistingSessions[remote];
|
|
if (until > curr)
|
|
curr = until;
|
|
LogDebug("persist session to ", remote, " until ", curr - time_now_ms());
|
|
}
|
|
|
|
void
|
|
LinkManager::ForEachPeer(
|
|
std::function<void(const ILinkSession*, bool)> visit, bool randomize) const
|
|
{
|
|
if (stopping)
|
|
return;
|
|
|
|
for (const auto& link : outboundLinks)
|
|
{
|
|
link->ForEachSession([visit](const ILinkSession* peer) { visit(peer, true); }, randomize);
|
|
}
|
|
for (const auto& link : inboundLinks)
|
|
{
|
|
link->ForEachSession([visit](const ILinkSession* peer) { visit(peer, false); }, randomize);
|
|
}
|
|
}
|
|
|
|
void
|
|
LinkManager::ForEachPeer(std::function<void(ILinkSession*)> visit)
|
|
{
|
|
if (stopping)
|
|
return;
|
|
|
|
for (const auto& link : outboundLinks)
|
|
{
|
|
link->ForEachSession([visit](ILinkSession* peer) { visit(peer); });
|
|
}
|
|
for (const auto& link : inboundLinks)
|
|
{
|
|
link->ForEachSession([visit](ILinkSession* peer) { visit(peer); });
|
|
}
|
|
}
|
|
|
|
void
|
|
LinkManager::ForEachInboundLink(std::function<void(LinkLayer_ptr)> visit) const
|
|
{
|
|
for (const auto& link : inboundLinks)
|
|
{
|
|
visit(link);
|
|
}
|
|
}
|
|
|
|
size_t
|
|
LinkManager::NumberOfConnectedRouters() const
|
|
{
|
|
std::set<RouterID> connectedRouters;
|
|
|
|
auto fn = [&connectedRouters](const ILinkSession* session, bool) {
|
|
if (session->IsEstablished())
|
|
{
|
|
const RouterContact rc(session->GetRemoteRC());
|
|
if (rc.IsPublicRouter())
|
|
{
|
|
connectedRouters.insert(rc.pubkey);
|
|
}
|
|
}
|
|
};
|
|
|
|
ForEachPeer(fn);
|
|
|
|
return connectedRouters.size();
|
|
}
|
|
|
|
size_t
|
|
LinkManager::NumberOfConnectedClients() const
|
|
{
|
|
std::set<RouterID> connectedClients;
|
|
|
|
auto fn = [&connectedClients](const ILinkSession* session, bool) {
|
|
if (session->IsEstablished())
|
|
{
|
|
const RouterContact rc(session->GetRemoteRC());
|
|
if (!rc.IsPublicRouter())
|
|
{
|
|
connectedClients.insert(rc.pubkey);
|
|
}
|
|
}
|
|
};
|
|
|
|
ForEachPeer(fn);
|
|
|
|
return connectedClients.size();
|
|
}
|
|
|
|
size_t
|
|
LinkManager::NumberOfPendingConnections() const
|
|
{
|
|
size_t pending = 0;
|
|
for (const auto& link : inboundLinks)
|
|
{
|
|
pending += link->NumberOfPendingSessions();
|
|
}
|
|
|
|
for (const auto& link : outboundLinks)
|
|
{
|
|
pending += link->NumberOfPendingSessions();
|
|
}
|
|
|
|
return pending;
|
|
}
|
|
|
|
bool
|
|
LinkManager::GetRandomConnectedRouter(RouterContact& router) const
|
|
{
|
|
std::unordered_map<RouterID, RouterContact, RouterID::Hash> connectedRouters;
|
|
|
|
ForEachPeer(
|
|
[&connectedRouters](const ILinkSession* peer, bool unused) {
|
|
(void)unused;
|
|
connectedRouters[peer->GetPubKey()] = peer->GetRemoteRC();
|
|
},
|
|
false);
|
|
|
|
const auto sz = connectedRouters.size();
|
|
if (sz)
|
|
{
|
|
auto itr = connectedRouters.begin();
|
|
if (sz > 1)
|
|
{
|
|
std::advance(itr, randint() % sz);
|
|
}
|
|
|
|
router = itr->second;
|
|
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
void
|
|
LinkManager::CheckPersistingSessions(llarp_time_t now)
|
|
{
|
|
if (stopping)
|
|
return;
|
|
|
|
std::vector<RouterID> sessionsNeeded;
|
|
|
|
{
|
|
util::Lock l(_mutex);
|
|
|
|
auto itr = m_PersistingSessions.begin();
|
|
while (itr != m_PersistingSessions.end())
|
|
{
|
|
if (now < itr->second)
|
|
{
|
|
auto link = GetLinkWithSessionTo(itr->first);
|
|
if (link)
|
|
{
|
|
link->KeepAliveSessionTo(itr->first);
|
|
}
|
|
else
|
|
{
|
|
sessionsNeeded.push_back(itr->first);
|
|
}
|
|
++itr;
|
|
}
|
|
else
|
|
{
|
|
const RouterID r(itr->first);
|
|
LogInfo("commit to ", r, " expired");
|
|
itr = m_PersistingSessions.erase(itr);
|
|
for (const auto& link : outboundLinks)
|
|
{
|
|
link->CloseSessionTo(r);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
for (const auto& router : sessionsNeeded)
|
|
{
|
|
_sessionMaker->CreateSessionTo(router, nullptr);
|
|
}
|
|
}
|
|
|
|
void
|
|
LinkManager::updatePeerDb(std::shared_ptr<PeerDb> peerDb)
|
|
{
|
|
std::vector<std::pair<RouterID, SessionStats>> statsToUpdate;
|
|
|
|
int64_t diffTotalTX = 0;
|
|
|
|
ForEachPeer([&](ILinkSession* session) {
|
|
// derive RouterID
|
|
RouterID id = RouterID(session->GetRemoteRC().pubkey);
|
|
|
|
SessionStats sessionStats = session->GetSessionStats();
|
|
SessionStats diff;
|
|
SessionStats& lastStats = m_lastRouterStats[id];
|
|
|
|
// TODO: operator overloads / member func for diff
|
|
diff.currentRateRX = std::max(sessionStats.currentRateRX, lastStats.currentRateRX);
|
|
diff.currentRateTX = std::max(sessionStats.currentRateTX, lastStats.currentRateTX);
|
|
diff.totalPacketsRX = sessionStats.totalPacketsRX - lastStats.totalPacketsRX;
|
|
diff.totalAckedTX = sessionStats.totalAckedTX - lastStats.totalAckedTX;
|
|
diff.totalDroppedTX = sessionStats.totalDroppedTX - lastStats.totalDroppedTX;
|
|
|
|
diffTotalTX = diff.totalAckedTX + diff.totalDroppedTX + diff.totalInFlightTX;
|
|
|
|
lastStats = sessionStats;
|
|
|
|
// TODO: if we have both inbound and outbound session, this will overwrite
|
|
statsToUpdate.push_back({id, diff});
|
|
});
|
|
|
|
for (auto& routerStats : statsToUpdate)
|
|
{
|
|
peerDb->modifyPeerStats(routerStats.first, [&](PeerStats& stats) {
|
|
// TODO: store separate stats for up vs down
|
|
const auto& diff = routerStats.second;
|
|
|
|
// note that 'currentRateRX' and 'currentRateTX' are per-second
|
|
stats.peakBandwidthBytesPerSec = std::max(
|
|
stats.peakBandwidthBytesPerSec,
|
|
(double)std::max(diff.currentRateRX, diff.currentRateTX));
|
|
stats.numPacketsDropped += diff.totalDroppedTX;
|
|
stats.numPacketsSent = diff.totalAckedTX;
|
|
stats.numPacketsAttempted = diffTotalTX;
|
|
|
|
// TODO: others -- we have slight mismatch on what we store
|
|
});
|
|
}
|
|
}
|
|
|
|
util::StatusObject
|
|
LinkManager::ExtractStatus() const
|
|
{
|
|
std::vector<util::StatusObject> ob_links, ib_links;
|
|
std::transform(
|
|
inboundLinks.begin(),
|
|
inboundLinks.end(),
|
|
std::back_inserter(ib_links),
|
|
[](const auto& link) -> util::StatusObject { return link->ExtractStatus(); });
|
|
std::transform(
|
|
outboundLinks.begin(),
|
|
outboundLinks.end(),
|
|
std::back_inserter(ob_links),
|
|
[](const auto& link) -> util::StatusObject { return link->ExtractStatus(); });
|
|
|
|
util::StatusObject obj{{"outbound", ob_links}, {"inbound", ib_links}};
|
|
|
|
return obj;
|
|
}
|
|
|
|
void
|
|
LinkManager::Init(IOutboundSessionMaker* sessionMaker)
|
|
{
|
|
stopping = false;
|
|
_sessionMaker = sessionMaker;
|
|
}
|
|
|
|
LinkLayer_ptr
|
|
LinkManager::GetLinkWithSessionTo(const RouterID& remote) const
|
|
{
|
|
if (stopping)
|
|
return nullptr;
|
|
|
|
for (const auto& link : outboundLinks)
|
|
{
|
|
if (link->HasSessionTo(remote))
|
|
{
|
|
return link;
|
|
}
|
|
}
|
|
for (const auto& link : inboundLinks)
|
|
{
|
|
if (link->HasSessionTo(remote))
|
|
{
|
|
return link;
|
|
}
|
|
}
|
|
return nullptr;
|
|
}
|
|
|
|
} // namespace llarp
|