<commit_before>#include <chrono>
#include <cstdio>
#include <unistd.h>
#include <thread>
#include <vector>

#include "raft_go_if.h"
#include "raft_shm.h"

extern "C" {
#include "_cgo_export.h"
}

using namespace raft;

using boost::interprocess::anonymous_instance;

namespace {

void dispatch_apply(CallSlot<ApplyArgs, true>& slot);
void run_worker();

const static uint32_t N_WORKERS = 4;

std::vector<std::thread> workers;

}

void raft_ready()
{
    workers.reserve(N_WORKERS);
    for (uint32_t i = 0; i < N_WORKERS; ++i) {
        workers.emplace_back(run_worker);
    }
    
    raft::scoreboard->is_raft_running = true;
}

namespace {

void run_worker()
{
    fprintf(stderr, "Starting worker.\n");
    for (;;) {
        auto rec = scoreboard->api_queue.take();
        CallTag tag = rec.first;
        BaseSlot::pointer slot = rec.second;
        fprintf(stderr, "API call received, tag %d, call %p.\n",
                tag, slot.get());
        raft::mutex_lock l(slot->owned);
        slot->timings.record("call received");
        assert(slot->state == raft::CallState::Pending);

        switch (tag) {
        case CallTag::Apply:
            dispatch_apply((CallSlot<ApplyArgs, true>&) *slot);
            break;
        default:
            fprintf(stderr, "Unhandled call type: %d\n",
                    tag);
            abort();
        }
    }
}

uintptr_t shm_offset(void* ptr)
{
    uintptr_t shm_base = (uintptr_t) shm.get_address();
    uintptr_t shm_end = (uintptr_t) shm_base + shm.get_size();
    uintptr_t address = (uintptr_t) ptr;
    assert(address >= shm_base && address < shm_end);
    return address - shm_base;
}

void dispatch_apply(CallSlot<ApplyArgs, true>& slot)
{
    assert(slot.tag == CallTag::Apply);
    
    size_t cmd_offset = shm_offset(slot.args.cmd_buf.get());
    fprintf(stderr, "Command offset: %#lx\n", cmd_offset);

    slot.timings.record("RaftApply call");
    RaftApply(&slot, cmd_offset, slot.args.cmd_len, slot.args.timeout_ns);
}

}

void raft_reply_apply(raft_call call_p, uint64_t retval, RaftError error)
{
    auto* slot = (BaseSlot*) call_p;
    mutex_lock lock(slot->owned);
    slot->timings.record("RaftApply return");
    assert(slot->tag == CallTag::Apply);
    slot->retval = retval;
    slot->error = error;
    slot->state = error ? CallState::Error : CallState::Success;

    slot->ret_ready = true;
    slot->ret_cond.notify_one();
    slot->timings.record("result dispatched");
}

