Archives
/
lokinet
mirror of https://github.com/oxen-io/lokinet
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

Partial implementation of libquic as wire protocol

Browse Source 
TODO:

- set up all the callbacks for libquic

- define control message requests, responses, commands

- plug new control messages into lokinet (path creation, network state, etc)

- plug connection state changes (established, failed, closed, etc.) into lokinet

- lots of cleanup and miscellanea
pull/2196/head
Thomas Winget 9 months ago
parent ad5055b85f
commit ab86318404
13 changed files with 337 additions and 596 deletions
Show Stats Download Patch File Download Diff File

13
llarp/link/connection.hpp
Unescape Escape View File

@ -8,23 +8,14 @@
namespace llarp::link
{
class Connection
struct Connection
{
std::shared_ptr<oxen::quic::connection_interface> conn;
std::shared_ptr<oxen::quic::Stream> control_stream;
RouterID remote_id;
RouterContact remote_rc;
AddressInfo remote_addr_info; // RC may have many, this is the one in use for this connection
bool inbound; // one side of a connection will be responsible for some things, e.g. heartbeat
bool remote_is_relay;
public:
const RouterContact& RemoteRC() { return remote_rc; }
const RouterID& RemoteID() { return remote_id; }
bool RemoteIsRelay() { return remote_is_relay; }
};
} // namespace llarp::link

38
llarp/link/endpoint.hpp
Unescape Escape View File

@ -10,48 +10,16 @@
namespace llarp::link
{
#ifndef NDEBUG
struct debug_hooks
{
oxen::quic::dgram_data_callback incoming_datagram;
oxen::quic::stream_data_callback incoming_stream_packet;
};
#endif
class Endpoint
struct Endpoint
{
std::shared_ptr<oxen::quic::Endpoint> endpoint;
bool inbound {false};
// for outgoing packets, we route via RouterID; map RouterID->Connection
// for incoming packets, we get a ConnectionID; map ConnectionID->RouterID
std::unordered_map<RouterID, std::shared_ptr<llarp::link::Connection>> connections;
std::unordered_map<RouterID, llarp::link::Connection> connections;
std::unordered_map<oxen::quic::ConnectionID, RouterID> connid_map;
AbstractRouter* router;
oxen::quic::Address bind_addr;
public:
#ifndef NDEBUG
debug_hooks debug;
// could obviously set directly because public, but setter doesn't hurt
void SetDebugHooks(debug_hooks hooks) { debug = std::move(hooks); }
#endif
Endpoint(AbstractRouter* router, oxen::quic::Address bind_addr);
// Establish a connection to the remote `rc`.
//
// If already connected (or pending), returns existing connection.
// If connection not possible (e.g. no suitable remote address), returns nullptr.
// Otherwise, creates and returns Connection, usable right away.
std::shared_ptr<llarp::link::Connection> Connect(RouterContact rc);
// Return the Connection to `remote` if we have one, else nullptr.
std::shared_ptr<llarp::link::Connection> GetConnection(RouterID remote);
void HandleIncomingDataMessage(oxen::quic::dgram_interface& dgi, bstring dgram);
void HandleIncomingControlMessage(oxen::quic::Stream& stream, bstring_view packet);
};
} // namespace llarp::link

33
llarp/link/i_link_manager.hpp
Unescape Escape View File

@ -20,7 +20,7 @@ namespace llarp
{
virtual ~ILinkManager() = default;
virtual LinkLayer_ptr
virtual llarp::link::Endpoint*
GetCompatibleLink(const RouterContact& rc) const = 0;
virtual IOutboundSessionMaker*
@ -34,28 +34,12 @@ namespace llarp
uint16_t priority = 0) = 0;
virtual bool
HasSessionTo(const RouterID& remote) const = 0;
HaveConnection(const RouterID& remote) const = 0;
// it is fine to have both an inbound and outbound session with
// another relay, and is useful for network testing. This test
// is more specific for use with "should we connect outbound?"
virtual bool
HasOutboundSessionTo(const RouterID& remote) const = 0;
/// return true if the session with this pubkey is a client
/// return false if the session with this pubkey is a router
/// return std::nullopt we have no session with this pubkey
virtual std::optional<bool>
SessionIsClient(RouterID remote) const = 0;
virtual void
PumpLinks() = 0;
virtual void
AddLink(LinkLayer_ptr link, bool inbound = false) = 0;
virtual bool
StartLinks() = 0;
/// return true if we have a connection to the remote and it is not a relay,
/// else return false
bool
HaveClientConnection(const RouterID& remote) const = 0;
virtual void
Stop() = 0;
@ -82,14 +66,11 @@ namespace llarp
DeregisterPeer(RouterID remote) = 0;
virtual size_t
NumberOfConnectedRouters() const = 0;
NumberOfConnectedRouters(bool clients_only = false) const = 0;
virtual size_t
NumberOfConnectedClients() const = 0;
virtual size_t
NumberOfPendingConnections() const = 0;
virtual bool
GetRandomConnectedRouter(RouterContact& router) const = 0;

413
llarp/link/link_manager.cpp
Unescape Escape View File

