StreamNode.hpp

// <StreamNode> -*- C++ -*-
 
#pragma once
 
#include "sparta/report/Report.hpp"
#include "sparta/utils/SpartaAssert.hpp"
 
#include <queue>
#include <mutex>
 
#include <vector>
#include <string>
#include <sstream>
#include <algorithm>
 
namespace sparta {
 
class Report;
class StatisticInstance;
 
namespace statistics {
 
class StreamController;
 
class StreamNode
{
public:
    virtual ~StreamNode() {}
    StreamNode(const StreamNode &) = delete;
    StreamNode & operator=(const StreamNode &) = delete;
 
    const std::string & getName() const {
        return name_;
    }
 
    const std::string & getFullPath() {
        if (!full_path_.empty()) {
            return full_path_;
        }
        resolveFullPath_();
        return full_path_;
    }
 
    std::vector<std::shared_ptr<StreamNode>> & getChildren() {
        return children_;
    }
 
    const std::vector<std::shared_ptr<StreamNode>> & getChildren() const {
        return children_;
    }
 
    void setParent(StreamNode * parent) {
        sparta_assert(parent_ == nullptr || parent_ == parent,
                    "Cannot reassign a StreamNode's parent node");
        parent_ = parent;
    }
 
    StreamNode * getRoot() {
        if (cached_root_ != nullptr) {
            return cached_root_;
        }
 
        //Walk to the top node, and cache it for faster
        //access later on
        StreamNode * parent = this;
        while (parent != nullptr) {
            cached_root_ = parent;
            parent = cached_root_->parent_;
        }
        return cached_root_;
    }
 
    void initialize() {
        if (is_initialized_) {
            return;
        }
 
        //Subclasses now turn their vector/scalar SI(s) into a single
        //vector<double>, with all of their SI's connected to that
        //data vector via SnapshotLogger's. When anyone asks those
        //SI(s) what their current value is, the double value will
        //be written into our vector. It also goes into that vector
        //at the same index every time, so we can safely send the
        //entire vector to any sink for faster processing than if
        //we had to process just one point at a time.
        initialize_();
 
        //The reporting infrastructure will notify our root node
        //whenever a report write/update was made. We need the root
        //to forward that call to us so we can put the SI data vector
        //into a buffer queue for asynchronous processing.
        getRoot()->addStreamListener_(this);
 
        is_initialized_ = true;
    }
 
    bool notifyListenersOfStreamUpdate();
 
    void pushStreamUpdateToListeners();
 
    void getBufferedStreamData(std::queue<std::vector<double>> & data_queue);
 
    void setStreamController(const std::shared_ptr<StreamController> & controller) {
        sparta_assert(controller_ == nullptr,
                    "You cannot reset a StreamNode's stream controller");
        controller_ = controller;
    }
 
protected:
    explicit StreamNode(const std::string & name) :
        name_(name)
    {
        sparta_assert(!name_.empty(), "You may not create a StreamNode without a name");
    }
 
private:
    //When a report update occurs, the root stream node will be notified.
    //This root node contains all of the child nodes that have a streaming
    //client attached to them. We call these children "listeners".
    //
    //    Report update -> Root StreamNode
    //                              |
    //                    ---------------------
    //                    |                   |
    //                  ChildA             ChildB
    //                (no client)       (has clients)
    //
    //In this example, listeners_ = {ChildB*}
    void addStreamListener_(StreamNode * listener) {
        std::lock_guard<std::mutex> guard(listeners_mutex_);
        listeners_.emplace_back(listener);
    }
 
    //The simulation synchronously pushes packets of data into
    //the root StreamNode, and we keep that data organized in
    //a map of child StreamNode* -> queue<packet>
    //
    //This data can be consumed on a separate thread if desired.
    //
    // ** TEMPORARY: While asynchronous C++/Python communication
    // ** is developed, we will process this buffered data from
    // ** the main thread.
    void appendDataValuesForListener_(StreamNode * listener,
                                      const std::vector<double> & data)
    {
        std::lock_guard<std::mutex> guard(listeners_mutex_);
        listeners_data_[listener].push(data);
    }
 
    //The consumer thread (or the main thread during a forced
    //synchronous flush) is requesting all buffered data for
    //a particular client. We do not do any bookkeeping to
    //account for that released data. As far as the StreamNode
    //is concerned, the data is gone forever.
    void releaseDataBufferForListener_(
        StreamNode * listener,
        std::queue<std::vector<double>> & data_queue)
    {
        std::lock_guard<std::mutex> guard(listeners_mutex_);
        auto iter = listeners_data_.find(listener);
        if (iter != listeners_data_.end()) {
            std::swap(data_queue, iter->second);
        }
    }
 
    //Walk up to the topmost node in our hierarchy, and create
    //a string for the full node path.
    void resolveFullPath_() {
        std::vector<std::string> path;
        path.push_back(name_);
 
        const StreamNode * parent = parent_;
        while (parent) {
            path.emplace_back(parent->name_);
            parent = parent->parent_;
        }
 
        //We now have a vector that looks like:
        //    "ipc.rob.core0.top"
        //
        //So flip it around and dot-delimit it like TreeNode's do
        std::reverse(path.begin(), path.end());
        std::ostringstream oss;
        for (const auto & p : path) {
            oss << p << ".";
        }
 
        full_path_ = oss.str();
        if (!full_path_.empty() && full_path_.back() == '.') {
            full_path_.pop_back();
        }
    }
 
    virtual void initialize_() = 0;
 
    virtual const std::vector<double> & readFromStream_() = 0;
 
    const std::string name_;
    std::string full_path_;
 
    StreamNode * parent_ = nullptr;
    StreamNode * cached_root_ = nullptr;
 
    std::vector<std::shared_ptr<StreamNode>> children_;
    bool is_initialized_ = false;
 
    std::vector<StreamNode*> listeners_;
 
    std::unordered_map<
        StreamNode*,
        std::queue<std::vector<double>>> listeners_data_;
 
    std::mutex listeners_mutex_;
 
    std::shared_ptr<StreamController> controller_;
};
 
class ReportStreamNode : public StreamNode
{
public:
    ReportStreamNode(const std::string & name,
                     const Report * r) :
        StreamNode(name),
        report_(r)
    {
        sparta_assert(report_, "Null Report given to a ReportStreamNode");
    }
 
    ReportStreamNode(const ReportStreamNode &) = delete;
    ReportStreamNode & operator=(const ReportStreamNode &) = delete;
 
private:
    void initialize_() override;
 
    const std::vector<double> & readFromStream_() override;
 
    const Report * report_ = nullptr;
    std::vector<const StatisticInstance*> stat_insts_;
    std::vector<double> aggregated_si_values_;
};
 
class StatisticInstStreamNode : public StreamNode
{
public:
    StatisticInstStreamNode(const std::string & name,
                            const StatisticInstance * si) :
        StreamNode(name),
        stat_inst_(si)
    {
        sparta_assert(stat_inst_, "Null StatisticInstance given to a "
                                "StatisticInstStreamNode");
    }
 
    StatisticInstStreamNode(const StatisticInstStreamNode &) = delete;
    StatisticInstStreamNode & operator=(const StatisticInstStreamNode &) = delete;
 
private:
    void initialize_() override;
 
    const std::vector<double> & readFromStream_() override;
 
    const StatisticInstance * stat_inst_ = nullptr;
    std::vector<double> one_si_value_;
};
 
} // namespace statistics
} // namespace sparta