Collectable.hpp

Go to the documentation of this file.

// <Collectable.hpp>  -*- C++ -*-
 
#pragma once
 
#include <algorithm>
#include <sstream>
#include <functional>
#include <type_traits>
#include <iomanip>
 
#include "sparta/collection/PipelineCollector.hpp"
#include "sparta/pipeViewer/transaction_structures.hpp"
#include "sparta/events/PayloadEvent.hpp"
#include "sparta/events/EventSet.hpp"
#include "sparta/events/SchedulingPhases.hpp"
#include "sparta/pairs/SpartaKeyPairs.hpp"
#include "sparta/utils/Utils.hpp"
#include "sparta/utils/MetaStructs.hpp"
 
#include "boost/numeric/conversion/converter.hpp"
 
namespace sparta{
    namespace collection
    {
 
        template<typename DataT, SchedulingPhase collection_phase = SchedulingPhase::Collection, typename = void>
        class Collectable : public CollectableTreeNode
        {
        public:
            static constexpr uint64_t BAD_DISPLAY_ID = 0x1000;
 
            Collectable(sparta::TreeNode* parent,
                        const std::string& name,
                        const std::string& group,
                        uint32_t index,
                        uint64_t parentid = 0,
                        const std::string & desc = "Collectable <manual, no desc>") :
                CollectableTreeNode(sparta::notNull(parent), name, group, index, desc),
                event_set_(this),
                ev_close_record_(&event_set_, name + "_pipeline_collectable_close_event",
                                 CREATE_SPARTA_HANDLER_WITH_DATA(Collectable, closeRecord, bool))
            {
                sparta_assert(getNodeUID() < ~(decltype(transaction_t::location_ID))0,
                            "Tree has at least " << getNodeUID() << " in it. Because pipeouts "
                            "use these locations and only have 32b location IDs, the pipeline "
                            "collection system imposes a limit of " << ~(decltype(transaction_t::location_ID))0
                            << " nodes in the tree. This is expected to be an unattainable number "
                            "without some kind of runaway allocation bug. If you are sure you "
                            << " nodes in the tree. This is expected to be an unattainable number "
                            "without some kind of runaway allocation bug. If you are sure you "
                            "need more nodes than this, contact the SPARTA team. In the meantime, "
                            "it is probably safe to comment out this assertion if you do not "
                            "plan to use the pipeline collection functionality");
 
                argos_record_.location_ID        = boost::numeric::converter<decltype(argos_record_.location_ID),
                                                                             decltype(getNodeUID())>::convert(getNodeUID());
                argos_record_.control_Process_ID = 0; // what is this for?
                argos_record_.parent_ID          = parentid;
                argos_record_.flags              = is_Annotation;
                argos_record_.time_Start         = 0;
                argos_record_.time_End           = 1;
 
 
            }
 
            Collectable(sparta::TreeNode* parent,
                        const std::string& name,
                        const DataT * collected_object,
                        uint64_t parentid = 0,
                        const std::string & desc = "Collectable <no desc>") :
                Collectable(parent, name,
                            TreeNode::GROUP_NAME_NONE,
                            TreeNode::GROUP_IDX_NONE,
                            parentid,
                            desc)
            {
                collected_object_ = collected_object;
 
                // Get an initial value, if available
                if(collected_object) {
                    std::ostringstream ss;
                    ss << *collected_object;
                    prev_annot_ = ss.str();
                }
            }
 
            Collectable(sparta::TreeNode* parent,
                        const std::string& name,
                        uint64_t parentid = 0,
                        const std::string & desc = "Collectable <manual, no desc>") :
                Collectable(parent, name, nullptr, parentid, desc)
            {
                // Can't auto collect without setting collected_object_
                setManualCollection();
            }
 
            virtual ~Collectable() {}
 
            void initialize(const DataT & val) {
                std::ostringstream ss;
                ss << val;
                prev_annot_ = ss.str();
            }
 
            void collect(const DataT & val)
            {
                if(SPARTA_EXPECT_FALSE(isCollected()))
                {
                    std::ostringstream ss;
                    ss << val;
                    if((ss.str() != prev_annot_) && !record_closed_)
                    {
                        // Close the old record (if there is one)
                        closeRecord();
                    }
 
                    // Remember the new string for a new record and start
                    // a new record if not empty.
                    prev_annot_ = ss.str();
                    if(!prev_annot_.empty() && record_closed_) {
                        startNewRecord_();
                        record_closed_ = false;
                    }
                }
            }
 
