DatabaseCheckpointer.hpp

// <DatabaseCheckpointer> -*- C++ -*-
 
#pragma once
 
#include "sparta/serialization/checkpoint/Checkpointer.hpp"
#include "sparta/serialization/checkpoint/DatabaseCheckpoint.hpp"
#include "simdb/apps/App.hpp"
#include "simdb/pipeline/Pipeline.hpp"
#include <unordered_set>
 
namespace sparta::serialization::checkpoint
{
 
class DatabaseCheckpointer;
 
class DatabaseCheckpointer final : public simdb::App, public Checkpointer
{
public:
    static constexpr auto NAME = "db-checkpointer";
 
    using checkpoint_type = DatabaseCheckpoint;
    using checkpoint_ptr = std::shared_ptr<checkpoint_type>;
    using checkpoint_ptrs = std::vector<checkpoint_ptr>;
    using window_id_t = uint64_t;
 
    DatabaseCheckpointer(simdb::DatabaseManager* db_mgr, const std::vector<sparta::TreeNode*> & roots,
                         sparta::Scheduler* sched = nullptr);
 
    static void defineSchema(simdb::Schema& schema);
 
    void createPipeline(simdb::pipeline::PipelineManager* pipeline_mgr) override;
 
    void preTeardown() override;
 
    uint32_t getSnapshotThreshold() const;
 
    void setSnapshotThreshold(uint32_t thresh);
 
    void setMaxCachedWindows(uint32_t max_windows);
 
    uint64_t getTotalMemoryUse() const noexcept override;
 
    uint64_t getContentMemoryUse() const noexcept override;
 
    void deleteCheckpoint(chkpt_id_t) override final;
 
    void loadCheckpoint(chkpt_id_t id) override;
 
    std::vector<chkpt_id_t> getCheckpointsAt(tick_t t) override;
 
    std::vector<chkpt_id_t> getCheckpoints() override;
 
    uint32_t getNumCheckpoints() const noexcept override;
 
    uint32_t getNumSnapshots() const noexcept;
 
    uint32_t getNumDeltas() const noexcept;
 
    std::deque<chkpt_id_t> getCheckpointChain(chkpt_id_t id) override;
 
    std::shared_ptr<DatabaseCheckpoint> findCheckpoint(chkpt_id_t id, bool must_exist=false);
 
    std::shared_ptr<DatabaseCheckpoint> findLatestCheckpointAtOrBefore(tick_t tick, chkpt_id_t from);
 
    bool hasCheckpoint(chkpt_id_t id) noexcept override;
 
    void dumpRestoreChain(std::ostream& o, chkpt_id_t id);
 
    std::stack<chkpt_id_t> getHistoryChain(chkpt_id_t id);
 
    std::stack<chkpt_id_t> getRestoreChain(chkpt_id_t id);
 
    std::vector<chkpt_id_t> getNextIDs(chkpt_id_t id) override;
 
    uint32_t getDistanceToPrevSnapshot(chkpt_id_t id) noexcept;
 
    bool isSnapshot(chkpt_id_t id) noexcept;
 
    std::string stringize() const override;
 
    void dumpList(std::ostream& o) override;
 
    void dumpData(std::ostream& o) override;
 
    void dumpAnnotatedData(std::ostream& o) override;
 
    void traceValue(std::ostream& o, chkpt_id_t id, const ArchData* container, uint32_t offset, uint32_t size) override;
 
    bool isCheckpointCached(chkpt_id_t id) const noexcept;
 
private:
 
    void createHead_() override;
 
    chkpt_id_t createCheckpoint_(bool force_snapshot=false) override;
 
    void deleteCheckpoint_(chkpt_id_t id);
 
    void dumpCheckpointNode_(const chkpt_id_t id, std::ostream& o) override;
 
    void setHead_(DatabaseCheckpoint* head);
 
    void setCurrent_(DatabaseCheckpoint* current);
 
    void addToCache_(std::shared_ptr<checkpoint_type> chkpt);
 
    window_id_t getWindowID_(chkpt_id_t id) const;
 
    void touchWindow_(window_id_t id);
 
    void evictWindowsIfNeeded_(bool force_flush=false);
 
    bool ensureWindowLoaded_(chkpt_id_t id, bool must_succeed=true);
 
    std::unique_ptr<ChkptWindow> deserializeWindow_(
        const std::vector<char>& window_bytes,
        bool requires_decompression = true) const;
 
    bool loadWindowIntoCache_(window_id_t win_id, bool must_succeed=true);
 
    simdb::SnooperCallbackOutcome handleLoadWindowIntoCacheSnooper_(
        std::deque<ChkptWindow>& queue);
 
    simdb::SnooperCallbackOutcome handleLoadWindowIntoCacheSnooper_(
        std::deque<ChkptWindowBytes>& queue,
        bool requires_decompression);
 
    simdb::SnooperCallbackOutcome handleDeleteCheckpointSnooper_(
        std::deque<ChkptWindow>& queue);
 
    simdb::SnooperCallbackOutcome handleDeleteCheckpointSnooper_(
        std::deque<ChkptWindowBytes>& queue,
        bool requires_decompression);
 
    void forEachCheckpoint_(const std::function<void(const DatabaseCheckpoint*)>& cb);
 
    chkpt_id_t head_id_ = checkpoint_type::UNIDENTIFIED_CHECKPOINT;
 
    chkpt_id_t current_id_ = checkpoint_type::UNIDENTIFIED_CHECKPOINT;
 
    simdb::ConcurrentQueue<ChkptWindow>* pipeline_head_ = nullptr;
 
    simdb::ConcurrentQueue<chkpt_id_t>* pending_deletions_head_ = nullptr;
 
    std::unordered_map<window_id_t, checkpoint_ptrs> chkpts_cache_;
 
    std::list<window_id_t> lru_list_;
 
    std::unordered_map<window_id_t, std::list<window_id_t>::iterator> lru_map_;
 
    utils::ValidValue<uint32_t> max_cached_windows_;
 
    mutable std::recursive_mutex cache_mutex_;
 
    simdb::DatabaseManager* db_mgr_ = nullptr;
 
    simdb::pipeline::PipelineManager* pipeline_mgr_ = nullptr;
 
    simdb::pipeline::AsyncDatabaseAccessor* db_accessor_ = nullptr;
 
    std::unique_ptr<simdb::pipeline::RunnableFlusher> pipeline_flusher_;
 
    utils::ValidValue<window_id_t> snoop_win_id_for_retrieval_;
 
    utils::ValidValue<chkpt_id_t> snoop_chkpt_id_for_deletion_;
 
    utils::ValidValue<uint32_t> snap_thresh_;
 
    chkpt_id_t next_chkpt_id_;
};
 
} // namespace sparta::serialization::checkpoint
 
namespace simdb
{
 
template <>
class AppFactory<sparta::serialization::checkpoint::DatabaseCheckpointer> : public AppFactoryBase
{
public:
    using AppT = sparta::serialization::checkpoint::DatabaseCheckpointer;
 
    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