uint64_t raft_fsm_apply(uint64_t index, uint64_t term, RaftLogType type,
                        char* cmd_buf, size_t cmd_len)
{
    auto start_t = Timings::clock::now();
    shm_handle cmd_handle;
    char* shm_buf = nullptr;

    if (in_shm_bounds(cmd_buf)) {
        cmd_handle = raft::shm.get_handle_from_address(cmd_buf);
    } else {
        shm_buf = (char*) raft::shm.allocate(cmd_len);
        assert(shm_buf);
        fprintf(stderr, "Allocated %lu-byte buffer for log command at %p.\n",
                cmd_len, shm_buf);
        memcpy(shm_buf, cmd_buf, cmd_len);
        cmd_handle = raft::shm.get_handle_from_address(shm_buf);
    }
    auto cmd_buf_t = Timings::clock::now();

    auto* slot = shm.construct< CallSlot<LogEntry, true> >(anonymous_instance)
        (CallTag::FSMApply, index, term, type, cmd_handle, cmd_len);
    slot->timings = Timings(start_t);
    slot->timings.record("command buffer ready", cmd_buf_t);
    slot->timings.record("slot ready");
    auto rec = slot->rec();
    fprintf(stderr, "Issuing FSMApply, tag %d.\n", rec.first);
    scoreboard->fsm_queue.put(slot->rec());
<commit_msg>Refactor reply code.<commit_after>#include <chrono>
#include <cstdio>
#include <unistd.h>
#include <thread>
#include <vector>

#include "raft_go_if.h"
#include "raft_shm.h"

extern "C" {
#include "_cgo_export.h"
}

using namespace raft;

using boost::interprocess::anonymous_instance;

namespace {

void dispatch_apply(CallSlot<ApplyArgs, true>& slot);
void run_worker();

const static uint32_t N_WORKERS = 4;

std::vector<std::thread> workers;

}

void raft_ready()
{
    workers.reserve(N_WORKERS);
    for (uint32_t i = 0; i < N_WORKERS; ++i) {
        workers.emplace_back(run_worker);
    }
    
    raft::scoreboard->is_raft_running = true;
}

namespace {

void run_worker()
{
    fprintf(stderr, "Starting worker.\n");
    for (;;) {
        auto rec = scoreboard->api_queue.take();
        auto recv_ts = Timings::clock::now();
        CallTag tag = rec.first;
        BaseSlot::pointer slot = rec.second;
        raft::mutex_lock l(slot->owned);
        slot->timings.record("call received", recv_ts);
        slot->timings.record("call locked");
        fprintf(stderr, "API call received, tag %d, call %p.\n",
                tag, slot.get());
        assert(slot->state == raft::CallState::Pending);

        switch (tag) {
        case CallTag::Apply:
            dispatch_apply((CallSlot<ApplyArgs, true>&) *slot);
            break;
        default:
            fprintf(stderr, "Unhandled call type: %d\n",
                    tag);
            abort();
        }
    }
}

uintptr_t shm_offset(void* ptr)
{
    uintptr_t shm_base = (uintptr_t) shm.get_address();
    uintptr_t shm_end = (uintptr_t) shm_base + shm.get_size();
    uintptr_t address = (uintptr_t) ptr;
    assert(address >= shm_base && address < shm_end);
    return address - shm_base;
}

void dispatch_apply(CallSlot<ApplyArgs, true>& slot)
{
    assert(slot.tag == CallTag::Apply);
    
    size_t cmd_offset = shm_offset(slot.args.cmd_buf.get());
    fprintf(stderr, "Command offset: %#lx\n", cmd_offset);

    slot.timings.record("RaftApply call");
    RaftApply(&slot, cmd_offset, slot.args.cmd_len, slot.args.timeout_ns);
}

}

void raft_reply_apply(raft_call call_p, uint64_t retval, RaftError error)
{
    auto* slot = (BaseSlot*) call_p;
    mutex_lock lock(slot->owned);
    slot->timings.record("RaftApply return");
    assert(slot->tag == CallTag::Apply);
    if (!error)
        slot->reply(retval);
    else
        slot->reply(error);
}

uint64_t raft_fsm_apply(uint64_t index, uint64_t term, RaftLogType type,
                        char* cmd_buf, size_t cmd_len)
{
    auto start_t = Timings::clock::now();
    shm_handle cmd_handle;
    char* shm_buf = nullptr;

    if (in_shm_bounds(cmd_buf)) {
        cmd_handle = raft::shm.get_handle_from_address(cmd_buf);
    } else {
        shm_buf = (char*) raft::shm.allocate(cmd_len);
        assert(shm_buf);
        fprintf(stderr, "Allocated %lu-byte buffer for log command at %p.\n",
                cmd_len, shm_buf);
        memcpy(shm_buf, cmd_buf, cmd_len);
        cmd_handle = raft::shm.get_handle_from_address(shm_buf);
    }
    auto cmd_buf_t = Timings::clock::now();

    auto* slot = shm.construct< CallSlot<LogEntry, true> >(anonymous_instance)
        (CallTag::FSMApply, index, term, type, cmd_handle, cmd_len);
    slot->timings = Timings(start_t);
    slot->timings.record("command buffer ready", cmd_buf_t);
    slot->timings.record("slot ready");
    auto rec = slot->rec();
    fprintf(stderr, "Issuing FSMApply, tag %d.\n", rec.first);
    scoreboard->fsm_queue.put(slot->rec());

{'fraction_non_alphanumeric': 0.14225648213034336, 'fraction_numerical': 0.003924316748423266, 'mean_word_length': 3.182883939038687, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 64, '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': '3163763', 'n_tokens_mistral': 2548, 'n_tokens_neox': 2493, 'n_words': 496}