uhs/twophase/coordinator/controller.hpp

Go to the documentation of this file.

// Copyright (c) 2021 MIT Digital Currency Initiative,
//                    Federal Reserve Bank of Boston
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
#ifndef OPENCBDC_TX_SRC_COORDINATOR_CONTROLLER_H_
#define OPENCBDC_TX_SRC_COORDINATOR_CONTROLLER_H_
 
#include "distributed_tx.hpp"
#include "interface.hpp"
#include "server.hpp"
#include "state_machine.hpp"
#include "uhs/twophase/locking_shard/locking_shard.hpp"
#include "util/common/buffer.hpp"
#include "util/common/random_source.hpp"
#include "util/network/connection_manager.hpp"
#include "util/raft/node.hpp"
 
#include <secp256k1.h>
 
namespace cbdc::coordinator {
    class controller : public interface {
      public:
        using send_fn_t = std::function<void(const std::shared_ptr<buffer>&,
                                             cbdc::network::peer_id_t)>;
 
        controller(size_t node_id,
                   size_t coordinator_id,
                   cbdc::config::options opts,
                   std::shared_ptr<logging::log> logger);
 
        controller() = delete;
        controller(const controller&) = delete;
        auto operator=(const controller&) -> controller& = delete;
        controller(controller&&) = delete;
        auto operator=(controller&&) -> controller& = delete;
 
        ~controller() override;
 
        auto init() -> bool;
 
        void quit();
 
        using prepare_tx = std::vector<transaction::compact_tx>;
 
        using prepare_txs = std::
            unordered_map<hash_t, prepare_tx, hashing::const_sip_hash<hash_t>>;
 
        using commit_tx
            = std::pair<std::vector<bool>, std::vector<std::vector<uint64_t>>>;
 
        using commit_txs = std::
            unordered_map<hash_t, commit_tx, hashing::const_sip_hash<hash_t>>;
 
        using discard_txs
            = std::unordered_set<hash_t, hashing::const_sip_hash<hash_t>>;
 
        struct sm_command_header {
            state_machine::command m_comm{};
 
            std::optional<hash_t> m_dtx_id{};
 
            auto operator==(const sm_command_header& rhs) const -> bool;
        };
 
        struct sm_command {
            sm_command_header m_header{};
 
            std::optional<std::variant<prepare_tx, commit_tx>> m_data{};
        };
 
        struct coordinator_state {
            prepare_txs m_prepare_txs{};
 
            commit_txs m_commit_txs{};
 
            discard_txs m_discard_txs{};
 
            auto operator==(const coordinator_state& rhs) const -> bool;
        };
 
        auto execute_transaction(transaction::compact_tx tx,
                                 callback_type result_callback)
            -> bool override;
 
      private:
        size_t m_node_id;
        size_t m_coordinator_id;
        cbdc::config::options m_opts;
        std::shared_ptr<logging::log> m_logger;
 
        nuraft::ptr<state_machine> m_state_machine;
        std::shared_ptr<raft::node> m_raft_serv;
        nuraft::raft_params m_raft_params{};
        std::atomic_bool m_running{false};
        std::vector<std::shared_ptr<cbdc::locking_shard::interface>> m_shards;
        std::vector<std::vector<network::endpoint_t>> m_shard_endpoints;
        std::vector<cbdc::config::shard_range_t> m_shard_ranges;
        random_source m_rnd{config::random_source};
        std::mutex m_batch_mut;
        std::condition_variable m_batch_cv;
        std::shared_ptr<distributed_tx> m_current_batch;
        std::shared_ptr<std::unordered_map<hash_t,
                                           std::pair<callback_type, size_t>,
                                           hashing::const_sip_hash<hash_t>>>
            m_current_txs;
        size_t m_batch_size;
        std::shared_mutex m_shards_mut;
        std::thread m_batch_exec_thread;
        std::unique_ptr<rpc::server> m_rpc_server;
        network::endpoint_t m_handler_endpoint;
        std::vector<std::pair<std::shared_ptr<std::thread>, std::atomic_bool>>
            m_exec_threads;
        std::shared_mutex m_exec_mut;
 
        std::thread m_start_thread;
        bool m_start_flag{false};
        bool m_stop_flag{false};
        std::condition_variable m_start_cv;
        std::mutex m_start_mut;
        bool m_quit{false};
 
        void start_stop_func();
 
        void start();
 
        void stop();
 
        auto recovery_func() -> bool;
 
        void batch_executor_func();
 
        auto raft_callback(nuraft::cb_func::Type type,
                           nuraft::cb_func::Param* param)
            -> nuraft::cb_func::ReturnCode;
 
        auto prepare_cb(const hash_t& dtx_id,
                        const std::vector<transaction::compact_tx>& txs)
            -> bool;
        auto commit_cb(const hash_t& dtx_id,
                       const std::vector<bool>& complete_txs,
                       const std::vector<std::vector<uint64_t>>& tx_idxs)
            -> bool;
        auto discard_cb(const hash_t& dtx_id) -> bool;
        auto done_cb(const hash_t& dtx_id) -> bool;
 
        void batch_set_cbs(distributed_tx& c);
 
        [[nodiscard]] auto replicate_sm_command(const sm_command& c)
            -> std::optional<nuraft::ptr<nuraft::buffer>>;
 
        void connect_shards();
 
        void schedule_exec(std::function<void(size_t)>&& f);
 
        void join_execs();
    };
}
 
#endif // OPENCBDC_TX_SRC_COORDINATOR_CONTROLLER_H_