@ -8,32 +8,25 @@
namespace llarp
{
LinkLayer_ptr
llarp::link::Endpoint*
LinkManager::GetCompatibleLink(const RouterContact& rc) const
{
if (stopping)
return nullptr;
for (auto& link : outboundLinks)
for (const auto& ep : endpoints)
{
// 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 (not link->IsCompatable(rc))
continue;
return link;
//TODO: need some notion of "is this link compatible with that address".
// iwp just checks that the link dialect ("iwp") matches the address info dialect,
// but that feels insufficient. For now, just return the first endpoint we have;
// we should probably only have 1 for now anyway until we make ipv6 work.
return &ep;
}
return nullptr;
}
IOutboundSessionMaker*
LinkManager::GetSessionMaker() const
{
return _sessionMaker;
}
//TODO: replace with control/data message sending with libquic
bool
LinkManager::SendTo(
const RouterID& remote,
@ -58,108 +51,48 @@ namespace llarp
}
bool
LinkManager::HasSessionTo(const RouterID& remote) const
{
return GetLinkWithSessionTo(remote) != nullptr;
}
bool
LinkManager::HasOutboundSessionTo(const RouterID& remote) const
LinkManager::HaveClientConnection(const RouterID& remote) const
{
for (const auto& link : outboundLinks)
for (const auto& ep : endpoints)
{
if (link->HasSessionTo(remote))
if (auto itr = ep.connections.find(remote); itr != ep.connections.end())
{
if (itr->second.remote_is_relay)
return false;
return true;
}
}
return false;
}
std::optional<bool>
LinkManager::SessionIsClient(RouterID remote) const
{
for (const auto& link : inboundLinks)
{
const auto session = link->FindSessionByPubkey(remote);
if (session)
return not session->IsRelay();
}
if (HasOutboundSessionTo(remote))
return false;
return std::nullopt;
}
void
LinkManager::DeregisterPeer(RouterID remote)
{
m_PersistingSessions.erase(remote);
for (const auto& link : inboundLinks)
for (const auto& ep : endpoints)
{
link->CloseSessionTo(remote);
if (auto itr = ep.connections.find(remote); itr != ep.connections.end())
itr->second.conn->close(); //TODO: libquic needs some function for this
}
for (const auto& link : outboundLinks)
{
link->CloseSessionTo(remote);
}
LogInfo(remote, " has been de-registered");
}
void
LinkManager::PumpLinks()
{
for (const auto& link : inboundLinks)
{
link->Pump();
}
for (const auto& link : outboundLinks)
{
link->Pump();
}
LogInfo(remote, " has been de-registered");
}
void
LinkManager::AddLink(LinkLayer_ptr link, bool inbound)
AddLink(oxen::quic::Address bind, bool inbound = false)
{
util::Lock l(_mutex);
//TODO: libquic callbacks: new_conn_alpn_notify, new_conn_pubkey_ok, new_conn_established/ready
auto ep = quic->endpoint(bind);
endpoints.emplace_back();
auto& endp = endpoints.back();
endp.endpoint = std::move(ep);
if (inbound)
{
inboundLinks.emplace(link);
}
else
{
outboundLinks.emplace(link);
}
}
bool
LinkManager::StartLinks()
{
LogInfo("starting ", outboundLinks.size(), " outbound links");
for (const auto& link : outboundLinks)
{
if (!link->Start())
{
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())
{
LogWarn("Link ", link->Name(), " failed to start");
return false;
}
LogDebug("Inbound Link ", link->Name(), " started");
}
oxen::quic::dgram_data_callback dgram_cb = [this](oxen::quic::dgram_interface& dgi, bstring dgram){ HandleIncomingDataMessage(dgi, dgram); };
oxen::quic::stream_data_callback stream_cb = [this](oxen::quic::Stream& stream, bstring_view packet){ HandleIncomingControlMessage(stream, packet); };
endp.endpoint->listen(tls_creds, dgram_cb, stream_cb);
endp.inbound = true;
}
return true;
}
void
@ -175,10 +108,7 @@ namespace llarp
LogInfo("stopping links");
stopping = true;
for (const auto& link : outboundLinks)
link->Stop();
for (const auto& link : inboundLinks)
link->Stop();
quic.reset();
}
void
@ -190,13 +120,10 @@ namespace llarp
util::Lock l(_mutex);
m_PersistingSessions[remote] = std::max(until, m_PersistingSessions[remote]);
if (auto maybe = SessionIsClient(remote))
if (HaveClientConnection(remote))
{
if (*maybe)
{
// mark this as a client so we don't try to back connect
m_Clients.Upsert(remote);
}
// mark this as a client so we don't try to back connect
m_Clients.Upsert(remote);
}
}
@ -252,64 +179,28 @@ namespace llarp
}
size_t
LinkManager::NumberOfConnectedRouters() const
LinkManager::NumberOfConnectedRouters(bool clients_only) const
{
std::set<RouterID> connectedRouters;
auto fn = [&connectedRouters](const ILinkSession* session, bool) {
if (session->IsEstablished())
size_t count{0};
for (const auto& ep : endpoints)
{
for (const auto& conn : ep.connections)
{
const RouterContact rc(session->GetRemoteRC());
if (rc.IsPublicRouter())
{
connectedRouters.insert(rc.pubkey);
}
if (not (conn.remote_is_relay and clients_only))
count++;
}
};
ForEachPeer(fn);
}
return connectedRouters.size();
return count;
}
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;
return NumberOfConnectedRouters(true);
}
//TODO: libquic
bool
LinkManager::GetRandomConnectedRouter(RouterContact& router) const
{
@ -339,148 +230,142 @@ namespace llarp
return false;
}
//TODO: this? perhaps no longer necessary in the same way?
void
LinkManager::CheckPersistingSessions(llarp_time_t now)
{
if (stopping)
return;
std::vector<RouterID> sessionsNeeded;
std::vector<RouterID> sessionsClosed;
{
util::Lock l(_mutex);
for (auto [remote, until] : m_PersistingSessions)
{
if (now < until)
{
auto link = GetLinkWithSessionTo(remote);
if (link)
{
link->KeepAliveSessionTo(remote);
}
else if (not m_Clients.Contains(remote))
{
sessionsNeeded.push_back(remote);
}
}
else if (not m_Clients.Contains(remote))
{
sessionsClosed.push_back(remote);
}
}
}
for (const auto& router : sessionsNeeded)
{
LogDebug("ensuring session to ", router, " for previously made commitment");
_sessionMaker->CreateSessionTo(router, nullptr);
}
for (const auto& router : sessionsClosed)
{
m_PersistingSessions.erase(router);
ForEachOutboundLink([router](auto link) { link->CloseSessionTo(router); });
}
}
//TODO: do we still need this concept?
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
});
}
}
//TODO: this
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;
return {};
}
void
LinkManager::Init(IOutboundSessionMaker* sessionMaker)
LinkManager::Init(I_RCLookupHandler* rcLookup)
{
stopping = false;
_sessionMaker = sessionMaker;
_rcLookup = rcLookup;
_nodedb = router->nodedb();
}
LinkLayer_ptr
LinkManager::GetLinkWithSessionTo(const RouterID& remote) const
void
LinkManager::Connect(RouterID router)
{
if (stopping)
return nullptr;
auto fn = [this](const RouterID& r, const RouterContact* const rc, const RCRequestResult res){
if (res == RCRequestResult::Success)
Connect(*rc);
/* TODO:
else
RC lookup failure callback here
*/
};
for (const auto& link : outboundLinks)
_rcLookup->GetRC(router, fn);
}
// This function assumes the RC has already had its signature verified and connection is allowed.
void
LinkManager::Connect(RouterContact rc)
{
//TODO: connection failed callback
if (HaveConnection(rc.pubkey))
return;
// RC shouldn't be valid if this is the case, but may as well sanity check...
//TODO: connection failed callback
if (rc.addrs.empty())
return;
//TODO: connection failed callback
auto* ep = GetCompatibleLink(rc);
if (ep == nullptr)
return;
//TODO: connection established/failed callbacks
oxen::quic::dgram_data_callback dgram_cb = [this](oxen::quic::dgram_interface& dgi, bstring dgram){ HandleIncomingDataMessage(dgi, dgram); };
oxen::quic::stream_data_callback stream_cb = [this](oxen::quic::Stream& stream, bstring_view packet){ HandleIncomingControlMessage(stream, packet); };
//TODO: once "compatible link" cares about address, actually choose addr to connect to
// based on which one is compatible with the link we chose. For now, just use
// the first one.
auto& selected = rc.addrs[0];
llarp::quic::opt::remote_addr remote{selected.IPString(), selected.port};
//TODO: confirm remote end is using the expected pubkey (RouterID).
//TODO: ALPN for "client" vs "relay" (could just be set on endpoint creation)
auto conn_interface = ep->connect(remote, dgram_cb, stream_cb, tls_creds);
std::shared_ptr<oxen::quic::Stream> stream = conn_interface->get_new_stream();
llarp::link::Connection conn;
conn.conn = conn_interface;
conn.control_stream = stream;
conn.remote_rc = rc;
conn.inbound = false;
conn.remote_is_relay = true;
ep->connections[rc.pubkey] = std::move(conn);
ep->connid_map[conn_interface->scid()] = rc.pubkey;
}
void
LinkManager::ConnectToRandomRouters(int numDesired)
{
std::set<RouterID> exclude;
do
{
if (link->HasSessionTo(remote))
auto filter = [exclude](const auto& rc) -> bool { return exclude.count(rc.pubkey) == 0; };
RouterContact other;
if (const auto maybe = _nodedb->GetRandom(filter))
{
return link;
other = *maybe;
}
}
for (const auto& link : inboundLinks)
else
break;
exclude.insert(other.pubkey);
if (not _rcLookup->SessionIsAllowed(other.pubkey))
continue;
Connect(other);
--remainingDesired;
} while (remainingDesired > 0);
}
bool
LinkManager::HaveConnection(const RouterID& remote)
{
for (const auto& ep : endpoints)
{
if (link->HasSessionTo(remote))
if (ep.connections.contains(remote))
{
return link;
return true;
}
}
return nullptr;
return false;
}
void
LinkManager::HandleIncomingDataMessage(oxen::quic::dgram_interface& dgi, bstring dgram)
{
//TODO: this
}
void
LinkManager::HandleIncomingControlMessage(oxen::quic::Stream& stream, bstring_view packet)
{
//TODO: this
}
} // namespace llarp

