mirror of https://github.com/oxen-io/lokinet
Jeff
5 years ago
committed by
GitHub
commit
f7e05ad13a
@ -0,0 +1 @@
|
||||
#include <link/i_link_manager.hpp>
|
||||
@ -0,0 +1,85 @@
|
||||
#ifndef LLARP_I_LINK_MANAGER_HPP
|
||||
#define LLARP_I_LINK_MANAGER_HPP
|
||||
|
||||
#include <link/server.hpp>
|
||||
#include <util/types.hpp>
|
||||
#include <util/logic.hpp>
|
||||
|
||||
#include <functional>
|
||||
|
||||
struct llarp_buffer_t;
|
||||
|
||||
namespace llarp
|
||||
{
|
||||
using Logic_ptr = std::shared_ptr< Logic >;
|
||||
|
||||
struct RouterContact;
|
||||
struct ILinkSession;
|
||||
struct IOutboundSessionMaker;
|
||||
struct RouterID;
|
||||
|
||||
namespace util
|
||||
{
|
||||
struct StatusObject;
|
||||
} // namespace util
|
||||
|
||||
struct ILinkManager
|
||||
{
|
||||
virtual ~ILinkManager() = default;
|
||||
|
||||
virtual LinkLayer_ptr
|
||||
GetCompatibleLink(const RouterContact &rc) const = 0;
|
||||
|
||||
virtual IOutboundSessionMaker *
|
||||
GetSessionMaker() const = 0;
|
||||
|
||||
virtual bool
|
||||
SendTo(const RouterID &remote, const llarp_buffer_t &buf) = 0;
|
||||
|
||||
virtual bool
|
||||
HasSessionTo(const RouterID &remote) const = 0;
|
||||
|
||||
virtual void
|
||||
PumpLinks() = 0;
|
||||
|
||||
virtual void
|
||||
AddLink(LinkLayer_ptr link, bool inbound = false) = 0;
|
||||
|
||||
virtual bool
|
||||
StartLinks(Logic_ptr logic) = 0;
|
||||
|
||||
virtual void
|
||||
Stop() = 0;
|
||||
|
||||
virtual void
|
||||
PersistSessionUntil(const RouterID &remote, llarp_time_t until) = 0;
|
||||
|
||||
virtual void
|
||||
ForEachPeer(std::function< void(const ILinkSession *, bool) > visit,
|
||||
bool randomize = false) const = 0;
|
||||
|
||||
virtual void
|
||||
ForEachPeer(std::function< void(ILinkSession *) > visit) = 0;
|
||||
|
||||
virtual void
|
||||
ForEachInboundLink(std::function< void(LinkLayer_ptr) > visit) const = 0;
|
||||
|
||||
virtual size_t
|
||||
NumberOfConnectedRouters() const = 0;
|
||||
|
||||
virtual size_t
|
||||
NumberOfConnectedClients() const = 0;
|
||||
|
||||
virtual bool
|
||||
GetRandomConnectedRouter(RouterContact &router) const = 0;
|
||||
|
||||
virtual void
|
||||
CheckPersistingSessions(llarp_time_t now) = 0;
|
||||
|
||||
virtual util::StatusObject
|
||||
ExtractStatus() const = 0;
|
||||
};
|
||||
|
||||
} // namespace llarp
|
||||
|
||||
#endif // LLARP_I_LINK_MANAGER_HPP
|
||||
@ -0,0 +1,362 @@
|
||||
#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)
|
||||
{
|
||||
if(stopping)
|
||||
return false;
|
||||
|
||||
auto link = GetLinkWithSessionTo(remote);
|
||||
if(link == nullptr)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
return link->SendTo(remote, buf);
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
m_PersistingSessions[remote] =
|
||||
std::max(until, m_PersistingSessions[remote]);
|
||||
LogDebug("persist session to ", remote, " until ",
|
||||
m_PersistingSessions[remote]);
|
||||
}
|
||||
|
||||
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();
|
||||
}
|
||||
|
||||
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())
|
||||
{
|
||||
auto link = GetLinkWithSessionTo(itr->first);
|
||||
if(now < itr->second)
|
||||
{
|
||||
if(link)
|
||||
{
|
||||
LogDebug("keepalive to ", itr->first);
|
||||
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 &router : sessionsNeeded)
|
||||
{
|
||||
_sessionMaker->CreateSessionTo(router, nullptr);
|
||||
}
|
||||
}
|
||||
|
||||
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 : inboundLinks)
|
||||
{
|
||||
if(link->HasSessionTo(remote))
|
||||
{
|
||||
return link;
|
||||
}
|
||||
}
|
||||
for(const auto &link : outboundLinks)
|
||||
{
|
||||
if(link->HasSessionTo(remote))
|
||||
{
|
||||
return link;
|
||||
}
|
||||
}
|
||||
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
} // namespace llarp
|
||||
@ -0,0 +1,100 @@
|
||||
#ifndef LLARP_LINK_MANAGER_HPP
|
||||
#define LLARP_LINK_MANAGER_HPP
|
||||
|
||||
#include <link/i_link_manager.hpp>
|
||||
|
||||
#include <util/compare_ptr.hpp>
|
||||
#include <util/threading.hpp>
|
||||
#include <link/server.hpp>
|
||||
|
||||
#include <unordered_map>
|
||||
#include <set>
|
||||
#include <atomic>
|
||||
|
||||
namespace llarp
|
||||
{
|
||||
struct IRouterContactManager;
|
||||
|
||||
struct LinkManager final : public ILinkManager
|
||||
{
|
||||
public:
|
||||
~LinkManager() = default;
|
||||
|
||||
LinkLayer_ptr
|
||||
GetCompatibleLink(const RouterContact &rc) const override;
|
||||
|
||||
IOutboundSessionMaker *
|
||||
GetSessionMaker() const override;
|
||||
|
||||
bool
|
||||
SendTo(const RouterID &remote, const llarp_buffer_t &buf) override;
|
||||
|
||||
bool
|
||||
HasSessionTo(const RouterID &remote) const override;
|
||||
|
||||
void
|
||||
PumpLinks() override;
|
||||
|
||||
void
|
||||
AddLink(LinkLayer_ptr link, bool inbound = false) override;
|
||||
|
||||
bool
|
||||
StartLinks(Logic_ptr logic) override;
|
||||
|
||||
void
|
||||
Stop() override;
|
||||
|
||||
void
|
||||
PersistSessionUntil(const RouterID &remote, llarp_time_t until) override;
|
||||
|
||||
void
|
||||
ForEachPeer(std::function< void(const ILinkSession *, bool) > visit,
|
||||
bool randomize = false) const override;
|
||||
|
||||
void
|
||||
ForEachPeer(std::function< void(ILinkSession *) > visit) override;
|
||||
|
||||
void
|
||||
ForEachInboundLink(
|
||||
std::function< void(LinkLayer_ptr) > visit) const override;
|
||||
|
||||
size_t
|
||||
NumberOfConnectedRouters() const override;
|
||||
|
||||
size_t
|
||||
NumberOfConnectedClients() const override;
|
||||
|
||||
bool
|
||||
GetRandomConnectedRouter(RouterContact &router) const override;
|
||||
|
||||
void
|
||||
CheckPersistingSessions(llarp_time_t now) override;
|
||||
|
||||
virtual util::StatusObject
|
||||
ExtractStatus() const override;
|
||||
|
||||
void
|
||||
Init(IOutboundSessionMaker *sessionMaker);
|
||||
|
||||
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, RouterID::Hash >
|
||||
m_PersistingSessions GUARDED_BY(_mutex);
|
||||
|
||||
IOutboundSessionMaker *_sessionMaker;
|
||||
};
|
||||
|
||||
} // namespace llarp
|
||||
|
||||
#endif // LLARP_LINK_MANAGER_HPP
|
||||
@ -0,0 +1 @@
|
||||
#include <router/i_outbound_message_handler.hpp>
|
||||
@ -0,0 +1,43 @@
|
||||
#ifndef LLARP_ROUTER_I_OUTBOUND_MESSAGE_HANDLER_HPP
|
||||
#define LLARP_ROUTER_I_OUTBOUND_MESSAGE_HANDLER_HPP
|
||||
|
||||
#include <cstdint>
|
||||
#include <functional>
|
||||
|
||||
namespace llarp
|
||||
{
|
||||
enum class SendStatus
|
||||
{
|
||||
Success,
|
||||
Timeout,
|
||||
NoLink,
|
||||
InvalidRouter,
|
||||
RouterNotFound,
|
||||
Congestion
|
||||
};
|
||||
|
||||
struct ILinkMessage;
|
||||
struct RouterID;
|
||||
|
||||
namespace util
|
||||
{
|
||||
struct StatusObject;
|
||||
}
|
||||
|
||||
using SendStatusHandler = std::function< void(SendStatus) >;
|
||||
|
||||
struct IOutboundMessageHandler
|
||||
{
|
||||
virtual ~IOutboundMessageHandler() = default;
|
||||
|
||||
virtual bool
|
||||
QueueMessage(const RouterID &remote, const ILinkMessage *msg,
|
||||
SendStatusHandler callback) = 0;
|
||||
|
||||
virtual util::StatusObject
|
||||
ExtractStatus() const = 0;
|
||||
};
|
||||
|
||||
} // namespace llarp
|
||||
|
||||
#endif // LLARP_ROUTER_I_OUTBOUND_MESSAGE_HANDLER_HPP
|
||||
@ -0,0 +1 @@
|
||||
#include <router/i_outbound_session_maker.hpp>
|
||||
@ -0,0 +1,59 @@
|
||||
#ifndef LLARP_ROUTER_I_OUTBOUND_SESSION_MAKER_HPP
|
||||
#define LLARP_ROUTER_I_OUTBOUND_SESSION_MAKER_HPP
|
||||
|
||||
#include <util/types.hpp>
|
||||
|
||||
#include <functional>
|
||||
|
||||
namespace llarp
|
||||
{
|
||||
namespace util
|
||||
{
|
||||
struct StatusObject;
|
||||
} // namespace util
|
||||
|
||||
struct ILinkSession;
|
||||
struct RouterID;
|
||||
struct RouterContact;
|
||||
|
||||
enum class SessionResult
|
||||
{
|
||||
Establish,
|
||||
Timeout,
|
||||
RouterNotFound,
|
||||
InvalidRouter,
|
||||
NoLink
|
||||
};
|
||||
|
||||
using RouterCallback =
|
||||
std::function< void(const RouterID &, const SessionResult) >;
|
||||
|
||||
struct IOutboundSessionMaker
|
||||
{
|
||||
virtual ~IOutboundSessionMaker() = default;
|
||||
|
||||
virtual bool
|
||||
OnSessionEstablished(ILinkSession *session) = 0;
|
||||
|
||||
virtual void
|
||||
OnConnectTimeout(ILinkSession *session) = 0;
|
||||
|
||||
virtual void
|
||||
CreateSessionTo(const RouterID &router, RouterCallback on_result) = 0;
|
||||
|
||||
virtual void
|
||||
CreateSessionTo(const RouterContact &rc, RouterCallback on_result) = 0;
|
||||
|
||||
virtual bool
|
||||
HavePendingSessionTo(const RouterID &router) const = 0;
|
||||
|
||||
virtual void
|
||||
ConnectToRandomRouters(int numDesired, llarp_time_t now) = 0;
|
||||
|
||||
virtual util::StatusObject
|
||||
ExtractStatus() const = 0;
|
||||
};
|
||||
|
||||
} // namespace llarp
|
||||
|
||||
#endif // LLARP_ROUTER_I_OUTBOUND_SESSION_MAKER_HPP
|
||||
@ -0,0 +1 @@
|
||||
#include <router/i_rc_lookup_handler.hpp>
|
||||
@ -0,0 +1,63 @@
|
||||
#ifndef LLARP_I_RC_LOOKUP_HANDLER_HPP
|
||||
#define LLARP_I_RC_LOOKUP_HANDLER_HPP
|
||||
|
||||
#include <util/types.hpp>
|
||||
#include <router_id.hpp>
|
||||
|
||||
#include <memory>
|
||||
#include <set>
|
||||
#include <vector>
|
||||
|
||||
namespace llarp
|
||||
{
|
||||
struct RouterContact;
|
||||
|
||||
enum class RCRequestResult
|
||||
{
|
||||
Success,
|
||||
InvalidRouter,
|
||||
RouterNotFound,
|
||||
BadRC
|
||||
};
|
||||
|
||||
using RCRequestCallback = std::function< void(
|
||||
const RouterID &, const RouterContact *const, const RCRequestResult) >;
|
||||
|
||||
struct I_RCLookupHandler
|
||||
{
|
||||
virtual ~I_RCLookupHandler() = default;
|
||||
|
||||
virtual void
|
||||
AddValidRouter(const RouterID &router) = 0;
|
||||
|
||||
virtual void
|
||||
RemoveValidRouter(const RouterID &router) = 0;
|
||||
|
||||
virtual void
|
||||
SetRouterWhitelist(const std::vector< RouterID > &routers) = 0;
|
||||
|
||||
virtual void
|
||||
GetRC(const RouterID &router, RCRequestCallback callback) = 0;
|
||||
|
||||
virtual bool
|
||||
RemoteIsAllowed(const RouterID &remote) const = 0;
|
||||
|
||||
virtual bool
|
||||
CheckRC(const RouterContact &rc) const = 0;
|
||||
|
||||
virtual bool
|
||||
GetRandomWhitelistRouter(RouterID &router) const = 0;
|
||||
|
||||
virtual bool
|
||||
CheckRenegotiateValid(RouterContact newrc, RouterContact oldrc) = 0;
|
||||
|
||||
virtual void
|
||||
PeriodicUpdate(llarp_time_t now) = 0;
|
||||
|
||||
virtual void
|
||||
ExploreNetwork() = 0;
|
||||
};
|
||||
|
||||
} // namespace llarp
|
||||
|
||||
#endif // LLARP_I_RC_LOOKUP_HANDLER_HPP
|
||||
@ -0,0 +1,224 @@
|
||||
#include <router/outbound_message_handler.hpp>
|
||||
|
||||
#include <messages/link_message.hpp>
|
||||
#include <router/i_outbound_session_maker.hpp>
|
||||
#include <link/i_link_manager.hpp>
|
||||
#include <constants/link_layer.hpp>
|
||||
#include <util/memfn.hpp>
|
||||
#include <util/status.hpp>
|
||||
|
||||
#include <algorithm>
|
||||
#include <cstdlib>
|
||||
|
||||
namespace llarp
|
||||
{
|
||||
bool
|
||||
OutboundMessageHandler::QueueMessage(const RouterID &remote,
|
||||
const ILinkMessage *msg,
|
||||
SendStatusHandler callback)
|
||||
{
|
||||
std::array< byte_t, MAX_LINK_MSG_SIZE > linkmsg_buffer;
|
||||
llarp_buffer_t buf(linkmsg_buffer);
|
||||
|
||||
if(!EncodeBuffer(msg, buf))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
Message message;
|
||||
message.first.reserve(buf.sz);
|
||||
message.second = callback;
|
||||
|
||||
std::copy_n(buf.base, buf.sz, message.first.data());
|
||||
|
||||
if(SendIfSession(remote, message))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
bool shouldCreateSession = false;
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
|
||||
// create queue for <remote> if it doesn't exist, and get iterator
|
||||
auto itr_pair = outboundMessageQueue.emplace(remote, MessageQueue());
|
||||
|
||||
itr_pair.first->second.push_back(std::move(message));
|
||||
|
||||
shouldCreateSession = itr_pair.second;
|
||||
}
|
||||
|
||||
if(shouldCreateSession)
|
||||
{
|
||||
QueueSessionCreation(remote);
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
// TODO: this
|
||||
util::StatusObject
|
||||
OutboundMessageHandler::ExtractStatus() const
|
||||
{
|
||||
util::StatusObject status{};
|
||||
return status;
|
||||
}
|
||||
|
||||
void
|
||||
OutboundMessageHandler::Init(ILinkManager *linkManager,
|
||||
std::shared_ptr< Logic > logic)
|
||||
{
|
||||
_linkManager = linkManager;
|
||||
_logic = logic;
|
||||
}
|
||||
|
||||
void
|
||||
OutboundMessageHandler::OnSessionEstablished(const RouterID &router)
|
||||
{
|
||||
FinalizeRequest(router, SendStatus::Success);
|
||||
}
|
||||
|
||||
void
|
||||
OutboundMessageHandler::OnConnectTimeout(const RouterID &router)
|
||||
{
|
||||
FinalizeRequest(router, SendStatus::Timeout);
|
||||
}
|
||||
|
||||
void
|
||||
OutboundMessageHandler::OnRouterNotFound(const RouterID &router)
|
||||
{
|
||||
FinalizeRequest(router, SendStatus::RouterNotFound);
|
||||
}
|
||||
|
||||
void
|
||||
OutboundMessageHandler::OnInvalidRouter(const RouterID &router)
|
||||
{
|
||||
FinalizeRequest(router, SendStatus::InvalidRouter);
|
||||
}
|
||||
|
||||
void
|
||||
OutboundMessageHandler::OnNoLink(const RouterID &router)
|
||||
{
|
||||
FinalizeRequest(router, SendStatus::NoLink);
|
||||
}
|
||||
|
||||
void
|
||||
OutboundMessageHandler::OnSessionResult(const RouterID &router,
|
||||
const SessionResult result)
|
||||
{
|
||||
switch(result)
|
||||
{
|
||||
case SessionResult::Establish:
|
||||
OnSessionEstablished(router);
|
||||
break;
|
||||
case SessionResult::Timeout:
|
||||
OnConnectTimeout(router);
|
||||
break;
|
||||
case SessionResult::RouterNotFound:
|
||||
OnRouterNotFound(router);
|
||||
break;
|
||||
case SessionResult::InvalidRouter:
|
||||
OnInvalidRouter(router);
|
||||
break;
|
||||
case SessionResult::NoLink:
|
||||
OnNoLink(router);
|
||||
break;
|
||||
default:
|
||||
LogError("Impossible situation: enum class value out of bounds.");
|
||||
std::abort();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
void
|
||||
OutboundMessageHandler::DoCallback(SendStatusHandler callback,
|
||||
SendStatus status)
|
||||
{
|
||||
if(callback)
|
||||
{
|
||||
auto func = std::bind(callback, status);
|
||||
_logic->queue_func(func);
|
||||
}
|
||||
}
|
||||
|
||||
void
|
||||
OutboundMessageHandler::QueueSessionCreation(const RouterID &remote)
|
||||
{
|
||||
auto fn = util::memFn(&OutboundMessageHandler::OnSessionResult, this);
|
||||
_linkManager->GetSessionMaker()->CreateSessionTo(remote, fn);
|
||||
}
|
||||
|
||||
bool
|
||||
OutboundMessageHandler::EncodeBuffer(const ILinkMessage *msg,
|
||||
llarp_buffer_t &buf)
|
||||
{
|
||||
if(!msg->BEncode(&buf))
|
||||
{
|
||||
LogWarn("failed to encode outbound message, buffer size left: ",
|
||||
buf.size_left());
|
||||
return false;
|
||||
}
|
||||
// set size of message
|
||||
buf.sz = buf.cur - buf.base;
|
||||
buf.cur = buf.base;
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool
|
||||
OutboundMessageHandler::Send(const RouterID &remote, const Message &msg)
|
||||
{
|
||||
llarp_buffer_t buf(msg.first);
|
||||
if(_linkManager->SendTo(remote, buf))
|
||||
{
|
||||
DoCallback(msg.second, SendStatus::Success);
|
||||
return true;
|
||||
}
|
||||
DoCallback(msg.second, SendStatus::Congestion);
|
||||
return false;
|
||||
}
|
||||
|
||||
bool
|
||||
OutboundMessageHandler::SendIfSession(const RouterID &remote,
|
||||
const Message &msg)
|
||||
{
|
||||
if(_linkManager->HasSessionTo(remote))
|
||||
{
|
||||
return Send(remote, msg);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
void
|
||||
OutboundMessageHandler::FinalizeRequest(const RouterID &router,
|
||||
SendStatus status)
|
||||
{
|
||||
MessageQueue movedMessages;
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
auto itr = outboundMessageQueue.find(router);
|
||||
|
||||
if(itr == outboundMessageQueue.end())
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
movedMessages.splice(movedMessages.begin(), itr->second);
|
||||
|
||||
outboundMessageQueue.erase(itr);
|
||||
}
|
||||
|
||||
for(const auto &msg : movedMessages)
|
||||
{
|
||||
if(status == SendStatus::Success)
|
||||
{
|
||||
Send(router, msg);
|
||||
}
|
||||
else
|
||||
{
|
||||
DoCallback(msg.second, status);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace llarp
|
||||
@ -0,0 +1,88 @@
|
||||
#ifndef LLARP_ROUTER_OUTBOUND_MESSAGE_HANDLER_HPP
|
||||
#define LLARP_ROUTER_OUTBOUND_MESSAGE_HANDLER_HPP
|
||||
|
||||
#include <router/i_outbound_message_handler.hpp>
|
||||
|
||||
#include <util/threading.hpp>
|
||||
#include <util/logic.hpp>
|
||||
#include <router_id.hpp>
|
||||
|
||||
#include <list>
|
||||
#include <unordered_map>
|
||||
#include <utility>
|
||||
|
||||
struct llarp_buffer_t;
|
||||
|
||||
namespace llarp
|
||||
{
|
||||
struct ILinkManager;
|
||||
struct Logic;
|
||||
enum class SessionResult;
|
||||
|
||||
struct OutboundMessageHandler final : public IOutboundMessageHandler
|
||||
{
|
||||
public:
|
||||
~OutboundMessageHandler() = default;
|
||||
|
||||
bool
|
||||
QueueMessage(const RouterID &remote, const ILinkMessage *msg,
|
||||
SendStatusHandler callback) override;
|
||||
|
||||
util::StatusObject
|
||||
ExtractStatus() const override;
|
||||
|
||||
void
|
||||
Init(ILinkManager *linkManager, std::shared_ptr< Logic > logic);
|
||||
|
||||
private:
|
||||
using Message = std::pair< std::vector< byte_t >, SendStatusHandler >;
|
||||
using MessageQueue = std::list< Message >;
|
||||
|
||||
void
|
||||
OnSessionEstablished(const RouterID &router);
|
||||
|
||||
void
|
||||
OnConnectTimeout(const RouterID &router);
|
||||
|
||||
void
|
||||
OnRouterNotFound(const RouterID &router);
|
||||
|
||||
void
|
||||
OnInvalidRouter(const RouterID &router);
|
||||
|
||||
void
|
||||
OnNoLink(const RouterID &router);
|
||||
|
||||
void
|
||||
OnSessionResult(const RouterID &router, const SessionResult result);
|
||||
|
||||
void
|
||||
DoCallback(SendStatusHandler callback, SendStatus status);
|
||||
|
||||
void
|
||||
QueueSessionCreation(const RouterID &remote);
|
||||
|
||||
bool
|
||||
EncodeBuffer(const ILinkMessage *msg, llarp_buffer_t &buf);
|
||||
|
||||
bool
|
||||
Send(const RouterID &remote, const Message &msg);
|
||||
|
||||
bool
|
||||
SendIfSession(const RouterID &remote, const Message &msg);
|
||||
|
||||
void
|
||||
FinalizeRequest(const RouterID &router, SendStatus status);
|
||||
|
||||
mutable util::Mutex _mutex; // protects outboundMessageQueue
|
||||
|
||||
std::unordered_map< RouterID, MessageQueue, RouterID::Hash >
|
||||
outboundMessageQueue GUARDED_BY(_mutex);
|
||||
|
||||
ILinkManager *_linkManager;
|
||||
std::shared_ptr< Logic > _logic;
|
||||
};
|
||||
|
||||
} // namespace llarp
|
||||
|
||||
#endif // LLARP_ROUTER_OUTBOUND_MESSAGE_HANDLER_HPP
|
||||
@ -0,0 +1,317 @@
|
||||
#include <router/outbound_session_maker.hpp>
|
||||
|
||||
#include <link/server.hpp>
|
||||
#include <router_contact.hpp>
|
||||
#include <nodedb.hpp>
|
||||
#include <router/i_rc_lookup_handler.hpp>
|
||||
#include <link/i_link_manager.hpp>
|
||||
#include <util/logic.hpp>
|
||||
#include <util/memfn.hpp>
|
||||
#include <util/threading.hpp>
|
||||
#include <util/status.hpp>
|
||||
#include <crypto/crypto.hpp>
|
||||
|
||||
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(const RouterContact &rc, LinkLayer_ptr link)
|
||||
: rc(rc), link(link)
|
||||
{
|
||||
}
|
||||
};
|
||||
|
||||
bool
|
||||
OutboundSessionMaker::OnSessionEstablished(ILinkSession *session)
|
||||
{
|
||||
// TODO: do we want to keep it
|
||||
|
||||
const auto router = RouterID(session->GetPubKey());
|
||||
|
||||
const std::string remoteType =
|
||||
session->GetRemoteRC().IsPublicRouter() ? "router" : "client";
|
||||
LogInfo("session with ", remoteType, " [", router, "] established");
|
||||
|
||||
if(not _rcLookup->RemoteIsAllowed(router))
|
||||
{
|
||||
FinalizeRequest(router, SessionResult::InvalidRouter);
|
||||
return false;
|
||||
}
|
||||
|
||||
auto func = std::bind(&OutboundSessionMaker::VerifyRC, this,
|
||||
session->GetRemoteRC());
|
||||
_threadpool->addJob(func);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::OnConnectTimeout(ILinkSession *session)
|
||||
{
|
||||
// TODO: retry/num attempts
|
||||
|
||||
LogWarn("Session establish attempt to ", RouterID(session->GetPubKey()),
|
||||
" timed out.");
|
||||
FinalizeRequest(session->GetPubKey(), SessionResult::Timeout);
|
||||
}
|
||||
|
||||
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(HavePendingSessionTo(router))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
CreatePendingSession(router);
|
||||
|
||||
LogDebug("Creating session establish attempt to ", router, " .");
|
||||
|
||||
auto fn = util::memFn(&OutboundSessionMaker::OnRouterContactResult, this);
|
||||
|
||||
_rcLookup->GetRC(router, fn);
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::CreateSessionTo(const RouterContact &rc,
|
||||
RouterCallback on_result)
|
||||
{
|
||||
if(on_result)
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
|
||||
auto itr_pair = pendingCallbacks.emplace(rc.pubkey, CallbacksQueue{});
|
||||
itr_pair.first->second.push_back(on_result);
|
||||
}
|
||||
|
||||
if(not HavePendingSessionTo(rc.pubkey))
|
||||
{
|
||||
LogDebug("Creating session establish attempt to ", rc.pubkey, " .");
|
||||
CreatePendingSession(rc.pubkey);
|
||||
}
|
||||
|
||||
GotRouterContact(rc.pubkey, rc);
|
||||
}
|
||||
|
||||
bool
|
||||
OutboundSessionMaker::HavePendingSessionTo(const RouterID &router) const
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
return pendingSessions.find(router) != pendingSessions.end();
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::ConnectToRandomRouters(int numDesired, llarp_time_t now)
|
||||
{
|
||||
int remainingDesired = numDesired;
|
||||
|
||||
_nodedb->visit([&](const RouterContact &other) -> bool {
|
||||
// check if we really remainingDesired to
|
||||
if(other.ExpiresSoon(now, 30000)) // TODO: make delta configurable
|
||||
{
|
||||
return remainingDesired > 0;
|
||||
}
|
||||
if(!_rcLookup->RemoteIsAllowed(other.pubkey))
|
||||
{
|
||||
return remainingDesired > 0;
|
||||
}
|
||||
if(randint() % 2 == 0
|
||||
&& !(_linkManager->HasSessionTo(other.pubkey)
|
||||
|| HavePendingSessionTo(other.pubkey)))
|
||||
{
|
||||
CreateSessionTo(other, nullptr);
|
||||
--remainingDesired;
|
||||
}
|
||||
return remainingDesired > 0;
|
||||
});
|
||||
LogDebug("connecting to ", numDesired - remainingDesired, " out of ",
|
||||
numDesired, " random routers");
|
||||
}
|
||||
|
||||
// TODO: this
|
||||
util::StatusObject
|
||||
OutboundSessionMaker::ExtractStatus() const
|
||||
{
|
||||
util::StatusObject status{};
|
||||
return status;
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::Init(
|
||||
ILinkManager *linkManager, I_RCLookupHandler *rcLookup,
|
||||
std::shared_ptr< Logic > logic, llarp_nodedb *nodedb,
|
||||
std::shared_ptr< llarp::thread::ThreadPool > threadpool)
|
||||
{
|
||||
_linkManager = linkManager;
|
||||
_rcLookup = rcLookup;
|
||||
_logic = logic;
|
||||
_nodedb = nodedb;
|
||||
_threadpool = threadpool;
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::DoEstablish(const RouterID &router)
|
||||
{
|
||||
util::ReleasableLock l(&_mutex);
|
||||
|
||||
auto itr = pendingSessions.find(router);
|
||||
|
||||
if(itr == pendingSessions.end())
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
const auto &job = itr->second;
|
||||
if(!job->link->TryEstablishTo(job->rc))
|
||||
{
|
||||
// TODO: maybe different failure type?
|
||||
|
||||
l.Release();
|
||||
FinalizeRequest(router, SessionResult::NoLink);
|
||||
}
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::GotRouterContact(const RouterID &router,
|
||||
const RouterContact &rc)
|
||||
{
|
||||
{
|
||||
util::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(!link)
|
||||
{
|
||||
FinalizeRequest(router, SessionResult::NoLink);
|
||||
return;
|
||||
}
|
||||
|
||||
auto session = std::make_shared< PendingSession >(rc, link);
|
||||
|
||||
itr->second = session;
|
||||
}
|
||||
|
||||
auto fn = std::bind(&OutboundSessionMaker::DoEstablish, this, router);
|
||||
_logic->queue_func(fn);
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::InvalidRouter(const RouterID &router)
|
||||
{
|
||||
FinalizeRequest(router, SessionResult::InvalidRouter);
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::RouterNotFound(const RouterID &router)
|
||||
{
|
||||
FinalizeRequest(router, SessionResult::RouterNotFound);
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::OnRouterContactResult(const RouterID &router,
|
||||
const RouterContact *const rc,
|
||||
const RCRequestResult result)
|
||||
{
|
||||
if(not HavePendingSessionTo(router))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
switch(result)
|
||||
{
|
||||
case RCRequestResult::Success:
|
||||
if(rc)
|
||||
{
|
||||
GotRouterContact(router, *rc);
|
||||
}
|
||||
else
|
||||
{
|
||||
LogError("RCRequestResult::Success but null rc pointer given");
|
||||
}
|
||||
break;
|
||||
case RCRequestResult::InvalidRouter:
|
||||
InvalidRouter(router);
|
||||
break;
|
||||
case RCRequestResult::RouterNotFound:
|
||||
RouterNotFound(router);
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::VerifyRC(const RouterContact rc)
|
||||
{
|
||||
if(not _rcLookup->CheckRC(rc))
|
||||
{
|
||||
FinalizeRequest(rc.pubkey, SessionResult::InvalidRouter);
|
||||
return;
|
||||
}
|
||||
|
||||
FinalizeRequest(rc.pubkey, SessionResult::Establish);
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::CreatePendingSession(const RouterID &router)
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
pendingSessions.emplace(router, nullptr);
|
||||
}
|
||||
|
||||
void
|
||||
OutboundSessionMaker::FinalizeRequest(const RouterID &router,
|
||||
const SessionResult type)
|
||||
{
|
||||
CallbacksQueue movedCallbacks;
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
|
||||
// TODO: Router profiling stuff
|
||||
|
||||
auto itr = pendingCallbacks.find(router);
|
||||
|
||||
if(itr != pendingCallbacks.end())
|
||||
{
|
||||
movedCallbacks.splice(movedCallbacks.begin(), itr->second);
|
||||
pendingCallbacks.erase(itr);
|
||||
}
|
||||
}
|
||||
|
||||
for(const auto &callback : movedCallbacks)
|
||||
{
|
||||
auto func = std::bind(callback, router, type);
|
||||
_logic->queue_func(func);
|
||||
}
|
||||
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
pendingSessions.erase(router);
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace llarp
|
||||
@ -0,0 +1,103 @@
|
||||
#ifndef LLARP_ROUTER_OUTBOUND_SESSION_MAKER_HPP
|
||||
#define LLARP_ROUTER_OUTBOUND_SESSION_MAKER_HPP
|
||||
|
||||
#include <router/i_outbound_session_maker.hpp>
|
||||
|
||||
#include <router/i_rc_lookup_handler.hpp>
|
||||
#include <util/threading.hpp>
|
||||
#include <util/thread_pool.hpp>
|
||||
#include <util/logic.hpp>
|
||||
|
||||
#include <unordered_map>
|
||||
#include <list>
|
||||
#include <memory>
|
||||
|
||||
struct llarp_nodedb;
|
||||
|
||||
namespace llarp
|
||||
{
|
||||
struct PendingSession;
|
||||
|
||||
struct ILinkManager;
|
||||
struct I_RCLookupHandler;
|
||||
|
||||
struct OutboundSessionMaker final : public IOutboundSessionMaker
|
||||
{
|
||||
using CallbacksQueue = std::list< RouterCallback >;
|
||||
|
||||
public:
|
||||
~OutboundSessionMaker() = default;
|
||||
|
||||
bool
|
||||
OnSessionEstablished(ILinkSession *session) override;
|
||||
|
||||
void
|
||||
OnConnectTimeout(ILinkSession *session) override;
|
||||
|
||||
void
|
||||
CreateSessionTo(const RouterID &router,
|
||||
RouterCallback on_result) /* override */;
|
||||
|
||||
void
|
||||
CreateSessionTo(const RouterContact &rc,
|
||||
RouterCallback on_result) /* override */;
|
||||
|
||||
bool
|
||||
HavePendingSessionTo(const RouterID &router) const override;
|
||||
|
||||
void
|
||||
ConnectToRandomRouters(int numDesired, llarp_time_t now) override;
|
||||
|
||||
util::StatusObject
|
||||
ExtractStatus() const override;
|
||||
|
||||
void
|
||||
Init(ILinkManager *linkManager, I_RCLookupHandler *rcLookup,
|
||||
std::shared_ptr< Logic > logic, llarp_nodedb *nodedb,
|
||||
std::shared_ptr< llarp::thread::ThreadPool > threadpool);
|
||||
|
||||
private:
|
||||
void
|
||||
DoEstablish(const RouterID &router);
|
||||
|
||||
void
|
||||
GotRouterContact(const RouterID &router, const RouterContact &rc);
|
||||
|
||||
void
|
||||
InvalidRouter(const RouterID &router);
|
||||
|
||||
void
|
||||
RouterNotFound(const RouterID &router);
|
||||
|
||||
void
|
||||
OnRouterContactResult(const RouterID &router, const RouterContact *const rc,
|
||||
const RCRequestResult result);
|
||||
|
||||
void
|
||||
VerifyRC(const RouterContact rc);
|
||||
|
||||
void
|
||||
CreatePendingSession(const RouterID &router);
|
||||
|
||||
void
|
||||
FinalizeRequest(const RouterID &router, const SessionResult type);
|
||||
|
||||
mutable util::Mutex _mutex; // protects pendingSessions, pendingCallbacks
|
||||
|
||||
std::unordered_map< RouterID, std::shared_ptr< PendingSession >,
|
||||
RouterID::Hash >
|
||||
pendingSessions GUARDED_BY(_mutex);
|
||||
|
||||
std::unordered_map< RouterID, CallbacksQueue, RouterID::Hash >
|
||||
pendingCallbacks GUARDED_BY(_mutex);
|
||||
|
||||
ILinkManager *_linkManager;
|
||||
I_RCLookupHandler *_rcLookup;
|
||||
std::shared_ptr< Logic > _logic;
|
||||
llarp_nodedb *_nodedb;
|
||||
std::shared_ptr< llarp::thread::ThreadPool > _threadpool;
|
||||
};
|
||||
|
||||
} // namespace llarp
|
||||
|
||||
#endif // LLARP_ROUTER_OUTBOUND_SESSION_MAKER_HPP
|
||||
@ -0,0 +1,337 @@
|
||||
#include <router/rc_lookup_handler.hpp>
|
||||
|
||||
#include <link/i_link_manager.hpp>
|
||||
#include <link/server.hpp>
|
||||
#include <crypto/crypto.hpp>
|
||||
#include <service/context.hpp>
|
||||
#include <router_contact.hpp>
|
||||
#include <util/memfn.hpp>
|
||||
#include <util/types.hpp>
|
||||
#include <util/threading.hpp>
|
||||
#include <nodedb.hpp>
|
||||
#include <dht/context.hpp>
|
||||
|
||||
#include <iterator>
|
||||
#include <functional>
|
||||
|
||||
namespace llarp
|
||||
{
|
||||
void
|
||||
RCLookupHandler::AddValidRouter(const RouterID &router)
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
whitelistRouters.insert(router);
|
||||
}
|
||||
|
||||
void
|
||||
RCLookupHandler::RemoveValidRouter(const RouterID &router)
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
whitelistRouters.erase(router);
|
||||
}
|
||||
|
||||
void
|
||||
RCLookupHandler::SetRouterWhitelist(const std::vector< RouterID > &routers)
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
|
||||
whitelistRouters.clear();
|
||||
for(auto &router : routers)
|
||||
{
|
||||
whitelistRouters.emplace(router);
|
||||
}
|
||||
|
||||
LogInfo("lokinet service node list now has ", whitelistRouters.size(),
|
||||
" routers");
|
||||
}
|
||||
|
||||
void
|
||||
RCLookupHandler::GetRC(const RouterID &router, RCRequestCallback callback)
|
||||
{
|
||||
RouterContact remoteRC;
|
||||
|
||||
if(_nodedb->Get(router, remoteRC))
|
||||
{
|
||||
if(callback)
|
||||
{
|
||||
callback(router, &remoteRC, RCRequestResult::Success);
|
||||
}
|
||||
FinalizeRequest(router, &remoteRC, RCRequestResult::Success);
|
||||
return;
|
||||
}
|
||||
|
||||
bool shouldDoLookup = false;
|
||||
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
|
||||
auto itr_pair = pendingCallbacks.emplace(router, CallbacksQueue{});
|
||||
|
||||
if(callback)
|
||||
{
|
||||
itr_pair.first->second.push_back(callback);
|
||||
}
|
||||
shouldDoLookup = itr_pair.second;
|
||||
}
|
||||
|
||||
if(shouldDoLookup)
|
||||
{
|
||||
auto fn = std::bind(&RCLookupHandler::HandleDHTLookupResult, this, router,
|
||||
std::placeholders::_1);
|
||||
|
||||
// if we are a client try using the hidden service endpoints
|
||||
if(!isServiceNode)
|
||||
{
|
||||
bool sent = false;
|
||||
LogInfo("Lookup ", router, " anonymously");
|
||||
_hiddenServiceContext->ForEachService(
|
||||
[&](const std::string &,
|
||||
const std::shared_ptr< service::Endpoint > &ep) -> bool {
|
||||
const bool success = ep->LookupRouterAnon(router, fn);
|
||||
sent = sent || success;
|
||||
return !success;
|
||||
});
|
||||
if(sent)
|
||||
return;
|
||||
LogWarn("cannot lookup ", router, " anonymously");
|
||||
}
|
||||
|
||||
if(!_dht->impl->LookupRouter(router, fn))
|
||||
{
|
||||
FinalizeRequest(router, nullptr, RCRequestResult::RouterNotFound);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
bool
|
||||
RCLookupHandler::RemoteIsAllowed(const RouterID &remote) const
|
||||
{
|
||||
if(_strictConnectPubkeys.size() && _strictConnectPubkeys.count(remote) == 0
|
||||
&& !RemoteInBootstrap(remote))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
util::Lock l(&_mutex);
|
||||
|
||||
if(useWhitelist && whitelistRouters.find(remote) == whitelistRouters.end())
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool
|
||||
RCLookupHandler::CheckRC(const RouterContact &rc) const
|
||||
{
|
||||
if(not RemoteIsAllowed(rc.pubkey))
|
||||
{
|
||||
_dht->impl->DelRCNodeAsync(dht::Key_t{rc.pubkey});
|
||||
return false;
|
||||
}
|
||||
|
||||
if(not rc.Verify(_dht->impl->Now()))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
// update nodedb if required
|
||||
if(rc.IsPublicRouter())
|
||||
{
|
||||
LogInfo("Adding or updating RC for ", RouterID(rc.pubkey),
|
||||
" to nodedb and dht.");
|
||||
_nodedb->UpdateAsyncIfNewer(rc);
|
||||
_dht->impl->PutRCNodeAsync(rc);
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool
|
||||
RCLookupHandler::GetRandomWhitelistRouter(RouterID &router) const
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
|
||||
const auto sz = whitelistRouters.size();
|
||||
auto itr = whitelistRouters.begin();
|
||||
if(sz == 0)
|
||||
return false;
|
||||
if(sz > 1)
|
||||
std::advance(itr, randint() % sz);
|
||||
router = *itr;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool
|
||||
RCLookupHandler::CheckRenegotiateValid(RouterContact newrc,
|
||||
RouterContact oldrc)
|
||||
{
|
||||
// missmatch of identity ?
|
||||
if(newrc.pubkey != oldrc.pubkey)
|
||||
return false;
|
||||
|
||||
if(!RemoteIsAllowed(newrc.pubkey))
|
||||
return false;
|
||||
|
||||
auto func = std::bind(&RCLookupHandler::CheckRC, this, newrc);
|
||||
_threadpool->addJob(func);
|
||||
|
||||
// update dht if required
|
||||
if(_dht->impl->Nodes()->HasNode(dht::Key_t{newrc.pubkey}))
|
||||
{
|
||||
_dht->impl->Nodes()->PutNode(newrc);
|
||||
}
|
||||
|
||||
// TODO: check for other places that need updating the RC
|
||||
return true;
|
||||
}
|
||||
|
||||
void
|
||||
RCLookupHandler::PeriodicUpdate(llarp_time_t now)
|
||||
{
|
||||
// try looking up stale routers
|
||||
std::set< RouterID > routersToLookUp;
|
||||
|
||||
_nodedb->VisitInsertedBefore(
|
||||
[&](const RouterContact &rc) {
|
||||
if(HavePendingLookup(rc.pubkey))
|
||||
return;
|
||||
routersToLookUp.insert(rc.pubkey);
|
||||
},
|
||||
now - RouterContact::UpdateInterval);
|
||||
|
||||
for(const auto &router : routersToLookUp)
|
||||
{
|
||||
GetRC(router, nullptr);
|
||||
}
|
||||
|
||||
_nodedb->RemoveStaleRCs(_bootstrapRouterIDList,
|
||||
now - RouterContact::StaleInsertionAge);
|
||||
}
|
||||
|
||||
void
|
||||
RCLookupHandler::ExploreNetwork()
|
||||
{
|
||||
if(_bootstrapRCList.size())
|
||||
{
|
||||
for(const auto &rc : _bootstrapRCList)
|
||||
{
|
||||
LogInfo("Doing explore via bootstrap node: ", RouterID(rc.pubkey));
|
||||
_dht->impl->ExploreNetworkVia(dht::Key_t{rc.pubkey});
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
LogError("we have no bootstrap nodes specified");
|
||||
}
|
||||
|
||||
// TODO: only explore via random subset
|
||||
// explore via every connected peer
|
||||
_linkManager->ForEachPeer([&](ILinkSession *s) {
|
||||
if(!s->IsEstablished())
|
||||
return;
|
||||
const RouterContact rc = s->GetRemoteRC();
|
||||
if(rc.IsPublicRouter()
|
||||
&& (_bootstrapRCList.find(rc) == _bootstrapRCList.end()))
|
||||
{
|
||||
LogInfo("Doing explore via public node: ", RouterID(rc.pubkey));
|
||||
_dht->impl->ExploreNetworkVia(dht::Key_t{rc.pubkey});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void
|
||||
RCLookupHandler::Init(llarp_dht_context *dht, llarp_nodedb *nodedb,
|
||||
std::shared_ptr< llarp::thread::ThreadPool > threadpool,
|
||||
ILinkManager *linkManager,
|
||||
service::Context *hiddenServiceContext,
|
||||
const std::set< RouterID > &strictConnectPubkeys,
|
||||
const std::set< RouterContact > &bootstrapRCList,
|
||||
bool useWhitelist_arg, bool isServiceNode_arg)
|
||||
{
|
||||
_dht = dht;
|
||||
_nodedb = nodedb;
|
||||
_threadpool = threadpool;
|
||||
_hiddenServiceContext = hiddenServiceContext;
|
||||
_strictConnectPubkeys = strictConnectPubkeys;
|
||||
_bootstrapRCList = bootstrapRCList;
|
||||
_linkManager = linkManager;
|
||||
useWhitelist = useWhitelist_arg;
|
||||
isServiceNode = isServiceNode_arg;
|
||||
|
||||
for(const auto &rc : _bootstrapRCList)
|
||||
{
|
||||
_bootstrapRouterIDList.insert(rc.pubkey);
|
||||
}
|
||||
}
|
||||
|
||||
void
|
||||
RCLookupHandler::HandleDHTLookupResult(
|
||||
RouterID remote, const std::vector< RouterContact > &results)
|
||||
{
|
||||
if(not results.size())
|
||||
{
|
||||
FinalizeRequest(remote, nullptr, RCRequestResult::RouterNotFound);
|
||||
return;
|
||||
}
|
||||
|
||||
if(not RemoteIsAllowed(remote))
|
||||
{
|
||||
FinalizeRequest(remote, &results[0], RCRequestResult::InvalidRouter);
|
||||
return;
|
||||
}
|
||||
|
||||
if(not CheckRC(results[0]))
|
||||
{
|
||||
FinalizeRequest(remote, &results[0], RCRequestResult::BadRC);
|
||||
return;
|
||||
}
|
||||
|
||||
FinalizeRequest(remote, &results[0], RCRequestResult::Success);
|
||||
}
|
||||
|
||||
bool
|
||||
RCLookupHandler::HavePendingLookup(RouterID remote) const
|
||||
{
|
||||
return pendingCallbacks.find(remote) != pendingCallbacks.end();
|
||||
}
|
||||
|
||||
bool
|
||||
RCLookupHandler::RemoteInBootstrap(const RouterID &remote) const
|
||||
{
|
||||
for(const auto &rc : _bootstrapRCList)
|
||||
{
|
||||
if(rc.pubkey == remote)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
void
|
||||
RCLookupHandler::FinalizeRequest(const RouterID &router,
|
||||
const RouterContact *const rc,
|
||||
RCRequestResult result)
|
||||
{
|
||||
CallbacksQueue movedCallbacks;
|
||||
{
|
||||
util::Lock l(&_mutex);
|
||||
|
||||
auto itr = pendingCallbacks.find(router);
|
||||
|
||||
if(itr != pendingCallbacks.end())
|
||||
{
|
||||
movedCallbacks.splice(movedCallbacks.begin(), itr->second);
|
||||
pendingCallbacks.erase(itr);
|
||||
}
|
||||
} // lock
|
||||
|
||||
for(const auto &callback : movedCallbacks)
|
||||
{
|
||||
callback(router, rc, result);
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace llarp
|
||||
@ -0,0 +1,112 @@
|
||||
#ifndef LLARP_RC_LOOKUP_HANDLER_HPP
|
||||
#define LLARP_RC_LOOKUP_HANDLER_HPP
|
||||
|
||||
#include <router/i_rc_lookup_handler.hpp>
|
||||
|
||||
#include <util/threading.hpp>
|
||||
#include <util/thread_pool.hpp>
|
||||
|
||||
#include <unordered_map>
|
||||
#include <set>
|
||||
#include <list>
|
||||
|
||||
struct llarp_nodedb;
|
||||
struct llarp_dht_context;
|
||||
|
||||
namespace llarp
|
||||
{
|
||||
namespace service
|
||||
{
|
||||
struct Context;
|
||||
|
||||
} // namespace service
|
||||
|
||||
struct ILinkManager;
|
||||
|
||||
struct RCLookupHandler final : public I_RCLookupHandler
|
||||
{
|
||||
public:
|
||||
using CallbacksQueue = std::list< RCRequestCallback >;
|
||||
|
||||
~RCLookupHandler() = default;
|
||||
|
||||
void
|
||||
AddValidRouter(const RouterID &router) override;
|
||||
|
||||
void
|
||||
RemoveValidRouter(const RouterID &router) override;
|
||||
|
||||
void
|
||||
SetRouterWhitelist(const std::vector< RouterID > &routers) override;
|
||||
|
||||
void
|
||||
GetRC(const RouterID &router, RCRequestCallback callback) override;
|
||||
|
||||
bool
|
||||
RemoteIsAllowed(const RouterID &remote) const override;
|
||||
|
||||
bool
|
||||
CheckRC(const RouterContact &rc) const override;
|
||||
|
||||
bool
|
||||
GetRandomWhitelistRouter(RouterID &router) const override;
|
||||
|
||||
bool
|
||||
CheckRenegotiateValid(RouterContact newrc, RouterContact oldrc) override;
|
||||
|
||||
void
|
||||
PeriodicUpdate(llarp_time_t now) override;
|
||||
|
||||
void
|
||||
ExploreNetwork() override;
|
||||
|
||||
void
|
||||
Init(llarp_dht_context *dht, llarp_nodedb *nodedb,
|
||||
std::shared_ptr< llarp::thread::ThreadPool > threadpool,
|
||||
ILinkManager *linkManager, service::Context *hiddenServiceContext,
|
||||
const std::set< RouterID > &strictConnectPubkeys,
|
||||
const std::set< RouterContact > &bootstrapRCList,
|
||||
bool useWhitelist_arg, bool isServiceNode_arg);
|
||||
|
||||
private:
|
||||
void
|
||||
HandleDHTLookupResult(RouterID remote,
|
||||
const std::vector< RouterContact > &results);
|
||||
|
||||
bool
|
||||
HavePendingLookup(RouterID remote) const;
|
||||
|
||||
bool
|
||||
RemoteInBootstrap(const RouterID &remote) const;
|
||||
|
||||
void
|
||||
FinalizeRequest(const RouterID &router, const RouterContact *const rc,
|
||||
RCRequestResult result);
|
||||
|
||||
mutable util::Mutex _mutex; // protects pendingCallbacks, whitelistRouters
|
||||
|
||||
llarp_dht_context *_dht = nullptr;
|
||||
llarp_nodedb *_nodedb = nullptr;
|
||||
std::shared_ptr< llarp::thread::ThreadPool > _threadpool = nullptr;
|
||||
service::Context *_hiddenServiceContext = nullptr;
|
||||
ILinkManager *_linkManager = nullptr;
|
||||
|
||||
/// explicit whitelist of routers we will connect to directly (not for
|
||||
/// service nodes)
|
||||
std::set< RouterID > _strictConnectPubkeys;
|
||||
|
||||
std::set< RouterContact > _bootstrapRCList;
|
||||
std::set< RouterID > _bootstrapRouterIDList;
|
||||
|
||||
std::unordered_map< RouterID, CallbacksQueue, RouterID::Hash >
|
||||
pendingCallbacks GUARDED_BY(_mutex);
|
||||
|
||||
bool useWhitelist = false;
|
||||
bool isServiceNode = false;
|
||||
|
||||
std::set< RouterID > whitelistRouters GUARDED_BY(_mutex);
|
||||
};
|
||||
|
||||
} // namespace llarp
|
||||
|
||||
#endif // LLARP_RC_LOOKUP_HANDLER_HPP
|
||||
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue