<commit_before>/*
 * Application registration support for normal uipcps.
 *
 * Copyright (C) 2015-2016 Nextworks
 * Author: <NAME> <<EMAIL>>
 *
 * This file is part of rlite.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
 */

#include <ctime>
#include <iterator>
#include <cstdlib>

#include "uipcp-normal.hpp"

using namespace std;

static uint64_t
time64()
{
    struct timespec tv;

    if (clock_gettime(CLOCK_MONOTONIC, &tv)) {
        perror("clock_gettime() failed");
        tv.tv_sec  = 0;
        tv.tv_nsec = 0;
    }

    return (tv.tv_sec << 32) | (tv.tv_nsec & ((1L << 32) - 1L));
}

class FullyReplicatedDFT : public DFT {
    /* Directory Forwarding Table, mapping application name (std::string)
     * to a set of nodes that registered that name. All nodes are considered
     * equivalent. */
    std::multimap<std::string, DFTEntry> dft_table;

public:
    RL_NODEFAULT_NONCOPIABLE(FullyReplicatedDFT);
    FullyReplicatedDFT(struct uipcp_rib *_ur) : DFT(_ur) {}
    ~FullyReplicatedDFT() {}

    void dump(std::stringstream &ss) const override;

    int lookup_req(const std::string &appl_name, rlm_addr_t *dstaddr,
                   const rlm_addr_t preferred, uint32_t cookie) override;
    int appl_register(const struct rl_kmsg_appl_register *req) override;
    void update_address(rlm_addr_t new_addr) override;
    int rib_handler(const CDAPMessage *rm, NeighFlow *nf,
                    rlm_addr_t src_addr) override;
    int sync_neigh(NeighFlow *nf, unsigned int limit) const override;
    int neighs_refresh(size_t limit) override;

    /* Helper function shared with CentralizedFaultTolerantDFT::Replica. */
    void mod_table(const DFTEntry &e, bool add, DFTSlice *added,
                   DFTSlice *removed);
};

int
FullyReplicatedDFT::lookup_req(const std::string &appl_name,
                               rlm_addr_t *dstaddr, const rlm_addr_t preferred,
                               uint32_t cookie)
{
    /* Fetch all entries that hold 'appl_name'. */
    auto range = dft_table.equal_range(appl_name);
    int d;

    d = distance(range.first, range.second);

    if (d == 0) {
        /* No entry. */
        return -1;
    }

    auto mit = range.first;

    if (d > 1) {
        if (preferred != RL_ADDR_NULL) {
            /* Only accept the preferred address. */

            for (; mit != range.second; mit++) {
                if (mit->second.address == preferred) {
                    break;
                }
            }
        } else {
            /* Load balance by selecting an entry based on
             * the cookie value. */
            cookie %= d;
            while (cookie > 0 && mit != range.second) {
                mit++;
                cookie--;
            }
        }

        assert(mit != range.second);
    }

    *dstaddr = mit->second.address;

    return 0;
}

int
FullyReplicatedDFT::appl_register(const struct rl_kmsg_appl_register *req)
{
    multimap<string, DFTEntry>::iterator mit;
    string appl_name(req->appl_name);
    struct uipcp *uipcp = rib->uipcp;
    DFTSlice dft_slice;
    DFTEntry dft_entry;

    dft_entry.address   = rib->myaddr;
    dft_entry.appl_name = RinaName(appl_name);
    dft_entry.timestamp = time64();
    dft_entry.local     = true;

    /* Get all the entries for 'appl_name', and see if there
     * is an entry associated to this uipcp. */
    auto range = dft_table.equal_range(appl_name);
    for (mit = range.first; mit != range.second; mit++) {
        if (mit->second.address == rib->myaddr) {
            break;
        }
    }

    if (req->reg) {
        if (mit != range.second) { /* local collision */
            UPE(uipcp, "Application %s already registered on this uipcp\n",
                appl_name.c_str());
            return uipcp_appl_register_resp(uipcp, RLITE_ERR, req->event_id,
                                            req->appl_name);
        }

        if (req->reg) {
            /* Registration requires a response, while unregistrations doesn't.
             * Respond to the client before committing to the RIB, because the
             * response may fail. */
            int ret = uipcp_appl_register_resp(uipcp, RLITE_SUCC, req->event_id,
                                               req->appl_name);
            if (ret) {
                return ret;
            }
        }

        /* Insert the object into the RIB. */
        dft_table.insert(make_pair(appl_name, dft_entry));
    } else {
        if (mit == range.second) {
            UPE(uipcp, "Application %s was not registered here\n",
                appl_name.c_str());
            return 0;
        }

        /* Remove from the RIB. */
        dft_table.erase(mit);
    }

    dft_slice.entries.push_back(dft_entry);

    UPD(uipcp, "Application %s %sregistered\n", appl_name.c_str(),
        req->reg ? "" : "un");

    rib->neighs_sync_obj_all(req->reg != 0, obj_class::dft, obj_name::dft,
                             &dft_slice);

    return 0;
}

/* Tries ot add or remove an entry 'e' from the DFT multimap. If not nullptr,
 * the entries added and/or removed are appended to 'added' and 'removed'
 * respectively. */
void
FullyReplicatedDFT::mod_table(const DFTEntry &e, bool add, DFTSlice *added,
                              DFTSlice *removed)
{
    string key = static_cast<string>(e.appl_name);
    auto range = dft_table.equal_range(key);
    multimap<string, DFTEntry>::iterator mit;
    struct uipcp *uipcp = rib->uipcp;

    for (mit = range.first; mit != range.second; mit++) {
        if (mit->second.address == e.address) {
            break;
        }
    }

    if (add) {
        bool collision = (mit != range.second);

        if (!collision || e.timestamp > mit->second.timestamp) {
            if (collision) {
                /* Remove the collided entry. */
                if (removed) {
                    removed->entries.push_back(mit->second);
                }
                dft_table.erase(mit);
            }
            dft_table.insert(make_pair(key, e));
            if (added) {
                added->entries.push_back(e);
            }
            UPD(uipcp, "DFT entry %s --> %lu %s remotely\n", key.c_str(),
                e.address, (collision ? "updated" : "added"));
        }

    } else {
        if (mit == range.second) {
            UPI(uipcp, "DFT entry does not exist\n");
        } else {
            dft_table.erase(mit);
            if (removed) {
                removed->entries.push_back(e);
            }
            UPD(uipcp, "DFT entry %s --> %lu removed remotely\n", key.c_str(),
                e.address);
        }
    }
}

int
FullyReplicatedDFT::rib_handler(const CDAPMessage *rm, NeighFlow *nf,
                                rlm_addr_t src_addr)
{
    struct uipcp *uipcp = rib->uipcp;
    const char *objbuf;
    size_t objlen;
    bool add = true;

    if (rm->op_code != gpb::M_CREATE && rm->op_code != gpb::M_DELETE) {
        UPE(uipcp, "M_CREATE or M_DELETE expected\n");
        return 0;
    }

    if (rm->op_code == gpb::M_DELETE) {
        add = false;
    }

    rm->get_obj_value(objbuf, objlen);
    if (!objbuf) {
        UPE(uipcp, "M_START does not contain a nested message\n");
        abort();
        return 0;
    }

    DFTSlice dft_slice(objbuf, objlen);
    DFTSlice prop_dft_add, prop_dft_del;

    for (const DFTEntry &e : dft_slice.entries) {
        mod_table(e, add, &prop_dft_add, &prop_dft_del);
    }

    /* Propagate the DFT entries update to the other neighbors,
     * except for who told us. */
    if (prop_dft_add.entries.size()) {
        rib->neighs_sync_obj_excluding(nf->neigh, true, obj_class::dft,
                                       obj_name::dft, &prop_dft_add);
    }

    if (prop_dft_del.entries.size()) {
        rib->neighs_sync_obj_excluding(nf->neigh, false, obj_class::dft,
                                       obj_name::dft, &prop_dft_del);
    }

    return 0;
}

void
FullyReplicatedDFT::update_address(rlm_addr_t new_addr)
{
    DFTSlice prop_dft;
    DFTSlice del_dft;

    /* Update all the DFT entries corresponding to application that are
     * registered within us. */
    for (auto &kve : dft_table) {
        if (kve.second.local && kve.second.address == rib->myaddr) {
            del_dft.entries.push_back(kve.second);
            kve.second.address   = new_addr;
            kve.second.timestamp = time64();
            prop_dft.entries.push_back(kve.second);
            UPD(rib->uipcp, "Updated address for DFT entry %s\n",
                kve.first.c_str());
        }
    }

    /* Disseminate the update. */
    if (prop_dft.entries.size()) {
        rib->neighs_sync_obj_all(true, obj_class::dft, obj_name::dft,
                                 &prop_dft);
    }

    if (del_dft.entries.size()) {
        rib->neighs_sync_obj_all(false, obj_class::dft, obj_name::dft,
                                 &del_dft);
    }
}

void
FullyReplicatedDFT::dump(stringstream &ss) const
{
    ss << "Directory Forwarding Table:" << endl;
    for (const auto &kve : dft_table) {
        const DFTEntry &entry = kve.second;

        ss << "    Application: " << static_cast<string>(entry.appl_name)
           << ", Address: " << entry.address
           << ", Timestamp: " << entry.timestamp << endl;
    }

    ss << endl;
}

int
FullyReplicatedDFT::sync_neigh(NeighFlow *nf, unsigned int limit) const
{
    int ret = 0;

    for (auto eit = dft_table.begin(); eit != dft_table.end();) {
        DFTSlice dft_slice;

        while (dft_slice.entries.size() < limit && eit != dft_table.end()) {
            dft_slice.entries.push_back(eit->second);
            eit++;
        }

        ret |= nf->neigh->neigh_sync_obj(nf, true, obj_class::dft,
                                         obj_name::dft, &dft_slice);
    }

    return ret;
}

/* Propagate local entries (i.e. the ones corresponding to locally
 * registered applications) to all our neighbors. */
int
FullyReplicatedDFT::neighs_refresh(size_t limit)
{
    int ret = 0;

    for (auto eit = dft_table.begin(); eit != dft_table.end();) {
        DFTSlice dft_slice;

        while (dft_slice.entries.size() < limit && eit != dft_table.end()) {
            if (eit->second.local) {
                dft_slice.entries.push_back(eit->second);
            }
            eit++;
        }

        if (dft_slice.entries.size()) {
            ret |= rib->neighs_sync_obj_all(true, obj_class::dft, obj_name::dft,
                                            &dft_slice);
        }
    }

    return ret;
}

class CentralizedFaultTolerantDFT : public DFT {
    /* An instance of this class can be a state machine replica or it can just
     * be a client that will redirect requests to one of the replicas. */

    /* In case of state machine replica, a pointer to a Raft state
     * machine. */
    class Replica : public raft::RaftSM {
        CentralizedFaultTolerantDFT *parent = nullptr;

        /* The structure of a DFT command (i.e. a log entry for the Raft SM). */
        struct Command {
            rlm_addr_t address;
            char name[63];
            uint8_t opcode;
            static constexpr uint8_t OpcodeSet = 1;
            static constexpr uint8_t OpcodeDel = 2;
        } __attribute__((packed));
        static_assert(sizeof(struct Command) == sizeof(Command::address) +
                                                    sizeof(Command::name) +
                                                    sizeof(Command::opcode),
                      "Invalid memory layout for class Replica::Command");

        /* Timer needed by the Raft state machine. */
        std::unique_ptr<TimeoutEvent> timer;
        raft::RaftTimerType timer_type;

        /* State machine implementation. Just reuse the implementation of
         * a fully replicated DFT. */
        std::unique_ptr<FullyReplicatedDFT> impl;

    public:
        Replica(CentralizedFaultTolerantDFT *dft)
            : raft::RaftSM(std::string("ceft-dft-") + dft->rib->myname,
                           dft->rib->myname,
                           std::string("/tmp/ceft-dft-") +
                               std::to_string(dft->rib->uipcp->id) +
                               std::string("-") + dft->rib->myname,
                           sizeof(Command), std::cerr, std::cout),
              parent(dft),
              impl(make_unique<FullyReplicatedDFT>(dft->rib)){};
        int process_sm_output(raft::RaftSMOutput out);
        int process_timeout();
        int apply(const char *const serbuf) override;
        int lookup_req(const std::string &appl_name, rlm_addr_t *dstaddr,
                       const rlm_addr_t preferred, uint32_t cookie)
        {
            return impl->lookup_req(appl_name, dstaddr, preferred, cookie);
        }
        int appl_register(const struct rl_kmsg_appl_register *req);
        int rib_handler(const CDAPMessage *rm, NeighFlow *nf,
                        rlm_addr_t src_addr);
        void dump(std::stringstream &ss) const { impl->dump(ss); };
    };
    std::unique_ptr<Replica> raft;

