PhasedPayloadEvent.hpp

Go to the documentation of this file.

// <PayloadEvent.h> -*- C++ -*-
 
 
#pragma once
 
#include <memory>
#include "sparta/events/EventNode.hpp"
#include "sparta/events/Scheduleable.hpp"
#include "sparta/events/SchedulingPhases.hpp"
#include "sparta/utils/MetaStructs.hpp"
#include "sparta/utils/ValidValue.hpp"
#include "sparta/utils/FastList.hpp"
#include "sparta/events/StartupEvent.hpp"
 
namespace sparta
{
 
    template<class DataT>
    class PhasedPayloadEvent : public EventNode
    {
    private:
 
        class PayloadDeliveringProxy;
        using ProxyAllocation   = std::vector<std::unique_ptr<PayloadDeliveringProxy>>;
        using ProxyFreeList     = std::vector<PayloadDeliveringProxy *>;
        using ProxyInflightList = sparta::utils::FastList <PayloadDeliveringProxy *>;
 
        static constexpr size_t INITIAL_OUTSTANDING = 16;
 
        class PayloadDeliveringProxy : public Scheduleable
        {
        public:
 
            void scheduleRelativeTick(sparta::Scheduler::Tick rel_tick,
                                      sparta::Scheduler * scheduler) override
            {
                sparta_assert((cancelled_ == false) && (scheduled_ == false),
                            "This Payload handle is already scheduled or was previously cancelled.  "
                            "To schedule again, you must create a new one");
                Scheduleable::scheduleRelativeTick(rel_tick, scheduler);
                scheduled_ = true;
            }
 
 
            PayloadDeliveringProxy(const Scheduleable & prototype,
                                   PhasedPayloadEvent<DataT> * parent) :
                Scheduleable(prototype),
                parent_(parent),
                target_consumer_event_handler_(prototype.getHandler()),
                loc_(parent_->inflight_pl_.end())
            {
                // Reset the base class consumer event handler to be
                // this class' delivery proxy
                Scheduleable::consumer_event_handler_ =
                    CREATE_SPARTA_HANDLER(PayloadDeliveringProxy, deliverPayload_);
            }
 
        private:
 
            // Make the parent class a friend
            friend class PhasedPayloadEvent;
 
            // If this Scheduleable is managed by a
            // ScheduleableHandle, then this method is called when the
            // Handle goes out of scope.  The proxy is only reclaimed
            // if it's not on the Scheduler and there are not handles
            // pointing to it.
            void reclaim_() override {
                if(!scheduled_ && (getScheduleableHandleCount_() == 0)) {
                    parent_->reclaimProxy_(loc_);
                    cancelled_ = false;
                }
            }
 
            // This Scheduleable was cancelled via an indirect cancel
            // on the Scheduler
            void eventCancelled_() override {
                scheduled_ = false;
                cancelled_ = true;
                reclaim_();
            }
 
            // Set a payload for a delayed delivery
            void setPayload_(const DataT & pl) {
                sparta_assert(scheduled_ == false);
                payload_ = new (&payload_storage_) DataT(pl);
            }
 
            // Destroy payload
            void destroyPayload_() {
                sparta_assert(scheduled_ == false);
                std::destroy_at(payload_);
            }
 
            // Get a payload for a delayed delivery
            const DataT & getPayload_() const {
                return *payload_;
            }
 
 
            // Set the location in the inflight list of the parent for
            // quick removal
            void setInFlightLocation_(const typename ProxyInflightList::iterator & loc) {
                loc_ = loc;
            }
 
            // Called by the SPARTA scheduler.  Deliver the payload to
            // the user's callback.  Then, recycle this Proxy
            void deliverPayload_() {
                sparta_assert(scheduled_ == true,
                            "Some construct is trying to deliver a payload twice: "
                            << parent_->name_ << " to handler: "
                            << target_consumer_event_handler_.getName());
                scheduled_ = false;
                target_consumer_event_handler_((const void*)payload_);
                reclaim_();
            }
 
            PhasedPayloadEvent<DataT> * parent_ = nullptr;
            const SpartaHandler         target_consumer_event_handler_;
            DataT *                     payload_;
            alignas(DataT) std::byte    payload_storage_[sizeof(DataT)];
            typename ProxyInflightList::iterator loc_;
            bool scheduled_ = false;
            bool cancelled_ = false;
        };
 
 
        ScheduleableHandle allocateProxy_(const DataT & dat)
        {
            PayloadDeliveringProxy * proxy = nullptr;
            if(SPARTA_EXPECT_TRUE(free_idx_ != 0)) {
                --free_idx_;
                proxy = free_pl_[free_idx_];
            }
            else {
                // See if we hit the ceiling
                if(allocation_idx_ == allocated_proxies_.size()) {
                    addProxies_();
                }
 
                // Take one from the allocation list
                proxy = allocated_proxies_[allocation_idx_].get();
                ++allocation_idx_;
 
                sparta_assert(allocation_idx_ < inflight_pl_.max_size(),
                              "The PayloadEvent: '" << getLocation() <<
                              "' has allocated over " << inflight_pl_.max_size() <<
                              " outstanding events -- does that seem right?");
            }
            proxy->setInFlightLocation_(inflight_pl_.emplace_back(proxy));
            proxy->setPayload_(dat);
 
            return proxy;
        }
 
