DeltaCheckpoint.hpp

// <DeltaCheckpoint.hpp> -*- C++ -*-
 
#pragma once
 
#include <iostream>
#include <sstream>
#include <stack>
 
#include "sparta/simulation/TreeNode.hpp"
#include "sparta/functional/ArchData.hpp"
#include "sparta/utils/SpartaException.hpp"
#include "sparta/utils/SpartaAssert.hpp"
#include "sparta/kernel/Scheduler.hpp"
 
#include "sparta/serialization/checkpoint/Checkpointer.hpp"
#include "sparta/serialization/checkpoint/CheckpointExceptions.hpp"
#include "sparta/serialization/checkpoint/VectorStorage.hpp"
#include "sparta/serialization/checkpoint/StringStreamStorage.hpp"
 
namespace sparta::serialization::checkpoint
{
    class FastCheckpointer;
 
    template<typename StorageT=storage::StringStreamStorage>
    class DeltaCheckpoint : public Checkpoint
    {
    public:
 
 
        DeltaCheckpoint() = default;
 
        DeltaCheckpoint(const DeltaCheckpoint&) = delete;
 
        const DeltaCheckpoint& operator=(const DeltaCheckpoint&) = delete;
 
    private:
 
        DeltaCheckpoint(const std::vector<ArchData*>& dats,
                        chkpt_id_t id,
                        tick_t tick,
                        DeltaCheckpoint* prev_delta,
                        bool is_snapshot) :
            Checkpoint(id, tick, prev_delta),
            deleted_id_(UNIDENTIFIED_CHECKPOINT),
            is_snapshot_(is_snapshot)
        {
            if(nullptr == prev_delta){
                if(is_snapshot == false){
                    throw CheckpointError("Cannot create a DeltaCheckpoint id=")
                        << id << " at tick=" << tick << " which has no prev_delta and is not a snapshot";
                }
            }else{
                prev_delta->addNext(this);
            }
 
            // Store the checkpoint from root
            if(is_snapshot){
                storeSnapshot_(dats);
            }else{
                storeDelta_(dats);
            }
        }
 
        friend class FastCheckpointer;
 
    public:
 
        virtual ~DeltaCheckpoint() {
            if(!canDelete()){
                std::cerr << "WARNING: DeltaCheckpoint " << getID()
                          << " being destructed without being allowed to delete" << std::endl;
            }
        }
 
 
        virtual std::string stringize() const override {
            std::stringstream ss;
            ss << "<DeltaCheckpoint id=";
            if(isFlaggedDeleted()){
                ss << "DELETED";
            }else{
                ss << getID();
            }
            ss << " at t=" << getTick();
            if(isSnapshot()){
                ss << "(snapshot)";
            }
            ss << ' ' << getTotalMemoryUse()/1000.0f << "kB (" << getContentMemoryUse()/1000.0f << "kB Data)";
            ss << '>';
            return ss.str();
        }
 
        virtual void dumpData(std::ostream& o) const override {
            data_.dump(o);
        }
 
        void dumpRestoreChain(std::ostream& o) const {
            auto rc = getRestoreChain();
            while(1){
                const DeltaCheckpoint* cp = rc.top();
                rc.pop();
                if(cp->isSnapshot()){
                    o << '(';
                }
                if(cp->getID() == UNIDENTIFIED_CHECKPOINT){
                    o << "*" << getDeletedID();
                }else{
                    o << cp->getID();
                }
                if(cp->isSnapshot()){
                    o << ')';
                }
                if(rc.empty()){
                    break;
                }else{
                    o << " --> ";
                }
            }
        }
 
        virtual uint64_t getTotalMemoryUse() const noexcept override {
            return getContentMemoryUse() \
                + sizeof(decltype(*this)) \
                + (getNexts().size() * sizeof(typename std::remove_reference<decltype(*this)>::type*));
        }
 
