PipelineCollector.hpp

Go to the documentation of this file.

// <PipelineCollector.hpp> -*- C++ -*-
 
#pragma once
 
#include <set>
#include <map>
 
#include "sparta/simulation/TreeNode.hpp"
#include "sparta/simulation/Clock.hpp"
#include "sparta/utils/SpartaAssert.hpp"
#include "sparta/collection/CollectableTreeNode.hpp"
#include "sparta/collection/Collector.hpp"
#include "sparta/events/EventSet.hpp"
#include "sparta/events/UniqueEvent.hpp"
#include "sparta/events/GlobalOrderingPoint.hpp"
#include "sparta/kernel/Scheduler.hpp"
 
#include "sparta/pipeViewer/Outputter.hpp"
#include "sparta/pipeViewer/ClockFileWriter.hpp"
#include "sparta/pipeViewer/LocationFileWriter.hpp"
#include "sparta/simulation/TreeNodePrivateAttorney.hpp"
 
namespace sparta{
namespace collection
{
    class PipelineCollector : public Collector
    {
        class CollectablesByClock
        {
        public:
            CollectablesByClock(const Clock * clk,
                                SchedulingPhase collection_phase) :
                ev_set_(nullptr)
            {
                switch(collection_phase)
                {
                case SchedulingPhase::Trigger:
                    ev_collect_.reset(new sparta::UniqueEvent<SchedulingPhase::Trigger>
                                      (&ev_set_, sparta::notNull(clk)->getName() + "_auto_collection_event_trigger",
                                       CREATE_SPARTA_HANDLER(CollectablesByClock, performCollection), 1));
                    break;
                case SchedulingPhase::Update:
                    ev_collect_.reset(new sparta::UniqueEvent<SchedulingPhase::Update>
                                      (&ev_set_, sparta::notNull(clk)->getName() + "_auto_collection_event_update",
                                       CREATE_SPARTA_HANDLER(CollectablesByClock, performCollection), 1));
                    break;
                case SchedulingPhase::PortUpdate:
                    ev_collect_.reset(new sparta::UniqueEvent<SchedulingPhase::PortUpdate>
                                      (&ev_set_, sparta::notNull(clk)->getName() + "_auto_collection_event_portupdate",
                                       CREATE_SPARTA_HANDLER(CollectablesByClock, performCollection), 1));
                    break;
                case SchedulingPhase::Flush:
                    ev_collect_.reset(new sparta::UniqueEvent<SchedulingPhase::Flush>
                                      (&ev_set_, sparta::notNull(clk)->getName() + "_auto_collection_event_flush",
                                       CREATE_SPARTA_HANDLER(CollectablesByClock, performCollection), 1));
                    break;
                case SchedulingPhase::Collection:
                    ev_collect_.reset(new sparta::UniqueEvent<SchedulingPhase::Collection>
                                      (&ev_set_, sparta::notNull(clk)->getName() + "_auto_collection_event_collection",
                                       CREATE_SPARTA_HANDLER(CollectablesByClock, performCollection), 1));
                    break;
                case SchedulingPhase::Tick:
                    ev_collect_.reset(new sparta::UniqueEvent<SchedulingPhase::Tick>
                                      (&ev_set_, sparta::notNull(clk)->getName() + "_auto_collection_event_tick",
                                       CREATE_SPARTA_HANDLER(CollectablesByClock, performCollection), 1));
                    break;
                case SchedulingPhase::PostTick:
                    ev_collect_.reset(new sparta::UniqueEvent<SchedulingPhase::PostTick>
                                      (&ev_set_, sparta::notNull(clk)->getName() + "_auto_collection_event_posttick",
                                       CREATE_SPARTA_HANDLER(CollectablesByClock, performCollection), 1));
                    break;
                case SchedulingPhase::__last_scheduling_phase:
                    sparta_assert(!"Should not have gotten here");
                    break;
                    // NO DEFAULT!  Allows for compiler errors if the enum
                    // class is updated.
                }
                ev_collect_->setScheduleableClock(clk);
                ev_collect_->setScheduler(clk->getScheduler());
                ev_collect_->setContinuing(false);
            }
 