86
llarp/link/link_manager.hpp
Unescape Escape View File

@ -4,6 +4,9 @@
#include <llarp/util/compare_ptr.hpp>
#include "server.hpp"
#include "endpoint.hpp"
#include <external/oxen-libquic/include/quic.hpp>
#include <unordered_map>
#include <set>
@ -16,9 +19,11 @@ namespace llarp
struct LinkManager final : public ILinkManager
{
public:
LinkManager(AbstractRouter* r) : router(r) {}
~LinkManager() override = default;
LinkLayer_ptr
llarp::link::Endpoint*
GetCompatibleLink(const RouterContact& rc) const override;
IOutboundSessionMaker*
@ -32,25 +37,16 @@ namespace llarp
uint16_t priority) override;
bool
HasSessionTo(const RouterID& remote) const override;
HaveConnection(const RouterID& remote) const override;
bool
HasOutboundSessionTo(const RouterID& remote) const override;
std::optional<bool>
SessionIsClient(RouterID remote) const override;
HaveClientConnection(const RouterID& remote) const
void
DeregisterPeer(RouterID remote) override;
void
PumpLinks() override;
void
AddLink(LinkLayer_ptr link, bool inbound = false) override;
bool
StartLinks() override;
AddLink(oxen::quic::Address bind, bool inbound = false);
void
Stop() override;
@ -58,28 +54,29 @@ namespace llarp
void
PersistSessionUntil(const RouterID& remote, llarp_time_t until) override;
//TODO: change for libquic Connections
void
ForEachPeer(std::function<void(const ILinkSession*, bool)> visit, bool randomize = false)
const override;
//TODO: change for libquic Connections
void
ForEachPeer(std::function<void(ILinkSession*)> visit) override;
//TODO: change for libquic Endpoints
void
ForEachInboundLink(std::function<void(LinkLayer_ptr)> visit) const override;
//TODO: change for libquic Endpoints
void
ForEachOutboundLink(std::function<void(LinkLayer_ptr)> visit) const override;
size_t
NumberOfConnectedRouters() const override;
NumberOfConnectedRouters(bool clients_only = false) const override;
size_t
NumberOfConnectedClients() const override;
size_t
NumberOfPendingConnections() const override;
bool
GetRandomConnectedRouter(RouterContact& router) const override;
@ -93,20 +90,40 @@ namespace llarp
ExtractStatus() const override;
void
Init(IOutboundSessionMaker* sessionMaker);
Init(I_RCLookupHandler* rcLookup);
// Do an RC lookup for the given RouterID; the result will trigger
// Connect(RouterContact) on success (or if we already have it), and will
// trigger connection failure callback on lookup failure.
void
Connect(RouterID router);
// Establish a connection to the remote `rc`.
//
// Connection established/failed callbacks should be invoked when either happens,
// but this function should do nothing if already connected.
void
Connect(RouterContact rc);
// Attempts to connect to a number of random routers.
//
// This will try to connect to *up to* numDesired routers, but will not
// check if we already have a connection to any of the random set, as making
// that thread safe would be slow...I think.
void
ConnectToRandomRouters(int numDesired);
//TODO: tune these (maybe even remove max?) now that we're switching to quic
/// always maintain this many connections to other routers
size_t minConnectedRouters = 4;
/// hard upperbound limit on the number of router to router connections
size_t maxConnectedRouters = 6;
private:
LinkLayer_ptr
GetLinkWithSessionTo(const RouterID& remote) const;
std::atomic<bool> stopping;
mutable util::Mutex _mutex; // protects m_PersistingSessions
using LinkSet = std::set<LinkLayer_ptr, ComparePtr<LinkLayer_ptr>>;
LinkSet outboundLinks;
LinkSet inboundLinks;
// sessions to persist -> timestamp to end persist at
std::unordered_map<RouterID, llarp_time_t> m_PersistingSessions GUARDED_BY(_mutex);
@ -114,7 +131,24 @@ namespace llarp
util::DecayingHashSet<RouterID> m_Clients{path::default_lifetime};
IOutboundSessionMaker* _sessionMaker;
I_RCLookupHandler* _rcLookup;
std::shared_ptr<NodeDB> _nodedb;
AbstractRouter* router;
// FIXME: Lokinet currently expects to be able to kill all network functionality before
// finishing other shutdown things, including destroying this class, and that is all in
// Network's destructor, so we need to be able to destroy it before this class.
std::unique_ptr<oxen::quic::Network> quic { std::make_unique<oxen::quic::Network>() };
std::vector<Endpoint> endpoints;
//TODO: initialize creds
std::shared_ptr<oxen::quic::GNUTLSCreds> tls_creds;
void HandleIncomingDataMessage(oxen::quic::dgram_interface& dgi, bstring dgram);
void HandleIncomingControlMessage(oxen::quic::Stream& stream, bstring_view packet);
};
} // namespace llarp