    /* In case of client, a pointer to client-side data structures. */
    class Client {
        CentralizedFaultTolerantDFT *parent = nullptr;
        std::list<raft::ReplicaId> replicas;
        /* The leader, if we know who it is, otherwise the empty
         * string. */
        raft::ReplicaId leader_id;
        std::unique_ptr<TimeoutEvent> timer;

        struct PendingReq {
            gpb::opCode_t op_code;
            std::string appl_name;
            raft::ReplicaId replica;
            uint32_t kevent_id;
            std::chrono::system_clock::time_point t;
            PendingReq() = default;
            PendingReq(gpb::opCode_t op, const std::string &a,
                       const raft::ReplicaId &r, uint32_t event_id)
                : op_code(op), appl_name(a), replica(r), kevent_id(event_id)
            {
                t = std::chrono::system_clock::now();
            }
        };
        std::unordered_map</*invoke_id*/ int, PendingReq> pending;

        void rearm_pending_timer();

    public:
        Client(CentralizedFaultTolerantDFT *dft,
               std::list<raft::ReplicaId> names)
            : parent(dft), replicas(std::move(names))
        {
        }
        int lookup_req(const std::string &appl_name, rlm_addr_t *dstaddr,
                       const rlm_addr_t preferred, uint32_t cookie);
        int appl_register(const struct rl_kmsg_appl_register *req);
        int rib_handler(const CDAPMessage *rm, NeighFlow *nf,
                        rlm_addr_t src_addr);
        int process_timeout();
    };
    std::unique_ptr<Client> client;

public:
    RL_NODEFAULT_NONCOPIABLE(CentralizedFaultTolerantDFT);
    CentralizedFaultTolerantDFT(struct uipcp_rib *_ur) : DFT(_ur) {}
    int param_changed(const std::string &param_name) override;
    void dump(std::stringstream &ss) const override
    {
        if (client) {
            ss << "Directory Forwarding Table: not available locally" << endl
               << endl;
        } else {
            raft->dump(ss);
        }
    }

    int lookup_req(const std::string &appl_name, rlm_addr_t *dstaddr,
                   const rlm_addr_t preferred, uint32_t cookie) override
    {
        if (client) {
            return client->lookup_req(appl_name, dstaddr, preferred, cookie);
        }
        return raft->lookup_req(appl_name, dstaddr, preferred, cookie);
    }
    int appl_register(const struct rl_kmsg_appl_register *req) override
    {
        if (client) {
            return client->appl_register(req);
        }
        return raft->appl_register(req);
    }
    void update_address(rlm_addr_t new_addr) override;
    int rib_handler(const CDAPMessage *rm, NeighFlow *nf,
                    rlm_addr_t src_addr) override
    {
        if (client) {
            return client->rib_handler(rm, nf, src_addr);
        }
        return raft->rib_handler(rm, nf, src_addr);
    }
};