            void enable(CollectableTreeNode * ctn) {
                enabled_ctns_.insert(ctn);
                // Schedule collect event in the next cycle in case
                // this is called in an unavailable pphase.
                ev_collect_->schedule(sparta::Clock::Cycle(1));
            }
 
            void disable(CollectableTreeNode * ctn) {
                enabled_ctns_.erase(ctn);
            }
 
            bool anyCollected() const {
                return !enabled_ctns_.empty();
            }
 
            void performCollection() {
                // std::for_each(enabled_ctns_.begin(),
                //               enabled_ctns_.end(), [&](CollectableTreeNode * it) {
                //                   std::cout << it->getName() << " " << ev_collect_.getClock()->currentCycle() << std::endl;
                //               });
 
                for(auto & ctn : enabled_ctns_) {
                    if(ctn->isCollected()) {
                        ctn->collect();
                    }
                }
                if(!enabled_ctns_.empty()) {
                    ev_collect_->schedule();
                }
            }
 
            void print() {
                for(auto ctn : enabled_ctns_) {
                    std::cout << '\t' << ctn->getName() << std::endl;
                }
            }
        private:
            EventSet ev_set_;
 
            std::unique_ptr<sparta::Scheduleable> ev_collect_;
            std::set<CollectableTreeNode*> enabled_ctns_;
        };
 
        // A map of the clock pointer and the structures that
        // represent collectables on that clock.
        std::map<const sparta::Clock *,
                 std::array<std::unique_ptr<CollectablesByClock>,
                            sparta::NUM_SCHEDULING_PHASES>> clock_ctn_map_;
 
        // Registered collectables
        std::set<CollectableTreeNode*> registered_collectables_;
 
    public:
 