        friend class PayloadDeliveringProxy;
 
        void reclaimProxy_(typename ProxyInflightList::iterator & pl_location) {
            if (SPARTA_EXPECT_FALSE(pl_location == inflight_pl_.end())) {
                return;
            }
            (*pl_location)->destroyPayload_();
            free_pl_[free_idx_++] = *pl_location;
            inflight_pl_.erase(pl_location);
            pl_location = inflight_pl_.end();
        }
 
    public:
 
        static constexpr size_t DEFAULT_MAX_OUTSTANDING = 16384;
 
        /*
         * \brief Create a PhasedPayloadEvent to deliver data at a particular time
         * \param event_set The sparta::EventSet this PhasedPayloadEvent belongs to
         * \param name      The name of this event (as it shows in the EventSet)
         * \param sched_phase The SchedulingPhase this PhasedPayloadEvent belongs to
         * \param consumer_event_handler A SpartaHandler to the consumer's event_handler
         * \param delay The relative time (in Cycles) from "now" to schedule
         * \param max_outstanding Ceiling on simultaneously in-flight payloads
         *
         * The suggestion is to use the derived class sparta::PayloadEvent
         * instead of this class directly.
         */
        PhasedPayloadEvent(TreeNode            * event_set,
                           const std::string   & name,
                           SchedulingPhase       sched_phase,
                           const SpartaHandler & consumer_event_handler,
                           Clock::Cycle          delay = 0,
                           size_t                max_outstanding = DEFAULT_MAX_OUTSTANDING) :
            EventNode(event_set, name, sched_phase),
            name_(name + "[" + consumer_event_handler.getName() + "]"),
            prototype_(consumer_event_handler, delay, sched_phase),
            inflight_pl_(INITIAL_OUTSTANDING, max_outstanding)
        {
            sparta_assert(consumer_event_handler.argCount() == 1,
                          "You must assign a PhasedPayloadEvent a consumer handler "
                          "that takes exactly one argument");
            prototype_.setScheduleableClock(getClock());
            prototype_.setScheduler(determineScheduler(getClock()));
        }
 
        virtual ~PhasedPayloadEvent() {
            for(PayloadDeliveringProxy * proxy : inflight_pl_) {
                std::destroy_at(proxy->payload_);
            }
        }
 
        PhasedPayloadEvent<DataT> & operator=(const PhasedPayloadEvent<DataT> &) = delete;
        PhasedPayloadEvent(PhasedPayloadEvent<DataT> &&) = delete;
        PhasedPayloadEvent(const PhasedPayloadEvent<DataT> &) = delete;
 
        ScheduleableHandle preparePayload(const DataT & payload) {
            return allocateProxy_(payload);
        }
 
        Scheduleable& operator>>(Scheduleable & consumer)
        {
            prototype_ >> consumer;
            return consumer;
        }
 
        // Get the underlying Scheduleable prototype
        Scheduleable & getScheduleable() override {
            return prototype_;
        }
 
        void setContinuing(bool continuing) {
            getScheduleable().setContinuing(continuing);
        }
 
        uint32_t getNumOutstandingEvents() const {
            return inflight_pl_.size();
        }
 
        bool isScheduled(Clock::Cycle rel_cycle) const {
            for(auto * proxy : inflight_pl_) {
                if(proxy->isScheduled(rel_cycle)) {
                    return true;
                }
            }
            return false;
        }
 
        bool isScheduled() const {
            return getNumOutstandingEvents() > 0;
        }
 
        uint32_t cancel()
        {
            const uint32_t cancel_cnt = inflight_pl_.size();
            // Cancelling the event will change the inflight_pl_
            // list, hence we cannot use a range for loop
            auto bit = inflight_pl_.begin();
            while(bit != inflight_pl_.end()) {
                auto proxy = *bit;
                ++bit;
                proxy->cancel();
            }
            return cancel_cnt;
        }
 
        uint32_t cancel(Clock::Cycle rel_cycle) {
            const uint32_t cancel_cnt = inflight_pl_.size();
            auto bit = inflight_pl_.begin();
            while(bit != inflight_pl_.end()) {
                auto proxy = *bit;
                ++bit;
                // Cancelling the event will change the inflight_pl_
                // list, hence we cannot use a range for loop
                proxy->cancel(rel_cycle);
            }
            return cancel_cnt;
        }
 
