<commit_before>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include "reconfigurable_stateserver.h"
#include "sbenv.h"
#include "remote_check.h"
#include <vespa/vespalib/util/host_name.h>
#include <vespa/vespalib/util/exceptions.h>
#include <vespa/vespalib/stllike/asciistream.h>
#include <vespa/fnet/frt/supervisor.h>
#include <vespa/fnet/transport.h>
#include <vespa/config/helper/configfetcher.h>
#include <thread>
#include <sstream>

#include <vespa/log/log.h>
LOG_SETUP(".slobrok.server.sbenv");

using namespace std::chrono_literals;

namespace slobrok {

namespace {

std::string
createSpec(int port)
{
    if (port == 0) {
        return std::string();
    }
    std::ostringstream str;
    str << "tcp/";
    str << vespalib::HostName::get();
    str << ":";
    str << port;
    return str.str();
}

void
discard(std::vector<std::string> &vec, const std::string & val)
{
    uint32_t i = 0;
    uint32_t size = vec.size();
    while (i < size) {
        if (vec[i] == val) {
            std::swap(vec[i], vec[size - 1]);
            vec.pop_back();
            --size;
        } else {
            ++i;
        }
    }
    LOG_ASSERT(size == vec.size());
}


class ConfigTask : public FNET_Task
{
private:
    Configurator& _configurator;

    ConfigTask(const ConfigTask &);
    ConfigTask &operator=(const ConfigTask &);
public:
    ConfigTask(FNET_Scheduler *sched, Configurator& configurator);

    ~ConfigTask();
    void PerformTask() override;
};


ConfigTask::ConfigTask(FNET_Scheduler *sched, Configurator& configurator)
    : FNET_Task(sched),
      _configurator(configurator)
{
    Schedule(1.0);
}


ConfigTask::~ConfigTask()
{
    Kill();
}


void
ConfigTask::PerformTask()
{
    Schedule(1.0);
    LOG(spam, "checking for new config");
    try {
        _configurator.poll();
    } catch (std::exception &e) {
        LOG(warning, "ConfigTask: poll failed: %s", e.what());
        Schedule(10.0);
    }
}

} // namespace slobrok::<unnamed>

SBEnv::SBEnv(const ConfigShim &shim)
    : _transport(std::make_unique<FNET_Transport>(TransportConfig().drop_empty_buffers(true))),
      _supervisor(std::make_unique<FRT_Supervisor>(_transport.get())),
      _configShim(shim),
      _configurator(shim.factory().create(*this)),
      _shuttingDown(false),
      _partnerList(),
      _me(createSpec(_configShim.portNumber())),
      _rpcHooks(*this, _rpcsrvmap, _rpcsrvmanager),
      _remotechecktask(std::make_unique<RemoteCheck>(getSupervisor()->GetScheduler(), _rpcsrvmap, _rpcsrvmanager, _exchanger)),
      _health(),
      _metrics(_rpcHooks, *_transport),
      _components(),
      _localRpcMonitorMap(*_supervisor),
      _rpcsrvmanager(*this),
      _exchanger(*this, _rpcsrvmap),
      _rpcsrvmap()
{
    srandom(time(nullptr) ^ getpid());
    _localMonitorSubscription = MapSubscription::subscribe(_rpcsrvmap.proxy(), _localRpcMonitorMap);
    _consensusSubscription = MapSubscription::subscribe(_localRpcMonitorMap.dispatcher(), _consensusMap);
    // TODO: use consensus as source here:
    _globalHistorySubscription = MapSubscription::subscribe(_rpcsrvmap.proxy(), _globalVisibleHistory);
    _rpcHooks.initRPC(getSupervisor());
}


SBEnv::~SBEnv() = default;

FNET_Scheduler *
SBEnv::getScheduler() {
    return _transport->GetScheduler();
}

void
SBEnv::shutdown()
{
    _shuttingDown = true;
    getTransport()->ShutDown(false);
}

void
SBEnv::resume()
{
    // nop
}

namespace {

vespalib::string<commit_msg>stop listening to events from the old RpcServerMap, make feedback loop instead<commit_after>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include "reconfigurable_stateserver.h"
#include "sbenv.h"
#include "remote_check.h"
#include <vespa/vespalib/util/host_name.h>
#include <vespa/vespalib/util/exceptions.h>
#include <vespa/vespalib/stllike/asciistream.h>
#include <vespa/fnet/frt/supervisor.h>
#include <vespa/fnet/transport.h>
#include <vespa/config/helper/configfetcher.h>
#include <thread>
#include <sstream>

#include <vespa/log/log.h>
LOG_SETUP(".slobrok.server.sbenv");

using namespace std::chrono_literals;

namespace slobrok {

namespace {

std::string
createSpec(int port)
{
    if (port == 0) {
        return std::string();
    }
    std::ostringstream str;
    str << "tcp/";
    str << vespalib::HostName::get();
    str << ":";
    str << port;
    return str.str();
}

void
discard(std::vector<std::string> &vec, const std::string & val)
{
    uint32_t i = 0;
    uint32_t size = vec.size();
    while (i < size) {
        if (vec[i] == val) {
            std::swap(vec[i], vec[size - 1]);
            vec.pop_back();
            --size;
        } else {
            ++i;
        }
    }
    LOG_ASSERT(size == vec.size());
}


class ConfigTask : public FNET_Task
{
private:
    Configurator& _configurator;