        PipelineCollector(const std::string& filepath, Scheduler::Tick heartbeat_interval,
                          const sparta::Clock* root_clk, const sparta::TreeNode* root,
                          Scheduler* scheduler=nullptr) :
            Collector("PipelineCollector"),
            scheduler_(scheduler != nullptr ? scheduler : root_clk->getScheduler()),
            collector_events_(nullptr),
            ev_heartbeat_(&collector_events_, Collector::getName() + "_heartbeat_event",
                          CREATE_SPARTA_HANDLER(PipelineCollector, performHeartBeat_), 0)
        {
            // Sanity check - pipeline collection cannot occur without a scheduler
            sparta_assert(scheduler_);
 
            ev_heartbeat_.setScheduleableClock(root_clk);
            ev_heartbeat_.setScheduler(scheduler_);
 
            // We need to reverse the dependency order for the PostTick GOP and ev_heartbeat_ so that every other
            // event happens *before* ev_heartbeat_.
            // Doing this ensures that we don't accidentally mark a 1-cycle transaction as continued.
            DAG* dag = scheduler_->getDAG(); // Get a handle to the DAG
            sparta_assert(dag);
            Vertex* post_tick_gop = dag->getGOPoint("PostTick"); // Get a handle to the PostTick GOP
            sparta_assert(post_tick_gop);
 
            dag->unlink(ev_heartbeat_.getVertex(), post_tick_gop); // Undo the ev_heartbeat_ >> heartbeat_gop link
            post_tick_gop->precedes(ev_heartbeat_); // Set post_tick_gop >> heartbeat_gop
 
            sparta_assert(root != nullptr, "Pipeline Collection will not be able to create location file because it was passed a nullptr root treenode.");
            sparta_assert(root->isFinalized(), "Pipeline collection cannot be constructed until the sparta tree has been finalized.");
            // Assert that we got valid pointers necessary for pipeline collection.
            sparta_assert(root_clk != nullptr, "Cannot construct PipelineCollector because root clock is a nullptr");
            sparta_assert(scheduler_->isFinalized() == false, "Pipeline Collection cannot be instantiated after scheduler finalization -- it creates events");
 
            // Initialize the clock/collectable map and find the fastest clock
            const sparta::Clock* fastest_clk = nullptr;
            std::function<void (const sparta::Clock*)> addClks;
            addClks = [&addClks, this, &fastest_clk] (const sparta::Clock* clk)
                {
                    if(clk != nullptr){
                        auto & u_p = clock_ctn_map_[clk];
                        for(uint32_t i = 0; i < NUM_SCHEDULING_PHASES; ++i) {
                            u_p[i].reset(new CollectablesByClock(clk, static_cast<SchedulingPhase>(i)));
                        }
                        for(const sparta::TreeNode* child : sparta::TreeNodePrivateAttorney::getAllChildren(clk)) {
                            const sparta::Clock* child_clk = dynamic_cast<const sparta::Clock*>(child);
                            if(child_clk){
                                auto clk_period = child_clk->getPeriod();
                                // If this clock has a non-1 period (i.e. not the root clock)
                                // AND there is either
                                //   (A) no fastest clock yet
                                //    or
                                //   (B) this clock is a higher frequency than the fastest clock.
                                // then choose this as the fastest
                                if(clk_period != 1 && (!fastest_clk || (clk_period < fastest_clk->getPeriod()))) {
                                    fastest_clk = child_clk;
                                }
                                addClks(child_clk);
                            }
                        }
                    }
                };
            addClks(root_clk);
            if(fastest_clk == nullptr){
                fastest_clk = root_clk;
            }
 
            // A multiple to multiply against the fastest clock when no heartbeat was set.
            static const uint32_t heartbeat_multiplier = 200;
            // round heartbeat using a multiple of the fastest clock if one was not set.
            if (heartbeat_interval == 0)
            {
                sparta_assert(fastest_clk->getPeriod() != 0);
                heartbeat_interval = fastest_clk->getPeriod() * heartbeat_multiplier;                //round up to the nearest multiple of 100
                heartbeat_interval = heartbeat_interval + (100 - (heartbeat_interval % 100));
                // pipeViewer requires that intervals be a multiple of 100.
                sparta_assert(heartbeat_interval % 100 == 0)
            }
 
            // We are gonna subtract one from the heartbeat_interval
            // later.. Better be greater than one.
            sparta_assert(heartbeat_interval > 1);
            // Initialize some values.
            filepath_           = filepath;
            heartbeat_interval_ = heartbeat_interval;
            closing_time_       = heartbeat_interval;
            root_clk_           = root_clk;
 
            ev_heartbeat_.setContinuing(false); // This event does not keep simulation going
        }
 
        ~PipelineCollector() {
            // XXX This is a little goofy looking.  Should we make sparta_abort that takes conditional
            //    be sparta_abort_unless()?
            sparta_abort(collection_active_ != true,
                         "The PipelineCollector was not torn down properly. Before "
                         "tearing down the simulation tree, you must call "
                         "destroy() on the collector");
        }
 
        void destroy()
        {
            if(collection_active_) {
                sparta_assert(writer_ != nullptr, "Somehow collection is active, but we have a null writer");
                for(auto & ctn : registered_collectables_) {
                    if(ctn->isCollected()) {
                        ctn->closeRecord(true); // set true for simulation termination
                    }
                }
            }
            registered_collectables_.clear();
            writer_.reset();
            collection_active_ = false;
        }
 
        void reactivate(const std::string& filepath)
        {
            sparta_assert(!collection_active_,
                          "You can only reactivate the PipelineCollector after you destroy it");
            filepath_ = filepath;
        }
 
