CherryPickFastCheckpointer.hpp

// <CherryPickFastCheckpointer> -*- C++ -*-
 
#pragma once
 
#include "sparta/serialization/checkpoint/FastCheckpointer.hpp"
#include "sparta/utils/SpartaException.hpp"
#include "simdb/apps/App.hpp"
#include "simdb/utils/ConcurrentQueue.hpp"
#include <map>
 
namespace sparta
{
    class TreeNode;
    class Scheduler;
}
 
namespace simdb::pipeline
{
    class Flusher;
}
 
namespace sparta::serialization::checkpoint
{
 
class CherryPickFastCheckpointer final : public simdb::App
{
public:
    using checkpoint_type = typename FastCheckpointer::checkpoint_type;
    using checkpoint_ptr = typename FastCheckpointer::checkpoint_ptr;
    using checkpoint_ptrs = typename FastCheckpointer::checkpoint_ptrs;
    using chkpt_id_t = typename Checkpointer::chkpt_id_t;
    using arch_id_t = uint64_t;
    using tick_t = typename Checkpointer::tick_t;
 
    static constexpr auto NAME = "cherry-pick-fast-checkpointer";
 
    CherryPickFastCheckpointer(simdb::DatabaseManager* db_mgr, const std::vector<TreeNode*> & roots,
                               Scheduler* sched = nullptr);
 
    static void defineSchema(simdb::Schema& schema);
 
    void createPipeline(simdb::pipeline::PipelineManager* pipeline_mgr) override;
 
    FastCheckpointer& getFastCheckpointer() noexcept {
        return checkpointer_;
    }
 
    void commitCurrentBranch(bool force_new_head_chkpt = false);
 
    void saveCheckpoints(checkpoint_ptrs&& checkpoints);
 
    size_t getNumCheckpoints() const;
 
    std::string stringize() const;
 
    struct ChkptWindow {
        arch_id_t start_arch_id = UINT64_MAX;
        arch_id_t end_arch_id = UINT64_MAX;
        tick_t start_tick = UINT64_MAX;
        tick_t end_tick = UINT64_MAX;
        checkpoint_ptrs checkpoints;
 
        template <typename Archive>
        void serialize(Archive & ar, const unsigned int /*version*/) {
            ar & start_arch_id;
            ar & end_arch_id;
            ar & start_tick;
            ar & end_tick;
 
            if (checkpoints.empty()) {
                // We are loading a checkpoint window from disk
                const auto num_chkpts = end_arch_id - start_arch_id + 1;
                for (size_t i = 0; i < num_chkpts; ++i) {
                    checkpoints.emplace_back(new checkpoint_type);
                    ar & *checkpoints.back();
                }
            } else {
                // We are saving a checkpoint window to disk
                for (auto& chkpt : checkpoints) {
                    ar & *chkpt;
                }
            }
        }
    };
 
    struct ChkptWindowBytes {
        arch_id_t start_arch_id = UINT64_MAX;
        arch_id_t end_arch_id = UINT64_MAX;
        tick_t start_tick = UINT64_MAX;
        tick_t end_tick = UINT64_MAX;
        size_t num_chkpts = 0;
        std::vector<char> chkpt_bytes;
    };
 
private:
    FastCheckpointer checkpointer_;
    simdb::DatabaseManager* db_mgr_ = nullptr;
    simdb::ConcurrentQueue<ChkptWindow>* pipeline_head_ = nullptr;
    std::unique_ptr<simdb::pipeline::Flusher> pipeline_flusher_;
};
 
} // namespace sparta::serialization::checkpoint
 
namespace simdb
{
 
template <>
class AppFactory<sparta::serialization::checkpoint::CherryPickFastCheckpointer> : public AppFactoryBase
{
public:
    using AppT = sparta::serialization::checkpoint::CherryPickFastCheckpointer;
 
    void setArchDataRoots(size_t instance_num, const std::vector<sparta::TreeNode*>& roots)
    {
        roots_by_inst_num_[instance_num] = roots;
    }
 
    void setScheduler(sparta::Scheduler& sched)
    {
        scheduler_ = &sched;
    }
 
    AppT* createApp(DatabaseManager* db_mgr, size_t instance_num = 0) override
    {
        auto it = roots_by_inst_num_.find(instance_num);
        if (it == roots_by_inst_num_.end()) {
            throw sparta::SpartaException(
                "No TreeNode (ArchData root) set for DatabaseCheckpointer instance number ")
                << instance_num << ". Did you forget to call setArchDataRoot()?";
        }
 
        const auto & roots = it->second;
 
        // Make the ctor call that the default AppFactory cannot make.
        return new AppT(db_mgr, roots, scheduler_);
    }
 
    void defineSchema(Schema& schema) const override
    {
        AppT::defineSchema(schema);
    }
 
private:
    sparta::Scheduler* scheduler_ = nullptr;
    std::map<size_t, std::vector<sparta::TreeNode*>> roots_by_inst_num_;
};
 
} // namespace simdb