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"
 
 
namespace sparta::serialization::checkpoint
{
    namespace storage
    {
        class VectorStorage
        {
            class Segment{
                ArchData::line_idx_type idx_;
                std::unique_ptr<char []> data_;
                uint32_t bytes_;
            public:
 
                Segment(const Segment&) = delete;
 
                Segment(Segment&& rhp) :
                    idx_(rhp.idx_),
                    data_(std::move(rhp.data_)),
                    bytes_(rhp.bytes_)
                {
                    rhp.idx_ = ArchData::INVALID_LINE_IDX;
                    rhp.bytes_ = 0;
                }
 
                Segment() :
                    idx_(ArchData::INVALID_LINE_IDX),
                    bytes_(0)
                {;}
 
                Segment& operator=(const Segment& rhp) = delete;
 
                Segment(ArchData::line_idx_type idx, const char* data, size_t bytes) :
                    idx_(idx), bytes_(bytes)
                {
                    sparta_assert(idx != ArchData::INVALID_LINE_IDX,
                                  "Attempted to create segment of " << bytes << " bytes with invalid line index");
                    data_.reset(new char[bytes]);
                    ::memcpy(data_.get(), data, bytes);
                }
 
                ArchData::line_idx_type getLineIdx() const {
                    return idx_;
                }
 
                uint32_t getSize() const {
                    return sizeof(decltype(*this)) + bytes_;
                }
 
                void copyTo(char* buf, uint32_t size) const {
                    sparta_assert(size == bytes_, \
                                  "Attempted to restore checkpoint data for a line where the "
                                  "data was " << bytes_ << " bytes but the loader requested "
                                  << size << " bytes. The sizes must match up or something is "
                                  "wrong");
                    memcpy(buf, data_.get(), bytes_);
                }
 
                void dump(std::ostream& o) const {
                    if(idx_ == ArchData::INVALID_LINE_IDX){
                        std::cout << "\nEnd of ArchData";
                        return;
                    }
 
                    std::cout << "\nLine: " << std::dec << idx_ << " (" << bytes_ << ") bytes";
                    for(uint32_t off = 0; off < bytes_;){
                        char chr = data_[off];
                        if(off % 32 == 0){
                            o << std::endl << std::setw(7) << std::hex << off;
                        }
                        if(chr == 0){
                            o << ' ' << "..";
                        }else{
                            o << ' ' << std::setfill('0') << std::setw(2) << std::hex << (0xff & (uint16_t)chr);
                        }
                        off++;
                    }
                }
            };
 
            std::vector<Segment> data_;
 
            ArchData::line_idx_type next_idx_ = ArchData::INVALID_LINE_IDX;
 
            uint32_t next_restore_idx_ = 0;
 
            decltype(data_)::const_iterator cur_restore_itr_;
 
        public:
            VectorStorage() {
            }
 
            ~VectorStorage() {
            }
 
            void dump(std::ostream& o) const {
                for(auto const &seg : data_){
                    seg.dump(o);
                }
            }
 
            uint32_t getSize() const {
                uint32_t bytes = sizeof(decltype(*this));
                for(Segment const & seg : data_){
                    bytes += seg.getSize();
                }
                return bytes;
            }
 
            void prepareForLoad() {
                next_restore_idx_ = 0;
                cur_restore_itr_ = data_.begin();
            }
 
            void beginLine(ArchData::line_idx_type idx) {
                sparta_assert(idx != ArchData::INVALID_LINE_IDX,
                              "Cannot begin line with INVALID_LINE_IDX index");
                next_idx_ = idx;
            }
 
            void writeLineBytes(const char* data, size_t size) {
                sparta_assert(data_.size() == 0 || data_.back().getLineIdx() != next_idx_,
                              "Cannot store the same line idx twice in a checkpoint. Line "
                              << next_idx_ << " detected twice in a row");
                sparta_assert(next_idx_ != ArchData::INVALID_LINE_IDX,
                              "Cannot write line bytes with INVALID_LINE_IDX index");
                data_.emplace_back(next_idx_, data, size);
            }
 
            void endArchData() {
                data_.emplace_back();
            }
 
            bool good() const {
                return next_restore_idx_ <= data_.size(); // Not past end of stream
            }
 
            ArchData::line_idx_type getNextRestoreLine() {
                if(next_restore_idx_ == data_.size()){
                    next_restore_idx_++; // Increment to detect errors
                    return ArchData::INVALID_LINE_IDX; // Done with restore
                }else if(next_restore_idx_ > data_.size()){ // Past the end
                    throw SpartaException("Failed to restore a checkpoint because ")
                        << "caller tried to keep getting next line even after "
                        "reaching the end of the restore data";
                }
                if(next_restore_idx_ != 0){
                    cur_restore_itr_++;
                }
                next_restore_idx_++;
 
                const auto next_line_idx = cur_restore_itr_->getLineIdx(); // May be invalid to indicate end of ArchData
                return next_line_idx;
            };
 