        void startCollection(sparta::TreeNode* starting_node)
        {
            if(collection_active_ == false)
            {
                // Create the outputter used for writing transactions to disk.
                writer_.reset(new pipeViewer::Outputter(filepath_, heartbeat_interval_));
 
                // We need to write an index on the start BEFORE any transactions have been written.
                writer_->writeIndex();
 
                // Write the clock information out
                writeClockFile_();
 
                // Open the locations file
                location_writer_.reset(new pipeViewer::LocationFileWriter(filepath_));
 
                // The reader needs heartbeat indexes up to the current
                // collection point.  This can happen on delayed pipeline
                // collection.  Start the heartbeats at the first interval
                // as the Outputter class handles hb 0
                last_heartbeat_ = 0;
                const uint64_t num_hb = scheduler_->getCurrentTick()/heartbeat_interval_;
                uint64_t cnt = 0;
                while(cnt != num_hb) {
                    // write an index
                    writer_->writeIndex();
                    last_heartbeat_ += heartbeat_interval_;
                    ++cnt;
                }
 
                // Schedule a heartbeat at the next interval, offset from
                // the current tick.  For example, if the scheduler is at
                // 6,600,456, then we want to schedule at 7,000,000 if the
                // interval is 1M and the num_hb is 6
                ev_heartbeat_.scheduleRelativeTick(((num_hb + 1) * heartbeat_interval_) -
                                                   scheduler_->getCurrentTick(), scheduler_);
 
                collection_active_ = true;
            }
 
            *(location_writer_.get()) << (*starting_node);
 
            // Recursively collect the start node and children
            std::function<void (sparta::TreeNode* starting_node)> recursiveCollect;
            recursiveCollect = [&recursiveCollect, this] (sparta::TreeNode* starting_node)
                {
                    // First turn on this node if it's actually a CollectableTreeNode
                    CollectableTreeNode* c_node = dynamic_cast<CollectableTreeNode*>(starting_node);
                    if(c_node != nullptr) {
                        c_node->startCollecting(this);
                        registered_collectables_.insert(c_node);
                    }
 
                    // Recursive step. Go through the children and turn them on as well.
                    for(sparta::TreeNode* node : sparta::TreeNodePrivateAttorney::getAllChildren(starting_node))
                    {
                        recursiveCollect(node);
                    }
                };
 
            recursiveCollect(starting_node);
        }
 
        void stopCollection(sparta::TreeNode* starting_node)
        {
            std::function<void (sparta::TreeNode* starting_node)> recursiveStopCollect;
            recursiveStopCollect = [&recursiveStopCollect, this] (sparta::TreeNode* starting_node) {
                // First turn off this node if it's actually a CollectableTreeNode
                CollectableTreeNode* c_node = dynamic_cast<CollectableTreeNode*>(starting_node);
                if(c_node != nullptr)
                {
                    c_node->stopCollecting(this);
                    registered_collectables_.erase(c_node);
                }
 
                // Recursive step. Go through the children and turn them on as well.
                for(sparta::TreeNode* node : sparta::TreeNodePrivateAttorney::getAllChildren(starting_node))
                {
                    recursiveStopCollect(node);
                }
            };
            recursiveStopCollect(starting_node);
 
            bool still_active = !registered_collectables_.empty();
            // for(auto & cp : clock_ctn_map_) {
            //     if(cp.second->anyCollected()) {
            //         still_active = true;
            //         break;
            //     }
            // }
            collection_active_ = still_active;
        }
 
        void stopCollection() {
            for(auto & col : registered_collectables_) {
                col->stopCollecting(this);
            }
            registered_collectables_.clear();
        }
 
        void addToAutoCollection(CollectableTreeNode * ctn,
                                 SchedulingPhase collection_phase = SchedulingPhase::Tick)
        {
            auto ccm_pair = clock_ctn_map_.find(ctn->getClock());
            sparta_assert(ccm_pair != clock_ctn_map_.end());
            ccm_pair->second[static_cast<uint32_t>(collection_phase)]->enable(ctn);
        }
 
        void removeFromAutoCollection(CollectableTreeNode * ctn)
        {
            auto ccm_pair = clock_ctn_map_.find(ctn->getClock());
            sparta_assert(ccm_pair != clock_ctn_map_.end());
            for(auto & u_p : ccm_pair->second) {
                u_p->disable(ctn);
            }
        }
 
