mirror of https://github.com/oxen-io/lokinet
Review commit
parent
66ec886480
commit
c0b19de963
@ -0,0 +1,5 @@
|
|||||||
|
set(GRAPHVIZ_MODULE_LIBS OFF CACHE BOOL "")
|
||||||
|
set(GRAPHVIZ_EXTERNAL_LIBS OFF CACHE BOOL "")
|
||||||
|
set(GRAPHVIZ_GENERATE_PER_TARGET OFF CACHE BOOL "")
|
||||||
|
set(GRAPHVIZ_GENERATE_DEPENDERS OFF CACHE BOOL "")
|
||||||
|
set(GRAPHVIZ_OBJECT_LIBS OFF CACHE BOOL "")
|
@ -0,0 +1,421 @@
|
|||||||
|
include(Version)
|
||||||
|
|
||||||
|
add_library(lokinet-cryptography
|
||||||
|
STATIC
|
||||||
|
crypto/crypto.cpp
|
||||||
|
crypto/encrypted_frame.cpp
|
||||||
|
crypto/types.cpp
|
||||||
|
)
|
||||||
|
|
||||||
|
add_library(lokinet-util
|
||||||
|
STATIC
|
||||||
|
${CMAKE_CURRENT_BINARY_DIR}/constants/version.cpp
|
||||||
|
util/bencode.cpp
|
||||||
|
util/buffer.cpp
|
||||||
|
util/file.cpp
|
||||||
|
util/json.cpp
|
||||||
|
util/logging/buffer.cpp
|
||||||
|
util/easter_eggs.cpp
|
||||||
|
util/mem.cpp
|
||||||
|
util/str.cpp
|
||||||
|
util/thread/queue_manager.cpp
|
||||||
|
util/thread/threading.cpp
|
||||||
|
util/time.cpp)
|
||||||
|
|
||||||
|
add_dependencies(lokinet-util genversion)
|
||||||
|
|
||||||
|
# lokinet-platform holds all platform specific code
|
||||||
|
add_library(lokinet-platform
|
||||||
|
STATIC
|
||||||
|
# for networking
|
||||||
|
ev/ev.cpp
|
||||||
|
ev/libuv.cpp
|
||||||
|
net/interface_info.cpp
|
||||||
|
net/ip.cpp
|
||||||
|
net/ip_address.cpp
|
||||||
|
net/ip_packet.cpp
|
||||||
|
net/ip_range.cpp
|
||||||
|
net/net_int.cpp
|
||||||
|
net/sock_addr.cpp
|
||||||
|
vpn/packet_router.cpp
|
||||||
|
vpn/egres_packet_router.cpp
|
||||||
|
vpn/platform.cpp
|
||||||
|
)
|
||||||
|
|
||||||
|
if (ANDROID)
|
||||||
|
target_sources(lokinet-platform PRIVATE android/ifaddrs.c util/nop_service_manager.cpp)
|
||||||
|
endif()
|
||||||
|
|
||||||
|
if(CMAKE_SYSTEM_NAME MATCHES "Linux")
|
||||||
|
target_sources(lokinet-platform PRIVATE linux/dbus.cpp)
|
||||||
|
if(WITH_SYSTEMD)
|
||||||
|
target_sources(lokinet-platform PRIVATE linux/sd_service_manager.cpp)
|
||||||
|
else()
|
||||||
|
target_sources(lokinet-platform PRIVATE util/nop_service_manager.cpp)
|
||||||
|
endif()
|
||||||
|
endif()
|
||||||
|
|
||||||
|
if (WIN32)
|
||||||
|
target_sources(lokinet-platform PRIVATE
|
||||||
|
net/win32.cpp
|
||||||
|
vpn/win32.cpp
|
||||||
|
win32/service_manager.cpp
|
||||||
|
win32/exec.cpp
|
||||||
|
win32/dll.cpp
|
||||||
|
win32/exception.cpp
|
||||||
|
win32/wintun.cpp
|
||||||
|
win32/windivert.cpp)
|
||||||
|
target_include_directories(lokinet-platform PRIVATE ${CMAKE_BINARY_DIR}/wintun/include/ ${CMAKE_BINARY_DIR}/WinDivert-${WINDIVERT_VERSION}/include/)
|
||||||
|
else()
|
||||||
|
target_sources(lokinet-platform PRIVATE
|
||||||
|
net/posix.cpp)
|
||||||
|
endif()
|
||||||
|
|
||||||
|
if(APPLE)
|
||||||
|
add_subdirectory(apple)
|
||||||
|
target_sources(lokinet-platform PRIVATE util/nop_service_manager.cpp)
|
||||||
|
endif()
|
||||||
|
|
||||||
|
# lokinet-dns is the dns parsing and hooking library that we use to
|
||||||
|
# parse modify and reconstitute dns wire proto, dns queries and RR
|
||||||
|
# should have no concept of dns caching, this is left as an implementation
|
||||||
|
# detail of dns resolvers (LATER: make separate lib for dns resolvers)
|
||||||
|
add_library(lokinet-dns
|
||||||
|
STATIC
|
||||||
|
dns/message.cpp
|
||||||
|
dns/name.cpp
|
||||||
|
dns/platform.cpp
|
||||||
|
dns/question.cpp
|
||||||
|
dns/rr.cpp
|
||||||
|
dns/serialize.cpp
|
||||||
|
dns/server.cpp
|
||||||
|
dns/srv_data.cpp)
|
||||||
|
|
||||||
|
# platform specific bits and bobs for setting dns
|
||||||
|
add_library(lokinet-dns-platform INTERFACE)
|
||||||
|
if(WITH_SYSTEMD)
|
||||||
|
add_library(lokinet-dns-systemd STATIC dns/nm_platform.cpp dns/sd_platform.cpp)
|
||||||
|
target_link_libraries(lokinet-dns-platform INTERFACE lokinet-dns-systemd)
|
||||||
|
endif()
|
||||||
|
|
||||||
|
# lokinet-nodedb holds all types and logic for storing parsing and constructing
|
||||||
|
# nodedb data published to the network and versions of it stored locally
|
||||||
|
add_library(lokinet-nodedb
|
||||||
|
STATIC
|
||||||
|
bootstrap.cpp
|
||||||
|
net/exit_info.cpp
|
||||||
|
net/traffic_policy.cpp
|
||||||
|
nodedb.cpp
|
||||||
|
pow.cpp
|
||||||
|
profiling.cpp
|
||||||
|
router_contact.cpp
|
||||||
|
router_id.cpp
|
||||||
|
router_version.cpp
|
||||||
|
)
|
||||||
|
|
||||||
|
set(BOOTSTRAP_FALLBACKS)
|
||||||
|
foreach(bs IN ITEMS MAINNET TESTNET)
|
||||||
|
if(BOOTSTRAP_FALLBACK_${bs})
|
||||||
|
message(STATUS "Building with ${bs} fallback boostrap path \"${BOOTSTRAP_FALLBACK_${bs}}\"")
|
||||||
|
file(READ "${BOOTSTRAP_FALLBACK_${bs}}" bs_data HEX)
|
||||||
|
if(bs STREQUAL TESTNET)
|
||||||
|
set(network "gamma")
|
||||||
|
elseif(bs STREQUAL MAINNET)
|
||||||
|
set(network "lokinet")
|
||||||
|
else()
|
||||||
|
string(TOLOWER "${bs}" network)
|
||||||
|
endif()
|
||||||
|
string(REGEX REPLACE "([0-9a-f][0-9a-f])" "\\\\x\\1" bs_data "${bs_data}")
|
||||||
|
set(BOOTSTRAP_FALLBACKS "${BOOTSTRAP_FALLBACKS}{\"${network}\"s, \"${bs_data}\"sv},\n")
|
||||||
|
endif()
|
||||||
|
endforeach()
|
||||||
|
configure_file("bootstrap-fallbacks.cpp.in" "${CMAKE_CURRENT_BINARY_DIR}/bootstrap-fallbacks.cpp" @ONLY)
|
||||||
|
target_sources(lokinet-nodedb PRIVATE "${CMAKE_CURRENT_BINARY_DIR}/bootstrap-fallbacks.cpp")
|
||||||
|
|
||||||
|
# lokinet-config is for all configuration types and parsers
|
||||||
|
add_library(lokinet-config
|
||||||
|
STATIC
|
||||||
|
config/config.cpp
|
||||||
|
config/definition.cpp
|
||||||
|
config/ini.cpp
|
||||||
|
config/key_manager.cpp
|
||||||
|
)
|
||||||
|
|
||||||
|
# lokinet-consensus is for deriving and tracking network consensus state for both service nodes and clients
|
||||||
|
add_library(lokinet-consensus
|
||||||
|
STATIC
|
||||||
|
consensus/reachability_testing.cpp
|
||||||
|
)
|
||||||
|
|
||||||
|
# lokinet-layer-onion is the "dumb" onion routing layer with builds manages and does i/o
|
||||||
|
# with onion paths. onion paths anonymize routing layer pdu.
|
||||||
|
add_library(lokinet-layer-onion
|
||||||
|
STATIC
|
||||||
|
path/abstracthophandler.cpp
|
||||||
|
path/path_context.cpp
|
||||||
|
path/path.cpp
|
||||||
|
path/pathbuilder.cpp
|
||||||
|
path/pathset.cpp
|
||||||
|
path/transit_hop.cpp
|
||||||
|
messages/relay.cpp
|
||||||
|
)
|
||||||
|
|
||||||
|
# lokinet-layer-link is for our layer 2 analog which splits up layer 2 frames into
|
||||||
|
# a series of layer 1 symbols which are then transmitted between lokinet instances
|
||||||
|
add_library(lokinet-layer-link
|
||||||
|
STATIC
|
||||||
|
link/connection.cpp
|
||||||
|
link/contacts.cpp
|
||||||
|
link/link_manager.cpp
|
||||||
|
# link/tunnel.cpp
|
||||||
|
)
|
||||||
|
|
||||||
|
# lokinet-context holds the contextualized god objects for a lokinet instance
|
||||||
|
# it is what any main function would link to in practice but it is hidden behind an interface library (lokinet-amalgum)
|
||||||
|
add_library(lokinet-context
|
||||||
|
STATIC
|
||||||
|
context.cpp
|
||||||
|
link/link_manager.cpp
|
||||||
|
router/rc_lookup_handler.cpp
|
||||||
|
router/rc_gossiper.cpp
|
||||||
|
router/router.cpp
|
||||||
|
router/route_poker.cpp
|
||||||
|
)
|
||||||
|
|
||||||
|
# lokinet-rpc holds all rpc related compilation units
|
||||||
|
add_library(lokinet-rpc
|
||||||
|
STATIC
|
||||||
|
rpc/json_binary_proxy.cpp
|
||||||
|
rpc/json_conversions.cpp
|
||||||
|
rpc/lokid_rpc_client.cpp
|
||||||
|
rpc/rpc_request_parser.cpp
|
||||||
|
rpc/rpc_server.cpp
|
||||||
|
rpc/endpoint_rpc.cpp
|
||||||
|
)
|
||||||
|
|
||||||
|
# optional peer stats library
|
||||||
|
add_library(lokinet-peerstats
|
||||||
|
STATIC
|
||||||
|
peerstats/peer_db.cpp
|
||||||
|
peerstats/types.cpp
|
||||||
|
)
|
||||||
|
|
||||||
|
# kitchen sink to be removed after refactor
|
||||||
|
add_library(lokinet-service-deprecated-kitchensink
|
||||||
|
STATIC
|
||||||
|
endpoint_base.cpp
|
||||||
|
exit/context.cpp
|
||||||
|
exit/endpoint.cpp
|
||||||
|
exit/policy.cpp
|
||||||
|
exit/session.cpp
|
||||||
|
handlers/exit.cpp
|
||||||
|
handlers/tun.cpp
|
||||||
|
service/name.cpp
|
||||||
|
service/address.cpp
|
||||||
|
service/async_key_exchange.cpp
|
||||||
|
service/auth.cpp
|
||||||
|
service/convotag.cpp
|
||||||
|
service/context.cpp
|
||||||
|
service/endpoint_state.cpp
|
||||||
|
service/endpoint_util.cpp
|
||||||
|
service/endpoint.cpp
|
||||||
|
service/identity.cpp
|
||||||
|
service/info.cpp
|
||||||
|
service/intro_set.cpp
|
||||||
|
service/intro.cpp
|
||||||
|
service/lns_tracker.cpp
|
||||||
|
service/name.cpp
|
||||||
|
service/outbound_context.cpp
|
||||||
|
service/protocol.cpp
|
||||||
|
service/session.cpp
|
||||||
|
service/tag.cpp
|
||||||
|
)
|
||||||
|
|
||||||
|
# interface library for setting commone includes, linkage and flags.
|
||||||
|
add_library(lokinet-base INTERFACE)
|
||||||
|
target_include_directories(lokinet-base
|
||||||
|
INTERFACE ${PROJECT_SOURCE_DIR} ${PROJECT_SOURCE_DIR}/include
|
||||||
|
)
|
||||||
|
target_link_libraries(lokinet-base INTERFACE quic)
|
||||||
|
|
||||||
|
if(WITH_PEERSTATS)
|
||||||
|
target_compile_definitions(lokinet-base INTERFACE -DLOKINET_PEERSTATS_BACKEND)
|
||||||
|
target_link_libraries(lokinet-base INTERFACE sqlite_orm)
|
||||||
|
endif()
|
||||||
|
|
||||||
|
# interface libraries for internal linkage
|
||||||
|
add_library(lokinet-layers INTERFACE)
|
||||||
|
add_library(lokinet-amalgum INTERFACE)
|
||||||
|
|
||||||
|
# helper function to link a library to lokinet-base, enable lto, add to lokinet-amalgum and then link to other libs
|
||||||
|
function(lokinet_link_lib libname)
|
||||||
|
# Absolutely fuck this line, it doesn't "create" the target, as the target already exists
|
||||||
|
# message(DEBUG "created target: ${libname}")
|
||||||
|
enable_lto(${libname})
|
||||||
|
target_link_libraries(${libname} PUBLIC lokinet-base ${ARGN})
|
||||||
|
target_link_libraries(lokinet-amalgum INTERFACE ${libname})
|
||||||
|
endfunction()
|
||||||
|
|
||||||
|
# internal public linkages of components
|
||||||
|
lokinet_link_lib(lokinet-util lokinet-libntrup)
|
||||||
|
lokinet_link_lib(lokinet-cryptography lokinet-libcrypt lokinet-util)
|
||||||
|
lokinet_link_lib(lokinet-peerstats lokinet-context)
|
||||||
|
lokinet_link_lib(lokinet-consensus lokinet-context)
|
||||||
|
lokinet_link_lib(lokinet-layer-link lokinet-cryptography lokinet-peerstats)
|
||||||
|
|
||||||
|
if(TARGET lokinet-dns-systemd)
|
||||||
|
lokinet_link_lib(lokinet-dns-systemd
|
||||||
|
lokinet-dns
|
||||||
|
lokinet-platform
|
||||||
|
)
|
||||||
|
endif()
|
||||||
|
|
||||||
|
lokinet_link_lib(lokinet-platform lokinet-util)
|
||||||
|
|
||||||
|
lokinet_link_lib(lokinet-config
|
||||||
|
lokinet-util
|
||||||
|
lokinet-nodedb
|
||||||
|
lokinet-dns
|
||||||
|
lokinet-platform
|
||||||
|
)
|
||||||
|
|
||||||
|
lokinet_link_lib(lokinet-context
|
||||||
|
lokinet-config
|
||||||
|
lokinet-platform
|
||||||
|
lokinet-peerstats
|
||||||
|
lokinet-layers
|
||||||
|
lokinet-consensus
|
||||||
|
lokinet-rpc
|
||||||
|
)
|
||||||
|
|
||||||
|
lokinet_link_lib(
|
||||||
|
lokinet-platform
|
||||||
|
lokinet-config
|
||||||
|
)
|
||||||
|
|
||||||
|
lokinet_link_lib(lokinet-dns
|
||||||
|
lokinet-platform
|
||||||
|
lokinet-dns-platform
|
||||||
|
lokinet-config
|
||||||
|
)
|
||||||
|
|
||||||
|
lokinet_link_lib(lokinet-nodedb
|
||||||
|
lokinet-util
|
||||||
|
lokinet-platform
|
||||||
|
)
|
||||||
|
|
||||||
|
lokinet_link_lib(lokinet-util
|
||||||
|
lokinet-nodedb
|
||||||
|
lokinet-platform
|
||||||
|
)
|
||||||
|
|
||||||
|
lokinet_link_lib(lokinet-rpc
|
||||||
|
lokinet-context
|
||||||
|
lokinet-peerstats
|
||||||
|
lokinet-util
|
||||||
|
)
|
||||||
|
|
||||||
|
# inter lokinet-layer public/private linkage.
|
||||||
|
# when linking each layer, we consider the layer directly below private linkage and the layer above public linkage.
|
||||||
|
# this lets us hide functionality of layers below us when depended on by another component.
|
||||||
|
#
|
||||||
|
# from highest to lowest layer, the above layers are stacked as follows:
|
||||||
|
#
|
||||||
|
# platform (what lokinet snapps interact with, be it l3 os interaction or embedded lokinet)
|
||||||
|
# flow (how we want to route and stripe over our onion routing)
|
||||||
|
# routing (what we are onion routing)
|
||||||
|
# onion (how the onion routing happens)
|
||||||
|
# link (what we want to send over the wire and to where)
|
||||||
|
# wire (what is actually sent over the wire)
|
||||||
|
#
|
||||||
|
function(link_lokinet_layers)
|
||||||
|
set(lib ${ARGV0})
|
||||||
|
if(${ARGC} GREATER 1)
|
||||||
|
lokinet_link_lib(${ARGV1} ${lib})
|
||||||
|
list(REMOVE_AT ARGV 1)
|
||||||
|
target_link_libraries(${lib} PRIVATE ${ARGV1})
|
||||||
|
# recursion :D
|
||||||
|
link_lokinet_layers(${ARGV})
|
||||||
|
else()
|
||||||
|
lokinet_link_lib(${lib})
|
||||||
|
endif()
|
||||||
|
endfunction()
|
||||||
|
|
||||||
|
link_lokinet_layers(
|
||||||
|
lokinet-layer-onion
|
||||||
|
lokinet-layer-link
|
||||||
|
)
|
||||||
|
|
||||||
|
# set me to OFF to disable old codepath
|
||||||
|
set(use_old_impl ON)
|
||||||
|
if(use_old_impl)
|
||||||
|
# flow layer deprecated-kitchensink (remove me after refactor)
|
||||||
|
lokinet_link_lib(lokinet-service-deprecated-kitchensink
|
||||||
|
lokinet-dns
|
||||||
|
lokinet-nodedb
|
||||||
|
lokinet-context
|
||||||
|
lokinet-layer-onion
|
||||||
|
lokinet-platform
|
||||||
|
lokinet-rpc
|
||||||
|
)
|
||||||
|
target_link_libraries(lokinet-layers INTERFACE lokinet-service-deprecated-kitchensink)
|
||||||
|
endif()
|
||||||
|
|
||||||
|
target_link_libraries(lokinet-layers INTERFACE
|
||||||
|
lokinet-layer-onion
|
||||||
|
lokinet-layer-link
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# per component external deps
|
||||||
|
|
||||||
|
target_link_libraries(lokinet-config PUBLIC oxenmq::oxenmq)
|
||||||
|
target_link_libraries(lokinet-platform PUBLIC oxenmq::oxenmq)
|
||||||
|
target_link_libraries(lokinet-dns PUBLIC libunbound)
|
||||||
|
|
||||||
|
target_link_libraries(lokinet-cryptography PUBLIC
|
||||||
|
oxenc::oxenc
|
||||||
|
sodium
|
||||||
|
)
|
||||||
|
|
||||||
|
target_link_libraries(lokinet-context PUBLIC
|
||||||
|
CLI11
|
||||||
|
oxenmq::oxenmq
|
||||||
|
uvw
|
||||||
|
)
|
||||||
|
|
||||||
|
target_link_libraries(lokinet-platform PUBLIC
|
||||||
|
Threads::Threads
|
||||||
|
base_libs
|
||||||
|
uvw
|
||||||
|
)
|
||||||
|
|
||||||
|
target_link_libraries(lokinet-util PUBLIC
|
||||||
|
nlohmann_json::nlohmann_json
|
||||||
|
# filesystem
|
||||||
|
oxenc::oxenc
|
||||||
|
)
|
||||||
|
|
||||||
|
target_link_libraries(lokinet-layer-link PUBLIC
|
||||||
|
quic
|
||||||
|
uvw
|
||||||
|
)
|
||||||
|
|
||||||
|
if(WITH_EMBEDDED_LOKINET)
|
||||||
|
include(GNUInstallDirs)
|
||||||
|
add_library(lokinet-shared SHARED lokinet_shared.cpp)
|
||||||
|
target_link_libraries(lokinet-shared PUBLIC lokinet-amalgum)
|
||||||
|
if(WIN32)
|
||||||
|
set(CMAKE_SHARED_LIBRARY_PREFIX_CXX "")
|
||||||
|
endif()
|
||||||
|
set_target_properties(lokinet-shared PROPERTIES OUTPUT_NAME lokinet)
|
||||||
|
if(WIN32)
|
||||||
|
target_link_libraries(lokinet-shared PUBLIC ws2_32 iphlpapi -fstack-protector)
|
||||||
|
install(TARGETS lokinet-shared DESTINATION bin COMPONENT liblokinet)
|
||||||
|
elseif(NOT APPLE)
|
||||||
|
install(TARGETS lokinet-shared LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR} COMPONENT liblokinet)
|
||||||
|
endif()
|
||||||
|
endif()
|
||||||
|
|
||||||
|
file(GLOB_RECURSE docs_SRC */*.hpp *.hpp)
|
||||||
|
set(DOCS_SRC ${docs_SRC} PARENT_SCOPE)
|
@ -0,0 +1,25 @@
|
|||||||
|
#include <initializer_list>
|
||||||
|
#include "llarp/bootstrap.hpp"
|
||||||
|
|
||||||
|
namespace llarp
|
||||||
|
{
|
||||||
|
using namespace std::literals;
|
||||||
|
|
||||||
|
std::unordered_map<std::string, BootstrapList>
|
||||||
|
load_bootstrap_fallbacks()
|
||||||
|
{
|
||||||
|
std::unordered_map<std::string, BootstrapList> fallbacks;
|
||||||
|
using init_list = std::initializer_list<std::pair<std::string, std::string_view>>;
|
||||||
|
// clang-format off
|
||||||
|
for (const auto& [network, bootstrap] : init_list{
|
||||||
|
//
|
||||||
|
})
|
||||||
|
// clang-format on
|
||||||
|
{
|
||||||
|
llarp_buffer_t buf{bootstrap.data(), bootstrap.size()};
|
||||||
|
auto& bsl = fallbacks[network];
|
||||||
|
bsl.BDecode(&buf);
|
||||||
|
}
|
||||||
|
return fallbacks;
|
||||||
|
}
|
||||||
|
} // namespace llarp
|
@ -1 +0,0 @@
|
|||||||
#pragma once
|
|
@ -1,162 +0,0 @@
|
|||||||
#include <llarp/crypto/crypto.hpp>
|
|
||||||
#include <llarp/router_contact.hpp>
|
|
||||||
#include <llarp/router/router.hpp>
|
|
||||||
#include <llarp/util/bencode.h>
|
|
||||||
#include <llarp/util/logging.hpp>
|
|
||||||
|
|
||||||
#include <oxenc/bt_producer.h>
|
|
||||||
|
|
||||||
namespace llarp
|
|
||||||
{
|
|
||||||
bool
|
|
||||||
LinkIntroMessage::decode_key(const llarp_buffer_t& key, llarp_buffer_t* buf)
|
|
||||||
{
|
|
||||||
if (key.startswith("a"))
|
|
||||||
{
|
|
||||||
llarp_buffer_t strbuf;
|
|
||||||
if (!bencode_read_string(buf, &strbuf))
|
|
||||||
return false;
|
|
||||||
if (strbuf.sz != 1)
|
|
||||||
return false;
|
|
||||||
return *strbuf.cur == 'i';
|
|
||||||
}
|
|
||||||
if (key.startswith("n"))
|
|
||||||
{
|
|
||||||
if (nonce.BDecode(buf))
|
|
||||||
return true;
|
|
||||||
llarp::LogWarn("failed to decode nonce in LIM");
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
if (key.startswith("p"))
|
|
||||||
{
|
|
||||||
return bencode_read_integer(buf, &session_period);
|
|
||||||
}
|
|
||||||
if (key.startswith("r"))
|
|
||||||
{
|
|
||||||
if (rc.BDecode(buf))
|
|
||||||
return true;
|
|
||||||
llarp::LogWarn("failed to decode RC in LIM");
|
|
||||||
llarp::DumpBuffer(*buf);
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
if (key.startswith("v"))
|
|
||||||
{
|
|
||||||
if (!bencode_read_integer(buf, &version))
|
|
||||||
return false;
|
|
||||||
if (version != llarp::constants::proto_version)
|
|
||||||
{
|
|
||||||
llarp::LogWarn(
|
|
||||||
"llarp protocol version mismatch ", version, " != ", llarp::constants::proto_version);
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
llarp::LogDebug("LIM version ", version);
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
if (key.startswith("z"))
|
|
||||||
{
|
|
||||||
return sig.BDecode(buf);
|
|
||||||
}
|
|
||||||
|
|
||||||
llarp::LogWarn("invalid LIM key: ", *key.cur);
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
std::string
|
|
||||||
LinkIntroMessage::bt_encode() const
|
|
||||||
{
|
|
||||||
oxenc::bt_dict_producer btdp;
|
|
||||||
|
|
||||||
try
|
|
||||||
{
|
|
||||||
btdp.append("a", "i");
|
|
||||||
btdp.append("n", nonce.ToView());
|
|
||||||
btdp.append("p", session_period);
|
|
||||||
|
|
||||||
{
|
|
||||||
auto subdict = btdp.append_list("r");
|
|
||||||
rc.bt_encode_subdict(subdict);
|
|
||||||
}
|
|
||||||
|
|
||||||
btdp.append("v", llarp::constants::proto_version);
|
|
||||||
btdp.append("z", sig.ToView());
|
|
||||||
}
|
|
||||||
catch (...)
|
|
||||||
{
|
|
||||||
log::critical(link_cat, "Error: LinkIntroMessage failed to bt encode contents!");
|
|
||||||
}
|
|
||||||
|
|
||||||
return std::move(btdp).str();
|
|
||||||
}
|
|
||||||
|
|
||||||
bool
|
|
||||||
LinkIntroMessage::handle_message(Router* /*router*/) const
|
|
||||||
{
|
|
||||||
if (!verify())
|
|
||||||
return false;
|
|
||||||
return true;
|
|
||||||
// return conn->GotLIM(this);
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
LinkIntroMessage::clear()
|
|
||||||
{
|
|
||||||
session_period = 0;
|
|
||||||
nonce.Zero();
|
|
||||||
rc.Clear();
|
|
||||||
sig.Zero();
|
|
||||||
version = 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
bool
|
|
||||||
LinkIntroMessage::sign(std::function<bool(Signature&, const llarp_buffer_t&)> signer)
|
|
||||||
{
|
|
||||||
sig.Zero();
|
|
||||||
// need to keep this as a llarp_buffer_t for now, as all the crypto code expects
|
|
||||||
// byte_t types -- fix this later
|
|
||||||
std::array<byte_t, MAX_MSG_SIZE> tmp;
|
|
||||||
llarp_buffer_t buf(tmp);
|
|
||||||
|
|
||||||
auto bte = bt_encode();
|
|
||||||
buf.write(bte.begin(), bte.end());
|
|
||||||
|
|
||||||
buf.sz = buf.cur - buf.base;
|
|
||||||
buf.cur = buf.base;
|
|
||||||
|
|
||||||
return signer(sig, buf);
|
|
||||||
}
|
|
||||||
|
|
||||||
bool
|
|
||||||
LinkIntroMessage::verify() const
|
|
||||||
{
|
|
||||||
LinkIntroMessage copy;
|
|
||||||
copy = *this;
|
|
||||||
copy.sig.Zero();
|
|
||||||
|
|
||||||
// need to keep this as a llarp_buffer_t for now, as all the crypto code expects
|
|
||||||
// byte_t types -- fix this later
|
|
||||||
std::array<byte_t, MAX_MSG_SIZE> tmp;
|
|
||||||
llarp_buffer_t buf(tmp);
|
|
||||||
|
|
||||||
auto bte = copy.bt_encode();
|
|
||||||
buf.write(bte.begin(), bte.end());
|
|
||||||
|
|
||||||
buf.sz = buf.cur - buf.base;
|
|
||||||
buf.cur = buf.base;
|
|
||||||
|
|
||||||
// outer signature
|
|
||||||
if (!CryptoManager::instance()->verify(
|
|
||||||
rc.pubkey, reinterpret_cast<uint8_t*>(bte.data()), bte.size(), sig))
|
|
||||||
{
|
|
||||||
log::error(link_cat, "Error: outer signature failed!");
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
// verify RC
|
|
||||||
if (!rc.Verify(llarp::time_now_ms()))
|
|
||||||
{
|
|
||||||
log::error(link_cat, "Error: invalid RC in link intro!");
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
|
|
||||||
} // namespace llarp
|
|
@ -1,133 +0,0 @@
|
|||||||
#pragma once
|
|
||||||
|
|
||||||
#include <sqlite_orm/sqlite_orm.h>
|
|
||||||
|
|
||||||
#include "types.hpp"
|
|
||||||
|
|
||||||
/// Contains some code to help deal with sqlite_orm in hopes of keeping other headers clean
|
|
||||||
|
|
||||||
namespace llarp
|
|
||||||
{
|
|
||||||
inline auto
|
|
||||||
initStorage(const std::string& file)
|
|
||||||
{
|
|
||||||
using namespace sqlite_orm;
|
|
||||||
return make_storage(
|
|
||||||
file,
|
|
||||||
make_table(
|
|
||||||
"peerstats",
|
|
||||||
make_column("routerId", &PeerStats::routerId, primary_key(), unique()),
|
|
||||||
make_column("numConnectionAttempts", &PeerStats::numConnectionAttempts),
|
|
||||||
make_column("numConnectionSuccesses", &PeerStats::numConnectionSuccesses),
|
|
||||||
make_column("numConnectionRejections", &PeerStats::numConnectionRejections),
|
|
||||||
make_column("numConnectionTimeouts", &PeerStats::numConnectionTimeouts),
|
|
||||||
make_column("numPathBuilds", &PeerStats::numPathBuilds),
|
|
||||||
make_column("numPacketsAttempted", &PeerStats::numPacketsAttempted),
|
|
||||||
make_column("numPacketsSent", &PeerStats::numPacketsSent),
|
|
||||||
make_column("numPacketsDropped", &PeerStats::numPacketsDropped),
|
|
||||||
make_column("numPacketsResent", &PeerStats::numPacketsResent),
|
|
||||||
make_column("numDistinctRCsReceived", &PeerStats::numDistinctRCsReceived),
|
|
||||||
make_column("numLateRCs", &PeerStats::numLateRCs),
|
|
||||||
make_column("peakBandwidthBytesPerSec", &PeerStats::peakBandwidthBytesPerSec),
|
|
||||||
make_column("longestRCReceiveInterval", &PeerStats::longestRCReceiveInterval),
|
|
||||||
make_column("leastRCRemainingLifetime", &PeerStats::leastRCRemainingLifetime)));
|
|
||||||
}
|
|
||||||
|
|
||||||
using PeerDbStorage = decltype(initStorage(""));
|
|
||||||
|
|
||||||
} // namespace llarp
|
|
||||||
|
|
||||||
/// "custom" types for sqlite_orm
|
|
||||||
/// reference: https://github.com/fnc12/sqlite_orm/blob/master/examples/enum_binding.cpp
|
|
||||||
namespace sqlite_orm
|
|
||||||
{
|
|
||||||
/// llarp_time_t serialization
|
|
||||||
template <>
|
|
||||||
struct type_printer<llarp_time_t> : public integer_printer
|
|
||||||
{};
|
|
||||||
|
|
||||||
template <>
|
|
||||||
struct statement_binder<llarp_time_t>
|
|
||||||
{
|
|
||||||
int
|
|
||||||
bind(sqlite3_stmt* stmt, int index, const llarp_time_t& value)
|
|
||||||
{
|
|
||||||
return statement_binder<int64_t>().bind(stmt, index, value.count());
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
template <>
|
|
||||||
struct field_printer<llarp_time_t>
|
|
||||||
{
|
|
||||||
std::string
|
|
||||||
operator()(const llarp_time_t& value) const
|
|
||||||
{
|
|
||||||
return fmt::format("{}", value.count());
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
template <>
|
|
||||||
struct row_extractor<llarp_time_t>
|
|
||||||
{
|
|
||||||
llarp_time_t
|
|
||||||
extract(const char* row_value)
|
|
||||||
{
|
|
||||||
int64_t raw = static_cast<int64_t>(atoi(row_value));
|
|
||||||
return llarp_time_t(raw);
|
|
||||||
}
|
|
||||||
|
|
||||||
llarp_time_t
|
|
||||||
extract(sqlite3_stmt* stmt, int columnIndex)
|
|
||||||
{
|
|
||||||
auto str = sqlite3_column_text(stmt, columnIndex);
|
|
||||||
return this->extract((const char*)str);
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
/// RouterID serialization
|
|
||||||
template <>
|
|
||||||
struct type_printer<llarp::RouterID> : public text_printer
|
|
||||||
{};
|
|
||||||
|
|
||||||
template <>
|
|
||||||
struct statement_binder<llarp::RouterID>
|
|
||||||
{
|
|
||||||
int
|
|
||||||
bind(sqlite3_stmt* stmt, int index, const llarp::RouterID& value)
|
|
||||||
{
|
|
||||||
return statement_binder<std::string>().bind(stmt, index, value.ToString());
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
template <>
|
|
||||||
struct field_printer<llarp::RouterID>
|
|
||||||
{
|
|
||||||
std::string
|
|
||||||
operator()(const llarp::RouterID& value) const
|
|
||||||
{
|
|
||||||
return value.ToString();
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
template <>
|
|
||||||
struct row_extractor<llarp::RouterID>
|
|
||||||
{
|
|
||||||
llarp::RouterID
|
|
||||||
extract(const char* row_value)
|
|
||||||
{
|
|
||||||
llarp::RouterID id;
|
|
||||||
if (not id.FromString(row_value))
|
|
||||||
throw std::runtime_error("Invalid RouterID in sqlite3 database");
|
|
||||||
|
|
||||||
return id;
|
|
||||||
}
|
|
||||||
|
|
||||||
llarp::RouterID
|
|
||||||
extract(sqlite3_stmt* stmt, int columnIndex)
|
|
||||||
{
|
|
||||||
auto str = sqlite3_column_text(stmt, columnIndex);
|
|
||||||
return this->extract((const char*)str);
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
} // namespace sqlite_orm
|
|
@ -1,380 +0,0 @@
|
|||||||
#include "peer_db.hpp"
|
|
||||||
|
|
||||||
#include <llarp/util/logging.hpp>
|
|
||||||
#include <llarp/util/status.hpp>
|
|
||||||
#include <llarp/util/str.hpp>
|
|
||||||
|
|
||||||
namespace llarp
|
|
||||||
{
|
|
||||||
#ifdef LOKINET_PEERSTATS_BACKEND
|
|
||||||
|
|
||||||
PeerDb::PeerDb()
|
|
||||||
{
|
|
||||||
m_lastFlush.store({});
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerDb::loadDatabase(std::optional<fs::path> file)
|
|
||||||
{
|
|
||||||
std::lock_guard guard(m_statsLock);
|
|
||||||
|
|
||||||
if (m_storage)
|
|
||||||
throw std::runtime_error("Reloading database not supported"); // TODO
|
|
||||||
|
|
||||||
m_peerStats.clear();
|
|
||||||
|
|
||||||
// sqlite_orm treats empty-string as an indicator to load a memory-backed database, which we'll
|
|
||||||
// use if file is an empty-optional
|
|
||||||
std::string fileString;
|
|
||||||
if (file.has_value())
|
|
||||||
{
|
|
||||||
fileString = file->string();
|
|
||||||
LogInfo("Loading PeerDb from file ", fileString);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
LogInfo("Loading memory-backed PeerDb");
|
|
||||||
}
|
|
||||||
|
|
||||||
m_storage = std::make_unique<PeerDbStorage>(initStorage(fileString));
|
|
||||||
m_storage->sync_schema(true); // true for "preserve" as in "don't nuke" (how cute!)
|
|
||||||
|
|
||||||
auto allStats = m_storage->get_all<PeerStats>();
|
|
||||||
LogInfo("Loading ", allStats.size(), " PeerStats from table peerstats...");
|
|
||||||
for (PeerStats& stats : allStats)
|
|
||||||
{
|
|
||||||
// we cleared m_peerStats, and the database should enforce that routerId is unique...
|
|
||||||
assert(m_peerStats.find(stats.routerId) == m_peerStats.end());
|
|
||||||
|
|
||||||
stats.stale = false;
|
|
||||||
m_peerStats[stats.routerId] = stats;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerDb::flushDatabase()
|
|
||||||
{
|
|
||||||
LogDebug("flushing PeerDb...");
|
|
||||||
|
|
||||||
auto start = time_now_ms();
|
|
||||||
if (not shouldFlush(start))
|
|
||||||
{
|
|
||||||
LogWarn("Call to flushDatabase() while already in progress, ignoring");
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (not m_storage)
|
|
||||||
throw std::runtime_error("Cannot flush database before it has been loaded");
|
|
||||||
|
|
||||||
std::vector<PeerStats> staleStats;
|
|
||||||
|
|
||||||
{
|
|
||||||
std::lock_guard guard(m_statsLock);
|
|
||||||
|
|
||||||
// copy all stale entries
|
|
||||||
for (auto& entry : m_peerStats)
|
|
||||||
{
|
|
||||||
if (entry.second.stale)
|
|
||||||
{
|
|
||||||
staleStats.push_back(entry.second);
|
|
||||||
entry.second.stale = false;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
LogDebug("Updating ", staleStats.size(), " stats");
|
|
||||||
|
|
||||||
{
|
|
||||||
auto guard = m_storage->transaction_guard();
|
|
||||||
|
|
||||||
for (const auto& stats : staleStats)
|
|
||||||
{
|
|
||||||
m_storage->replace(stats);
|
|
||||||
}
|
|
||||||
|
|
||||||
guard.commit();
|
|
||||||
}
|
|
||||||
|
|
||||||
auto end = time_now_ms();
|
|
||||||
|
|
||||||
auto elapsed = end - start;
|
|
||||||
LogDebug("PeerDb flush took about ", elapsed, " seconds");
|
|
||||||
|
|
||||||
m_lastFlush.store(end);
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerDb::accumulatePeerStats(const RouterID& routerId, const PeerStats& delta)
|
|
||||||
{
|
|
||||||
if (routerId != delta.routerId)
|
|
||||||
throw std::invalid_argument{
|
|
||||||
fmt::format("routerId {} doesn't match {}", routerId, delta.routerId)};
|
|
||||||
|
|
||||||
std::lock_guard guard(m_statsLock);
|
|
||||||
auto itr = m_peerStats.find(routerId);
|
|
||||||
if (itr == m_peerStats.end())
|
|
||||||
itr = m_peerStats.insert({routerId, delta}).first;
|
|
||||||
else
|
|
||||||
itr->second += delta;
|
|
||||||
|
|
||||||
itr->second.stale = true;
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerDb::modifyPeerStats(const RouterID& routerId, std::function<void(PeerStats&)> callback)
|
|
||||||
{
|
|
||||||
std::lock_guard guard(m_statsLock);
|
|
||||||
|
|
||||||
PeerStats& stats = m_peerStats[routerId];
|
|
||||||
stats.routerId = routerId;
|
|
||||||
stats.stale = true;
|
|
||||||
callback(stats);
|
|
||||||
}
|
|
||||||
|
|
||||||
std::optional<PeerStats>
|
|
||||||
PeerDb::getCurrentPeerStats(const RouterID& routerId) const
|
|
||||||
{
|
|
||||||
std::lock_guard guard(m_statsLock);
|
|
||||||
auto itr = m_peerStats.find(routerId);
|
|
||||||
if (itr == m_peerStats.end())
|
|
||||||
return std::nullopt;
|
|
||||||
else
|
|
||||||
return itr->second;
|
|
||||||
}
|
|
||||||
|
|
||||||
std::vector<PeerStats>
|
|
||||||
PeerDb::listAllPeerStats() const
|
|
||||||
{
|
|
||||||
std::lock_guard guard(m_statsLock);
|
|
||||||
|
|
||||||
std::vector<PeerStats> statsList;
|
|
||||||
statsList.reserve(m_peerStats.size());
|
|
||||||
|
|
||||||
for (const auto& [routerId, stats] : m_peerStats)
|
|
||||||
{
|
|
||||||
statsList.push_back(stats);
|
|
||||||
}
|
|
||||||
|
|
||||||
return statsList;
|
|
||||||
}
|
|
||||||
|
|
||||||
std::vector<PeerStats>
|
|
||||||
PeerDb::listPeerStats(const std::vector<RouterID>& ids) const
|
|
||||||
{
|
|
||||||
std::lock_guard guard(m_statsLock);
|
|
||||||
|
|
||||||
std::vector<PeerStats> statsList;
|
|
||||||
statsList.reserve(ids.size());
|
|
||||||
|
|
||||||
for (const auto& id : ids)
|
|
||||||
{
|
|
||||||
const auto itr = m_peerStats.find(id);
|
|
||||||
if (itr != m_peerStats.end())
|
|
||||||
statsList.push_back(itr->second);
|
|
||||||
}
|
|
||||||
|
|
||||||
return statsList;
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Assume we receive an RC at some point `R` in time which was signed at some point `S` in time
|
|
||||||
/// and expires at some point `E` in time, as depicted below:
|
|
||||||
///
|
|
||||||
/// +-----------------------------+
|
|
||||||
/// | signed rc | <- useful lifetime of RC
|
|
||||||
/// +-----------------------------+
|
|
||||||
/// ^ [ . . . . . . . . ] <----------- window in which we receive this RC gossiped to us
|
|
||||||
/// | ^ ^
|
|
||||||
/// | | |
|
|
||||||
/// S R E
|
|
||||||
///
|
|
||||||
/// One useful metric from this is the difference between (E - R), the useful contact time of this
|
|
||||||
/// RC. As we track this metric over time, the high and low watermarks serve to tell us how
|
|
||||||
/// quickly we receive signed RCs from a given router and how close to expiration they are when
|
|
||||||
/// we receive them. The latter is particularly useful, and should always be a positive number for
|
|
||||||
/// a healthy router. A negative number indicates that we are receiving an expired RC.
|
|
||||||
///
|
|
||||||
/// TODO: we actually discard expired RCs, so we currently would not detect a negative value for
|
|
||||||
/// (E - R)
|
|
||||||
///
|
|
||||||
/// Another related metric is the distance between a newly received RC and the previous RC's
|
|
||||||
/// expiration, which represents how close we came to having no useful RC to work with. This
|
|
||||||
/// should be a high (positive) number for a healthy router, and if negative indicates that we
|
|
||||||
/// had no way to contact this router for a period of time.
|
|
||||||
///
|
|
||||||
/// E1 E2 E3
|
|
||||||
/// | | |
|
|
||||||
/// v | |
|
|
||||||
/// +-----------------------------+ | |
|
|
||||||
/// | signed rc 1 | | |
|
|
||||||
/// +-----------------------------+ | |
|
|
||||||
/// [ . . . . . ] v |
|
|
||||||
/// ^ +-----------------------------+ |
|
|
||||||
/// | | signed rc 2 | |
|
|
||||||
/// | +-----------------------------+ |
|
|
||||||
/// | [ . . . . . . . . . . ] v
|
|
||||||
/// | ^ +-----------------------------+
|
|
||||||
/// | | | signed rc 3 |
|
|
||||||
/// | | +-----------------------------+
|
|
||||||
/// | | [ . . ]
|
|
||||||
/// | | ^
|
|
||||||
/// | | |
|
|
||||||
/// R1 R2 R3
|
|
||||||
///
|
|
||||||
/// Example: the delta between (E1 - R2) is healthy, but the delta between (E2 - R3) is indicates
|
|
||||||
/// that we had a brief period of time where we had no valid (non-expired) RC for this router
|
|
||||||
/// (because it is negative).
|
|
||||||
void
|
|
||||||
PeerDb::handleGossipedRC(const RouterContact& rc, llarp_time_t now)
|
|
||||||
{
|
|
||||||
std::lock_guard guard(m_statsLock);
|
|
||||||
|
|
||||||
RouterID id(rc.pubkey);
|
|
||||||
auto& stats = m_peerStats[id];
|
|
||||||
stats.routerId = id;
|
|
||||||
|
|
||||||
const bool isNewRC = (stats.lastRCUpdated < rc.last_updated);
|
|
||||||
|
|
||||||
if (isNewRC)
|
|
||||||
{
|
|
||||||
stats.numDistinctRCsReceived++;
|
|
||||||
|
|
||||||
if (stats.numDistinctRCsReceived > 1)
|
|
||||||
{
|
|
||||||
auto prevRCExpiration = (stats.lastRCUpdated + RouterContact::Lifetime);
|
|
||||||
|
|
||||||
// we track max expiry as the delta between (last expiration time - time received),
|
|
||||||
// and this value will be negative for an unhealthy router
|
|
||||||
// TODO: handle case where new RC is also expired? just ignore?
|
|
||||||
auto expiry = prevRCExpiration - now;
|
|
||||||
|
|
||||||
if (stats.numDistinctRCsReceived == 2)
|
|
||||||
stats.leastRCRemainingLifetime = expiry;
|
|
||||||
else
|
|
||||||
stats.leastRCRemainingLifetime = std::min(stats.leastRCRemainingLifetime, expiry);
|
|
||||||
}
|
|
||||||
|
|
||||||
stats.lastRCUpdated = rc.last_updated;
|
|
||||||
stats.stale = true;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerDb::configure(const RouterConfig& routerConfig)
|
|
||||||
{
|
|
||||||
fs::path dbPath = routerConfig.m_dataDir / "peerstats.sqlite";
|
|
||||||
|
|
||||||
loadDatabase(dbPath);
|
|
||||||
}
|
|
||||||
|
|
||||||
bool
|
|
||||||
PeerDb::shouldFlush(llarp_time_t now)
|
|
||||||
{
|
|
||||||
constexpr llarp_time_t TargetFlushInterval = 30s;
|
|
||||||
return (now - m_lastFlush.load() >= TargetFlushInterval);
|
|
||||||
}
|
|
||||||
|
|
||||||
util::StatusObject
|
|
||||||
PeerDb::ExtractStatus() const
|
|
||||||
{
|
|
||||||
std::lock_guard guard(m_statsLock);
|
|
||||||
|
|
||||||
bool loaded = (m_storage.get() != nullptr);
|
|
||||||
util::StatusObject dbFile = nullptr;
|
|
||||||
if (loaded)
|
|
||||||
dbFile = m_storage->filename();
|
|
||||||
|
|
||||||
std::vector<util::StatusObject> statsObjs;
|
|
||||||
statsObjs.reserve(m_peerStats.size());
|
|
||||||
for (const auto& pair : m_peerStats)
|
|
||||||
{
|
|
||||||
statsObjs.push_back(pair.second.toJson());
|
|
||||||
}
|
|
||||||
|
|
||||||
util::StatusObject obj{
|
|
||||||
{"dbLoaded", loaded},
|
|
||||||
{"dbFile", dbFile},
|
|
||||||
{"lastFlushMs", m_lastFlush.load().count()},
|
|
||||||
{"stats", statsObjs},
|
|
||||||
};
|
|
||||||
return obj;
|
|
||||||
}
|
|
||||||
|
|
||||||
#else // !LOKINET_PEERSTATS
|
|
||||||
|
|
||||||
// Empty stubs
|
|
||||||
|
|
||||||
PeerDb::PeerDb()
|
|
||||||
{
|
|
||||||
throw std::logic_error{"Peer stats backend not enabled!"};
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerDb::loadDatabase(std::optional<fs::path>)
|
|
||||||
{}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerDb::flushDatabase()
|
|
||||||
{}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerDb::accumulatePeerStats(const RouterID&, const PeerStats&)
|
|
||||||
{}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerDb::modifyPeerStats(const RouterID&, std::function<void(PeerStats&)>)
|
|
||||||
{}
|
|
||||||
|
|
||||||
std::optional<PeerStats>
|
|
||||||
PeerDb::getCurrentPeerStats(const RouterID&) const
|
|
||||||
{
|
|
||||||
return std::nullopt;
|
|
||||||
}
|
|
||||||
|
|
||||||
std::vector<PeerStats>
|
|
||||||
PeerDb::listAllPeerStats() const
|
|
||||||
{
|
|
||||||
return {};
|
|
||||||
}
|
|
||||||
|
|
||||||
std::vector<PeerStats>
|
|
||||||
PeerDb::listPeerStats(const std::vector<RouterID>&) const
|
|
||||||
{
|
|
||||||
return {};
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerDb::handleGossipedRC(const RouterContact&, llarp_time_t)
|
|
||||||
{}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerDb::configure(const RouterConfig&)
|
|
||||||
{}
|
|
||||||
|
|
||||||
bool
|
|
||||||
PeerDb::shouldFlush(llarp_time_t)
|
|
||||||
{
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
util::StatusObject
|
|
||||||
PeerDb::ExtractStatus() const
|
|
||||||
{
|
|
||||||
return {};
|
|
||||||
}
|
|
||||||
|
|
||||||
#endif
|
|
||||||
|
|
||||||
}; // namespace llarp
|
|
||||||
|
|
||||||
/*
|
|
||||||
- How does a SN/quorum of SN determine if an operator is acting in good faith on the network
|
|
||||||
- w/o the other operator knowing
|
|
||||||
- Observing and gathering data in a meaningful way
|
|
||||||
|
|
||||||
- What type of stats could you even collect off lokinet?
|
|
||||||
- Collecting network-wide data can't be automatically actioned unless the SN is collecting
|
|
||||||
|
|
||||||
- Redundancy model
|
|
||||||
- Don't rely on an individual, but a collection of them
|
|
||||||
|
|
||||||
*/
|
|
@ -1,141 +0,0 @@
|
|||||||
#pragma once
|
|
||||||
|
|
||||||
#include <filesystem>
|
|
||||||
#include <functional>
|
|
||||||
#include <unordered_map>
|
|
||||||
|
|
||||||
#include <llarp/util/fs.hpp>
|
|
||||||
#include <llarp/config/config.hpp>
|
|
||||||
#include <llarp/router_id.hpp>
|
|
||||||
#include <llarp/util/time.hpp>
|
|
||||||
#include <llarp/util/status.hpp>
|
|
||||||
#include "types.hpp"
|
|
||||||
#ifdef LOKINET_PEERSTATS_BACKEND
|
|
||||||
#include "orm.hpp"
|
|
||||||
#endif
|
|
||||||
|
|
||||||
namespace llarp
|
|
||||||
{
|
|
||||||
/// Maintains a database of stats collected about the connections with our Service Node peers.
|
|
||||||
/// This uses a sqlite3 database behind the scenes as persistance, but this database is
|
|
||||||
/// periodically flushed to, meaning that it will become stale as PeerDb accumulates stats without
|
|
||||||
/// a flush.
|
|
||||||
struct PeerDb
|
|
||||||
{
|
|
||||||
/// Constructor
|
|
||||||
PeerDb();
|
|
||||||
|
|
||||||
/// Loads the database from disk using the provided filepath. If the file is equal to
|
|
||||||
/// `std::nullopt`, the database will be loaded into memory (useful for testing).
|
|
||||||
///
|
|
||||||
/// This must be called prior to calling flushDatabase(), and will truncate any existing data.
|
|
||||||
///
|
|
||||||
/// This is a blocking call, both in the sense that it blocks on disk/database I/O and that it
|
|
||||||
/// will sit on a mutex while the database is loaded.
|
|
||||||
///
|
|
||||||
/// @param file is an optional file which doesn't have to exist but must be writable, if a value
|
|
||||||
/// is provided. If no value is provided, the database will be memory-backed.
|
|
||||||
/// @throws if sqlite_orm/sqlite3 is unable to open or create a database at the given file
|
|
||||||
void
|
|
||||||
loadDatabase(std::optional<fs::path> file);
|
|
||||||
|
|
||||||
/// Flushes the database. Must be called after loadDatabase(). This call will block during I/O
|
|
||||||
/// and should be called in an appropriate threading context. However, it will make a temporary
|
|
||||||
/// copy of the peer stats so as to avoid sitting on a mutex lock during disk I/O.
|
|
||||||
///
|
|
||||||
/// @throws if the database could not be written to (esp. if loadDatabase() has not been called)
|
|
||||||
void
|
|
||||||
flushDatabase();
|
|
||||||
|
|
||||||
/// Add the given stats to the cummulative stats for the given peer. For cummulative stats, the
|
|
||||||
/// stats are added together; for watermark stats, the max is kept.
|
|
||||||
///
|
|
||||||
/// This is intended to be used in the following pattern:
|
|
||||||
///
|
|
||||||
/// 1) Initialize an empty PeerStats
|
|
||||||
/// 2) Collect relevant stats
|
|
||||||
/// 3) Call accumulatePeerStats() with the stats
|
|
||||||
/// 4) Reset the stats to 0
|
|
||||||
/// 5) <Repeat 2-4 periodically>
|
|
||||||
///
|
|
||||||
/// @param routerId is the id of the router whose stats should be modified.
|
|
||||||
/// @param delta is the stats to add to the existing stats
|
|
||||||
void
|
|
||||||
accumulatePeerStats(const RouterID& routerId, const PeerStats& delta);
|
|
||||||
|
|
||||||
/// Allows write-access to the stats for a given peer while appropriate mutex lock is held. This
|
|
||||||
/// is an alternative means of incrementing peer stats that is suitable for one-off
|
|
||||||
/// modifications.
|
|
||||||
///
|
|
||||||
/// Note that this holds m_statsLock during the callback invocation, so the callback should
|
|
||||||
/// return as quickly as possible.
|
|
||||||
///
|
|
||||||
/// @param routerId is the id of the router whose stats should be modified.
|
|
||||||
/// @param callback is a function which will be called immediately with mutex held
|
|
||||||
void
|
|
||||||
modifyPeerStats(const RouterID& routerId, std::function<void(PeerStats&)> callback);
|
|
||||||
|
|
||||||
/// Provides a snapshot of the most recent PeerStats we have for the given peer. If we don't
|
|
||||||
/// have any stats for the peer, std::nullopt
|
|
||||||
///
|
|
||||||
/// @param routerId is the RouterID of the requested peer
|
|
||||||
/// @return a copy of the most recent peer stats or an empty one if no such peer is known
|
|
||||||
std::optional<PeerStats>
|
|
||||||
getCurrentPeerStats(const RouterID& routerId) const;
|
|
||||||
|
|
||||||
/// Lists all peer stats. This essentially dumps the database into a list of PeerStats objects.
|
|
||||||
///
|
|
||||||
/// Note that this avoids disk I/O by copying from our cached map of peers.
|
|
||||||
///
|
|
||||||
/// @return a list of all PeerStats we have maintained
|
|
||||||
std::vector<PeerStats>
|
|
||||||
listAllPeerStats() const;
|
|
||||||
|
|
||||||
/// Lists specific peer stats.
|
|
||||||
///
|
|
||||||
/// @param peers is list of RouterIDs which are desired
|
|
||||||
/// @return a list of the requested peers. Peers not found will be omitted.
|
|
||||||
std::vector<PeerStats>
|
|
||||||
listPeerStats(const std::vector<RouterID>& ids) const;
|
|
||||||
|
|
||||||
/// Handles a new gossiped RC, updating stats as needed. The database tracks the last
|
|
||||||
/// advertised update time, so it knows whether this is a new RC or not.
|
|
||||||
///
|
|
||||||
/// The given RC is assumed to be valid.
|
|
||||||
///
|
|
||||||
/// @param rc is the RouterContact to handle
|
|
||||||
/// @param now is an optional time representing the current time
|
|
||||||
void
|
|
||||||
handleGossipedRC(const RouterContact& rc, llarp_time_t now = time_now_ms());
|
|
||||||
|
|
||||||
/// Configures the PeerDb based on RouterConfig
|
|
||||||
///
|
|
||||||
/// @param routerConfig
|
|
||||||
void
|
|
||||||
configure(const RouterConfig& routerConfig);
|
|
||||||
|
|
||||||
/// Returns whether or not we should flush, as determined by the last time we flushed and the
|
|
||||||
/// configured flush interval.
|
|
||||||
///
|
|
||||||
/// @param now is the current[-ish] time
|
|
||||||
bool
|
|
||||||
shouldFlush(llarp_time_t now);
|
|
||||||
|
|
||||||
/// Get JSON status for API
|
|
||||||
///
|
|
||||||
/// @return JSON object representing our current status
|
|
||||||
util::StatusObject
|
|
||||||
ExtractStatus() const;
|
|
||||||
|
|
||||||
#ifdef LOKINET_PEERSTATS_BACKEND
|
|
||||||
private:
|
|
||||||
std::unordered_map<RouterID, PeerStats> m_peerStats;
|
|
||||||
mutable std::mutex m_statsLock;
|
|
||||||
|
|
||||||
std::unique_ptr<PeerDbStorage> m_storage;
|
|
||||||
|
|
||||||
std::atomic<llarp_time_t> m_lastFlush;
|
|
||||||
#endif
|
|
||||||
};
|
|
||||||
|
|
||||||
} // namespace llarp
|
|
@ -1,146 +0,0 @@
|
|||||||
#include "types.hpp"
|
|
||||||
|
|
||||||
#include <llarp/util/str.hpp>
|
|
||||||
#include <oxenc/bt_serialize.h>
|
|
||||||
#include <stdexcept>
|
|
||||||
|
|
||||||
namespace llarp
|
|
||||||
{
|
|
||||||
constexpr auto RouterIdKey = "routerId";
|
|
||||||
constexpr auto NumConnectionAttemptsKey = "numConnectionAttempts";
|
|
||||||
constexpr auto NumConnectionSuccessesKey = "numConnectionSuccesses";
|
|
||||||
constexpr auto NumConnectionRejectionsKey = "numConnectionRejections";
|
|
||||||
constexpr auto NumConnectionTimeoutsKey = "numConnectionTimeouts";
|
|
||||||
constexpr auto NumPathBuildsKey = "numPathBuilds";
|
|
||||||
constexpr auto NumPacketsAttemptedKey = "numPacketsAttempted";
|
|
||||||
constexpr auto NumPacketsSentKey = "numPacketsSent";
|
|
||||||
constexpr auto NumPacketsDroppedKey = "numPacketsDropped";
|
|
||||||
constexpr auto NumPacketsResentKey = "numPacketsResent";
|
|
||||||
constexpr auto NumDistinctRCsReceivedKey = "numDistinctRCsReceived";
|
|
||||||
constexpr auto NumLateRCsKey = "numLateRCs";
|
|
||||||
constexpr auto PeakBandwidthBytesPerSecKey = "peakBandwidthBytesPerSec";
|
|
||||||
constexpr auto LongestRCReceiveIntervalKey = "longestRCReceiveInterval";
|
|
||||||
constexpr auto LeastRCRemainingLifetimeKey = "leastRCRemainingLifetime";
|
|
||||||
constexpr auto LastRCUpdatedKey = "lastRCUpdated";
|
|
||||||
|
|
||||||
PeerStats::PeerStats() = default;
|
|
||||||
|
|
||||||
PeerStats::PeerStats(const RouterID& routerId_) : routerId(routerId_)
|
|
||||||
{}
|
|
||||||
|
|
||||||
PeerStats&
|
|
||||||
PeerStats::operator+=(const PeerStats& other)
|
|
||||||
{
|
|
||||||
numConnectionAttempts += other.numConnectionAttempts;
|
|
||||||
numConnectionSuccesses += other.numConnectionSuccesses;
|
|
||||||
numConnectionRejections += other.numConnectionRejections;
|
|
||||||
numConnectionTimeouts += other.numConnectionTimeouts;
|
|
||||||
|
|
||||||
numPathBuilds += other.numPathBuilds;
|
|
||||||
numPacketsAttempted += other.numPacketsAttempted;
|
|
||||||
numPacketsSent += other.numPacketsSent;
|
|
||||||
numPacketsDropped += other.numPacketsDropped;
|
|
||||||
numPacketsResent += other.numPacketsResent;
|
|
||||||
|
|
||||||
numDistinctRCsReceived += other.numDistinctRCsReceived;
|
|
||||||
numLateRCs += other.numLateRCs;
|
|
||||||
|
|
||||||
peakBandwidthBytesPerSec = std::max(peakBandwidthBytesPerSec, other.peakBandwidthBytesPerSec);
|
|
||||||
longestRCReceiveInterval = std::max(longestRCReceiveInterval, other.longestRCReceiveInterval);
|
|
||||||
leastRCRemainingLifetime = std::max(leastRCRemainingLifetime, other.leastRCRemainingLifetime);
|
|
||||||
lastRCUpdated = std::max(lastRCUpdated, other.lastRCUpdated);
|
|
||||||
|
|
||||||
return *this;
|
|
||||||
}
|
|
||||||
|
|
||||||
bool
|
|
||||||
PeerStats::operator==(const PeerStats& other) const
|
|
||||||
{
|
|
||||||
return routerId == other.routerId and numConnectionAttempts == other.numConnectionAttempts
|
|
||||||
and numConnectionSuccesses == other.numConnectionSuccesses
|
|
||||||
and numConnectionRejections == other.numConnectionRejections
|
|
||||||
and numConnectionTimeouts == other.numConnectionTimeouts
|
|
||||||
|
|
||||||
and numPathBuilds == other.numPathBuilds
|
|
||||||
and numPacketsAttempted == other.numPacketsAttempted
|
|
||||||
and numPacketsSent == other.numPacketsSent and numPacketsDropped == other.numPacketsDropped
|
|
||||||
and numPacketsResent == other.numPacketsResent
|
|
||||||
|
|
||||||
and numDistinctRCsReceived == other.numDistinctRCsReceived
|
|
||||||
and numLateRCs == other.numLateRCs
|
|
||||||
|
|
||||||
and peakBandwidthBytesPerSec == other.peakBandwidthBytesPerSec
|
|
||||||
and longestRCReceiveInterval == other.longestRCReceiveInterval
|
|
||||||
and leastRCRemainingLifetime == other.leastRCRemainingLifetime
|
|
||||||
and lastRCUpdated == other.lastRCUpdated;
|
|
||||||
}
|
|
||||||
|
|
||||||
util::StatusObject
|
|
||||||
PeerStats::toJson() const
|
|
||||||
{
|
|
||||||
return {
|
|
||||||
{RouterIdKey, routerId.ToString()},
|
|
||||||
{NumConnectionAttemptsKey, numConnectionAttempts},
|
|
||||||
{NumConnectionSuccessesKey, numConnectionSuccesses},
|
|
||||||
{NumConnectionRejectionsKey, numConnectionRejections},
|
|
||||||
{NumConnectionTimeoutsKey, numConnectionTimeouts},
|
|
||||||
{NumPathBuildsKey, numPathBuilds},
|
|
||||||
{NumPacketsAttemptedKey, numPacketsAttempted},
|
|
||||||
{NumPacketsSentKey, numPacketsSent},
|
|
||||||
{NumPacketsDroppedKey, numPacketsDropped},
|
|
||||||
{NumPacketsResentKey, numPacketsResent},
|
|
||||||
{NumDistinctRCsReceivedKey, numDistinctRCsReceived},
|
|
||||||
{NumLateRCsKey, numLateRCs},
|
|
||||||
{PeakBandwidthBytesPerSecKey, peakBandwidthBytesPerSec},
|
|
||||||
{LongestRCReceiveIntervalKey, longestRCReceiveInterval.count()},
|
|
||||||
{LeastRCRemainingLifetimeKey, leastRCRemainingLifetime.count()},
|
|
||||||
{LastRCUpdatedKey, lastRCUpdated.count()},
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerStats::BEncode(llarp_buffer_t* buf) const
|
|
||||||
{
|
|
||||||
if (not buf)
|
|
||||||
throw std::runtime_error("PeerStats: Can't use null buf");
|
|
||||||
const oxenc::bt_dict data = {
|
|
||||||
{NumConnectionAttemptsKey, numConnectionAttempts},
|
|
||||||
{NumConnectionSuccessesKey, numConnectionSuccesses},
|
|
||||||
{NumConnectionRejectionsKey, numConnectionRejections},
|
|
||||||
{NumConnectionTimeoutsKey, numConnectionTimeouts},
|
|
||||||
{NumPathBuildsKey, numPathBuilds},
|
|
||||||
{NumPacketsAttemptedKey, numPacketsAttempted},
|
|
||||||
{NumPacketsSentKey, numPacketsSent},
|
|
||||||
{NumPacketsDroppedKey, numPacketsDropped},
|
|
||||||
{NumPacketsResentKey, numPacketsResent},
|
|
||||||
{NumDistinctRCsReceivedKey, numDistinctRCsReceived},
|
|
||||||
{NumLateRCsKey, numLateRCs},
|
|
||||||
{PeakBandwidthBytesPerSecKey, (uint64_t)peakBandwidthBytesPerSec},
|
|
||||||
{LongestRCReceiveIntervalKey, longestRCReceiveInterval.count()},
|
|
||||||
{LeastRCRemainingLifetimeKey, leastRCRemainingLifetime.count()},
|
|
||||||
{LastRCUpdatedKey, lastRCUpdated.count()},
|
|
||||||
};
|
|
||||||
const auto serialized = oxenc::bt_serialize(data);
|
|
||||||
if (not buf->write(serialized.begin(), serialized.end()))
|
|
||||||
throw std::runtime_error("PeerStats: buffer too small");
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
PeerStats::BEncodeList(const std::vector<PeerStats>& statsList, llarp_buffer_t* buf)
|
|
||||||
{
|
|
||||||
if (not buf)
|
|
||||||
throw std::runtime_error("PeerStats: Can't use null buf");
|
|
||||||
|
|
||||||
if (not bencode_start_list(buf))
|
|
||||||
throw std::runtime_error("PeerStats: Could not create bencode dict");
|
|
||||||
|
|
||||||
for (const auto& stats : statsList)
|
|
||||||
{
|
|
||||||
stats.BEncode(buf);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (not bencode_end(buf))
|
|
||||||
throw std::runtime_error("PeerStats: Could not end bencode dict");
|
|
||||||
}
|
|
||||||
|
|
||||||
}; // namespace llarp
|
|
@ -1,59 +0,0 @@
|
|||||||
#pragma once
|
|
||||||
|
|
||||||
#include <chrono>
|
|
||||||
#include <unordered_map>
|
|
||||||
|
|
||||||
#include <llarp/router_id.hpp>
|
|
||||||
#include <llarp/util/status.hpp>
|
|
||||||
#include <llarp/util/time.hpp>
|
|
||||||
|
|
||||||
/// Types stored in our peerstats database are declared here
|
|
||||||
|
|
||||||
namespace llarp
|
|
||||||
{
|
|
||||||
// Struct containing stats we know about a peer
|
|
||||||
struct PeerStats
|
|
||||||
{
|
|
||||||
RouterID routerId;
|
|
||||||
|
|
||||||
int32_t numConnectionAttempts = 0;
|
|
||||||
int32_t numConnectionSuccesses = 0;
|
|
||||||
int32_t numConnectionRejections = 0;
|
|
||||||
int32_t numConnectionTimeouts = 0;
|
|
||||||
|
|
||||||
int32_t numPathBuilds = 0;
|
|
||||||
int64_t numPacketsAttempted = 0;
|
|
||||||
int64_t numPacketsSent = 0;
|
|
||||||
int64_t numPacketsDropped = 0;
|
|
||||||
int64_t numPacketsResent = 0;
|
|
||||||
|
|
||||||
int32_t numDistinctRCsReceived = 0;
|
|
||||||
int32_t numLateRCs = 0;
|
|
||||||
|
|
||||||
double peakBandwidthBytesPerSec = 0;
|
|
||||||
llarp_time_t longestRCReceiveInterval = 0ms;
|
|
||||||
llarp_time_t leastRCRemainingLifetime = 0ms;
|
|
||||||
llarp_time_t lastRCUpdated = 0ms;
|
|
||||||
|
|
||||||
// not serialized
|
|
||||||
bool stale = true;
|
|
||||||
|
|
||||||
PeerStats();
|
|
||||||
PeerStats(const RouterID& routerId);
|
|
||||||
|
|
||||||
PeerStats&
|
|
||||||
operator+=(const PeerStats& other);
|
|
||||||
bool
|
|
||||||
operator==(const PeerStats& other) const;
|
|
||||||
|
|
||||||
util::StatusObject
|
|
||||||
toJson() const;
|
|
||||||
|
|
||||||
void
|
|
||||||
BEncode(llarp_buffer_t* buf) const;
|
|
||||||
|
|
||||||
static void
|
|
||||||
BEncodeList(const std::vector<PeerStats>& statsList, llarp_buffer_t* buf);
|
|
||||||
};
|
|
||||||
|
|
||||||
} // namespace llarp
|
|
@ -1,389 +0,0 @@
|
|||||||
|
|
||||||
#include <llarp/messages/link_message.hpp>
|
|
||||||
#include "router.hpp"
|
|
||||||
#include <llarp/constants/link_layer.hpp>
|
|
||||||
#include <llarp/util/meta/memfn.hpp>
|
|
||||||
#include <llarp/util/status.hpp>
|
|
||||||
|
|
||||||
#include <algorithm>
|
|
||||||
#include <cstdlib>
|
|
||||||
|
|
||||||
namespace llarp
|
|
||||||
{
|
|
||||||
const PathID_t OutboundMessageHandler::zeroID;
|
|
||||||
|
|
||||||
using namespace std::chrono_literals;
|
|
||||||
|
|
||||||
OutboundMessageHandler::OutboundMessageHandler(size_t maxQueueSize)
|
|
||||||
: outboundQueue(maxQueueSize), recentlyRemovedPaths(5s), removedSomePaths(false)
|
|
||||||
{}
|
|
||||||
|
|
||||||
bool
|
|
||||||
OutboundMessageHandler::QueueMessage(
|
|
||||||
const RouterID& remote, const AbstractLinkMessage& msg, SendStatusHandler callback)
|
|
||||||
{
|
|
||||||
// if the destination is invalid, callback with failure and return
|
|
||||||
if (not _router->link_manager().have_client_connection_to(remote)
|
|
||||||
and not _router->rc_lookup_handler().is_session_allowed(remote))
|
|
||||||
{
|
|
||||||
DoCallback(callback, SendStatus::InvalidRouter);
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
MessageQueueEntry ent;
|
|
||||||
ent.router = remote;
|
|
||||||
ent.inform = std::move(callback);
|
|
||||||
ent.pathid = msg.pathid;
|
|
||||||
ent.priority = msg.priority();
|
|
||||||
|
|
||||||
std::string _buf;
|
|
||||||
_buf.reserve(MAX_LINK_MSG_SIZE);
|
|
||||||
|
|
||||||
if (!EncodeBuffer(msg, _buf))
|
|
||||||
return false;
|
|
||||||
|
|
||||||
ent.message.resize(_buf.size());
|
|
||||||
|
|
||||||
std::copy_n(_buf.data(), _buf.size(), ent.message.data());
|
|
||||||
|
|
||||||
// if we have a session to the destination, queue the message and return
|
|
||||||
if (_router->link_manager().have_connection_to(remote))
|
|
||||||
{
|
|
||||||
QueueOutboundMessage(std::move(ent));
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
|
|
||||||
// if we don't have a session to the destination, queue the message onto
|
|
||||||
// a special pending session queue for that destination, and then create
|
|
||||||
// that pending session if there is not already a session establish attempt
|
|
||||||
// in progress.
|
|
||||||
bool shouldCreateSession = false;
|
|
||||||
{
|
|
||||||
util::Lock l{_mutex};
|
|
||||||
|
|
||||||
// create queue for <remote> if it doesn't exist, and get iterator
|
|
||||||
auto [queue_itr, is_new] = pendingSessionMessageQueues.emplace(remote, MessageQueue());
|
|
||||||
queue_itr->second.push(std::move(ent));
|
|
||||||
|
|
||||||
shouldCreateSession = is_new;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (shouldCreateSession)
|
|
||||||
{
|
|
||||||
QueueSessionCreation(remote);
|
|
||||||
}
|
|
||||||
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundMessageHandler::Pump()
|
|
||||||
{
|
|
||||||
m_Killer.TryAccess([this]() {
|
|
||||||
recentlyRemovedPaths.Decay();
|
|
||||||
ProcessOutboundQueue();
|
|
||||||
// TODO: this probably shouldn't be pumping, as it defeats the purpose
|
|
||||||
// of having a limit on sends per tick, but chaning it is potentially bad
|
|
||||||
// and requires testing so it should be changed later.
|
|
||||||
if (/*bool more = */ SendRoundRobin())
|
|
||||||
_router->TriggerPump();
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundMessageHandler::RemovePath(const PathID_t& pathid)
|
|
||||||
{
|
|
||||||
m_Killer.TryAccess([this, pathid]() {
|
|
||||||
/* add the path id to a list of recently removed paths to act as a filter
|
|
||||||
* for messages that are queued but haven't been sorted into path queues yet.
|
|
||||||
*
|
|
||||||
* otherwise these messages would re-create the path queue we just removed, and
|
|
||||||
* those path queues would be leaked / never removed.
|
|
||||||
*/
|
|
||||||
recentlyRemovedPaths.Insert(pathid);
|
|
||||||
auto itr = outboundMessageQueues.find(pathid);
|
|
||||||
if (itr != outboundMessageQueues.end())
|
|
||||||
{
|
|
||||||
outboundMessageQueues.erase(itr);
|
|
||||||
}
|
|
||||||
removedSomePaths = true;
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
util::StatusObject
|
|
||||||
OutboundMessageHandler::ExtractStatus() const
|
|
||||||
{
|
|
||||||
util::StatusObject status{
|
|
||||||
"queueStats",
|
|
||||||
{{"queued", m_queueStats.queued},
|
|
||||||
{"dropped", m_queueStats.dropped},
|
|
||||||
{"sent", m_queueStats.sent},
|
|
||||||
{"queueWatermark", m_queueStats.queueWatermark},
|
|
||||||
{"perTickMax", m_queueStats.perTickMax},
|
|
||||||
{"numTicks", m_queueStats.numTicks}}};
|
|
||||||
|
|
||||||
return status;
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundMessageHandler::Init(Router* router)
|
|
||||||
{
|
|
||||||
_router = router;
|
|
||||||
outboundMessageQueues.emplace(zeroID, MessageQueue());
|
|
||||||
}
|
|
||||||
|
|
||||||
static inline SendStatus
|
|
||||||
ToSendStatus(const SessionResult result)
|
|
||||||
{
|
|
||||||
switch (result)
|
|
||||||
{
|
|
||||||
case SessionResult::Establish:
|
|
||||||
return SendStatus::Success;
|
|
||||||
case SessionResult::Timeout:
|
|
||||||
case SessionResult::EstablishFail:
|
|
||||||
return SendStatus::Timeout;
|
|
||||||
case SessionResult::RouterNotFound:
|
|
||||||
return SendStatus::RouterNotFound;
|
|
||||||
case SessionResult::InvalidRouter:
|
|
||||||
return SendStatus::InvalidRouter;
|
|
||||||
case SessionResult::NoLink:
|
|
||||||
return SendStatus::NoLink;
|
|
||||||
}
|
|
||||||
throw std::invalid_argument{
|
|
||||||
fmt::format("SessionResult {} has no corresponding SendStatus when transforming", result)};
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundMessageHandler::OnSessionResult(const RouterID& router, const SessionResult result)
|
|
||||||
{
|
|
||||||
FinalizeSessionRequest(router, ToSendStatus(result));
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundMessageHandler::DoCallback(SendStatusHandler callback, SendStatus status)
|
|
||||||
{
|
|
||||||
if (callback)
|
|
||||||
_router->loop()->call([f = std::move(callback), status] { f(status); });
|
|
||||||
}
|
|
||||||
|
|
||||||
// TODO: still necessary/desired?
|
|
||||||
void
|
|
||||||
OutboundMessageHandler::QueueSessionCreation(const RouterID& remote)
|
|
||||||
{
|
|
||||||
_router->link_manager().Connect(remote);
|
|
||||||
}
|
|
||||||
|
|
||||||
/** Note: This is where AbstractLinkMessage::bt_encode() is called. Contextually, this is
|
|
||||||
different than how the other Abstract message types invoke ::bt_encode(), namely that
|
|
||||||
there is no bt_dict_producer already being appended to. As a result, this use case
|
|
||||||
likely requires a span backport and/or re-designed llarp_buffer. Until then, the
|
|
||||||
::bt_encode() override that returns an std::string upon destruction of its bt_dict_producer
|
|
||||||
will be used
|
|
||||||
*/
|
|
||||||
bool
|
|
||||||
OutboundMessageHandler::EncodeBuffer(const AbstractLinkMessage& msg, std::string& buf)
|
|
||||||
{
|
|
||||||
if (buf = msg.bt_encode(); not buf.empty())
|
|
||||||
return true;
|
|
||||||
|
|
||||||
log::error(link_cat, "Error: OutboundMessageHandler failed to encode outbound message!");
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
bool
|
|
||||||
OutboundMessageHandler::Send(const MessageQueueEntry& ent)
|
|
||||||
{
|
|
||||||
const llarp_buffer_t buf{ent.message};
|
|
||||||
m_queueStats.sent++;
|
|
||||||
SendStatusHandler callback = ent.inform;
|
|
||||||
return _router->link_manager().send_to(
|
|
||||||
ent.router,
|
|
||||||
buf,
|
|
||||||
[this, callback](AbstractLinkSession::DeliveryStatus status) {
|
|
||||||
if (status == AbstractLinkSession::DeliveryStatus::eDeliverySuccess)
|
|
||||||
DoCallback(callback, SendStatus::Success);
|
|
||||||
else
|
|
||||||
{
|
|
||||||
DoCallback(callback, SendStatus::Congestion);
|
|
||||||
}
|
|
||||||
},
|
|
||||||
ent.priority);
|
|
||||||
}
|
|
||||||
|
|
||||||
bool
|
|
||||||
OutboundMessageHandler::SendIfSession(const MessageQueueEntry& ent)
|
|
||||||
{
|
|
||||||
if (_router->link_manager().have_connection_to(ent.router))
|
|
||||||
{
|
|
||||||
return Send(ent);
|
|
||||||
}
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
bool
|
|
||||||
OutboundMessageHandler::QueueOutboundMessage(MessageQueueEntry entry)
|
|
||||||
{
|
|
||||||
// copy callback in case we need to call it, so we can std::move(entry)
|
|
||||||
auto callback = entry.inform;
|
|
||||||
if (outboundQueue.tryPushBack(std::move(entry)) != llarp::thread::QueueReturn::Success)
|
|
||||||
{
|
|
||||||
m_queueStats.dropped++;
|
|
||||||
DoCallback(callback, SendStatus::Congestion);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
m_queueStats.queued++;
|
|
||||||
uint32_t queueSize = outboundQueue.size();
|
|
||||||
m_queueStats.queueWatermark = std::max(queueSize, m_queueStats.queueWatermark);
|
|
||||||
}
|
|
||||||
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundMessageHandler::ProcessOutboundQueue()
|
|
||||||
{
|
|
||||||
while (not outboundQueue.empty())
|
|
||||||
{
|
|
||||||
MessageQueueEntry entry = outboundQueue.popFront();
|
|
||||||
|
|
||||||
// messages may still be queued for processing when a pathid is removed,
|
|
||||||
// so check here if the pathid was recently removed.
|
|
||||||
if (recentlyRemovedPaths.Contains(entry.pathid))
|
|
||||||
{
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
|
|
||||||
auto [queue_itr, is_new] = outboundMessageQueues.emplace(entry.pathid, MessageQueue());
|
|
||||||
|
|
||||||
if (is_new && !entry.pathid.IsZero())
|
|
||||||
{
|
|
||||||
roundRobinOrder.push(entry.pathid);
|
|
||||||
}
|
|
||||||
|
|
||||||
MessageQueue& path_queue = queue_itr->second;
|
|
||||||
|
|
||||||
if (path_queue.size() < MAX_PATH_QUEUE_SIZE || entry.pathid.IsZero())
|
|
||||||
{
|
|
||||||
path_queue.push(std::move(entry));
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
DoCallback(entry.inform, SendStatus::Congestion);
|
|
||||||
m_queueStats.dropped++;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
bool
|
|
||||||
OutboundMessageHandler::SendRoundRobin()
|
|
||||||
{
|
|
||||||
m_queueStats.numTicks++;
|
|
||||||
|
|
||||||
// send routing messages first priority
|
|
||||||
auto& routing_mq = outboundMessageQueues[zeroID];
|
|
||||||
while (not routing_mq.empty())
|
|
||||||
{
|
|
||||||
const MessageQueueEntry& entry = routing_mq.top();
|
|
||||||
Send(entry);
|
|
||||||
routing_mq.pop();
|
|
||||||
}
|
|
||||||
|
|
||||||
size_t num_queues = roundRobinOrder.size();
|
|
||||||
|
|
||||||
// if any paths have been removed since last tick, remove any stale
|
|
||||||
// entries from the round-robin ordering
|
|
||||||
if (removedSomePaths)
|
|
||||||
{
|
|
||||||
for (size_t i = 0; i < num_queues; i++)
|
|
||||||
{
|
|
||||||
PathID_t pathid = std::move(roundRobinOrder.front());
|
|
||||||
roundRobinOrder.pop();
|
|
||||||
|
|
||||||
if (outboundMessageQueues.find(pathid) != outboundMessageQueues.end())
|
|
||||||
{
|
|
||||||
roundRobinOrder.push(std::move(pathid));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
removedSomePaths = false;
|
|
||||||
|
|
||||||
num_queues = roundRobinOrder.size();
|
|
||||||
if (num_queues == 0)
|
|
||||||
{
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
// send messages for each pathid in roundRobinOrder, stopping when
|
|
||||||
// either every path's queue is empty or a set maximum amount of
|
|
||||||
// messages have been sent.
|
|
||||||
size_t consecutive_empty = 0;
|
|
||||||
for (size_t sent_count = 0; sent_count < MAX_OUTBOUND_MESSAGES_PER_TICK;)
|
|
||||||
{
|
|
||||||
PathID_t pathid = std::move(roundRobinOrder.front());
|
|
||||||
roundRobinOrder.pop();
|
|
||||||
|
|
||||||
auto& message_queue = outboundMessageQueues[pathid];
|
|
||||||
if (message_queue.size() > 0)
|
|
||||||
{
|
|
||||||
const MessageQueueEntry& entry = message_queue.top();
|
|
||||||
|
|
||||||
Send(entry);
|
|
||||||
message_queue.pop();
|
|
||||||
|
|
||||||
consecutive_empty = 0;
|
|
||||||
consecutive_empty++;
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
consecutive_empty++;
|
|
||||||
}
|
|
||||||
|
|
||||||
roundRobinOrder.push(std::move(pathid));
|
|
||||||
|
|
||||||
// if num_queues empty queues in a row, all queues empty.
|
|
||||||
if (consecutive_empty == num_queues)
|
|
||||||
{
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
m_queueStats.perTickMax = std::max((uint32_t)consecutive_empty, m_queueStats.perTickMax);
|
|
||||||
|
|
||||||
return consecutive_empty != num_queues;
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundMessageHandler::FinalizeSessionRequest(const RouterID& router, SendStatus status)
|
|
||||||
{
|
|
||||||
MessageQueue movedMessages;
|
|
||||||
{
|
|
||||||
util::Lock l(_mutex);
|
|
||||||
auto itr = pendingSessionMessageQueues.find(router);
|
|
||||||
|
|
||||||
if (itr == pendingSessionMessageQueues.end())
|
|
||||||
{
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
movedMessages.swap(itr->second);
|
|
||||||
|
|
||||||
pendingSessionMessageQueues.erase(itr);
|
|
||||||
}
|
|
||||||
|
|
||||||
while (!movedMessages.empty())
|
|
||||||
{
|
|
||||||
const MessageQueueEntry& entry = movedMessages.top();
|
|
||||||
|
|
||||||
if (status == SendStatus::Success)
|
|
||||||
{
|
|
||||||
Send(entry);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
DoCallback(entry.inform, status);
|
|
||||||
}
|
|
||||||
movedMessages.pop();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
} // namespace llarp
|
|
@ -1,319 +0,0 @@
|
|||||||
#include "outbound_session_maker.hpp"
|
|
||||||
|
|
||||||
#include "router.hpp"
|
|
||||||
#include <llarp/tooling/peer_stats_event.hpp>
|
|
||||||
#include <llarp/router_contact.hpp>
|
|
||||||
#include <llarp/nodedb.hpp>
|
|
||||||
#include "rc_lookup_handler.hpp"
|
|
||||||
#include <llarp/link/link_manager.hpp>
|
|
||||||
#include <llarp/util/meta/memfn.hpp>
|
|
||||||
#include <llarp/util/thread/threading.hpp>
|
|
||||||
#include <llarp/util/status.hpp>
|
|
||||||
#include <llarp/crypto/crypto.hpp>
|
|
||||||
#include <utility>
|
|
||||||
|
|
||||||
#include <llarp/rpc/lokid_rpc_client.hpp>
|
|
||||||
|
|
||||||
namespace llarp
|
|
||||||
{
|
|
||||||
|
|
||||||
bool
|
|
||||||
OutboundSessionMaker::OnSessionEstablished(AbstractLinkSession* session)
|
|
||||||
{
|
|
||||||
// TODO: do we want to keep it
|
|
||||||
const RouterContact rc = session->GetRemoteRC();
|
|
||||||
const auto router = RouterID(session->GetPubKey());
|
|
||||||
const bool isOutbound = not session->IsInbound();
|
|
||||||
const std::string remoteType = rc.IsPublicRouter() ? "router" : "client";
|
|
||||||
LogInfo(
|
|
||||||
"session with ", remoteType, " [", router, "] ", isOutbound ? "established" : "received");
|
|
||||||
|
|
||||||
if (not _rcLookup->SessionIsAllowed(router))
|
|
||||||
{
|
|
||||||
FinalizeRequest(router, SessionResult::InvalidRouter);
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (isOutbound)
|
|
||||||
{
|
|
||||||
work([this, rc] { VerifyRC(rc); });
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
return _rcLookup->CheckRC(rc);
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundSessionMaker::OnConnectTimeout(AbstractLinkSession* session)
|
|
||||||
{
|
|
||||||
const auto router = RouterID(session->GetPubKey());
|
|
||||||
LogWarn("Session establish attempt to ", router, " timed out.", session->GetRemoteEndpoint());
|
|
||||||
FinalizeRequest(router, SessionResult::Timeout);
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundSessionMaker::CreateSessionTo(const RouterID& router, RouterCallback on_result)
|
|
||||||
{
|
|
||||||
{
|
|
||||||
util::Lock l(_mutex);
|
|
||||||
|
|
||||||
auto itr_pair = pendingCallbacks.emplace(router, CallbacksQueue{});
|
|
||||||
if (on_result)
|
|
||||||
itr_pair.first->second.push_back(on_result);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (HavePendingSessionTo(router))
|
|
||||||
{
|
|
||||||
LogDebug("has pending session to", router);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
CreatePendingSession(router);
|
|
||||||
|
|
||||||
// short-circuit to success callback if we already have an outbound session
|
|
||||||
// to the remote
|
|
||||||
if (_linkManager->HasConnection(router))
|
|
||||||
{
|
|
||||||
FinalizeRequest(router, SessionResult::Establish);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
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)
|
|
||||||
{
|
|
||||||
const RouterID router{rc.pubkey};
|
|
||||||
|
|
||||||
{
|
|
||||||
util::Lock l(_mutex);
|
|
||||||
|
|
||||||
auto itr_pair = pendingCallbacks.emplace(router, CallbacksQueue{});
|
|
||||||
if (on_result)
|
|
||||||
itr_pair.first->second.push_back(on_result);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (not HavePendingSessionTo(router))
|
|
||||||
{
|
|
||||||
LogDebug("Creating session establish attempt to ", router);
|
|
||||||
CreatePendingSession(router);
|
|
||||||
}
|
|
||||||
|
|
||||||
GotRouterContact(router, rc);
|
|
||||||
}
|
|
||||||
|
|
||||||
bool
|
|
||||||
OutboundSessionMaker::HavePendingSessionTo(const RouterID& router) const
|
|
||||||
{
|
|
||||||
util::Lock l(_mutex);
|
|
||||||
return pendingCallbacks.find(router) != pendingCallbacks.end();
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundSessionMaker::ConnectToRandomRouters(int numDesired)
|
|
||||||
{
|
|
||||||
int remainingDesired = numDesired;
|
|
||||||
std::set<RouterID> exclude;
|
|
||||||
do
|
|
||||||
{
|
|
||||||
auto filter = [exclude](const auto& rc) -> bool { return exclude.count(rc.pubkey) == 0; };
|
|
||||||
|
|
||||||
RouterContact other;
|
|
||||||
if (const auto maybe = _nodedb->GetRandom(filter))
|
|
||||||
{
|
|
||||||
other = *maybe;
|
|
||||||
}
|
|
||||||
else
|
|
||||||
break;
|
|
||||||
|
|
||||||
exclude.insert(other.pubkey);
|
|
||||||
if (not _rcLookup->SessionIsAllowed(other.pubkey))
|
|
||||||
{
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
if (not(_linkManager->HasSessionTo(other.pubkey) || HavePendingSessionTo(other.pubkey)))
|
|
||||||
{
|
|
||||||
CreateSessionTo(other, nullptr);
|
|
||||||
--remainingDesired;
|
|
||||||
}
|
|
||||||
|
|
||||||
} while (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(
|
|
||||||
Router* router,
|
|
||||||
LinkManager* linkManager,
|
|
||||||
RCLookupHandler* rcLookup,
|
|
||||||
Profiling* profiler,
|
|
||||||
EventLoop_ptr loop,
|
|
||||||
WorkerFunc_t dowork)
|
|
||||||
{
|
|
||||||
_router = router;
|
|
||||||
_linkManager = linkManager;
|
|
||||||
_rcLookup = rcLookup;
|
|
||||||
_loop = std::move(loop);
|
|
||||||
_nodedb = router->nodedb();
|
|
||||||
_profiler = profiler;
|
|
||||||
work = std::move(dowork);
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundSessionMaker::GotRouterContact(const RouterID& router, const RouterContact& rc)
|
|
||||||
{
|
|
||||||
if (not _rcLookup->CheckRC(rc))
|
|
||||||
{
|
|
||||||
FinalizeRequest(rc.pubkey, SessionResult::InvalidRouter);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (not ShouldConnectTo(router))
|
|
||||||
{
|
|
||||||
FinalizeRequest(router, SessionResult::NoLink);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
auto result = _linkManager->Connect(rc);
|
|
||||||
if (result)
|
|
||||||
FinalizeRequest(router, SessionResult::Establish);
|
|
||||||
else
|
|
||||||
FinalizeRequest(router, SessionResult::EstablishFail);
|
|
||||||
}
|
|
||||||
|
|
||||||
bool
|
|
||||||
OutboundSessionMaker::ShouldConnectTo(const RouterID& router) const
|
|
||||||
{
|
|
||||||
if (router == us or not _rcLookup->SessionIsAllowed(router))
|
|
||||||
return false;
|
|
||||||
if (_router->IsServiceNode())
|
|
||||||
return true;
|
|
||||||
|
|
||||||
size_t numPending = 0;
|
|
||||||
{
|
|
||||||
util::Lock lock(_mutex);
|
|
||||||
if (pendingCallbacks.find(router) == pendingCallbacks.end())
|
|
||||||
numPending += pendingCallbacks.size();
|
|
||||||
}
|
|
||||||
|
|
||||||
return _linkManager->get_num_connected() + numPending < maxConnectedRouters;
|
|
||||||
}
|
|
||||||
|
|
||||||
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))
|
|
||||||
{
|
|
||||||
LogError("no pending session to ", router);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
switch (result)
|
|
||||||
{
|
|
||||||
case RCRequestResult::Success:
|
|
||||||
if (rc)
|
|
||||||
{
|
|
||||||
GotRouterContact(router, *rc);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
LogError("RCRequestResult::Success but null rc pointer given");
|
|
||||||
InvalidRouter(router);
|
|
||||||
}
|
|
||||||
break;
|
|
||||||
case RCRequestResult::InvalidRouter:
|
|
||||||
InvalidRouter(router);
|
|
||||||
break;
|
|
||||||
case RCRequestResult::RouterNotFound:
|
|
||||||
RouterNotFound(router);
|
|
||||||
break;
|
|
||||||
default:
|
|
||||||
RouterNotFound(router);
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundSessionMaker::VerifyRC(const RouterContact rc)
|
|
||||||
{
|
|
||||||
if (not _rcLookup->CheckRC(rc))
|
|
||||||
{
|
|
||||||
FinalizeRequest(rc.pubkey, SessionResult::InvalidRouter);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
FinalizeRequest(rc.pubkey, SessionResult::Establish);
|
|
||||||
}
|
|
||||||
|
|
||||||
// TODO: rename this, if we even want to keep it
|
|
||||||
void
|
|
||||||
OutboundSessionMaker::CreatePendingSession(const RouterID& router)
|
|
||||||
{
|
|
||||||
auto peerDb = _router->peerDb();
|
|
||||||
if (peerDb)
|
|
||||||
{
|
|
||||||
peerDb->modifyPeerStats(router, [](PeerStats& stats) { stats.numConnectionAttempts++; });
|
|
||||||
}
|
|
||||||
|
|
||||||
_router->NotifyRouterEvent<tooling::ConnectionAttemptEvent>(_router->pubkey(), router);
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
OutboundSessionMaker::FinalizeRequest(const RouterID& router, const SessionResult type)
|
|
||||||
{
|
|
||||||
CallbacksQueue movedCallbacks;
|
|
||||||
{
|
|
||||||
util::Lock l(_mutex);
|
|
||||||
|
|
||||||
if (type == SessionResult::Establish)
|
|
||||||
{
|
|
||||||
_profiler->MarkConnectSuccess(router);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
// TODO: add non timeout related fail case
|
|
||||||
_profiler->MarkConnectTimeout(router);
|
|
||||||
}
|
|
||||||
|
|
||||||
auto itr = pendingCallbacks.find(router);
|
|
||||||
|
|
||||||
if (itr != pendingCallbacks.end())
|
|
||||||
{
|
|
||||||
movedCallbacks.splice(movedCallbacks.begin(), itr->second);
|
|
||||||
pendingCallbacks.erase(itr);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
for (const auto& callback : movedCallbacks)
|
|
||||||
{
|
|
||||||
_loop->call([callback, router, type] { return callback(router, type); });
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
} // namespace llarp
|
|
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in New Issue