            void copyLineBytes(char* buf, uint32_t size) {
                sparta_assert(cur_restore_itr_ != data_.end(),
                              "Attempted to copy line bytes from an invalid line iterator");
                sparta_assert(cur_restore_itr_->getLineIdx() != ArchData::INVALID_LINE_IDX,
                              "About to return line from checkpoint data segment with INVALID_LINE_IDX index");
                cur_restore_itr_->copyTo(buf, size);
            }
 
            //void stealLineBytes(char*& buf_ptr, uint32_t size) {
            //    cur_restore_itr_->stealBuffer(buf_ptr, size);
            //}
        };
 
        class StringStreamStorage
        {
            std::stringstream ss_;
 
        public:
            StringStreamStorage() {
                ss_.exceptions(std::ostream::eofbit | std::ostream::badbit |
                               std::ostream::failbit | std::ostream::goodbit);
            }
 
            void dump(std::ostream& o) const {
                auto s = ss_.str();
                auto itr = s.begin();
                for(; itr != s.end(); itr++){
                    char chr = *itr;
                    if(chr == 'L'){
                        uint32_t off = 0;
                        ArchData::line_idx_type ln_idx;
                        strncpy((char*)&ln_idx, s.substr(itr-s.begin(), sizeof(ln_idx)).c_str(), sizeof(ln_idx));
                        std::cout << "\nLine: " << ln_idx << std::endl;
                        itr += sizeof(ArchData::line_idx_type);
 
                        for(uint16_t i=0; i<64; ++i){
                            chr = *itr;
                            if(off % 32 == 0){
                                o << std::setw(7) << std::hex << off;
                            }
                            if(chr == 0){
                                o << ' ' << "..";
                            }else{
                                o << ' ' << std::setfill('0') << std::setw(2) << std::hex << (0xff & (uint16_t)chr);
                            }
                            off++;
                            if(off % 32 == 0){
                                o << std::endl;
                            }
                            ++itr;
                        }
                    }
                }
            }
 
            uint32_t getSize() const {
                return ss_.str().size() + sizeof(decltype(*this));
            }
 
            void prepareForLoad() {
                ss_.seekg(0); // Seek to start with get pointer before consuming
            }
 
            void beginLine(ArchData::line_idx_type idx) {
                ss_ << 'L'; // Line start char
 
                ArchData::line_idx_type idx_repr = reorder<ArchData::line_idx_type, LE>(idx);
                ss_.write((char*)&idx_repr, sizeof(ArchData::line_idx_type));
            }
 
            void writeLineBytes(const char* data, size_t size) {
                ss_.write(data, size);
            }
 
            void endArchData() {
                ss_ << "E"; // Indicates end of this checkpoint data
 
                sparta_assert(ss_.good(),
                              "Ostream error while writing checkpoint data");
            }
 
            bool good() const {
                return ss_.good();
            }
 
            ArchData::line_idx_type getNextRestoreLine() {
                char ctrl;
                ss_ >> ctrl;
                sparta_assert(ss_.good(),
                              "Encountered checkpoint data stream error or eof");
                if(ctrl == 'L'){
                    ArchData::line_idx_type ln_idx = 0;
                    ss_.read((char*)&ln_idx, sizeof(ln_idx)); // Presumed LE encoding
                    return ln_idx;
                }else if(ctrl == 'E'){
                    return ArchData::INVALID_LINE_IDX; // Done with restore
                }else{
                    throw SpartaException("Failed to restore a checkpoint because a '")
                        << ctrl << "' control character was found where an 'L' or 'E' was found";
                }
            };
 
            void copyLineBytes(char* buf, uint32_t size) {
                ss_.read(buf, size);
            }
        };
 
    } // namespace storage
 
    class FastCheckpointer;
 
    template<typename StorageT=storage::StringStreamStorage>
    class DeltaCheckpoint : public Checkpoint
    {
    public:
 
 
        DeltaCheckpoint() = delete;
 
        DeltaCheckpoint(const DeltaCheckpoint&) = delete;
 
        const DeltaCheckpoint& operator=(const DeltaCheckpoint&) = delete;
 
    private:
 
        DeltaCheckpoint(TreeNode& root,
                        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)
        {
            (void) root;
            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;
        }
 
        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");
            */
        }
 
 
    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 const is_snapshot_; 
        StorageT data_; 
    };
 
} // namespace sparta::serialization::checkpoint