        virtual uint64_t getContentMemoryUse() const noexcept override {
            return data_.getSize();
        }
 
 
        void traceValue(std::ostream& o, const std::vector<ArchData*>& dats,
                        const ArchData* container, uint32_t offset, uint32_t size)
        {
            std::stack<DeltaCheckpoint*> dcps = getHistoryChain();
 
            std::vector<std::pair<uint8_t,bool>> bytes; // pair<value,valid>
            bytes.resize(size, decltype(bytes)::value_type(0,false));
 
            constexpr uint32_t BUF_SIZE = 8192*2;
            std::unique_ptr<char[]> buf(new char[BUF_SIZE]); // Line-reading buffer
 
            while(!dcps.empty()){
                DeltaCheckpoint* d = dcps.top();
                o << "trace: Checkpoint " << d->getDeletedRepr() << (d->isSnapshot()?" (snapshot)":"") << std::endl;
                dcps.pop();
                d->data_.prepareForLoad();
                bool found_ad = false;
                bool changed = false;
                for(ArchData* ad : dats){
                    if(ad == container){
                        found_ad = true;
                        if(d->isSnapshot()){
                            for(auto &x : bytes){
                                x.second = false; // Invalidate result due to snapshot load
                            }
                        }
                    }
                    while(1){
                        auto ln_idx = d->data_.getNextRestoreLine();
                        if(ln_idx == ArchData::INVALID_LINE_IDX){
                            if(ad == container){
                                //if(changed == false){
                                //    o << "trace: No data for this value" << std::endl;
                                //}
                            }
                            break; // Done with this ArchData
                        }
                        auto ln_off = ln_idx * ad->getLineSize();
                        sparta_assert(BUF_SIZE > ad->getLineSize(),
                                      "Cannot trace value on ArchDatas with line sizes > " << BUF_SIZE << " (" << ad->getLineSize() << ")");
                        d->data_.copyLineBytes((char*)buf.get(), ad->getLineSize()); // Need to read regardless of usefulness of data
                        if(ad == container){
                            //o << "trace: Contains data for line idx " << std::dec << ln_idx
                            //  << " offsets [" << ln_off << "," << ln_off+ad->getLineSize() << ")" << std::endl;
                            if(offset >= ln_off && offset < ln_off + ad->getLineSize()){
                                sparta_assert(offset+size < ln_off + ad->getLineSize(),
                                              "Cannot trace value which spans multiple lines!");
                                sparta_assert(changed == false,
                                              "Value being traced changed twice in the same checkpoint");
                                changed = true;
                                auto off_in_line = offset - ln_off;
                                o << "trace: Value changed (line " << std::dec << ln_idx << ")" << std::endl;
                                for(uint32_t i=0; i<bytes.size(); i++){
                                    bytes[i].first = buf.get()[i+off_in_line];
                                    bytes[i].second = true; // Valid
                                }
                            }
                        }
                    }
                }
                if(!found_ad){
                    o << "trace: Could not find selected ArchData " << (const void*)container << " in this checkpoint!" << std::endl;
                }
                o << "trace: Value:";
                for(uint32_t i=0; i<bytes.size(); i++){
                    if(bytes[i].second){
                        o << ' ' << std::setfill('0') << std::setw(2) << std::hex << (uint16_t)bytes[i].first;
                    }else{
                        o << " xx"; // Invalid
                    }
                }
                o << std::endl;
            }
            o << std::endl;
        }
 
        std::stack<DeltaCheckpoint*> getHistoryChain() {
            // Build stack up to last snapshot
            DeltaCheckpoint* n = this;
            std::stack<DeltaCheckpoint*> dcps;
            while(n){
                dcps.push(n);
                n = static_cast<DeltaCheckpoint*>(n->getPrev());
            }
            return dcps;
        }
 
        std::stack<DeltaCheckpoint*> getRestoreChain() {
            // Build stack up to last snapshot
            DeltaCheckpoint* n = this;
            std::stack<DeltaCheckpoint*> dcps;
            while(1){
                dcps.push(n);
                if(n->isSnapshot()){
                    break;
                }
                n = static_cast<DeltaCheckpoint*>(n->getPrev());
            }
            return dcps;
        }
 
        std::stack<const DeltaCheckpoint*> getRestoreChain() const {
            // Build stack up to last snapshot
            const DeltaCheckpoint* n = this;
            std::stack<const DeltaCheckpoint*> dcps;
            while(1){
                dcps.push(n);
                if(n->isSnapshot()){
                    break;
                }
                n = static_cast<const DeltaCheckpoint*>(n->getPrev());
            }
            return dcps;
        }
 
        chkpt_id_t getPrevID() const override {
            if (auto prev = static_cast<const DeltaCheckpoint*>(getPrev())) {
                if (!prev->isFlaggedDeleted()) {
                    return prev->getID();
                }
            }
            return UNIDENTIFIED_CHECKPOINT;
        }
 
        std::vector<chkpt_id_t> getNextIDs() const override {
            std::vector<chkpt_id_t> next_ids;
            for (const auto chkpt : getNexts()) {
                const auto dcp = static_cast<const DeltaCheckpoint*>(chkpt);
                if (!dcp->isFlaggedDeleted()) {
                    next_ids.push_back(chkpt->getID());
                }
            }
            return next_ids;
        }
 
        virtual void load(const std::vector<ArchData*>& dats) override {
            // Build stack up to last snapshot
            std::stack<DeltaCheckpoint*> dcps = getRestoreChain();
 
            // Load in proper order
            while(!dcps.empty()){
                DeltaCheckpoint* d = dcps.top();
                dcps.pop();
                d->loadState(dats);
            }
        }
 
