Method PaxosAdd

phxqueue/store/store.cpp:483–535

}

comm::RetCode Store::PaxosAdd(const comm::proto::AddRequest &req, comm::proto::AddResponse &resp) {
    comm::StoreBP::GetThreadInstance()->OnPaxosAdd(req);

    int paxos_group_id{req.queue_id() % impl_->opt.ngroup};

    proto::StorePaxosArgs args;
    args.set_timestamp(time(nullptr));
    args.mutable_add_req()->CopyFrom(req);

    string buf;
    args.SerializeToString(&buf);

    StoreContext sc;
    sc.result = comm::RetCode::RET_OK;

    comm::StoreBP::GetThreadInstance()->OnBatchPropose(req);

    uint64_t instance_id;
    uint32_t batch_idx;
    phxpaxos::SMCtx ctx(StoreSM::ID, &sc);

    uint64_t t1{comm::utils::Time::GetSteadyClockMS()};
    int paxos_ret{impl_->node->BatchPropose(paxos_group_id, buf, instance_id, batch_idx, &ctx)};
    uint64_t t2{comm::utils::Time::GetSteadyClockMS()};
    uint64_t used_time_ms{t2 - t1};

    if (phxpaxos::PaxosTryCommitRet_OK != paxos_ret) {
        comm::StoreBP::GetThreadInstance()->OnBatchProposeErr(req, used_time_ms);
        QLErr("BatchPropose paxos_ret %d paxos_group_id %d buf_length %zu",
              paxos_ret, paxos_group_id, buf.length());
        switch (paxos_ret) {
        case phxpaxos::PaxosTryCommitRet_Timeout:
            comm::StoreBP::GetThreadInstance()->OnBatchProposeErrTimeout(req);
            return comm::RetCode::RET_ERR_PROPOSE_TIMEOUT;
        case phxpaxos::PaxosTryCommitRet_TooManyThreadWaiting_Reject:
            comm::StoreBP::GetThreadInstance()->OnBatchProposeErrTooManyThreadWaitingReject(req);
            return comm::RetCode::RET_ERR_PROPOSE_FAST_REJECT;
        case phxpaxos::PaxosTryCommitRet_Value_Size_TooLarge:
            comm::StoreBP::GetThreadInstance()->OnBatchProposeErrValueSizeTooLarge(req);
            return comm::RetCode::RET_ERR_SIZE_TOO_LARGE;
        default:
            comm::StoreBP::GetThreadInstance()->OnBatchProposeErrOther(req);
            return comm::RetCode::RET_ERR_PROPOSE;
        };
    }
    comm::StoreBP::GetThreadInstance()->
            OnBatchProposeSucc(req, instance_id, batch_idx, used_time_ms);


    resp.set_cursor_id(instance_id);

    return sc.result;
}

comm::RetCode Store::Get(const comm::proto::GetRequest &req, comm::proto::GetResponse &resp) {
    QLVerb("Get");