quorums_dkgsessionhandler.h

Go to the documentation of this file.

// Copyright (c) 2018-2021 The Dash Core developers
// Copyright (c) 2022 The PIVX Core developers
// Distributed under the MIT/X11 software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
#ifndef PIVX_LLMQ_QUORUMS_DKGSESSIONHANDLER_H
#define PIVX_LLMQ_QUORUMS_DKGSESSIONHANDLER_H
 
#include "ctpl_stl.h"
#include "llmq/quorums_dkgsession.h"
#include "validation.h"
 
namespace llmq
{
 
enum QuorumPhase {
    QuorumPhase_None = -1,
    QuorumPhase_Initialized = 1,
    QuorumPhase_Contribute,
    QuorumPhase_Complain,
    QuorumPhase_Justify,
    QuorumPhase_Commit,
    QuorumPhase_Finalize,
    QuorumPhase_Idle,
};
 
class CDKGPendingMessages
{
public:
    typedef std::pair<NodeId, std::shared_ptr<CDataStream>> BinaryMessage;
 
private:
    mutable RecursiveMutex cs;
    size_t maxMessagesPerNode;
    std::list<BinaryMessage> pendingMessages;
    std::map<NodeId, size_t> messagesPerNode;
    std::set<uint256> seenMessages;
 
public:
    explicit CDKGPendingMessages(size_t _maxMessagesPerNode);
 
    void PushPendingMessage(NodeId from, CDataStream& vRecv, int invType);
    std::list<BinaryMessage> PopPendingMessages(size_t maxCount);
    bool HasSeen(const uint256& hash) const;
    void Clear();
 
    template<typename Message>
    void PushPendingMessage(NodeId from, Message& msg, int invType)
    {
        CDataStream ds(SER_NETWORK, PROTOCOL_VERSION);
        ds << msg;
        PushPendingMessage(from, ds, invType);
    }
 
    // Might return nullptr messages, which indicates that deserialization failed for some reason
    template<typename Message>
    std::vector<std::pair<NodeId, std::shared_ptr<Message>>> PopAndDeserializeMessages(size_t maxCount)
    {
        auto binaryMessages = PopPendingMessages(maxCount);
        if (binaryMessages.empty()) {
            return {};
        }
 
        std::vector<std::pair<NodeId, std::shared_ptr<Message>>> ret;
        ret.reserve(binaryMessages.size());
        for (const auto& bm : binaryMessages) {
            auto msg = std::make_shared<Message>();
            try {
                *bm.second >> *msg;
            } catch (...) {
                msg = nullptr;
            }
            ret.emplace_back(std::make_pair(bm.first, std::move(msg)));
        }
 
        return ret;
    }
};
 
class CDKGSessionHandler
{
private:
    friend class CDKGSessionManager;
 
    mutable RecursiveMutex cs;
    std::atomic<bool> stopRequested{false};
 
    const Consensus::LLMQParams& params;
    CBLSWorker& blsWorker;
    CDKGSessionManager& dkgManager;
 
    QuorumPhase phase{QuorumPhase_Idle};
    int currentHeight{-1};
    int quorumHeight{-1};
    uint256 quorumHash{UINT256_ZERO};
    std::shared_ptr<CDKGSession> curSession;
    std::thread phaseHandlerThread;
 
    CDKGPendingMessages pendingContributions;
    CDKGPendingMessages pendingComplaints;
    CDKGPendingMessages pendingJustifications;
    CDKGPendingMessages pendingPrematureCommitments;
 
public:
    CDKGSessionHandler(const Consensus::LLMQParams& _params, CBLSWorker& blsWorker, CDKGSessionManager& _dkgManager);
    ~CDKGSessionHandler();
 
    void UpdatedBlockTip(const CBlockIndex *pindexNew);
    void ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStream& vRecv);
 
    void StartThread();
    void StopThread();
 
private:
    bool InitNewQuorum(const CBlockIndex* pindexQuorum);
 
    struct QuorumPhaseAndHash {
        QuorumPhase phase;
        uint256 quorumHash;
    };
    QuorumPhaseAndHash GetPhaseAndQuorumHash() const;
 
    typedef std::function<void()> StartPhaseFunc;
    typedef std::function<bool()> WhileWaitFunc;
    void WaitForNextPhase(QuorumPhase curPhase, QuorumPhase nextPhase, const uint256& expectedQuorumHash, const WhileWaitFunc& runWhileWaiting);
    void WaitForNewQuorum(const uint256& oldQuorumHash);
    void SleepBeforePhase(QuorumPhase curPhase, const uint256& expectedQuorumHash, double randomSleepFactor, const WhileWaitFunc& runWhileWaiting);
    void HandlePhase(QuorumPhase curPhase, QuorumPhase nextPhase, const uint256& expectedQuorumHash, double randomSleepFactor, const StartPhaseFunc& startPhaseFunc, const WhileWaitFunc& runWhileWaiting);
    void HandleDKGRound();
    void PhaseHandlerThread();
};
 
}
 
#endif // PIVX_LLMQ_QUORUMS_DKGSESSIONHANDLER_H