IterableCollector.hpp

// <IterableCollector.hpp> -*- C++ -*-
 
/*
 */
 
#pragma once
 
#include <memory>
#include <sstream>
#include <string>
#include <vector>
#include <iterator>
#include <type_traits>
#include "sparta/utils/Utils.hpp"
#include "sparta/log/MessageSource.hpp"
#include "sparta/events/SchedulingPhases.hpp"
#include "sparta/collection/Collectable.hpp"
#include "sparta/collection/CollectableTreeNode.hpp"
#include "sparta/collection/PipelineCollector.hpp"
 
namespace sparta {
namespace collection {
 
template <typename IterableType, SchedulingPhase collection_phase = SchedulingPhase::Collection, bool sparse_array_type = false>
class IterableCollector : public CollectableTreeNode
{
public:
    typedef typename IterableType::size_type size_type;
 
    IterableCollector (TreeNode * parent,
                       const std::string & name,
                       const std::string & group,
                       const uint32_t index,
                       const std::string & desc,
                       const IterableType * iterable,
                       const size_type expected_capacity) :
        CollectableTreeNode(parent, name, group, index, desc),
        iterable_object_(iterable),
        expected_capacity_(expected_capacity),
        event_set_(this),
        ev_close_record_(&event_set_, name + "_pipeline_collectable_close_event",
                         CREATE_SPARTA_HANDLER_WITH_DATA(IterableCollector, closeRecord, bool))
    {
        for (size_type i = 0; i < expected_capacity_; ++i)
        {
            std::stringstream name_str;
            name_str << name << i;
            positions_.emplace_back(new CollectableT(this, name_str.str(), group, i));
        }
    }
 
 
    IterableCollector (TreeNode * parent,
                       const std::string & name,
                       const std::string & group,
                       const uint32_t index,
                       const std::string & desc,
                       const IterableType & iterable,
                       const size_type expected_capacity) :
        IterableCollector(parent, name, group, index, desc, &iterable, expected_capacity)
    {}
 
    IterableCollector (TreeNode * parent,
                       const std::string & name,
                       const std::string & desc,
                       const IterableType * iterable,
                       const size_type expected_capacity) :
        IterableCollector(parent, name, name, 0, desc, iterable, expected_capacity)
    {}
 
    IterableCollector (TreeNode * parent,
                       const std::string & name,
                       const std::string & desc,
                       const IterableType & iterable,
                       const size_type expected_capacity) :
        IterableCollector(parent, name, name, 0, desc, &iterable, expected_capacity)
    {}
 
    IterableCollector (TreeNode * parent,
                       const std::string & name,
                       const IterableType * iterable,
                       const size_type expected_capacity) :
        IterableCollector (parent, name, name + " Iterable Collector",
                           iterable, expected_capacity)
    {
        // Delegated constructor
    }
 
    IterableCollector (TreeNode * parent,
                       const std::string & name,
                       const IterableType & iterable,
                       const size_type expected_capacity) :
        IterableCollector (parent, name, name + " Iterable Collector",
                           &iterable, expected_capacity)
    {
        // Delegated constructor
    }
 
    IterableCollector (TreeNode * parent,
                       const std::string & name,
                       const size_type expected_capacity) :
        IterableCollector (parent, name, name + " Iterable Collector",
                           nullptr, expected_capacity)
    {
        // Can't auto collect without setting iterable_object_
        setManualCollection();
    }
 
    void createSimDbEntryPoint(simdb::argos::ArgosCollector* argos_collector) override final
    {
        if constexpr (use_dynamic_fields<BinValueT>) {
            std::ostringstream oss;
            oss << "Collecting non-trivial classes using operator<< is no longer supported. Use PairDefinition.\n";
            oss << "  - path: " << getLocation() << "\n";
            oss << "  - type: " << simdb::demangle_type<BinValueT>() << " (container)";
            argos_collector->postNotif(oss.str(), simdb::argos::NotifType::WARNING);
        } else {
            auto loc = getLocation();
            auto clk_name = notNull(getClock())->getName();
            auto type = encodeCollectedType();
            entry_point_ = argos_collector->createContainerCollector(loc, clk_name, type);
 
            auto tiny_strings = argos_collector->getTinyStrings();
            auto enum_inspector = argos_collector->getEnumInspector();
            bit_bucket_ = std::make_shared<IterableCollectorBitBucket<sparse_array_type>>(tiny_strings, enum_inspector, expected_capacity_);
            for (auto & bin : positions_) {
                bin->setBitBucket(bit_bucket_);
            }
        }
    }
 
    void serializeStructSchema(simdb::DatabaseManager* db_mgr, std::set<std::string>& serialized_types) override final
    {
        positions_.at(0)->serializeStructSchema(db_mgr, serialized_types);
    }
 