int
CentralizedFaultTolerantDFT::param_changed(const std::string &param_name)
{
    list<raft::ReplicaId> peers;
    string param_value;

    if (param_name != "replicas") {
        return -1;
    }

    param_value = rib->get_param_value<std::string>("dft", "replicas");
    UPD(rib->uipcp, "replicas = %s\n", param_value.c_str());
    peers = strsplit(param_value, ',');

    auto it = peers.begin();
    for (; it != peers.end(); it++) {
        if (*it == rib->myname) {
            /* I'm one of the replicas. Create a Raft state machine and
             * initialize it. */
            client.reset();
            raft = make_unique<Replica>(this);
            peers.erase(it); /* remove myself */

            raft->set_election_timeout(std::chrono::milliseconds(1000),
                                       std::chrono::milliseconds(2000));
            raft->set_heartbeat_timeout(std::chrono::milliseconds(500));

            raft::RaftSMOutput out;
            if (raft->init(peers, &out)) {
                UPE(rib->uipcp, "Failed to init Raft state machine for DFT\n");
                return -1;
            }
            UPI(rib->uipcp, "Raft replica initialized\n");<commit_msg>uipcps: ceft dft: set raft log level and timeout<commit_after>/*
 * Application registration support for normal uipcps.
 *
 * Copyright (C) 2015-2016 Nextworks
 * Author: <NAME> <<EMAIL>>
 *
 * This file is part of rlite.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
 */

#include <ctime>
#include <iterator>
#include <cstdlib>

#include "uipcp-normal.hpp"

using namespace std;

static uint64_t
time64()
{
    struct timespec tv;

    if (clock_gettime(CLOCK_MONOTONIC, &tv)) {
        perror("clock_gettime() failed");
        tv.tv_sec  = 0;
        tv.tv_nsec = 0;
    }

    return (tv.tv_sec << 32) | (tv.tv_nsec & ((1L << 32) - 1L));
}

class FullyReplicatedDFT : public DFT {
    /* Directory Forwarding Table, mapping application name (std::string)
     * to a set of nodes that registered that name. All nodes are considered
     * equivalent. */
    std::multimap<std::string, DFTEntry> dft_table;

public:
    RL_NODEFAULT_NONCOPIABLE(FullyReplicatedDFT);
    FullyReplicatedDFT(struct uipcp_rib *_ur) : DFT(_ur) {}
    ~FullyReplicatedDFT() {}

    void dump(std::stringstream &ss) const override;

    int lookup_req(const std::string &appl_name, rlm_addr_t *dstaddr,
                   const rlm_addr_t preferred, uint32_t cookie) override;
    int appl_register(const struct rl_kmsg_appl_register *req) override;
    void update_address(rlm_addr_t new_addr) override;
    int rib_handler(const CDAPMessage *rm, NeighFlow *nf,
                    rlm_addr_t src_addr) override;
    int sync_neigh(NeighFlow *nf, unsigned int limit) const override;
    int neighs_refresh(size_t limit) override;

    /* Helper function shared with CentralizedFaultTolerantDFT::Replica. */
    void mod_table(const DFTEntry &e, bool add, DFTSlice *added,
                   DFTSlice *removed);
};

int
FullyReplicatedDFT::lookup_req(const std::string &appl_name,
                               rlm_addr_t *dstaddr, const rlm_addr_t preferred,
                               uint32_t cookie)
{
    /* Fetch all entries that hold 'appl_name'. */
    auto range = dft_table.equal_range(appl_name);
    int d;

    d = distance(range.first, range.second);

    if (d == 0) {
        /* No entry. */
        return -1;
    }

    auto mit = range.first;

    if (d > 1) {
        if (preferred != RL_ADDR_NULL) {
            /* Only accept the preferred address. */

            for (; mit != range.second; mit++) {
                if (mit->second.address == preferred) {
                    break;
                }
            }
        } else {
            /* Load balance by selecting an entry based on
             * the cookie value. */
            cookie %= d;
            while (cookie > 0 && mit != range.second) {
                mit++;
                cookie--;
            }
        }

        assert(mit != range.second);
    }

    *dstaddr = mit->second.address;

    return 0;
}

int
FullyReplicatedDFT::appl_register(const struct rl_kmsg_appl_register *req)
{
    multimap<string, DFTEntry>::iterator mit;
    string appl_name(req->appl_name);
    struct uipcp *uipcp = rib->uipcp;
    DFTSlice dft_slice;
    DFTEntry dft_entry;

    dft_entry.address   = rib->myaddr;
    dft_entry.appl_name = RinaName(appl_name);
    dft_entry.timestamp = time64();
    dft_entry.local     = true;

    /* Get all the entries for 'appl_name', and see if there
     * is an entry associated to this uipcp. */
    auto range = dft_table.equal_range(appl_name);
    for (mit = range.first; mit != range.second; mit++) {
        if (mit->second.address == rib->myaddr) {
            break;
        }
    }

    if (req->reg) {
        if (mit != range.second) { /* local collision */
            UPE(uipcp, "Application %s already registered on this uipcp\n",
                appl_name.c_str());
            return uipcp_appl_register_resp(uipcp, RLITE_ERR, req->event_id,
                                            req->appl_name);
        }

        if (req->reg) {
            /* Registration requires a response, while unregistrations doesn't.
             * Respond to the client before committing to the RIB, because the
             * response may fail. */
            int ret = uipcp_appl_register_resp(uipcp, RLITE_SUCC, req->event_id,
                                               req->appl_name);
            if (ret) {
                return ret;
            }
        }

        /* Insert the object into the RIB. */
        dft_table.insert(make_pair(appl_name, dft_entry));
    } else {
        if (mit == range.second) {
            UPE(uipcp, "Application %s was not registered here\n",
                appl_name.c_str());
            return 0;
        }

        /* Remove from the RIB. */
        dft_table.erase(mit);
    }

    dft_slice.entries.push_back(dft_entry);

    UPD(uipcp, "Application %s %sregistered\n", appl_name.c_str(),
        req->reg ? "" : "un");

    rib->neighs_sync_obj_all(req->reg != 0, obj_class::dft, obj_name::dft,
                             &dft_slice);

    return 0;
}

/* Tries ot add or remove an entry 'e' from the DFT multimap. If not nullptr,
 * the entries added and/or removed are appended to 'added' and 'removed'
 * respectively. */
void
FullyReplicatedDFT::mod_table(const DFTEntry &e, bool add, DFTSlice *added,
                              DFTSlice *removed)
{
    string key = static_cast<string>(e.appl_name);
    auto range = dft_table.equal_range(key);
    multimap<string, DFTEntry>::iterator mit;
    struct uipcp *uipcp = rib->uipcp;

    for (mit = range.first; mit != range.second; mit++) {
        if (mit->second.address == e.address) {
            break;
        }
    }

    if (add) {
        bool collision = (mit != range.second);

        if (!collision || e.timestamp > mit->second.timestamp) {
            if (collision) {
                /* Remove the collided entry. */
                if (removed) {
                    removed->entries.push_back(mit->second);
                }
                dft_table.erase(mit);
            }
            dft_table.insert(make_pair(key, e));
            if (added) {
                added->entries.push_back(e);
            }
            UPD(uipcp, "DFT entry %s --> %lu %s remotely\n", key.c_str(),
                e.address, (collision ? "updated" : "added"));
        }

    } else {
        if (mit == range.second) {
            UPI(uipcp, "DFT entry does not exist\n");
        } else {
            dft_table.erase(mit);
            if (removed) {
                removed->entries.push_back(e);
            }
            UPD(uipcp, "DFT entry %s --> %lu removed remotely\n", key.c_str(),
                e.address);
        }
    }
}

int
FullyReplicatedDFT::rib_handler(const CDAPMessage *rm, NeighFlow *nf,
                                rlm_addr_t src_addr)
{
    struct uipcp *uipcp = rib->uipcp;
    const char *objbuf;
    size_t objlen;
    bool add = true;

    if (rm->op_code != gpb::M_CREATE && rm->op_code != gpb::M_DELETE) {
        UPE(uipcp, "M_CREATE or M_DELETE expected\n");
        return 0;
    }

    if (rm->op_code == gpb::M_DELETE) {
        add = false;
    }

    rm->get_obj_value(objbuf, objlen);
    if (!objbuf) {
        UPE(uipcp, "M_START does not contain a nested message\n");
        abort();
        return 0;
    }

    DFTSlice dft_slice(objbuf, objlen);
    DFTSlice prop_dft_add, prop_dft_del;

    for (const DFTEntry &e : dft_slice.entries) {
        mod_table(e, add, &prop_dft_add, &prop_dft_del);
    }

    /* Propagate the DFT entries update to the other neighbors,
     * except for who told us. */
    if (prop_dft_add.entries.size()) {
        rib->neighs_sync_obj_excluding(nf->neigh, true, obj_class::dft,
                                       obj_name::dft, &prop_dft_add);
    }

    if (prop_dft_del.entries.size()) {
        rib->neighs_sync_obj_excluding(nf->neigh, false, obj_class::dft,
                                       obj_name::dft, &prop_dft_del);
    }

    return 0;
}

void
FullyReplicatedDFT::update_address(rlm_addr_t new_addr)
{
    DFTSlice prop_dft;
    DFTSlice del_dft;

    /* Update all the DFT entries corresponding to application that are
     * registered within us. */
    for (auto &kve : dft_table) {
        if (kve.second.local && kve.second.address == rib->myaddr) {
            del_dft.entries.push_back(kve.second);
            kve.second.address   = new_addr;
            kve.second.timestamp = time64();
            prop_dft.entries.push_back(kve.second);
            UPD(rib->uipcp, "Updated address for DFT entry %s\n",
                kve.first.c_str());
        }
    }

    /* Disseminate the update. */
    if (prop_dft.entries.size()) {
        rib->neighs_sync_obj_all(true, obj_class::dft, obj_name::dft,
                                 &prop_dft);
    }

    if (del_dft.entries.size()) {
        rib->neighs_sync_obj_all(false, obj_class::dft, obj_name::dft,
                                 &del_dft);
    }
}

void
FullyReplicatedDFT::dump(stringstream &ss) const
{
    ss << "Directory Forwarding Table:" << endl;
    for (const auto &kve : dft_table) {
        const DFTEntry &entry = kve.second;

        ss << "    Application: " << static_cast<string>(entry.appl_name)
           << ", Address: " << entry.address
           << ", Timestamp: " << entry.timestamp << endl;
    }

    ss << endl;
}

int
FullyReplicatedDFT::sync_neigh(NeighFlow *nf, unsigned int limit) const
{
    int ret = 0;

    for (auto eit = dft_table.begin(); eit != dft_table.end();) {
        DFTSlice dft_slice;

        while (dft_slice.entries.size() < limit && eit != dft_table.end()) {
            dft_slice.entries.push_back(eit->second);
            eit++;
        }

        ret |= nf->neigh->neigh_sync_obj(nf, true, obj_class::dft,
                                         obj_name::dft, &dft_slice);
    }

    return ret;
}

/* Propagate local entries (i.e. the ones corresponding to locally
 * registered applications) to all our neighbors. */
int
FullyReplicatedDFT::neighs_refresh(size_t limit)
{
    int ret = 0;

    for (auto eit = dft_table.begin(); eit != dft_table.end();) {
        DFTSlice dft_slice;

        while (dft_slice.entries.size() < limit && eit != dft_table.end()) {
            if (eit->second.local) {
                dft_slice.entries.push_back(eit->second);
            }
            eit++;
        }

        if (dft_slice.entries.size()) {
            ret |= rib->neighs_sync_obj_all(true, obj_class::dft, obj_name::dft,
                                            &dft_slice);
        }
    }

    return ret;
}

class CentralizedFaultTolerantDFT : public DFT {
    /* An instance of this class can be a state machine replica or it can just
     * be a client that will redirect requests to one of the replicas. */

    /* In case of state machine replica, a pointer to a Raft state
     * machine. */
    class Replica : public raft::RaftSM {
        CentralizedFaultTolerantDFT *parent = nullptr;

        /* The structure of a DFT command (i.e. a log entry for the Raft SM). */
        struct Command {
            rlm_addr_t address;
            char name[63];
            uint8_t opcode;
            static constexpr uint8_t OpcodeSet = 1;
            static constexpr uint8_t OpcodeDel = 2;
        } __attribute__((packed));
        static_assert(sizeof(struct Command) == sizeof(Command::address) +
                                                    sizeof(Command::name) +
                                                    sizeof(Command::opcode),
                      "Invalid memory layout for class Replica::Command");

        /* Timer needed by the Raft state machine. */
        std::unique_ptr<TimeoutEvent> timer;
        raft::RaftTimerType timer_type;

        /* State machine implementation. Just reuse the implementation of
         * a fully replicated DFT. */
        std::unique_ptr<FullyReplicatedDFT> impl;

    public:
        Replica(CentralizedFaultTolerantDFT *dft)
            : raft::RaftSM(std::string("ceft-dft-") + dft->rib->myname,
                           dft->rib->myname,
                           std::string("/tmp/ceft-dft-") +
                               std::to_string(dft->rib->uipcp->id) +
                               std::string("-") + dft->rib->myname,
                           sizeof(Command), std::cerr, std::cout),
              parent(dft),
              impl(make_unique<FullyReplicatedDFT>(dft->rib)){};
        int process_sm_output(raft::RaftSMOutput out);
        int process_timeout();
        int apply(const char *const serbuf) override;
        int lookup_req(const std::string &appl_name, rlm_addr_t *dstaddr,
                       const rlm_addr_t preferred, uint32_t cookie)
        {
            return impl->lookup_req(appl_name, dstaddr, preferred, cookie);
        }
        int appl_register(const struct rl_kmsg_appl_register *req);
        int rib_handler(const CDAPMessage *rm, NeighFlow *nf,
                        rlm_addr_t src_addr);
        void dump(std::stringstream &ss) const { impl->dump(ss); };
    };
    std::unique_ptr<Replica> raft;

    /* In case of client, a pointer to client-side data structures. */
    class Client {
        CentralizedFaultTolerantDFT *parent = nullptr;
        std::list<raft::ReplicaId> replicas;
        /* The leader, if we know who it is, otherwise the empty
         * string. */
        raft::ReplicaId leader_id;
        std::unique_ptr<TimeoutEvent> timer;

        struct PendingReq {
            gpb::opCode_t op_code;
            std::string appl_name;
            raft::ReplicaId replica;
            uint32_t kevent_id;
            std::chrono::system_clock::time_point t;
            PendingReq() = default;
            PendingReq(gpb::opCode_t op, const std::string &a,
                       const raft::ReplicaId &r, uint32_t event_id)
                : op_code(op), appl_name(a), replica(r), kevent_id(event_id)
            {
                t = std::chrono::system_clock::now();
            }
        };
        std::unordered_map</*invoke_id*/ int, PendingReq> pending;

        void rearm_pending_timer();

    public:
        Client(CentralizedFaultTolerantDFT *dft,
               std::list<raft::ReplicaId> names)
            : parent(dft), replicas(std::move(names))
        {
        }
        int lookup_req(const std::string &appl_name, rlm_addr_t *dstaddr,
                       const rlm_addr_t preferred, uint32_t cookie);
        int appl_register(const struct rl_kmsg_appl_register *req);
        int rib_handler(const CDAPMessage *rm, NeighFlow *nf,
                        rlm_addr_t src_addr);
        int process_timeout();
    };
    std::unique_ptr<Client> client;

public:
    RL_NODEFAULT_NONCOPIABLE(CentralizedFaultTolerantDFT);
    CentralizedFaultTolerantDFT(struct uipcp_rib *_ur) : DFT(_ur) {}
    int param_changed(const std::string &param_name) override;
    void dump(std::stringstream &ss) const override
    {
        if (client) {
            ss << "Directory Forwarding Table: not available locally" << endl
               << endl;
        } else {
            raft->dump(ss);
        }
    }

    int lookup_req(const std::string &appl_name, rlm_addr_t *dstaddr,
                   const rlm_addr_t preferred, uint32_t cookie) override
    {
        if (client) {
            return client->lookup_req(appl_name, dstaddr, preferred, cookie);
        }
        return raft->lookup_req(appl_name, dstaddr, preferred, cookie);
    }
    int appl_register(const struct rl_kmsg_appl_register *req) override
    {
        if (client) {
            return client->appl_register(req);
        }
        return raft->appl_register(req);
    }
    void update_address(rlm_addr_t new_addr) override;
    int rib_handler(const CDAPMessage *rm, NeighFlow *nf,
                    rlm_addr_t src_addr) override
    {
        if (client) {
            return client->rib_handler(rm, nf, src_addr);
        }
        return raft->rib_handler(rm, nf, src_addr);
    }
};

int
CentralizedFaultTolerantDFT::param_changed(const std::string &param_name)
{
    list<raft::ReplicaId> peers;
    string param_value;

    if (param_name != "replicas") {
        return -1;
    }

    param_value = rib->get_param_value<std::string>("dft", "replicas");
    UPD(rib->uipcp, "replicas = %s\n", param_value.c_str());
    peers = strsplit(param_value, ',');

    auto it = peers.begin();
    for (; it != peers.end(); it++) {
        if (*it == rib->myname) {
            /* I'm one of the replicas. Create a Raft state machine and
             * initialize it. */
            client.reset();
            raft = make_unique<Replica>(this);
            peers.erase(it); /* remove myself */

            raft->set_election_timeout(std::chrono::milliseconds(1000),
                                       std::chrono::milliseconds(2000));
            raft->set_heartbeat_timeout(std::chrono::milliseconds(100));
            raft->set_verbosity(raft::RaftSM::kVerboseInfo);

            raft::RaftSMOutput out;
            if (raft->init(peers, &out)) {
                UPE(rib->uipcp, "Failed to init Raft state machine for DFT\n");
                return -1;
            }
            UPI(rib->uipcp, "Raft replica initialized\n");
{'fraction_non_alphanumeric': 0.11259221998658618, 'fraction_numerical': 0.004694835680751174, 'mean_word_length': 2.0200860832137733, 'pattern_counts': {'":': 0, '<': 113, '<?xml version=': 0, '>': 198, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 34, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5985747', 'n_tokens_mistral': 11469, 'n_tokens_neox': 11074, 'n_words': 2670}