            void collectWithDuration(const DataT & val, sparta::Clock::Cycle duration)
            {
                if(SPARTA_EXPECT_FALSE(isCollected()))
                {
                    if(duration != 0) {
                        ev_close_record_.preparePayload(false)->schedule(duration);
                    }
                    collect(val);
                }
            }
 
            void collect() override final {
                // If pointer has become nullified, close the record
                if(nullptr == collected_object_) {
                    closeRecord();
                    return;
                }
                collect(*collected_object_);
            }
 
            void collectWithDuration(sparta::Clock::Cycle duration) {
                // If pointer has become nullified, close the record
                if(nullptr == collected_object_) {
                    closeRecord();
                    return;
                }
                collectWithDuration(*collected_object_, duration);
            }
 
            void restartRecord() override final {
                // Do not get a new ID when we're doing a heartbeat
                if(!record_closed_) {
                    argos_record_.flags |= CONTINUE_FLAG;
                    writeRecord_();
                    argos_record_.flags &= ~CONTINUE_FLAG;
                    startNewRecord_();
                }
            }
 
            void closeRecord(const bool & simulation_ending = false) override final
            {
                if(SPARTA_EXPECT_FALSE(isCollected()))
                {
                    if(!record_closed_ && writeRecord_(simulation_ending)) {
                        prev_annot_.clear();
                    }
                    record_closed_ = true;
                }
            }
 
            void setManualCollection() {
                auto_collect_ = false;
            }
 
        protected:
 
            EventSet & getEventSet_() {
                return event_set_;
            }
 
        private:
 
            bool writeRecord_(bool simulation_ending = false)
            {
                sparta_assert(pipeline_col_,
                            "Must startCollecting_ on this Collectable "
                            << getLocation() << " before a record can be written");
 
                // This record hasn't changed, don't write it
                if(SPARTA_EXPECT_FALSE(argos_record_.time_Start == pipeline_col_->getScheduler()->getCurrentTick())) {
                    return false;
                }
 
                // Set a new transaction ID since we're starting anew
                argos_record_.transaction_ID = pipeline_col_->getUniqueTransactionId();
 
                // Notice that we make the length +1 in order to null
                // terminate the data.
                argos_record_.length = static_cast<uint16_t>(prev_annot_.size());
 
                // We should just change the transaction structure to take
                // an std::string instead of char* in the future.
                argos_record_.annt = prev_annot_;
 
                // Capture the end time
                argos_record_.time_End =
                    pipeline_col_->getScheduler()->getCurrentTick() + (simulation_ending ? 1 : 0);;
 
                // If this assert fires, then a user is trying to collect
                // a record multiple times in their cycle window or
                // something else really bad has happened.
                sparta_assert(argos_record_.time_Start < argos_record_.time_End);
 
                // Write the old record to the file
                pipeline_col_->writeRecord(argos_record_);
 
                return true;
            }
 
            void startNewRecord_() {
                // Set up the new start time for the new record
                argos_record_.time_Start = pipeline_col_->getScheduler()->getCurrentTick();
            }
 
            void setCollecting_(bool collect, Collector * collector) override final
            {
                pipeline_col_ = dynamic_cast<PipelineCollector *>(collector);
                sparta_assert(pipeline_col_ != nullptr,
                            "Collectables can only added to PipelineCollectors... for now");
 
                if(collect && !prev_annot_.empty()) {
                    // Set the start time for this transaction to be
                    // moment collection is enabled.
                    startNewRecord_();
                }
 
                // If the collected object is null, this Collectable
                // object is to be explicitly collected
                if(collected_object_ && auto_collect_) {
                    if(collect) {
                        // Add this Collectable to the PipelineCollector's
                        // list of objects requiring collection
                        pipeline_col_->addToAutoCollection(this, collection_phase);
                    }
                    else {
                        // Remove this Collectable from the
                        // PipelineCollector's list of objects requiring
                        // collection
                        pipeline_col_->removeFromAutoCollection(this);
                    }
                }
 
                if(!collect && !record_closed_) {
                    // Force the record to be written
                    closeRecord();
                }
            }
 
            // The annotation object to be collected
            const DataT * collected_object_ = nullptr;
 
            // The live transaction record
            annotation_t argos_record_;
 
            // Store the result of the previous annotation, the
            // annotation_t struct holds a pointer to this
            std::string prev_annot_;
 
            // Ze Collec-tor
            PipelineCollector * pipeline_col_ = nullptr;
 