        uint32_t cancelIf(const DataT & criteria)
        {
            uint32_t cancel_cnt = 0;
            // Cancelling the event will change the inflight_pl_
            // list, hence we cannot use a range for loop
            auto bit = inflight_pl_.begin();
            while(bit != inflight_pl_.end()) {
                auto proxy = *bit;
                ++bit;
                if(proxy->getPayload_() == criteria) {
                    proxy->cancel();
                    ++cancel_cnt;
                }
            }
 
            return cancel_cnt;
        }
 
        std::vector<Scheduleable*> getHandleIf(const DataT & criteria) {
            std::vector<Scheduleable*> ple_vector;
            for(auto * proxy : inflight_pl_)
            {
                if(proxy->getPayload_() == criteria) {
                    ple_vector.push_back(proxy);
                }
            }
            return ple_vector;
        }
 
        bool confirmIf(const DataT & criteria) {
            for(auto * proxy : inflight_pl_)
            {
                if(proxy->getPayload_() == criteria) {
                    return true;
                }
            }
            return false;
        }
 
        uint32_t cancelIf(std::function<bool(const DataT &)> compare) {
            uint32_t cancel_cnt = 0;
            // Cancelling the event will change the inflight_pl_
            // list, hence we cannot use a range for loop
            auto bit = inflight_pl_.begin();
            while(bit != inflight_pl_.end()) {
                auto proxy = *bit;
                ++bit;
                if(compare(proxy->getPayload_())) {
                    proxy->cancel();
                    ++cancel_cnt;
                }
            }
            return cancel_cnt;
        }
 
        std::vector<Scheduleable*> getHandleIf(std::function<bool(const DataT &)> compare) {
            std::vector<Scheduleable*> ple_vector;
            for(auto * proxy : inflight_pl_) {
                if(compare(proxy->getPayload_())) {
                    ple_vector.push_back(proxy);
                }
            }
            return ple_vector;
        }
 
        uint32_t confirmIf(std::function<bool(const DataT &)> compare) {
            for(auto * proxy : inflight_pl_)
            {
                if(compare(proxy->getPayload_())) {
                    return true;
                }
            }
            return false;
        }
 
    private:
 
        friend class Scheduler;
        /*
         * \brief Create an PayloadEvent to deliver a payload in the future.
         *
         * \param event_set The sparta::EventSet this PhasedPayloadEvent belongs to
         * \param scheduler Pointer to the sparta::Scheduler that would schedule this event
         * \param name      The name of this event (as it shows in the EventSet)
         * \param sched_phase The SchedulingPhase this PhasedPayloadEvent belongs to
         * \param consumer_event_handler A SpartaHandler to the consumer's event_handler
         * \param delay The relative time (in Cycles) from "now" to schedule
         * \param max_outstanding Ceiling on simultaneously in-flight payloads
         *
         * \note This constructor is restricted to be used by the sparta::Scheduler only
         *       in order to support sparta::GlobalEvent
         */
        PhasedPayloadEvent(TreeNode            * event_set,
                           Scheduler           * scheduler,
                           const std::string   & name,
                           SchedulingPhase       sched_phase,
                           const SpartaHandler & consumer_event_handler,
                           Clock::Cycle          delay = 0,
                           size_t                max_outstanding = DEFAULT_MAX_OUTSTANDING) :
            EventNode(event_set, name, sched_phase),
            name_(name + "[" + consumer_event_handler.getName() + "]"),
            prototype_(consumer_event_handler, delay, sched_phase),
            inflight_pl_(INITIAL_OUTSTANDING, max_outstanding)
        {
 
            sparta_assert(consumer_event_handler.argCount() == 1,
                          "You must assign a PhasedPayloadEvent a consumer handler ""that takes exactly one argument");
            prototype_.setScheduler(scheduler);
        }
 
    private:
 
        void createResource_() override {
            prototype_.setScheduleableClock(getClock());
            prototype_.setScheduler(determineScheduler(getClock()));
 
            // Make sure no proxies are outstanding and all are allocated
            sparta_assert(inflight_pl_.empty());
 
        }
 
        void addProxies_()
        {
            const uint32_t old_size = allocated_proxies_.size();
            const uint32_t new_size = payload_proxy_allocation_cadence_ + old_size;
            allocated_proxies_.resize(new_size);
            for(uint32_t i = old_size; i < new_size; ++i) {
                allocated_proxies_[i].reset(new PayloadDeliveringProxy(prototype_, this));
            }
            free_pl_.resize(new_size, nullptr);
        }
 
        std::string       name_;
 
        Scheduleable      prototype_;
 
        ProxyAllocation   allocated_proxies_;
        ProxyFreeList     free_pl_;
        ProxyInflightList inflight_pl_;
 
        // Use 16, a power of 2 for allocation of more objects.  No
        // rhyme or reason, but this seems to be a sweet spot in
        // performance.
        const uint32_t payload_proxy_allocation_cadence_ = 16;
        uint32_t free_idx_           = 0;
        uint32_t allocation_idx_     = 0;
 
    };
}