<commit_before>/* discovery.cpp                                 -*- C++ -*-
   <NAME> (<EMAIL>), 26 Dec 2013
   FreeBSD-style copyright and disclaimer apply

   Implementation of the node discovery database.

*/

#include "discovery.h"
#include "pack.h"
#include "lockless/bits.h"

#include <set>
#include <vector>
#include <algorithm>
#include <functional>

namespace slick {


/******************************************************************************/
/* UTILS                                                                      */
/******************************************************************************/

template<typename It, typename Rng>
It pickRandom(It it, It last, Rng& rng)
{
    std::uniform_int_distribution<size_t> dist(std::distance(it, last) - 1);
    std::advance(it, dist(rng));
    return it;
}

template<typename T, typename It, typename Rng>
std::set<T> pickRandom(It first, It last, size_t n, Rng& rng)
{
    std::set<T> result;
    while (result.size() < n)
        result.insert(*pickRandom(first, last, rng));
    return std::move(result);
}


/******************************************************************************/
/* WATCH                                                                      */
/******************************************************************************/

DistributedDiscovery::Watch::
Watch(WatchFn watch) : watch(std::move(watch))
{
    static std::atomic<WatchHandle> handleCounter{0};
    handle = ++handleCounter;
}


/******************************************************************************/
/* PROTOCOL                                                                   */
/******************************************************************************/

namespace msg {

static const std::string Init =  "_slick_disc_";
static constexpr uint32_t Version = 1;

typedef uint16_t Type;
static constexpr Type Keys  = 1;
static constexpr Type Query = 2;
static constexpr Type Nodes = 3;
static constexpr Type Fetch = 4;
static constexpr Type Data  = 5;

} // namespace msg



/******************************************************************************/
/* DISTRIBUTED DISCOVERY                                                      */
/******************************************************************************/

size_t
DistributedDiscovery::
timerPeriod()
{
    enum { BasePeriod = 60 };
    std::uniform_int_distribution<size_t>dist(BasePeriod, BasePeriod * 2);
    return dist(rng);
}

DistributedDiscovery::
DistributedDiscovery(const std::vector<Address>& seed, Port port) :
    keyTTL_(DefaultKeyTTL),
    nodeTTL_(DefaultNodeTTL),
    myId(UUID::random()),
    rng(lockless::wall()),
    endpoint(port),
    timer(timerPeriod()) // \todo add randomness
{
    myNode = endpoint.interfaces();

    double now = lockless::wall();
    for (auto& addr : seed)
        nodes.emplace(UUID::random(), NodeLocation({addr}), SeedTTL, now);

    using namespace std::placeholders;

    endpoint.onPayload = bind(&DistributedDiscovery::onPayload, this, _1, _2);
    endpoint.onNewConnection = bind(&DistributedDiscovery::onConnect, this, _1);
    endpoint.onLostConnection = bind(&DistributedDiscovery::onDisconnect, this, _1);
    poller.add(endpoint);

    retracts.onOperation = std::bind(&DistributedDiscovery::retract, this, _1);
    poller.add(retracts);

    publishes.onOperation = std::bind(&DistributedDiscovery::publish, this, _1, _2);
    poller.add(publishes);

    typedef void (DistributedDiscovery::*DiscoverFn) (const std::string&, Watch&&);
    DiscoverFn discoverFn = &DistributedDiscovery::discover;
    discovers.onOperation = std::bind(discoverFn, this, _1, _2);
    poller.add(discovers);
<commit_msg>Tweaked the period calc.<commit_after>/* discovery.cpp                                 -*- C++ -*-
   <NAME> (<EMAIL>), 26 Dec 2013
   FreeBSD-style copyright and disclaimer apply

   Implementation of the node discovery database.

*/

#include "discovery.h"
#include "pack.h"
#include "lockless/bits.h"

#include <set>
#include <vector>
#include <algorithm>
#include <functional>

namespace slick {


/******************************************************************************/
/* UTILS                                                                      */
/******************************************************************************/

template<typename It, typename Rng>
It pickRandom(It it, It last, Rng& rng)
{
    std::uniform_int_distribution<size_t> dist(std::distance(it, last) - 1);
    std::advance(it, dist(rng));
    return it;
}

template<typename T, typename It, typename Rng>
std::set<T> pickRandom(It first, It last, size_t n, Rng& rng)
{
    std::set<T> result;
    while (result.size() < n)
        result.insert(*pickRandom(first, last, rng));
    return std::move(result);
}


/******************************************************************************/
/* WATCH                                                                      */
/******************************************************************************/

DistributedDiscovery::Watch::
Watch(WatchFn watch) : watch(std::move(watch))
{
    static std::atomic<WatchHandle> handleCounter{0};
    handle = ++handleCounter;
}


/******************************************************************************/
/* PROTOCOL                                                                   */
/******************************************************************************/

namespace msg {

static const std::string Init =  "_slick_disc_";
static constexpr uint32_t Version = 1;

typedef uint16_t Type;
static constexpr Type Keys  = 1;
static constexpr Type Query = 2;
static constexpr Type Nodes = 3;
static constexpr Type Fetch = 4;
static constexpr Type Data  = 5;

} // namespace msg



/******************************************************************************/
/* DISTRIBUTED DISCOVERY                                                      */
/******************************************************************************/

size_t
DistributedDiscovery::
timerPeriod()
{
    enum { BasePeriod = 60 };
    size_t min = BasePeriod / 2, max = min + BasePeriod;
    return std::uniform_int_distribution<size_t> dist(min, max)(rng);
}

DistributedDiscovery::
DistributedDiscovery(const std::vector<Address>& seed, Port port) :
    keyTTL_(DefaultKeyTTL),
    nodeTTL_(DefaultNodeTTL),
    myId(UUID::random()),
    rng(lockless::wall()),
    endpoint(port),
    timer(timerPeriod())
{
    myNode = endpoint.interfaces();

    double now = lockless::wall();
    for (auto& addr : seed)
        nodes.emplace(UUID::random(), NodeLocation({addr}), SeedTTL, now);

    using namespace std::placeholders;

    endpoint.onPayload = bind(&DistributedDiscovery::onPayload, this, _1, _2);
    endpoint.onNewConnection = bind(&DistributedDiscovery::onConnect, this, _1);
    endpoint.onLostConnection = bind(&DistributedDiscovery::onDisconnect, this, _1);
    poller.add(endpoint);

    retracts.onOperation = std::bind(&DistributedDiscovery::retract, this, _1);
    poller.add(retracts);

    publishes.onOperation = std::bind(&DistributedDiscovery::publish, this, _1, _2);
    poller.add(publishes);

    typedef void (DistributedDiscovery::*DiscoverFn) (const std::string&, Watch&&);
    DiscoverFn discoverFn = &DistributedDiscovery::discover;
    discovers.onOperation = std::bind(discoverFn, this, _1, _2);
    poller.add(discovers);

{'fraction_non_alphanumeric': 0.27682779049414297, 'fraction_numerical': 0.008078632018311566, 'mean_word_length': 3.5570552147239263, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '303899', 'n_tokens_mistral': 2046, 'n_tokens_neox': 1899, 'n_words': 440}