            // For those folks that want a value to automatically
            // disappear in the future
            sparta::EventSet event_set_;
            sparta::PayloadEvent<bool, sparta::SchedulingPhase::Trigger> ev_close_record_;
 
            // Is this collectable currently closed?
            bool record_closed_ = true;
 
            // Should we auto-collect?
            bool auto_collect_ = true;
        };
 
        class UniquePairIDGenerator {
        public:
 
            // We need to make Collectable a friend of this class, as it will directly call
            // the static private function from within itself.
            template<typename DataT, SchedulingPhase collection_phase, typename>
            friend class Collectable;
        private:
 
            // Any point this funtion is invoked, it releases a new 64bit integer which is unique.
            // So, we need to limit the access of this function.
            static uint64_t getUniquePairID_() {
                static uint64_t id = 0;
                return ++id;
             }
        };
 
        template<typename DataT, SchedulingPhase collection_phase>
        class Collectable<DataT, collection_phase,
                          MetaStruct::enable_if_t<
                              std::is_base_of<sparta::PairDefinition<MetaStruct::remove_any_pointer_t<DataT>>,
                                              typename MetaStruct::remove_any_pointer_t<DataT>::SpartaPairDefinitionType>::value>> :
            public sparta::PairCollector<typename MetaStruct::remove_any_pointer_t<DataT>::SpartaPairDefinitionType>, public CollectableTreeNode
        {
            // Aliasing the actual Datatype of the collectable being collected as Data_t
            typedef MetaStruct::remove_any_pointer_t<DataT> Data_t;
 
            // Aliasing the Datatype of the Pair Definition of the collectable being collected as PairDef_t
            typedef typename Data_t::SpartaPairDefinitionType PairDef_t;
 
            // Making a bunch of APIs of PairCollector class being available to us with the using directive
            using PairCollector<PairDef_t>::getNameStrings;
            using PairCollector<PairDef_t>::getDataVector;
            using PairCollector<PairDef_t>::getStringVector;
            using PairCollector<PairDef_t>::getFormatVector;
            using PairCollector<PairDef_t>::getSizeOfVector;
            using PairCollector<PairDef_t>::getPEventLogVector;
            using PairCollector<PairDef_t>::getArgosFormatGuide;
            using PairCollector<PairDef_t>::collect_;
            using PairCollector<PairDef_t>::isCollecting;
 
        public:
 
            Collectable(sparta::TreeNode* parent,
                        const std::string& name,
                        const std::string& group,
                        uint32_t index,
                        uint64_t parentid = 0,
                        const std::string & desc = "Collectable <manual, no desc>") :
                sparta::PairCollector<PairDef_t>(),
                CollectableTreeNode(sparta::notNull(parent), name, group, index, desc),
                argos_record_(pair_t(0, 1, parentid, 0, BAD_DISPLAY_ID,
                    boost::numeric::converter<decltype(argos_record_.location_ID),
                        decltype(getNodeUID())>::convert(getNodeUID()),
                            is_Pair, 0)),
                event_set_(this),
                ev_close_record_(&event_set_, name + "_pipeline_collectable_close_event",
                                 CREATE_SPARTA_HANDLER_WITH_DATA(Collectable, closeRecord, bool)) {
                static constexpr auto MAX_UID = std::numeric_limits<decltype(argos_record_.location_ID)>::max();
                sparta_assert(getNodeUID() < MAX_UID,
                            "Tree has at least " << getNodeUID() << " in it. Because pipeouts "
                            "use these locations and only have 32b location IDs, the pipeline "
                            "collection system imposes a limit of " << MAX_UID
                            << " nodes in the tree. This is expected to be an unattainable number "
                            "without some kind of runaway allocation bug. If you are sure you "
                            "need more nodes than this, contact the SPARTA team. In the meantime, "
                            "it is probably safe to comment out this assertion if you do not "
                            "plan to use the pipeline collection functionality");
            }
 
            Collectable(sparta::TreeNode* parent,
                        const std::string& name,
                        const DataT * collected_object,
                        uint64_t parentid = 0,
                        const std::string & desc = "Collectable <no desc>") :
                Collectable(parent, name,
                            TreeNode::GROUP_NAME_NONE,
                            TreeNode::GROUP_IDX_NONE,
                            parentid,
                            desc)
            {
                collected_object_ = collected_object;
 
                // Get an initial value, if available
                if(collected_object) {
                    initialize(*collected_object);
                }
            }
 
            Collectable(sparta::TreeNode* parent,
                        const std::string& name,
                        uint64_t parentid = 0,
                        const std::string & desc = "Collectable <manual, no desc>") :
                Collectable(parent, name, nullptr, parentid, desc)
            {
                // Can't auto collect without setting collected_object_
                setManualCollection();
            }
 
            virtual ~Collectable() {}
 
            template<typename T>
            MetaStruct::enable_if_t<!MetaStruct::is_any_pointer<T>::value, void>
            initialize(const T & val) {
                if(SPARTA_EXPECT_FALSE(isCollected())) {
                    collect(val);
                }
            }
 
            template<typename T>
            MetaStruct::enable_if_t<MetaStruct::is_any_pointer<T>::value, void>
            initialize(const T & val){
                if(SPARTA_EXPECT_FALSE(isCollected())) {
                    collect(*val);
                }
            }
 
            template<typename T>
            MetaStruct::enable_if_t<!MetaStruct::is_any_pointer<T>::value, void>
            collect(const T & val)
            {
                collect_(val);
 
                // if the value pairs are not the same as the
                // previous value pairs and the previous record is still open
                if(!isSameRecord() && !record_closed_)
                {
                    //Close the old record (if there is one)
                    closeRecord();
                }
 
                // Remember the new value pairs for a new record and start
                // a new record if not empty.
                updateLastRecord_();
 
                // If the new Value pairs are not empty and current record is closed, we start a new record.
                if(record_closed_ && hasData_()){
                    startNewRecord_();
                    record_closed_ = false;
                }
            }
 
            template <typename T>
            MetaStruct::enable_if_t<MetaStruct::is_any_pointer<T>::value, void>
            collect(const T & val){
                // If pointer has become nullified, close the record
                if(nullptr == val) {
                    closeRecord();
                    return;
                }
                collect(*val);
            }
 
            template<typename T>
            MetaStruct::enable_if_t<!MetaStruct::is_any_pointer<T>::value, void>
            collectWithDuration(const T & val, sparta::Clock::Cycle duration){
                if(SPARTA_EXPECT_FALSE(isCollected()))
                {
                    if(duration != 0) {
                        ev_close_record_.preparePayload(false)->schedule(duration);
                    }
                    collect(val);
                }
            }
 
            template<typename T>
            MetaStruct::enable_if_t<MetaStruct::is_any_pointer<T>::value, void>
            collectWithDuration(const T & val, sparta::Clock::Cycle duration){
                // If pointer has become nullified, close the record
                if(nullptr == val) {
                    closeRecord();
                    return;
                }
                collectWithDuration(*val, duration);
            }
 
            void collect() override final {
                // If pointer has become nullified, close the record
                if(nullptr == collected_object_) {
                    closeRecord();
                    return;
                }
                collect(*collected_object_);
            }
 
            void collectWithDuration(sparta::Clock::Cycle duration) {
                // If pointer has become nullified, close the record
                if(nullptr == collected_object_) {
                    closeRecord();
                    return;
                }
                collectWithDuration(*collected_object_, duration);
            }
 
            void restartRecord() override final {
 
                // Do not get a new ID when we're doing a heartbeat
               if(!record_closed_) {
                    argos_record_.flags |= CONTINUE_FLAG;
                    writeRecord_();
                    argos_record_.flags &= ~CONTINUE_FLAG;
                    startNewRecord_();
                }
            }
 
            void closeRecord(const bool & simulation_ending = false) override final
            {
                if(SPARTA_EXPECT_FALSE(isCollected()))
                {
                    if(!record_closed_ && writeRecord_(simulation_ending)) {
 
                        // Clear the previous vector containing the Name Value pairs.
                        argos_record_.valueVector.clear();
                        argos_record_.stringVector.clear();
                    }
                    record_closed_ = true;
                }
            }
 
            void setManualCollection() {
                auto_collect_ = false;
            }
 
            //  if any of the old Values have changed or not.
            bool isSameRecord() const {
                return (argos_record_.valueVector == getDataVector()) && (argos_record_.stringVector == getStringVector());
            }
 
            const std::string & dumpNameValuePairs(const DataT & val) {
                collect_(val);
                log_string_.clear();
                std::ostringstream ss;
                for(const auto & pairs : getPEventLogVector()){
                    ss << pairs.first << "(" << pairs.second << ") ";
                }
                log_string_ = ss.str();
                return log_string_;
            }
 
        protected:
 
            inline EventSet & getEventSet_() {
                return event_set_;
            }
 
            //  and this is a pure virtual function in base
            //  else Collectable will also become Abstract.
            virtual void generateCollectionString_() override {}
 
        private:
 
            inline static uint64_t getUniquePairID_() {
                static uint64_t id = 0;
                if(id){ return id; }
                id = sparta::collection::UniquePairIDGenerator::getUniquePairID_();
                return id;
            }
 
            //  we collected from the last collection
            //  this cycle of Collection.
            void updateLastRecord_(){
                // These values will change every time the transaction changes
                argos_record_.valueVector = getDataVector();
                argos_record_.stringVector = getStringVector();
                argos_record_.sizeOfVector = getSizeOfVector();
 
                if (has_display_id_field_) {
                    argos_record_.display_ID = argos_record_.valueVector[0].first & 0x0fff;
                }
            }
 
            bool hasData_() const {
                return !(argos_record_.valueVector.empty() && argos_record_.stringVector.empty());
            }
 
            //  for writing to the Transaction Database File.
            bool writeRecord_(bool simulation_ending = false)
            {
                sparta_assert(pipeline_col_,
                            "Must startCollecting_ on this Collectable "
                            << getLocation() << " before a record can be written");
 
                // This record hasn't changed, don't write it
                if(SPARTA_EXPECT_FALSE(argos_record_.time_Start == pipeline_col_->getScheduler()->getCurrentTick())) {
                    return false;
                }
 
                // Set a new transaction ID since we're starting anew
                argos_record_.transaction_ID = pipeline_col_->getUniqueTransactionId();
 
                argos_record_.pairId = getUniquePairID_();
 
                // Capture the end time
                argos_record_.time_End =
                    pipeline_col_->getScheduler()->getCurrentTick() + (simulation_ending ? 1 : 0);
 
                // If this assert fires, then a user is trying to collect
                // a record multiple times in their cycle window or
                // something else really bad has happened.
                sparta_assert(argos_record_.time_Start < argos_record_.time_End);
 
                // Write the old record to the file
                pipeline_col_->writeRecord(argos_record_);
                return true;
            }
 
            void startNewRecord_() {
 
                // Set up the new start time for the new record
                argos_record_.time_Start = pipeline_col_->getScheduler()->getCurrentTick();
            }
 
            void setCollecting_(bool collect, Collector * collector) override final
            {
                pipeline_col_ = dynamic_cast<PipelineCollector *>(collector);
                sparta_assert(pipeline_col_ != nullptr,
                            "Collectables can only added to PipelineCollectors... for now");
 
                // These fields only need to be set once - they define the format of the collectable
                // and remain constant for the duration of the simulation
                argos_record_.nameVector = getNameStrings();
                argos_record_.length = argos_record_.nameVector.size();
                has_display_id_field_ = (argos_record_.nameVector[0] == "DID");
                for(const auto formatter: getFormatVector()) {
                    argos_record_.delimVector.emplace_back(formatter);
                }
 
                if(collect && hasData_()){
 
                    // Set the start time for this transaction to be
                    // moment collection is enabled.
                    startNewRecord_();
                }
 
                // If the collected object is null, this Collectable
                // object is to be explicitly collected
                if(collected_object_ && auto_collect_) {
                    if(collect) {
 
                        // Add this Collectable to the PipelineCollector's
                        // list of objects requiring collection
                        pipeline_col_->addToAutoCollection(this, collection_phase);
                    }
                    else {
 
                        // Remove this Collectable from the
                        // PipelineCollector's list of objects requiring
                        // collection
                        pipeline_col_->removeFromAutoCollection(this);
                    }
                }
            }
            // The pair collectable object to be collected
            const DataT * collected_object_ = nullptr;
 
            // The live transaction record
            pair_t argos_record_;
 
            // Ze Collec-tor
            PipelineCollector * pipeline_col_ = nullptr;
 
            // For those folks that want a value to automatically
            // disappear in the future
            sparta::EventSet event_set_;
            sparta::PayloadEvent<bool, sparta::SchedulingPhase::Trigger> ev_close_record_;
 
            // Is this collectable currently closed?
            bool record_closed_ = true;
 
            // Should we auto-collect?
            bool auto_collect_ = true;
 
            // Is the first field the display ID (DID)?
            bool has_display_id_field_ = false;
 
            std::string log_string_;
        };
    }//namespace collection
}//namespace sparta