    std::string encodeCollectedType(bool human_readable = false) const override final
    {
        auto type = positions_.at(0)->encodeCollectedType();
        auto human_readable_type = positions_.at(0)->encodeCollectedType(true);
        auto human_readable_desc = human_readable_type.substr(type.size() + 1);
 
        if constexpr (sparse_array_type) {
            type += "_sparse";
        } else {
            type += "_contig";
        }
        type += "_capacity" + std::to_string(expected_capacity_);
        if (human_readable) {
            type += " " + human_readable_desc;
        }
        return type;
    }
 
    void collect(const IterableType * iterable_object)
    {
        // If pointer has become nullified, close the records
        if(nullptr == iterable_object) {
            closeRecord();
        }
        else if (SPARTA_EXPECT_FALSE(isCollected()))
        {
            if(SPARTA_EXPECT_FALSE(iterable_object->size() > expected_capacity_))
            {
                if(SPARTA_EXPECT_FALSE(warn_on_size_))
                {
                    sparta::log::MessageSource::getGlobalWarn()
                        << "WARNING! The collected object '"
                        << getLocation() << "' has grown beyond the "
                        << "expected capacity (given at construction) for collection. "
                        << "Expected " << expected_capacity_ << " but grew to "
                        << iterable_object->size()
                        << " This is your first and last warning.";
                    warn_on_size_ = false;
                }
            }
            collectImpl_(iterable_object, std::integral_constant<bool, sparse_array_type>());
        }
    }
 
    void collect() override {
        collect(iterable_object_);
    }
 
    void closeRecord(const bool & = false) override {
        if(SPARTA_EXPECT_FALSE(isCollected() && entry_point_)) {
            entry_point_->closeRecord();
        }
    }
 
    void scheduleCloseRecord(sparta::Clock::Cycle cycle) {
        if(SPARTA_EXPECT_FALSE(isCollected())) {
            ev_close_record_.preparePayload(false)->schedule(cycle);
        }
    }
 
    void setManualCollection() {
        auto_collect_ = false;
    }
 
    void collectWithDuration(sparta::Clock::Cycle duration) {
        if(SPARTA_EXPECT_FALSE(isCollected())) {
            collect();
            if(duration != 0) {
                ev_close_record_.preparePayload(false)->schedule(duration);
            }
        }
    }
 
    void reattach(const IterableType * obj) {
        iterable_object_ = obj;
    }
 
private:
    typedef typename std::iterator_traits<typename IterableType::iterator>::value_type BinT;
    typedef MetaStruct::remove_any_pointer_t<BinT> BinValueT;
    typedef Collectable<BinT> CollectableT;
    typedef IterableCollectorBitBucket<sparse_array_type> BitBucketT;
 
    // Standard walk of iterable types
    void collectImpl_(const IterableType * iterable_object, std::false_type)
    {
        // Unit tests may not have any entry point / bit bucket / simulation set up.
        if (!entry_point_) {
            return;
        }
 
        sparta_assert(bit_bucket_);
        sparta_assert(nullptr != iterable_object,
            "Can't collect iterable_object because it's a nullptr! How did we get here?");
        auto itr = iterable_object->begin();
        auto eitr = iterable_object->end();
        auto size = std::distance(itr, eitr);
        bit_bucket_->clear();
        for (uint32_t i = 0; i < size; ++i)
        {
            bit_bucket_->setActiveBinIdx(i);
            positions_[i]->collect(*itr);
            ++itr;
        }
 
        bit_bucket_->writeTo(entry_point_);
    }
 
    // Full iteration walk, checking validity of the iterator.  This
    // is used for Pipe and Array where the iterator points to valid
    // and not valid entries in the component
    void collectImpl_(const IterableType * iterable_object, std::true_type)
    {
        // Unit tests may not have any entry point / bit bucket / simulation set up.
        if (!entry_point_) {
            return;
        }
 
        sparta_assert(bit_bucket_);
        sparta_assert(nullptr != iterable_object,
            "Can't collect iterable_object because it's a nullptr! How did we get here?");
        auto itr = iterable_object->begin();
        bit_bucket_->clear();
        for (uint32_t i = 0; i < expected_capacity_; ++i)
        {
            sparta_assert(itr != iterable_object->end());
            if (itr.isValid()) {
                bit_bucket_->setActiveBinIdx(i);
                positions_[i]->collect(*itr);
            }
            ++itr;
        }
 
        bit_bucket_->writeTo(entry_point_);
    }
 
    void setCollecting_(bool collect, Collector * collector) override
    {
        if(collect && auto_collect_) {
            PipelineCollector * pipeline_col = dynamic_cast<PipelineCollector *>(collector);
            sparta_assert(pipeline_col != nullptr);
 
            // Add this Collectable to the PipelineCollector's
            // list of objects requiring collection
            pipeline_col->addToAutoCollection(this, collection_phase);
        }
        else {
            closeRecord();
        }
    }
 
    const IterableType * iterable_object_;
    std::vector<std::unique_ptr<CollectableT>> positions_;
    const size_type expected_capacity_ = 0;
    bool auto_collect_ = true;
    bool warn_on_size_ = true;
    std::shared_ptr<BitBucketT> bit_bucket_;
 
    // 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_;
 
};
} // namespace collection
} // namespace sparta