BatchPropose会造成状态机恢复时重复Execute或者丢失吗 about phxpaxos HOT 23 CLOSED

    for (int i = 0; i < oBatchValues.values_size(); i++)
    {
        const PaxosValue & oValue = oBatchValues.values(i);
        bool bExecuteSucc = DoExecuteForCheckpoint(iGroupIdx, llInstanceID, oValue.value(), oValue.smid());
        if (!bExecuteSucc)
        {
            return false;
        }
    }

#include <string>
#include <functional>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>

#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <phxpaxos/node.h>
#include <phxpaxos/sm.h>

using namespace std;

static const int gSMID = 123;
static const int gGroupId = 0;

class MockSM : public phxpaxos::StateMachine {
public:
    static const int mSMID = gSMID;
    uint64_t mCptVersion;
    MockSM(uint64_t cptVersion = phxpaxos::NoCheckpoint)
        : mCptVersion{cptVersion}
    {
    }
    bool Execute(const int iGroupIdx, const uint64_t llInstanceID, 
            const std::string & sPaxosValue, phxpaxos::SMCtx* poSMCtx) override
    {
        return true;
    }

    const int SMID() const override { return mSMID; }

    bool ExecuteForCheckpoint(const int iGroupIdx, const uint64_t llInstanceID,
            const std::string & sPaxosValue) override
    {
        bool ret = mCptVersion == phxpaxos::NoCheckpoint || llInstanceID > mCptVersion;
        printf("update checkpoint id = %lu, value = %s, ret = %s\n",
            llInstanceID, sPaxosValue.c_str(), ret ? "true" : "false");
        if (ret) {
            mCptVersion = llInstanceID;
        }
        return ret;
    }

    const uint64_t GetCheckpointInstanceID(const int iGroupIdx) const override
    { return mCptVersion; }
};

class ManualResetEvent {
private:
    std::mutex mMutex;
    std::condition_variable mCond;
    bool mSet{false};
public:
    void Set()
    {
        std::unique_lock<std::mutex> locker(mMutex);
        mSet = true;
        mCond.notify_all();
    }
    void Reset()
    {
        std::unique_lock<std::mutex> locker(mMutex);
        mSet = false;
    }
    void Wait()
    {
        std::unique_lock<std::mutex> locker(mMutex);
        mCond.wait(locker, [this]() { return mSet; });
    }
};

void BatchPropose(phxpaxos::Node *node, ManualResetEvent *e, std::string value)
{
    std::mutex m;
    std::unique_lock<std::mutex> locker(m);
    e->Wait();
    uint64_t id = (uint64_t)-1;
    uint32_t idx = (uint32_t)-1;
    phxpaxos::SMCtx ctx;
    ctx.m_iSMID = gSMID;
    int ret = node->BatchPropose(gGroupId, value, id, idx, &ctx);
    printf("batch propose id = %lu, idx = %u, value = %s, ret = %d\n", id, idx, value.c_str(), ret);
    ASSERT_EQ(0, ret);
}

TEST(phxpaxos, BatchTest)
{
    std::string ip = "127.0.0.1";
    int port = 12121;
    MockSM sm;
    phxpaxos::Node *node = nullptr;
    phxpaxos::Options options;

    {
        options.sLogStoragePath = "data/log_test";
        mkdir(options.sLogStoragePath.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
        options.oMyNode = phxpaxos::NodeInfo(ip, port);
        options.vecNodeInfoList.emplace_back(options.oMyNode);
        options.bUseBatchPropose = true;
        options.bUseCheckpointReplayer = true;

        phxpaxos::GroupSMInfo smInfo;
        smInfo.iGroupIdx = gGroupId;
        smInfo.vecSMList.push_back(&sm);
        smInfo.bIsUseMaster = false;

        options.vecGroupSMInfoList.push_back(smInfo);

    }

    // start a single paxos node
    int ret = phxpaxos::Node::RunNode(options, node);
    ASSERT_EQ(0, ret);

    // batch propose
    {
        ManualResetEvent e;

        std::vector<std::thread> workers;
        workers.emplace_back(&BatchPropose, node, &e, "a");
        workers.emplace_back(&BatchPropose, node, &e, "b");
        workers.emplace_back(&BatchPropose, node, &e, "c");
        workers.emplace_back(&BatchPropose, node, &e, "d");
        workers.emplace_back(&BatchPropose, node, &e, "e");

        e.Set();

        for (auto &worker : workers) {
            worker.join();
        }
    }

    node->ContinueCheckpointReplayer();
    puts("cpt replayer started");

    // propose
    {
        phxpaxos::SMCtx ctx;
        ctx.m_iSMID = gSMID;
        uint64_t id = (uint64_t)-1;
        std::string value = "single1";
        int ret = node->Propose(gGroupId, value, id, &ctx);
        printf("propose id = %lu, value = %s, ret = %d\n", id, value.c_str(), ret);
        ASSERT_EQ(0, ret);
        value = "single2";
        ret = node->Propose(gGroupId, value, id, &ctx);
        printf("propose id = %lu, value = %s, ret = %d\n", id, value.c_str(), ret);
        ASSERT_EQ(0, ret);
    }

    while (sm.mCptVersion == phxpaxos::NoCheckpoint) {
        sleep(1);
    }

    printf("now cpt version = %lu\n", sm.mCptVersion);

    delete node;
    node = nullptr;

    printf("starting new node\n");
    ret = phxpaxos::Node::RunNode(options, node);
    ASSERT_EQ(0, ret);

    node->ContinueCheckpointReplayer();
    puts("cpt replayer started");

    while (true) {
        sleep(1);
    }
}

[==========] Running 1 test from 1 test case.
[----------] Global test environment set-up.
[----------] 1 test from phxpaxos
[ RUN      ] phxpaxos.BatchTest
batch propose id = 0, idx = 2, value = b, ret = 0
batch propose id = 0, idx = 4, value = d, ret = 0
batch propose id = 0, idx = 3, value = c, ret = 0
batch propose id = 0, idx = 1, value = a, ret = 0
batch propose id = 0, idx = 0, value = e, ret = 0
cpt replayer started
propose id = 1, value = single1, ret = 0
propose id = 2, value = single2, ret = 0
update checkpoint id = 0, value = e, ret = true
update checkpoint id = 0, value = a, ret = false
now cpt version = 0
update checkpoint id = 0, value = e, ret = false
update checkpoint id = 0, value = e, ret = false
update checkpoint id = 0, value = e, ret = false
starting new node
cpt replayer started
update checkpoint id = 1, value = single1, ret = true
^C