        bool canDelete() const noexcept {
            if(!isFlaggedDeleted()){
                return false;
            }
            for(auto d : getNexts()){
                const DeltaCheckpoint* dcp = static_cast<const DeltaCheckpoint*>(d);
                if(!dcp->canDelete() && !dcp->isSnapshot()){
                    return false;
                }
            }
            return true; // This checkpoint was flagged deleted
        }
 
        void flagDeleted() {
            sparta_assert(not isFlaggedDeleted(),
                          "Cannot delete a checkpoint when it is already deleted: " << this);
            deleted_id_ = getID();
            setID_(UNIDENTIFIED_CHECKPOINT);
        }
 
        bool isFlaggedDeleted() const noexcept {
            return getID() == UNIDENTIFIED_CHECKPOINT;
        }
 
        chkpt_id_t getDeletedID() const noexcept {
            return deleted_id_;
        }
 
        virtual std::string getDeletedRepr() const override {
            std::stringstream ss;
            if(isFlaggedDeleted()){
                ss << "*" << getDeletedID();
            }else{
                ss << getID();
            }
            return ss.str();
        }
 
        bool isSnapshot() const noexcept { return is_snapshot_; }
 
        uint32_t getDistanceToPrevSnapshot() const noexcept {
            const DeltaCheckpoint* d = this;
            uint32_t dist = 0;
            while(d){
                if(d->isSnapshot()){
                    return dist;
                }
                d = static_cast<const DeltaCheckpoint*>(d->getPrev());
                ++dist;
            }
 
            // This will compile just fine....
#ifdef __clang__
            // This is known to be needed with Clang 8.0.0; not sure about other versions.
#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wexceptions"
#elif defined __GNUC__  // Note that clang defines both __clang__ and __GNUC__
#pragma GCC diagnostic push
#if __GNUC__ > 5
            // gcc 4.7 doesn't like this.
#pragma GCC diagnostic ignored "-Wterminate"
#endif  // __GNUC__ > 5
#endif  // ifdef __clang__ ... elif defined __GNUC__ ...
 
 
            throw CheckpointError() << "In getDistanceToPrevious, somehow reached null "
                                    << "previous-checkpoint without encountering a snapshot. This should "
                                    << "never occur and is a critical error";
#ifdef __clang__
#pragma clang diagnostic pop
#elif defined __GNUC__
#pragma GCC diagnostic pop
#endif
 
 
            // But using my macro that injects the pragmas doesn't work... ugh
            /*
              TERMINATING_THROW(CheckpointError, "In getDistanceToPrevious, somehow reached null "
              << "previous-checkpoint without encountering a snapshot. This should "
              << "never occur and is a critical error");
            */
        }
 
        void makeHeadCheckpoint() override {
            sparta_assert(is_snapshot_);
            Checkpoint::makeHeadCheckpoint();
        }
 
        template <typename Archive>
        void serialize(Archive& ar, const unsigned int version) {
            Checkpoint::serialize(ar, version);
            ar & deleted_id_;
            ar & is_snapshot_;
            ar & data_;
        }
 
 
    protected:
 
        void loadState(const std::vector<ArchData*>& dats) {
            data_.prepareForLoad();
            sparta_assert(data_.good(),
                          "Attempted to loadState from a DeltaCheckpoint with a bad data buffer");
            if(isSnapshot()){
                for(ArchData* ad : dats){
                    //std::cout << "Restoring for ArchData: " << (void*)ad << " " << ad->getOwnerNode()->getLocation() << std::endl;
                    ad->restoreAll(data_);
                }
            }else{
                for(ArchData* ad : dats){
                    //std::cout << "Restoring for ArchData: " << (void*)ad << " " << ad->getOwnerNode()->getLocation() << std::endl;
                    ad->restore(data_);
                }
            }
        }
 
    private:
 
 
        void storeSnapshot_(const std::vector<ArchData*>& dats) {
            sparta_assert(data_.good(),
                          "Attempted to storeSnapshot_ from a DeltaCheckpoint with a bad data buffer");
            // Cannot have stored already
            for(ArchData* ad : dats){
                //std::cout << "SaveAll for ArchData: " << ad->getOwnerNode()->getLocation() << std::endl;
                ad->saveAll(data_);
            }
        }
 
        void storeDelta_(const std::vector<ArchData*>& dats) {
            sparta_assert(data_.good(),
                          "Attempted to storeDelta_ from a DeltaCheckpoint with a bad data buffer");
            // Cannot have stored already
            for(ArchData* ad : dats){
                //std::cout << "Save for ArchData: " << ad->getOwnerNode()->getLocation() << std::endl;
                ad->save(data_);
            }
        }
 
 
 
        chkpt_id_t deleted_id_;
 
        bool is_snapshot_ = false;
 
        StorageT data_;
    };
 
} // namespace sparta::serialization::checkpoint