        uint64_t getUniqueTransactionId()
        {
            // make sure we are not going to overflow our int,
            // if we did overflow our id's are no longer unique!
            sparta_assert(last_transaction_id_ < (~(uint64_t)0));
            return ++last_transaction_id_;
        }
 
        template<class R_Type>
        void writeRecord(const R_Type& dat)
        {
            sparta_assert(collection_active_, "The pipeline head must be running in order to write a transaction");
 
            // Make sure it's within the heartbeat window
            sparta_assert(dat.time_End <= (last_heartbeat_ + heartbeat_interval_));
            sparta_assert(dat.time_Start >= last_heartbeat_);
 
 
            // Make sure transactions are exclusive.
            // transaction [4999-5000] should be written BEFORE the index is written for transactions
            // starting at 5000
            sparta_assert(closing_time_ < heartbeat_interval_  // Ignore first heartbeat
                              || dat.time_Start >= closing_time_ - heartbeat_interval_,
                              "Attempted to write a pipeout record with exclusive start =("
                              << dat.time_Start << "), less than closing of previous interval"
                              << closing_time_ - heartbeat_interval_ );
 
            // std::cout << "writing annt. " << "loc: " << dat.location_ID << " start: "
            //           << dat.time_Start << " end: " << dat.time_End
            //           << " parent: " << dat.parent_ID << std::endl;
 
            writer_->writeTransaction<R_Type>(dat);
            ++transactions_written_;
        }
 
        uint64_t numTransactionsWritten() const
        {
            return transactions_written_;
        }
 
        bool isCollectionActive() const {
            return collection_active_;
        }
 
        void printMap() {
            //std::cout << "Printing Map Not Supported" << std::endl;
            // for(auto & p : clock_ctn_map_) {
            //     std::cout << "\nClock            : " << p.first->getName()
            //               << "\nAuto collectables: " << std::endl;
            //     p.second->print();
            // }
        }
 
        const std::string & getFilePath() const {
            return filepath_;
        }
 
        Scheduler* getScheduler() const {
            return scheduler_;
        }
 
    private:
 
        void writeClockFile_()
        {
            // We only need the ClockFileWriter to exist during the writing of the clock file.
            // there for it was created on the stack.
            //std::cout << "Writing Pipeline Collection clock file. " << std::endl;
            pipeViewer::ClockFileWriter clock_writer(filepath_);
            clock_writer << (*root_clk_);
        }
 
        void performHeartBeat_()
        {
            if(collection_active_) {
                // Close all transactions
                for(auto & ctn : registered_collectables_) {
                    if(ctn->isCollected()) {
                        ctn->restartRecord();
                    }
                }
 
                // write an index
                writer_->writeIndex();
 
                // Remember the last time we recorded a heartbeat
                last_heartbeat_ = scheduler_->getCurrentTick();
 
                // Schedule another heartbeat
                ev_heartbeat_.schedule(heartbeat_interval_);
            }
        }
 
        std::unique_ptr<pipeViewer::Outputter> writer_;
 
        std::unique_ptr<pipeViewer::LocationFileWriter> location_writer_;
        //sparta::TreeNode* collected_treenode_ = nullptr;
 
        const sparta::Clock * root_clk_ = nullptr;
 
        std::string filepath_;
 
        uint64_t last_transaction_id_ = 0;
 
        uint64_t transactions_written_ = 0;
 
        uint64_t heartbeat_interval_ = 0;
 
        Scheduler::Tick last_heartbeat_ = 0;
 
        Scheduler * scheduler_;
 
        EventSet collector_events_;
        UniqueEvent<SchedulingPhase::PostTick> ev_heartbeat_;
 
        uint64_t closing_time_ = 0;
 
        bool collection_active_ = false;
 
    };
 
}// namespace collection
}// namespace sparta