    ConfigTask(const ConfigTask &);
    ConfigTask &operator=(const ConfigTask &);
public:
    ConfigTask(FNET_Scheduler *sched, Configurator& configurator);

    ~ConfigTask();
    void PerformTask() override;
};


ConfigTask::ConfigTask(FNET_Scheduler *sched, Configurator& configurator)
    : FNET_Task(sched),
      _configurator(configurator)
{
    Schedule(1.0);
}


ConfigTask::~ConfigTask()
{
    Kill();
}


void
ConfigTask::PerformTask()
{
    Schedule(1.0);
    LOG(spam, "checking for new config");
    try {
        _configurator.poll();
    } catch (std::exception &e) {
        LOG(warning, "ConfigTask: poll failed: %s", e.what());
        Schedule(10.0);
    }
}

} // namespace slobrok::<unnamed>

SBEnv::SBEnv(const ConfigShim &shim)
    : _transport(std::make_unique<FNET_Transport>(TransportConfig().drop_empty_buffers(true))),
      _supervisor(std::make_unique<FRT_Supervisor>(_transport.get())),
      _configShim(shim),
      _configurator(shim.factory().create(*this)),
      _shuttingDown(false),
      _partnerList(),
      _me(createSpec(_configShim.portNumber())),
      _rpcHooks(*this, _rpcsrvmap, _rpcsrvmanager),
      _remotechecktask(std::make_unique<RemoteCheck>(getSupervisor()->GetScheduler(), _rpcsrvmap, _rpcsrvmanager, _exchanger)),
      _health(),
      _metrics(_rpcHooks, *_transport),
      _components(),
      _localRpcMonitorMap(*_supervisor),
      _rpcsrvmanager(*this),
      _exchanger(*this, _rpcsrvmap),
      _rpcsrvmap()
{
    srandom(time(nullptr) ^ getpid());
    // note: feedback loop between these two:
    _localMonitorSubscription = MapSubscription::subscribe(_consensusMap, _localRpcMonitorMap);
    _consensusSubscription = MapSubscription::subscribe(_localRpcMonitorMap.dispatcher(), _consensusMap);
    // TODO: use consensus as source here:
    _globalHistorySubscription = MapSubscription::subscribe(_rpcsrvmap.proxy(), _globalVisibleHistory);
    _rpcHooks.initRPC(getSupervisor());
}


SBEnv::~SBEnv() = default;

FNET_Scheduler *
SBEnv::getScheduler() {
    return _transport->GetScheduler();
}

void
SBEnv::shutdown()
{
    _shuttingDown = true;
    getTransport()->ShutDown(false);
}

void
SBEnv::resume()
{
    // nop
}

namespace {

vespalib::string
{'fraction_non_alphanumeric': 0.14513422818791946, 'fraction_numerical': 0.005592841163310962, 'mean_word_length': 3.8926128590971274, 'pattern_counts': {'":': 2, '<': 49, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5703172', 'n_tokens_mistral': 2474, 'n_tokens_neox': 2453, 'n_words': 434}