8
llarp/net/address_info.cpp
Unescape Escape View File

@ -178,6 +178,14 @@ namespace llarp
return fmt::format("[{}]:{}", tmp, port);
}
std::string
AddressInfo::IPString() const
{
char tmp[INET6_ADDRSTRLEN] = {0};
inet_ntop(AF_INET6, (void*)&ip, tmp, sizeof(tmp));
return std::string{sizeof(tmp), tmp};
}
void
to_json(nlohmann::json& j, const AddressInfo& a)
{

3
llarp/net/address_info.hpp
Unescape Escape View File

@ -65,6 +65,9 @@ namespace llarp
std::string
ToString() const;
std::string
IPString() const;
};
void

4
llarp/router/abstractrouter.hpp
Unescape Escape View File

@ -40,7 +40,6 @@ namespace llarp
struct SecretKey;
struct Signature;
struct IOutboundMessageHandler;
struct IOutboundSessionMaker;
struct ILinkManager;
struct I_RCLookupHandler;
struct RoutePoker;
@ -175,9 +174,6 @@ namespace llarp
virtual IOutboundMessageHandler&
outboundMessageHandler() = 0;
virtual IOutboundSessionMaker&
outboundSessionMaker() = 0;
virtual ILinkManager&
linkManager() = 0;

6
llarp/router/outbound_message_handler.cpp
Unescape Escape View File

