使用C++20协程实现Raft共识算法

模板<类型名称 TSocket>
TValueTask<空>
TMessageWriter::Write(TMessageHolder 消息) {
    co_await TByteWriter(Socket).Write(message.Mes, message->Len);

    自动有效负载 = std::move(message.Payload);
    for (uint32_t i = 0; i < message.PayloadSize; ++i) {
        co_await Write(std::move(payload[i]));
    }

    共同返回；
}

模板<类型名称 TSocket>
TValueTask> TMessageReader::Read() {
    decltype(TMessage::Type) 类型； decltype(TMessage::Len) len;
    auto s = co_await Socket.ReadSome(&type, sizeof(type));
    if (s != sizeof(type)) { /* 抛出 */ }
    s = co_await Socket.ReadSome(&len, sizeof(len));
    if (s != sizeof(len)) { /* 抛出 */}
    自动消息 = NewHoldedMessage(type, len);
    co_await TByteReader(Socket).Read(mes->Value, len - sizeof(TMessage));
    auto MaybeAppendEntries = mes.Maybe();
    如果（也许AppendEntries）{
        自动appendEntries = MaybeAppendEntries.Cast();
        自动nentries =appendEntries->Nentries； mes.InitPayload(nentries);
        for (uint32_t i = 0; i < nentries; i++) mes.Payload[i] = co_await Read();
    }
    co_return me；
}

类 TRaftServer {
民众：
    无效服务（）{
        闲置的（）;
        入站服务（）；
    }

私人的：
    TVoidTask InboundServe();
    TVoidTask InboundConnection(TSocket 套接字);
    TVoidTask 空闲();
}

TVoidTask InboundServe() {
    而（真）{
        自动客户端 = co_await Socket.Accept();
        InboundConnection(std::move(client));
    }
    共同返回；
}

TVoidTask InboundConnection（TSocket套接字）{
    而（真）{
        auto mes = co_await TMessageReader(client->Sock()).Read();
        Raft->Process(std::chrono::steady_clock::now(), std::move(mes),
            客户）;
        Raft->ProcessTimeout(std::chrono::steady_clock::now());
        漏极节点();
    }
    共同返回；
}

while (true) {
    Raft->ProcessTimeout(std::chrono::steady_clock::now());
    漏极节点();
    自动 t1 = std::chrono::steady_clock::now();
    如果（t1 > t0 + dt）{
        调试打印();
        t0 = t1；
    }
    co_await Poller.Sleep(t1 + sleep);
}

尝试 {
    while (!Messages.empty()) {
        自动发送 = std::move(消息); Messages.clear();
        for (auto&& m : 发送) {
            co_await TMessageWriter(Socket).Write(std::move(m));
        }
    }
} catch (const std::Exception& ex) {
    连接（）;
}
co_return;

$ ./server --id 1 --node 127.0.0.1:8001:1 --node 127.0.0.1:8002:2 --node 127.0.0.1: 8003:3
...
候选人，期限：2，索引：0，CommitIndex：0，
...
领导者，任期：3，索引：1080175，提交索引：1080175，延迟：2:0 3:0
        匹配索引: 2:1080175 3:1080175 下一个索引: 2:1080176 3:1080176
....
$ ./server --id 2 --node 127.0.0.1:8001:1 --node 127.0.0.1:8002:2 --node 127.0.0.1:8003:3
...
$ ./server --id 3 --node 127.0.0.1:8001:1 --node 127.0.0.1:8002:2 --node 127.0.0.1:8003:3
...
关注者，期限：3，索引：1080175，CommitIndex：1080175，
...
$ dd if=/dev/urandom | 64 位 |光伏-l | ./client --node 127.0.0.1:8001:1 >log1
198k 0:00:03 [159.2k/s] [        <=>

任务<> tcp_echo_server() {
    字符数据[1024]；
    而（真）{
        std::size_t n = co_await socket.async_read_some(buffer(data));
        co_await async_write(socket, buffer(data, n));
    }
}

auto ReadSome(char* buf, size_t size) {
    结构体 TAwaitable {
        bool wait_ready() { 返回 false; /* 始终挂起 */ }
        无效等待_挂起（std :: coroutine_handle <> h）{
            轮询器->AddRead(fd, h);
        }
        int wait_resume() {
            返回读取（fd，b，s）；
        }
        TSelect* 轮询器； int fd;字符* b； size_t s；
    };
    返回 TAwaitable{Poller_,Fd_,buf,size};
}

结构 TEvent {
    int Fd; int 类型； // 读 = 1，写 = 2；
    std::coroutine_handle<> 句柄；
};
类 TSelect {
    无效民意调查（）{
        for (const auto& ch : 事件) { /* FD_SET(ReadFds); FD_SET(WriteFds);*/ }
        pselect(Size, ReadFds, WriteFds, nullptr, ts, nullptr);
        for (int k = 0; k < 大小; ++k) {
            if (FD_ISSET(k, WriteFds)) {
                事件[k].Handle.resume();
            }
            // ...
        }
    }
    std::vector 事件；
    // ...
};

TSimpleTask 任务(TSelect& poller) {
    TSocket 套接字（0，轮询器）；
    字符缓冲区[1024]；
    而（真）{
        auto readSize = co_await socket.ReadSome(buffer, sizeof(buffer));
    }
}
int main() {
    T选择轮询器；
    任务（轮询器）；
    while (true) { poller.Poll(); }
}

TValueTask Read() {
    特雷斯； size_t 大小 = sizeof(T);
    char* p = reinterpret_cast(&res);
    而（大小！= 0）{
        自动 readSize = co_await Socket.ReadSome(p, 大小);
        p += 读取大小；
        大小-=读取大小；
    }
    co_return 资源；
}
// 用法
T t = co_await Read();

模板<类型名称 T>
结构 TValueTask : std::coroutine_handle<> {
    bool wait_ready() { return !!this->promise().Value; } }
    void wait_suspend(std::coroutine_handle<> 调用者) {
        this->promise().Caller = 调用者；
    }
    T wait_resume() { return *this->promise().Value; } }
    使用 Promise_type = TValuePromise;
};

模板<类型名称 T>
结构 TValuePromise {
    void return_value(const T& t) { 值 = t; }
    std::suspend_never 初始挂起() { 返回 {}; }
    // 在这里恢复调用者
    TFinalSuspendContinuation Final_Suspend() noexcept;
    std::可选 值；
    std::coroutine_handle<> 调用者 = std::noop_coroutine();
};