mirror of
https://github.com/ZmnSCPxj/clboss.git
synced 2026-08-14 12:43:19 +02:00
497 lines
12 KiB
C++
497 lines
12 KiB
C++
#include"Boss/Mod/ChannelFinderByPopularity.hpp"
|
|
#include"Boss/Mod/Rpc.hpp"
|
|
#include"Boss/Mod/Waiter.hpp"
|
|
#include"Boss/Msg/Block.hpp"
|
|
#include"Boss/Msg/Init.hpp"
|
|
#include"Boss/Msg/ManifestNotification.hpp"
|
|
#include"Boss/Msg/Manifestation.hpp"
|
|
#include"Boss/Msg/Notification.hpp"
|
|
#include"Boss/Msg/ProposeChannelCandidates.hpp"
|
|
#include"Boss/Msg/SolicitChannelCandidates.hpp"
|
|
#include"Boss/Msg/TaskCompletion.hpp"
|
|
#include"Boss/concurrent.hpp"
|
|
#include"Boss/log.hpp"
|
|
#include"Boss/random_engine.hpp"
|
|
#include"Ev/Io.hpp"
|
|
#include"Ev/map.hpp"
|
|
#include"Ev/yield.hpp"
|
|
#include"Jsmn/Object.hpp"
|
|
#include"Json/Out.hpp"
|
|
#include"Ln/NodeId.hpp"
|
|
#include"S/Bus.hpp"
|
|
#include"Util/make_unique.hpp"
|
|
#include<algorithm>
|
|
#include<iterator>
|
|
#include<queue>
|
|
#include<random>
|
|
#include<set>
|
|
#include<sstream>
|
|
#include<vector>
|
|
|
|
namespace {
|
|
|
|
/** max_proposals
|
|
*
|
|
* @brief ChannelFinderByPopularity will only
|
|
* propose up to this number of nodes.
|
|
*/
|
|
auto const max_proposals = size_t(4);
|
|
|
|
/** min_channels_to_disable
|
|
*
|
|
* @brief once we have achieved this number of
|
|
* channels, ChannelFinderByPopularity will
|
|
* stop making proposals.
|
|
*/
|
|
auto const min_channels_to_disable = size_t(4);
|
|
|
|
/** min_nodes_to_process
|
|
*
|
|
* @brief if the number of nodes known by this node
|
|
* is less than this, go to sleep and try again
|
|
* later.
|
|
*/
|
|
auto const min_nodes_to_process = size_t(200);
|
|
|
|
/** seconds_after_connect
|
|
*
|
|
* @brief if we previously stopped looking for popular
|
|
* nodes, and a new connection appears, how many seconds
|
|
* should we wait for them to update us about the network?
|
|
*/
|
|
auto const seconds_after_connect = double(150);
|
|
|
|
}
|
|
|
|
namespace Boss { namespace Mod {
|
|
|
|
class ChannelFinderByPopularity::Impl {
|
|
private:
|
|
S::Bus& bus;
|
|
Boss::Mod::Waiter& waiter;
|
|
void* this_ptr;
|
|
Boss::Mod::Rpc* rpc;
|
|
Ln::NodeId self;
|
|
|
|
/* Set if we are already running. */
|
|
bool running;
|
|
/* Set if we are waiting for a while. */
|
|
bool try_later;
|
|
|
|
void start() {
|
|
running = false;
|
|
try_later = false;
|
|
bus.subscribe<Msg::Init>([this](Msg::Init const& init) {
|
|
rpc = &init.rpc;
|
|
self = init.self_id;
|
|
return Ev::lift();
|
|
});
|
|
bus.subscribe< Msg::SolicitChannelCandidates
|
|
>([this](Msg::SolicitChannelCandidates const& _) {
|
|
if (running || !rpc || !self)
|
|
return Ev::lift();
|
|
running = true;
|
|
return Boss::concurrent(Ev::lift().then([this]() {
|
|
return solicit();
|
|
}));
|
|
});
|
|
bus.subscribe< Msg::Block
|
|
>([this](Msg::Block const& _) {
|
|
if (try_later) {
|
|
try_later = false;
|
|
return Boss::concurrent(Ev::lift().then([this]() {
|
|
return solicit();
|
|
}));
|
|
}
|
|
return Ev::lift();
|
|
});
|
|
|
|
/* Also do try_later when we successfully connect. */
|
|
bus.subscribe< Msg::Manifestation
|
|
>([this](Msg::Manifestation const& _) {
|
|
return bus.raise(Msg::ManifestNotification{
|
|
"connect"
|
|
});
|
|
});
|
|
bus.subscribe< Msg::Notification
|
|
>([this](Msg::Notification const& n) {
|
|
if (n.notification != "connect")
|
|
return Ev::lift();
|
|
if (!try_later)
|
|
return Ev::lift();
|
|
try_later = false;
|
|
return Boss::concurrent(on_new_connect());
|
|
});
|
|
}
|
|
|
|
/* List of all nodes. */
|
|
std::queue<Ln::NodeId> all_nodes;
|
|
|
|
struct Popular {
|
|
Ln::NodeId node;
|
|
std::set<Ln::NodeId> peers;
|
|
};
|
|
/* A-Chao algorithm. */
|
|
double wsum;
|
|
std::vector<Popular> selected;
|
|
|
|
/* Called after a new connection while we were deferring
|
|
* a solicitation. */
|
|
Ev::Io<void> on_new_connect() {
|
|
return Boss::log( bus, Info
|
|
, "ChannelFinderByPopularity: "
|
|
"New connection, will wait %.0f seconds, "
|
|
"then re-find nodes."
|
|
, seconds_after_connect
|
|
).then([this]() {
|
|
return waiter.wait(seconds_after_connect);
|
|
}).then([this]() {
|
|
return solicit();
|
|
});
|
|
}
|
|
|
|
Ev::Io<void> solicit() {
|
|
/* First check if we have at least
|
|
* min_channels_to_disable channels.
|
|
* If so, finding by popularity self-disables.
|
|
*/
|
|
return rpc->command( "listchannels"
|
|
, Json::Out()
|
|
.start_object()
|
|
.field("source"
|
|
, std::string(self)
|
|
)
|
|
.end_object()
|
|
).then([this](Jsmn::Object res) {
|
|
if (!res.is_object() || !res.has("channels"))
|
|
return self_disable();
|
|
auto channels = res["channels"];
|
|
if (!channels.is_array())
|
|
return self_disable();
|
|
if (channels.size() >= min_channels_to_disable)
|
|
return self_disable();
|
|
return Boss::log( bus, Debug
|
|
, "ChannelFinderByPopularity: "
|
|
"Will find nodes."
|
|
).then([this]() {
|
|
return continue_solicit();
|
|
});
|
|
});
|
|
}
|
|
|
|
Ev::Io<void> self_disable() {
|
|
return Boss::log( bus, Debug
|
|
, "ChannelFinderByPopularity: "
|
|
"Voluntarily not operating, "
|
|
"we seem to have many channels."
|
|
).then([this]() {
|
|
/* Do not signal failed, since we successfully
|
|
* decided not to do anything. */
|
|
return signal_task_completion(false);
|
|
});
|
|
}
|
|
|
|
Ev::Io<void> continue_solicit() {
|
|
/* Now do listnodes for everything. */
|
|
return rpc->command( "listnodes"
|
|
, Json::Out::empty_object()
|
|
).then([this](Jsmn::Object res){
|
|
if (!res.is_object() || !res.has("nodes"))
|
|
return fail_solicit("Unexpected result from listnodes.");
|
|
|
|
auto nodes = res["nodes"];
|
|
if (!nodes.is_array())
|
|
return fail_solicit("Unexpected result from listnodes.");
|
|
|
|
if (nodes.size() < min_nodes_to_process)
|
|
return defer_solicit(nodes.size());
|
|
all_nodes = std::queue<Ln::NodeId>();
|
|
for (auto i = size_t(0); i < nodes.size(); ++i) {
|
|
auto node = nodes[i];
|
|
if (!node.is_object() || !node.has("nodeid"))
|
|
continue;
|
|
auto id = node["nodeid"];
|
|
if (!id.is_string())
|
|
continue;
|
|
all_nodes.emplace(std::string(id));
|
|
}
|
|
/* Initialize the A-Chao algorithm. */
|
|
wsum = 0;
|
|
selected.clear();
|
|
/* Print details. */
|
|
return Boss::log( bus, Debug
|
|
, "ChannelFinderByPopularity: "
|
|
"%zu nodes to be evaluated."
|
|
, all_nodes.size()
|
|
).then([this]() {
|
|
return select_by_popularity();
|
|
});
|
|
});
|
|
}
|
|
|
|
Ev::Io<void> defer_solicit(size_t n) {
|
|
try_later = true;
|
|
return Boss::log( bus, Info
|
|
, "ChannelFinderByPopularity: "
|
|
"Not enough nodes in routemap "
|
|
"(only %zu, need %zu), "
|
|
"will try again later."
|
|
, n, min_nodes_to_process
|
|
);
|
|
}
|
|
|
|
Ev::Io<void> fail_solicit(std::string const& reason) {
|
|
return Boss::log( bus, Error
|
|
, "ChannelFinderByPopularity: "
|
|
"Failed: %s"
|
|
, reason.c_str()
|
|
).then([this]() {
|
|
/* Failed! */
|
|
return signal_task_completion(true);
|
|
});
|
|
}
|
|
|
|
/* A-Chao Reservoir sampling algorithm. */
|
|
Ev::Io<void> select_by_popularity() {
|
|
return Ev::yield().then([this]() {
|
|
|
|
/* This is a loop. */
|
|
if (all_nodes.empty())
|
|
return complete_select_by_popularity();
|
|
/* Get channels of the current node. */
|
|
auto const& curr = all_nodes.front();
|
|
return rpc->command( "listchannels"
|
|
, Json::Out()
|
|
.start_object()
|
|
.field( "source"
|
|
, std::string(curr)
|
|
)
|
|
.end_object()
|
|
).then([this](Jsmn::Object res) {
|
|
auto entry = Popular();
|
|
/* Now move the current node off the queue. */
|
|
entry.node = std::move(all_nodes.front());
|
|
all_nodes.pop();
|
|
|
|
/* Now fill in the entry. */
|
|
if (!res.is_object() || !res.has("channels"))
|
|
/* Skip. */
|
|
return select_by_popularity();
|
|
auto channels = res["channels"];
|
|
if (!channels.is_array())
|
|
return select_by_popularity();
|
|
for (auto i = size_t(0); i < channels.size(); ++i) {
|
|
auto channel = channels[i];
|
|
if ( !channel.is_object()
|
|
|| !channel.has("destination")
|
|
)
|
|
continue;
|
|
auto destination = channel["destination"];
|
|
if (!destination.is_string())
|
|
continue;
|
|
auto dest_s = std::string(destination);
|
|
entry.peers.emplace(Ln::NodeId(dest_s));
|
|
}
|
|
|
|
/* ***Finally*** determine if we should select
|
|
* this entry. */
|
|
wsum += entry.peers.size();
|
|
/* If fewer are selected than max proposals, go extend
|
|
* it. */
|
|
if (selected.size() < max_proposals) {
|
|
selected.emplace_back(std::move(entry));
|
|
return select_by_popularity();
|
|
}
|
|
/* Otherwise see if we should replace. */
|
|
auto dist = std::uniform_real_distribution<double>(
|
|
0, 1
|
|
);
|
|
auto p = double(entry.peers.size()) / wsum;
|
|
auto j = dist(Boss::random_engine);
|
|
if (j <= p) {
|
|
/* Randomly replace an existing selection. */
|
|
auto dist_i = std::uniform_int_distribution<size_t>(
|
|
0, selected.size() - 1
|
|
);
|
|
auto i = dist_i(Boss::random_engine);
|
|
selected[i] = std::move(entry);
|
|
}
|
|
|
|
return select_by_popularity();
|
|
});
|
|
|
|
});
|
|
}
|
|
|
|
Ev::Io<void> complete_select_by_popularity() {
|
|
/* Generate a log. */
|
|
auto log = std::ostringstream();
|
|
log << "ChannelFinderByPopularity: "
|
|
<< "Random selection (by popularity): "
|
|
;
|
|
auto first = true;
|
|
for (auto const& s : selected) {
|
|
if (first)
|
|
first = false;
|
|
else
|
|
log << ", ";
|
|
log << s.node
|
|
<< " (" << s.peers.size() << " peers)"
|
|
;
|
|
}
|
|
return Boss::log( bus, Info
|
|
, "%s", log.str().c_str()
|
|
).then([this]() {
|
|
return try_connect();
|
|
});
|
|
}
|
|
|
|
/* Shuffles the peers of a particular selected node,
|
|
* then tries to connect each peer in sequence.
|
|
*/
|
|
class ConnectChecker {
|
|
private:
|
|
S::Bus& bus;
|
|
Boss::Mod::Rpc& rpc;
|
|
struct Impl {
|
|
Ln::NodeId node;
|
|
std::vector<Ln::NodeId> peers;
|
|
std::vector<Ln::NodeId>::const_iterator peer_it;
|
|
S::Bus* bus;
|
|
Boss::Mod::Rpc* rpc;
|
|
};
|
|
public:
|
|
ConnectChecker() =delete;
|
|
explicit
|
|
ConnectChecker( S::Bus& bus_
|
|
, Boss::Mod::Rpc& rpc_
|
|
) : bus(bus_)
|
|
, rpc(rpc_)
|
|
{ }
|
|
/* Functor. */
|
|
Ev::Io<bool> operator()(Popular p) {
|
|
auto self = std::make_shared<Impl>();
|
|
self->node = std::move(p.node);
|
|
std::copy( p.peers.begin(), p.peers.end()
|
|
, std::back_inserter(self->peers)
|
|
);
|
|
std::shuffle( self->peers.begin()
|
|
, self->peers.end()
|
|
, Boss::random_engine
|
|
);
|
|
self->peer_it = self->peers.begin();
|
|
self->bus = &bus;
|
|
self->rpc = &rpc;
|
|
return loop(std::move(self));
|
|
}
|
|
private:
|
|
static
|
|
Ev::Io<bool> loop(std::shared_ptr<Impl> self) {
|
|
if (self->peer_it == self->peers.end())
|
|
return Ev::lift(false);
|
|
/* FIXME: Use Connector object. */
|
|
auto params = Json::Out()
|
|
.start_object()
|
|
.field( "id"
|
|
, std::string(*self->peer_it)
|
|
)
|
|
.end_object()
|
|
;
|
|
return self->rpc->command("connect"
|
|
, params
|
|
).then([](Jsmn::Object _) {
|
|
return Ev::lift(true);
|
|
}).catching<RpcError>([](RpcError const& _) {
|
|
return Ev::lift(false);
|
|
}).then([self](bool success) {
|
|
/* If succeeded, send proposal and exit
|
|
* loop. */
|
|
if (success) {
|
|
return self->bus->raise(Msg::ProposeChannelCandidates{
|
|
*self->peer_it,
|
|
self->node
|
|
}).then([]() {
|
|
return Ev::lift(true);
|
|
});
|
|
} else {
|
|
/* Keep looping. */
|
|
++self->peer_it;
|
|
return loop(self);
|
|
}
|
|
});
|
|
}
|
|
};
|
|
|
|
size_t proposals_made;
|
|
|
|
Ev::Io<void> try_connect() {
|
|
/* Now we try to connect to each of the peers
|
|
* of each of the selected nodes.
|
|
* If we manage to connect to one of the peers,
|
|
* we propose that peer and set the popular node
|
|
* as the patron.
|
|
*
|
|
* We go over each of the selected nodes by Ev::map,
|
|
* which runs all the selected nodes in parallel.
|
|
*/
|
|
return Ev::map( ConnectChecker(bus, *rpc)
|
|
, std::move(selected)
|
|
).then([this](std::vector<bool> results) {
|
|
/* Count number of true. */
|
|
proposals_made = 0;
|
|
for (auto const& flag : results)
|
|
if (flag)
|
|
++proposals_made;
|
|
return completed_solicit();
|
|
});
|
|
}
|
|
|
|
|
|
Ev::Io<void> completed_solicit() {
|
|
return Ev::yield().then([this]() {
|
|
if (proposals_made != 0)
|
|
return Boss::log( bus, Info
|
|
, "ChannelFinderByPopularity: "
|
|
"Proposed %zu nodes to "
|
|
"channel to, finished."
|
|
, proposals_made
|
|
);
|
|
else
|
|
return Boss::log( bus, Warn
|
|
, "ChannelFinderByPopularity: "
|
|
"No nodes were proposed."
|
|
);
|
|
}).then([this]() {
|
|
return signal_task_completion(proposals_made == 0);
|
|
});
|
|
}
|
|
|
|
Ev::Io<void> signal_task_completion(bool failed) {
|
|
running = false;
|
|
return bus.raise(Msg::TaskCompletion{
|
|
"ChannelFinderByPopularity", this_ptr, failed
|
|
});
|
|
}
|
|
|
|
public:
|
|
Impl( S::Bus& bus_
|
|
, Boss::Mod::Waiter& waiter_
|
|
, void* this_ptr_
|
|
) : bus(bus_)
|
|
, waiter(waiter_)
|
|
, this_ptr(this_ptr_)
|
|
, rpc(nullptr)
|
|
, running(false)
|
|
{ start(); }
|
|
|
|
};
|
|
|
|
ChannelFinderByPopularity::ChannelFinderByPopularity
|
|
( S::Bus& bus
|
|
, Boss::Mod::Waiter& waiter
|
|
) : pimpl(Util::make_unique<Impl>(bus, waiter, this)) { }
|
|
ChannelFinderByPopularity::ChannelFinderByPopularity
|
|
( ChannelFinderByPopularity&& o
|
|
) : pimpl(std::move(o.pimpl)) { }
|
|
ChannelFinderByPopularity::~ChannelFinderByPopularity() { }
|
|
}}
|