@ -24,7 +24,7 @@ namespace llarp
const RouterID& remote, const ILinkMessage& msg, SendStatusHandler callback)
{
// if the destination is invalid, callback with failure and return
if (not _router->linkManager().SessionIsClient(remote)
if (not _router->linkManager().HaveClientConnection(remote)
and not _router->rcLookupHandler().SessionIsAllowed(remote))
{
DoCallback(callback, SendStatus::InvalidRouter);
@ -49,7 +49,7 @@ namespace llarp
std::copy_n(buf.base, buf.sz, ent.message.data());
// if we have a session to the destination, queue the message and return
if (_router->linkManager().HasSessionTo(remote))
if (_router->linkManager().HaveConnection(remote))
{
QueueOutboundMessage(std::move(ent));
return true;
@ -213,7 +213,7 @@ namespace llarp
bool
OutboundMessageHandler::SendIfSession(const MessageQueueEntry& ent)
{
if (_router->linkManager().HasSessionTo(ent.router))
if (_router->linkManager().HaveConnection(ent.router))
{
return Send(ent);
}

107
llarp/router/outbound_session_maker.cpp
Unescape Escape View File

@ -6,7 +6,7 @@
#include <llarp/router_contact.hpp>
#include <llarp/nodedb.hpp>
#include "i_rc_lookup_handler.hpp"
#include <llarp/link/i_link_manager.hpp>
#include <llarp/link/link_manager.hpp>
#include <llarp/util/meta/memfn.hpp>
#include <llarp/util/thread/threading.hpp>
#include <llarp/util/status.hpp>
@ -17,19 +17,6 @@
namespace llarp
{
struct PendingSession
{
// TODO: add session establish status metadata, e.g. num retries
const RouterContact rc;
LinkLayer_ptr link;
size_t attemptCount = 0;
PendingSession(RouterContact _rc, LinkLayer_ptr _link)
: rc(std::move(_rc)), link(std::move(_link))
{}
};
bool
OutboundSessionMaker::OnSessionEstablished(ILinkSession* session)
@ -67,12 +54,12 @@ namespace llarp
void
OutboundSessionMaker::CreateSessionTo(const RouterID& router, RouterCallback on_result)
{
if (on_result)
{
util::Lock l(_mutex);
auto itr_pair = pendingCallbacks.emplace(router, CallbacksQueue{});
itr_pair.first->second.push_back(on_result);
if (on_result)
itr_pair.first->second.push_back(on_result);
}
if (HavePendingSessionTo(router))
@ -85,7 +72,7 @@ namespace llarp
// short-circuit to success callback if we already have an outbound session
// to the remote
if (_linkManager->HasOutboundSessionTo(router))
if (_linkManager->HasConnection(router))
{
FinalizeRequest(router, SessionResult::Establish);
return;
@ -103,12 +90,12 @@ namespace llarp
{
const RouterID router{rc.pubkey};
if (on_result)
{
util::Lock l(_mutex);
auto itr_pair = pendingCallbacks.emplace(router, CallbacksQueue{});
itr_pair.first->second.push_back(on_result);
if (on_result)
itr_pair.first->second.push_back(on_result);
}
if (not HavePendingSessionTo(router))
@ -117,14 +104,6 @@ namespace llarp
CreatePendingSession(router);
}
// short-circuit to success callback if we already have an outbound session
// to the remote
if (_linkManager->HasOutboundSessionTo(router))
{
FinalizeRequest(router, SessionResult::Establish);
return;
}
GotRouterContact(router, rc);
}
@ -132,7 +111,7 @@ namespace llarp
OutboundSessionMaker::HavePendingSessionTo(const RouterID& router) const
{
util::Lock l(_mutex);
return pendingSessions.find(router) != pendingSessions.end();
return pendingCallbacks.find(router) != pendingCallbacks.end();
}
void
@ -195,60 +174,25 @@ namespace llarp
}
void
OutboundSessionMaker::DoEstablish(const RouterID& router)
OutboundSessionMaker::GotRouterContact(const RouterID& router, const RouterContact& rc)
{
std::unique_lock l{_mutex};
auto itr = pendingSessions.find(router);
if (itr == pendingSessions.end())
if (not _rcLookup->CheckRC(rc))
{
FinalizeRequest(rc.pubkey, SessionResult::InvalidRouter);
return;
}
const auto& job = itr->second;
if (not job->link->TryEstablishTo(job->rc))
if (not ShouldConnectTo(router))
{
l.unlock();
FinalizeRequest(router, SessionResult::EstablishFail);
FinalizeRequest(router, SessionResult::NoLink);
return;
}
}
void
OutboundSessionMaker::GotRouterContact(const RouterID& router, const RouterContact& rc)
{
{
std::unique_lock l{_mutex};
// in case other request found RC for this router after this request was
// made
auto itr = pendingSessions.find(router);
if (itr == pendingSessions.end())
{
return;
}
LinkLayer_ptr link = _linkManager->GetCompatibleLink(rc);
if (not link)
{
l.unlock();
FinalizeRequest(router, SessionResult::NoLink);
return;
}
auto session = std::make_shared<PendingSession>(rc, link);
itr->second = session;
}
if (ShouldConnectTo(router))
{
_loop->call([this, router] { DoEstablish(router); });
}
auto result = _linkManager->Connect(rc);
if (result)
FinalizeRequest(router, SessionResult::Establish);
else
{
FinalizeRequest(router, SessionResult::NoLink);
}
FinalizeRequest(router, SessionResult::EstablishFail);
}
bool
@ -256,16 +200,14 @@ namespace llarp
{
if (router == us or not _rcLookup->SessionIsAllowed(router))
return false;
if (_linkManager->HasOutboundSessionTo(router))
return false;
if (_router->IsServiceNode())
return true;
size_t numPending = 0;
{
util::Lock lock(_mutex);
if (pendingSessions.find(router) == pendingSessions.end())
numPending += pendingSessions.size();
if (pendingCallbacks.find(router) == pendingCallbacks.end())
numPending += pendingCallbacks.size();
}
return _linkManager->NumberOfConnectedRouters() + numPending < maxConnectedRouters;
@ -330,14 +272,10 @@ namespace llarp
FinalizeRequest(rc.pubkey, SessionResult::Establish);
}
//TODO: rename this, if we even want to keep it
void
OutboundSessionMaker::CreatePendingSession(const RouterID& router)
{
{
util::Lock l(_mutex);
pendingSessions.emplace(router, nullptr);
}
auto peerDb = _router->peerDb();
if (peerDb)
{
@ -377,11 +315,6 @@ namespace llarp
{
_loop->call([callback, router, type] { return callback(router, type); });
}
{
util::Lock l(_mutex);
pendingSessions.erase(router);
}
}
} // namespace llarp

12
llarp/router/outbound_session_maker.hpp
Unescape Escape View File

@ -15,7 +15,7 @@ namespace llarp
{
struct PendingSession;
struct ILinkManager;
struct LinkManager;
struct I_RCLookupHandler;
struct OutboundSessionMaker final : public IOutboundSessionMaker
@ -73,9 +73,6 @@ namespace llarp
size_t maxConnectedRouters = 6;
private:
void
DoEstablish(const RouterID& router) EXCLUDES(_mutex);
void
GotRouterContact(const RouterID& router, const RouterContact& rc) EXCLUDES(_mutex);
@ -98,15 +95,12 @@ namespace llarp
void
FinalizeRequest(const RouterID& router, const SessionResult type) EXCLUDES(_mutex);
mutable util::Mutex _mutex; // protects pendingSessions, pendingCallbacks
std::unordered_map<RouterID, std::shared_ptr<PendingSession>> pendingSessions
GUARDED_BY(_mutex);
mutable util::Mutex _mutex; // protects pendingCallbacks
std::unordered_map<RouterID, CallbacksQueue> pendingCallbacks GUARDED_BY(_mutex);
AbstractRouter* _router = nullptr;
ILinkManager* _linkManager = nullptr;
LinkManager* _linkManager = nullptr;
I_RCLookupHandler* _rcLookup = nullptr;
Profiling* _profiler = nullptr;
std::shared_ptr<NodeDB> _nodedb;

198
llarp/router/router.cpp
Unescape Escape View File

@ -79,6 +79,8 @@ namespace llarp
llarp_dht_context_free(_dht);
}
//TODO: investigate changes needed for libquic integration
// still needed at all?
void
Router::PumpLL()
{
@ -109,6 +111,7 @@ namespace llarp
{"outboundMessages", _outboundMessageHandler.ExtractStatus()}};
}
//TODO: investigate changes needed for libquic integration
util::StatusObject
Router::ExtractSummaryStatus() const
{
@ -206,6 +209,7 @@ namespace llarp
return stats;
}
//TODO: libquic change
bool
Router::HandleRecvLinkMessageBuffer(ILinkSession* session, const llarp_buffer_t& buf)
{
@ -219,6 +223,8 @@ namespace llarp
}
return inbound_link_msg_parser.ProcessFrom(session, buf);
}
//TODO: investigate changes needed for libquic integration
void
Router::Freeze()
{
@ -230,6 +236,7 @@ namespace llarp
});
}
//TODO: investigate changes needed for libquic integration
void
Router::Thaw()
{
@ -305,12 +312,14 @@ namespace llarp
return _outboundMessageHandler.QueueMessage(remote, msg, handler);
}
//TODO: if still needed/useful, replace this in line with libquic impl
void
Router::ForEachPeer(std::function<void(const ILinkSession*, bool)> visit, bool randomize) const
{
_linkManager.ForEachPeer(visit, randomize);
}
//TODO: if still needed/useful, replace this in line with libquic impl
void
Router::ForEachPeer(std::function<void(ILinkSession*)> visit)
{
@ -329,7 +338,7 @@ namespace llarp
if (remote.Verify(Now()))
{
LogDebug("verified signature");
_outboundSessionMaker.CreateSessionTo(remote, nullptr);
_linkManager->Connect(remote);
}
else
LogError(rcfile, " contains invalid RC");
@ -614,6 +623,7 @@ namespace llarp
_rc = std::move(nextRC);
if (rotateKeys)
{
//TODO: libquic change
// propagate RC by renegotiating sessions
ForEachPeer([](ILinkSession* s) {
if (s->RenegotiateSession())
@ -650,8 +660,8 @@ namespace llarp
// Router config
_rc.SetNick(conf.router.m_nickname);
_outboundSessionMaker.maxConnectedRouters = conf.router.m_maxConnectedRouters;
_outboundSessionMaker.minConnectedRouters = conf.router.m_minConnectedRouters;
_linkManager.maxConnectedRouters = conf.router.m_maxConnectedRouters;
_linkManager.minConnectedRouters = conf.router.m_minConnectedRouters;
encryption_keyfile = m_keyManager->m_encKeyPath;
our_rc_file = m_keyManager->m_rcPath;
@ -771,14 +781,7 @@ namespace llarp
// Init components after relevant config settings loaded
_outboundMessageHandler.Init(this);
_outboundSessionMaker.Init(
this,
&_linkManager,
&_rcLookupHandler,
&_routerProfiling,
_loop,
util::memFn(&AbstractRouter::QueueWork, this));
_linkManager.Init(&_outboundSessionMaker);
_linkManager.Init(&_rcLookupHandler);
_rcLookupHandler.Init(
_dht,
_nodedb,
@ -791,10 +794,11 @@ namespace llarp
whitelistRouters,
m_isServiceNode);
// inbound links
InitInboundLinks();
// outbound links
InitOutboundLinks();
//FIXME: kludge for now, will be part of larger cleanup effort.
if (m_isServiceNode)
InitInboundLinks();
else
InitOutboundLinks();
// profiling
_profilesFile = conf.router.m_dataDir / "profiles.dat";
@ -1047,10 +1051,6 @@ namespace llarp
_linkManager.CheckPersistingSessions(now);
size_t connected = NumberOfConnectedRouters();
if (not isSvcNode)
{
connected += _linkManager.NumberOfPendingConnections();
}
const int interval = isSvcNode ? 5 : 2;
const auto timepoint_now = Clock_t::now();
@ -1059,7 +1059,7 @@ namespace llarp
_rcLookupHandler.ExploreNetwork();
m_NextExploreAt = timepoint_now + std::chrono::seconds(interval);
}
size_t connectToNum = _outboundSessionMaker.minConnectedRouters;
size_t connectToNum = _linkManager.minConnectedRouters;
const auto strictConnect = _rcLookupHandler.NumberOfStrictConnectRouters();
if (strictConnect > 0 && connectToNum > strictConnect)
{
@ -1097,7 +1097,7 @@ namespace llarp
{
size_t dlt = connectToNum - connected;
LogDebug("connecting to ", dlt, " random routers to keep alive");
_outboundSessionMaker.ConnectToRandomRouters(dlt);
_linkManager.ConnectToRandomRouters(dlt);
}
_hiddenServiceContext.Tick(now);
@ -1134,6 +1134,7 @@ namespace llarp
}
}
//TODO: libquic change
// get connected peers
std::set<dht::Key_t> peersWeHave;
_linkManager.ForEachPeer([&peersWeHave](ILinkSession* s) {
@ -1156,6 +1157,7 @@ namespace llarp
return CryptoManager::instance()->sign(sig, identity(), buf);
}
//TODO: replace this in line with libquic impl
void
Router::SessionClosed(RouterID remote)
{
@ -1172,6 +1174,7 @@ namespace llarp
}
}
//TODO: replace this in line with libquic impl
void
Router::ConnectionTimedOut(ILinkSession* session)
{
@ -1195,6 +1198,7 @@ namespace llarp
}
}
//TODO: replace this in line with libquic impl
bool
Router::ConnectionEstablished(ILinkSession* session, bool inbound)
{
@ -1275,39 +1279,6 @@ namespace llarp
_rc.routerVersion = RouterVersion(llarp::VERSION, llarp::constants::proto_version);
}
_linkManager.ForEachInboundLink([&](LinkLayer_ptr link) {
AddressInfo ai;
if (link->GetOurAddressInfo(ai))
{
// override ip and port as needed
if (_ourAddress)
{
const auto ai_ip = ai.IP();
const auto override_ip = _ourAddress->getIP();
auto ai_ip_str = var::visit([](auto&& ip) { return ip.ToString(); }, ai_ip);
auto override_ip_str = var::visit([](auto&& ip) { return ip.ToString(); }, override_ip);
if ((not Net().IsBogonIP(ai_ip)) and (not Net().IsBogonIP(override_ip))
and ai_ip != override_ip)
throw std::runtime_error{
"Lokinet is bound to public IP '{}', but public-ip is set to '{}'. Either fix the "
"[router]:public-ip setting or set a bind address in the [bind] section of the "
"config."_format(ai_ip_str, override_ip_str)};
ai.fromSockAddr(*_ourAddress);
}
if (RouterContact::BlockBogons && Net().IsBogon(ai.ip))
throw std::runtime_error{var::visit(
[](auto&& ip) {
return "cannot use " + ip.ToString()
+ " as a public ip as it is in a non routable ip range";
},
ai.IP())};
LogInfo("adding address: ", ai);
_rc.addrs.push_back(ai);
}
});
if (IsServiceNode() and not _rc.IsPublicRouter())
{
LogError("we are configured as relay but have no reachable addresses");
@ -1332,12 +1303,6 @@ namespace llarp
return false;
}
}
_outboundSessionMaker.SetOurRouter(pubkey());
if (!_linkManager.StartLinks())
{
LogWarn("One or more links failed to start.");
return false;
}
if (IsServiceNode())
{
@ -1426,6 +1391,11 @@ namespace llarp
LogDebug("Establishing session to ", router, " for SN testing");
// try to make a session to this random router
// this will do a dht lookup if needed
_linkManager->Connect(router);
/*
* TODO: container of pending snode test routers to be queried on
* connection success/failure, then do this stuff there.
_outboundSessionMaker.CreateSessionTo(
router, [previous_fails = fails, this](const auto& router, const auto result) {
auto rpc = RpcClient();
@ -1465,6 +1435,7 @@ namespace llarp
rpc->InformConnection(router, result == SessionResult::Establish);
}
});
*/
}
});
}
@ -1569,7 +1540,7 @@ namespace llarp
bool
Router::HasSessionTo(const RouterID& remote) const
{
return _linkManager.HasSessionTo(remote);
return _linkManager.HaveConnection(remote);
}
std::string
@ -1589,13 +1560,9 @@ namespace llarp
{
const size_t want = _want;
auto connected = NumberOfConnectedRouters();
if (not IsServiceNode())
{
connected += _linkManager.NumberOfPendingConnections();
}
if (connected >= want)
return;
_outboundSessionMaker.ConnectToRandomRouters(want);
_linkManager.ConnectToRandomRouters(want);
}
bool
@ -1623,7 +1590,7 @@ namespace llarp
return false;
}
_outboundSessionMaker.CreateSessionTo(rc, nullptr);
_linkManager->Connect(rc);
return true;
}
@ -1649,6 +1616,8 @@ namespace llarp
return ep and ep->HasExit();
}
//TODO: change to use new LinkManager foreach semantics, or make function for this
// on LinkManager itself
std::optional<std::variant<nuint32_t, nuint128_t>>
Router::OurPublicIP() const
{
@ -1665,6 +1634,37 @@ namespace llarp
return found;
}
void
AddAddressToRC(AddressInfo& ai)
{
// override ip and port as needed
if (_ourAddress)
{
const auto ai_ip = ai.IP();
const auto override_ip = _ourAddress->getIP();
auto ai_ip_str = var::visit([](auto&& ip) { return ip.ToString(); }, ai_ip);
auto override_ip_str = var::visit([](auto&& ip) { return ip.ToString(); }, override_ip);
if ((not Net().IsBogonIP(ai_ip)) and (not Net().IsBogonIP(override_ip))
and ai_ip != override_ip)
throw std::runtime_error{
"Lokinet is bound to public IP '{}', but public-ip is set to '{}'. Either fix the "
"[router]:public-ip setting or set a bind address in the [bind] section of the "
"config."_format(ai_ip_str, override_ip_str)};
ai.fromSockAddr(*_ourAddress);
}
if (RouterContact::BlockBogons && Net().IsBogon(ai.ip))
throw std::runtime_error{var::visit(
[](auto&& ip) {
return "cannot use " + ip.ToString()
+ " as a public ip as it is in a non routable ip range";
},
ai.IP())};
LogInfo("adding address: ", ai);
_rc.addrs.push_back(ai);
}
void
Router::InitInboundLinks()
{
@ -1702,22 +1702,14 @@ namespace llarp
throw std::runtime_error{"no public ip provided for inbound socket"};
}
auto server = iwp::NewInboundLink(
m_keyManager,
loop(),
util::memFn(&AbstractRouter::rc, this),
util::memFn(&AbstractRouter::HandleRecvLinkMessageBuffer, this),
util::memFn(&AbstractRouter::Sign, this),
nullptr,
util::memFn(&Router::ConnectionEstablished, this),
util::memFn(&AbstractRouter::CheckRenegotiateValid, this),
util::memFn(&Router::ConnectionTimedOut, this),
util::memFn(&AbstractRouter::SessionClosed, this),
util::memFn(&AbstractRouter::TriggerPump, this),
util::memFn(&AbstractRouter::QueueWork, this));
_linkManager.AddLink(bind_addr.ToString(), true);
server->Bind(this, bind_addr);
_linkManager.AddLink(std::move(server), true);
AddressInfo ai;
ai.fromSockAddr(bind_addr);
ai.pubkey = llarp::seckey_topublic(_identity);
ai.dialect = "quicinet"; // FIXME: constant, also better name?
ai.rank = 2; // FIXME: hardcoded from the beginning...keep?
AddAddressToRC(ai);
}
}
@ -1730,47 +1722,7 @@ namespace llarp
for (auto& bind_addr : addrs)
{
auto link = iwp::NewOutboundLink(
m_keyManager,
loop(),
util::memFn(&AbstractRouter::rc, this),
util::memFn(&AbstractRouter::HandleRecvLinkMessageBuffer, this),
util::memFn(&AbstractRouter::Sign, this),
[this](llarp::RouterContact rc) {
if (IsServiceNode())
return;
for (const auto& addr : rc.addrs)
m_RoutePoker->AddRoute(addr.IPv4());
},
util::memFn(&Router::ConnectionEstablished, this),
util::memFn(&AbstractRouter::CheckRenegotiateValid, this),
util::memFn(&Router::ConnectionTimedOut, this),
util::memFn(&AbstractRouter::SessionClosed, this),
util::memFn(&AbstractRouter::TriggerPump, this),
util::memFn(&AbstractRouter::QueueWork, this));
const auto& net = Net();
// If outbound is set to wildcard and we have just one inbound, then bind to the inbound IP;
// if you have more than one inbound you have to be explicit about your outbound.
if (net.IsWildcardAddress(bind_addr.getIP()))
{
bool multiple = false;
_linkManager.ForEachInboundLink([&bind_addr, &multiple](const auto& link) {
if (multiple)
throw std::runtime_error{
"outbound= IP address must be specified when using multiple inbound= addresses"};
multiple = true;
bind_addr.setIP(link->LocalSocketAddr().getIP());
});
}
link->Bind(this, bind_addr);
if constexpr (llarp::platform::is_android)
m_OutboundUDPSocket = link->GetUDPFD().value_or(-1);
_linkManager.AddLink(std::move(link), false);
_linkManager.AddLink(bind_addr.ToString(), false);
}
}

12
llarp/router/router.hpp
Unescape Escape View File

@ -311,8 +311,7 @@ namespace llarp
Profiling _routerProfiling;
fs::path _profilesFile;
OutboundMessageHandler _outboundMessageHandler;
OutboundSessionMaker _outboundSessionMaker;
LinkManager _linkManager;
LinkManager _linkManager { this };
RCLookupHandler _rcLookupHandler;
RCGossiper _rcGossiper;
@ -330,12 +329,6 @@ namespace llarp
return _outboundMessageHandler;
}
IOutboundSessionMaker&
outboundSessionMaker() override
{
return _outboundSessionMaker;
}
ILinkManager&
linkManager() override
{
@ -595,6 +588,9 @@ namespace llarp
bool
TooFewPeers() const;
void
AddAddressToRC(AddressInfo& ai);
protected:
virtual void
HandleRouterEvent(tooling::RouterEventPtr event) const override;

Write Preview
Loading…
Cancel
Save