SyncPort.hpp

Go to the documentation of this file.

// <SyncPort.h> -*- C++ -*-
 
#pragma once
 
#include "sparta/simulation/TreeNode.hpp"
#include "sparta/utils/DataContainer.hpp"
#include "sparta/collection/DelayedCollectable.hpp"
#include "sparta/ports/Port.hpp"
#include "sparta/events/Precedence.hpp"
#include "sparta/events/EventSet.hpp"
 
namespace sparta
{
 
    // Forward declaration
    template<class DataT>
    class SyncInPort;
 
    template<class DataT>
    class SyncOutPort final : public OutPort
    {
        typedef collection::DelayedCollectable<DataT> CollectorType;
 
    public:
        typedef DataT DataType;
 
        SyncOutPort(TreeNode * portset, const std::string & name, const Clock * clk,
                    bool presume_zero_delay = true) :
            OutPort(portset, name, presume_zero_delay),
            clk_(clk), info_logger_(this, "pinfo", getLocation() + "_info")
        {
            sparta_assert(name.length() != 0, "You cannot have an unnamed port.");
            sparta_assert(clk_ != 0, "Clock ptr cannot be null in port: " << name);
 
            OutPort::sync_port_ = true; // prevent clock mismatch asserts
        }
 
        void bind(Port * in) override
        {
            SyncInPort<DataT> * inp;
            if((inp = dynamic_cast<SyncInPort<DataT> *>(in)) == 0) {
                throw SpartaException("ERROR: Attempt to bind SyncInPort of a disparate types: '" +
                                      in->getLocation() + "' to '" + getLocation() + "'");
            }
 
            // Sync ports only support one binding for now.
            sparta_assert(sync_in_port_ == 0, "Multiple bind attempts on port:" << getLocation());
            sync_in_port_ = inp;
            sparta_assert(sync_in_port_ != 0, "Cannot bind to null input port in:" << getLocation());
 
            OutPort::bind(in);
 
            bound_in_ports_.push_back(inp);
        }
 
        using Port::bind;
 
        bool isReady() const {
            sparta_assert(sync_in_port_ != 0, "isReady() check on unbound port:" << getLocation());
            return sync_in_port_->couldAccept_(clk_);
        }
 
        bool isReadyPS() const {
            sparta_assert(sync_in_port_ != 0, "isReadyPS() check on unbound port:" << getLocation());
            return sync_in_port_->getRawReady_();
        }
 
        bool isDriven(Clock::Cycle rel_cycle) const override {
            for(SyncInPort<DataT>* itr : bound_in_ports_) {
                if(itr->isDriven(rel_cycle, clk_)) {
                    return true;
                }
            }
            return false;
        }
 
        bool isDriven() const override {
            for(SyncInPort<DataT>* itr : bound_in_ports_) {
                if(itr->isDriven(clk_)) {
                    return true;
                }
            }
            return false;
        }
 
        uint64_t send(const DataT & dat)
        {
            return send(dat,0,false);
        }
 
        uint64_t send(const DataT & dat, sparta::Clock::Cycle send_delay_cycles)
        {
            return send(dat, send_delay_cycles, false);
        }
 
        uint64_t sendAndAllowSlide(const DataT & dat)
        {
            return send(dat, 0, true);
        }
 
        uint64_t sendAndAllowSlide(const DataT & dat, sparta::Clock::Cycle send_delay_cycles)
        {
            return send(dat, send_delay_cycles, true);
        }
 
        uint64_t send(const DataT & dat, sparta::Clock::Cycle send_delay_cycles, const bool allow_slide)
        {
            sparta_assert(sync_in_port_ != 0, "Attempting to send data on unbound port:" << getLocation());
            sparta_assert(clk_->isPosedge(), "Posedge check failed in port:" << getLocation());
 
            Clock::Cycle send_cycle = computeNextAvailableCycleForSend(send_delay_cycles, 0);
            
            if (SPARTA_EXPECT_FALSE(info_logger_)) {
                info_logger_ << "SEND @" << send_cycle
                             << " allow_slide=" << allow_slide
                             << " # " << dat;
            }
            bool is_fwd_progress = true;
            uint64_t sched_delay_ticks = sync_in_port_->send_(dat, clk_, send_delay_cycles,
                                                              allow_slide, is_fwd_progress);
 
            if(SPARTA_EXPECT_FALSE(collector_ != nullptr)) {
                collector_->collectWithDuration(dat, send_delay_cycles, 1);
            }
 
            sparta_assert(send_cycle > prev_data_send_cycle_ || prev_data_send_cycle_ == PREV_DATA_SEND_CYCLE_INIT, //init tick 0
                          getLocation()
                          << ": trying to send at cycle "
                          << send_cycle
                          << ", which is not later than the previous send cycle: "
                          << prev_data_send_cycle_
                          << "; SyncOutPorts are expected to send at most once per cycle");
            prev_data_send_cycle_ = send_cycle;
            return sched_delay_ticks;
        }
 
        Clock::Cycle computeNextAvailableCycleForSend(sparta::Clock::Cycle send_delay_cycles, const uint32_t num_beats)
        {
            sparta_assert(sync_in_port_ != 0, "Attempting to send data on unbound port:" << getLocation());
            sparta_assert(clk_->isPosedge(), "Posedge check failed in port:" << getLocation());
 
            // Start at the current clock with the specified delay
            Clock::Cycle current_cycle = clk_->currentCycle();
            Scheduler::Tick current_tick = clk_->currentTick();
 
            // Send each beat of data with a slide, so know previous slide
            Scheduler::Tick prev_data_arrival_tick =
                sync_in_port_->getPrevDataArrivalTick_();
 
            // Send one extra beat (<=) to represent the N+1 beat to be sent
            for(uint32_t beat_count = 0; beat_count <= num_beats; ++beat_count) {
                uint64_t sched_delay_ticks =
                    sync_in_port_->computeSendToReceiveTickDelay_(clk_,
                                                                  send_delay_cycles + beat_count,
                                                                  true /*allow_slide*/,
                                                                  prev_data_arrival_tick);
                prev_data_arrival_tick = current_tick + sched_delay_ticks;
            }
 
            // At this point, the prev_data_arrival_tick is the absolute
            // tick at which an N+1 beat would arrive. Now, find when that
            // would have been sent to have that arrival time.
            uint64_t num_ticks_before_arrival =
                sync_in_port_->computeReverseSendToReceiveTickDelay_(clk_,
                                                                     send_delay_cycles,
                                                                     prev_data_arrival_tick);
            uint64_t send_tick = prev_data_arrival_tick - num_ticks_before_arrival;
 
            // Convert the absolute tick of the send event into a current
            // cycle relative cycle and return that value
            Clock::Cycle next_send_cycle = clk_->getCycle(send_tick) + send_delay_cycles;
            sparta_assert(next_send_cycle >= current_cycle);
            return next_send_cycle;
        }
 
 
        SyncOutPort(const SyncOutPort &) = delete;
 
        SyncOutPort & operator=(const SyncOutPort &) = delete;
 
        void enableCollection(TreeNode* node) override {
            collector_.reset(new CollectorType(node, Port::name_, 0,
                                               "Data being sent out on this SyncOutPort"));
        }
 
    private:
 
        const sparta::Clock * clk_ = nullptr;
 
        SyncInPort<DataT> * sync_in_port_ = nullptr;
 
        std::unique_ptr<CollectorType> collector_;
 
        Clock::Cycle PREV_DATA_SEND_CYCLE_INIT = 0xffffffffffffffff; //init tick 0
        Clock::Cycle prev_data_send_cycle_ = PREV_DATA_SEND_CYCLE_INIT; //init tick 0
 
        sparta::log::MessageSource info_logger_;
 
        std::vector <SyncInPort<DataT>*> bound_in_ports_;
    };
 
 
    template<class DataT>
    class SyncInPort final : public InPort, public DataContainer<DataT>
    {
        typedef collection::Collectable<DataT> CollectorType;
    public:
        typedef DataT DataType;
 
        SyncInPort(TreeNode* portset, const std::string & name, const Clock * clk,
                   sparta::SchedulingPhase delivery_phase = sparta::SchedulingPhase::PortUpdate) :
            InPort(portset, name, delivery_phase),
            DataContainer<DataT>(clk),
            sync_port_events_(this),
            info_logger_(this, "pinfo", getLocation() + "_info")
        {
            receiver_clock_ = getClock();
            scheduler_ = receiver_clock_->getScheduler();
 
            sparta_assert(name.length() != 0, "You cannot have an unnamed port.");
            sparta_assert(receiver_clock_ != 0, "Clock ptr cannot be null in port: " << name);
 
            //sync_port_events_.setClock(clk);
            forward_event_.reset(new PhasedPayloadEvent<DataT>
                                 (&sync_port_events_, name + "_forward_event",
                                  delivery_phase,
                                  CREATE_SPARTA_HANDLER_WITH_DATA(SyncInPort<DataT>, forwardData_, DataT)));
        }
 
        SyncInPort(const SyncInPort &) = delete;
 
        SyncInPort & operator=(const SyncInPort &) = delete;
 
        ~SyncInPort()
        { }
 
        void bind(Port * out) override
        {
            SyncOutPort<DataT> * outp;
            if((outp = dynamic_cast<SyncOutPort<DataT> *>(out)) == 0) {
                throw SpartaException("ERROR: Attempt to bind SyncOutPort of a disparate types: '" +
                                      out->getLocation() + "' to '" + getLocation() + "'");
            }
            InPort::bind(out);
        }
 
        using Port::bind;
 
        bool isDriven(Clock::Cycle rel_cycle) const override {
            return forward_event_->isScheduled(rel_cycle);
        }
 
        bool isDriven() const override {
            return forward_event_->isScheduled();
        }
 
        bool isDriven(Clock::Cycle rel_cycle, const Clock * send_clk) {
            Scheduler::Tick num_delay_ticks =
                computeSendToReceiveTickDelay_(send_clk, rel_cycle, false, prev_data_arrival_tick_);
 
            sparta::Scheduler::Tick current_tick = scheduler_->getCurrentTick();
            sparta::Scheduler::Tick abs_scheduled_tick = num_delay_ticks + current_tick;
 
            bool is_already_driven =
                     !( prev_data_arrival_tick_ < abs_scheduled_tick ||
                        prev_data_arrival_tick_ == PREV_DATA_ARRIVAL_TICK_INIT );
 
            return is_already_driven;
        }
 
        bool isDriven(const Clock * send_clk) {
            Scheduler::Tick num_delay_ticks =
                computeSendToReceiveTickDelay_(send_clk, 0, false, prev_data_arrival_tick_);
 
            sparta::Scheduler::Tick current_tick = scheduler_->getCurrentTick();
            sparta::Scheduler::Tick abs_scheduled_tick = num_delay_ticks + current_tick;
 
            bool is_already_driven =
                     !( prev_data_arrival_tick_ < abs_scheduled_tick ||
                        prev_data_arrival_tick_ == PREV_DATA_ARRIVAL_TICK_INIT );
 
            return is_already_driven;
        }
 
        Clock::Cycle getPortDelay() const final {
            return receive_delay_cycles_;
        }
 
        void setPortDelay(sparta::Clock::Cycle delay_cycles) override {
            sparta_assert(isBound() == false, "Attempt to set a port delay after binding. \n"
                          "This can adversely affect precedence rules.  If possible call setPortDelay BEFORE\n"
                          "binding the port");
            sparta_assert(delay_was_set_ == false,
                          "Attempt to set port delay twice (that's not expected) for: " << getLocation());
            receive_delay_cycles_ = delay_cycles;
            receive_delay_ticks_  = receiver_clock_->getTick(delay_cycles);
            delay_was_set_ = true;
            if (SPARTA_EXPECT_FALSE(info_logger_)) {
                info_logger_ << "setPortDelay [cycles]: "
                             << " delay_cycles=" << delay_cycles
                             << " => receive_delay_ticks_=" << receive_delay_ticks_
                             << " receive_delay_cycles_=" << receive_delay_cycles_
                             << std::endl;
            }
        }
 
        void setPortDelay(double delay_cycles) override {
            sparta_assert(isBound() == false, "Attempt to set a port delay after binding. \n"
                          "This can adversely affect precedence rules.  If possible call setPortDelay BEFORE\n"
                          "binding the port");
            sparta_assert(delay_was_set_ == false,
                          "Attempt to set port delay twice (that's not expected) for: " << getLocation());
            receive_delay_ticks_ = receiver_clock_->getTick(delay_cycles);
            receive_delay_cycles_ = (receive_delay_ticks_ + (receiver_clock_->getPeriod() - 1)) / receiver_clock_->getPeriod();
            delay_was_set_ = true;
            if (SPARTA_EXPECT_FALSE(info_logger_)) {
                info_logger_ << "setPortDelay [double]: "
                             << " delay_cycles=" << delay_cycles
                             << " => receive_delay_ticks_=" << receive_delay_ticks_
                             << " receive_delay_cycles_=" << receive_delay_cycles_
                             << std::endl;
            }
        }
 
        void enableCollection(TreeNode* node) override
        {
            collector_.reset(new CollectorType(node, Port::name_, 0,
                                               "Data being recirculated on this SyncInPort"));
        }
 
        void setInitialReadyState(bool is_ready) {
            sparta_assert(scheduler_->isRunning() == false);
            sparta_assert(scheduler_->getCurrentTick() == 0); //sched init 0
            cur_is_ready_ = is_ready;
            prev_is_ready_ = is_ready;
        }
 
        void setReady(bool is_ready) {
            if (SPARTA_EXPECT_FALSE(info_logger_)) {
                info_logger_ << "setting ready to: " << is_ready << "; num_in_flight = " << num_in_flight_ << "\n";
            }
 
            Scheduler::Tick cur_tick = scheduler_->getCurrentTick();
            if (cur_tick > set_ready_tick_) {
                set_ready_tick_ = cur_tick;
                prev_is_ready_ = cur_is_ready_;
                cur_is_ready_ = is_ready;
            } else {
                sparta_assert(cur_tick == set_ready_tick_,
                              "Unexpected set-ready in the past for: " << getLocation());
                sparta_assert(is_ready == cur_is_ready_,
                              "Double-ready setting must be of the same value for: " << getLocation());
            }
 
 
        }
 
        bool getReady() const {
            return (cur_is_ready_);
        }
 
        void setContinuing(bool continuing) final {
            Port::setContinuing(continuing);
            forward_event_->setContinuing(continuing);
        }
 
        template<class ConsDataT, sparta::SchedulingPhase PhaseT>
        void precedes(PayloadEvent<ConsDataT, PhaseT> & consumer)
        {
            sparta_assert(PhaseT >= forward_event_->getSchedulingPhase(),
                          "The phase of the PayloadEvent is less than this Port's -- you cannot "
                          "force the Port to come before the PayloadEvent due to this constraint");
            if(PhaseT == forward_event_->getSchedulingPhase()) {
                forward_event_->getScheduleable().
                    precedes(consumer.getScheduleable());
            }
        }
 
        template<sparta::SchedulingPhase PhaseT>
        void precedes(UniqueEvent<PhaseT> & consumer)
        {
            sparta_assert(PhaseT >= forward_event_->getSchedulingPhase(),
                          "The phase of the PayloadEvent is less than this Port's -- you cannot "
                          "force the Port to come before the PayloadEvent due to this constraint");
            if(PhaseT == forward_event_->getSchedulingPhase()) {
                forward_event_->getScheduleable().precedes(consumer);
            }
        }
 
        using InPort::precedes;
 
        std::unique_ptr<PhasedPayloadEvent<DataT>> & getForwardingEvent() {
            return forward_event_;
        }
 
    private:
 
        Scheduleable & getScheduleable_() final {
            return forward_event_->getScheduleable();
        }
 
        void setProducerPrecedence_(Scheduleable * pd) final {
            if(pd->getSchedulingPhase() == forward_event_->getSchedulingPhase()) {
                pd->precedes(forward_event_->getScheduleable(), "Port::bind of OutPort to " + getName() + ": '" +
                             pd->getLabel() + "' is a registered driver");
            }
        }
 
        void registerConsumerHandler_(const SpartaHandler & handler) final
        {
            sparta_assert(handler.argCount() == 1,
                          "The handler associated with the SyncInPort must take at least one argument");
 
            // Help the user identify their events/callbacks from the Scheduler debug
            handler_name_ = getName() + "<SyncInPort>[" + explicit_consumer_handler_.getName() + "]";
            forward_event_->getScheduleable().setLabel(handler_name_.c_str());
        }
 
        void bind_(Port * outp) override final
        {
            InPort::bind_(outp);
            for(auto & consumer : port_consumers_)
            {
                if(consumer->getSchedulingPhase() == forward_event_->getSchedulingPhase()) {
                    forward_event_->getScheduleable().precedes(consumer, "Port::bind(" + getName() + "->" + outp->getName() + "),'"
                                                               + consumer->getLabel() + "' is registered driver");
                }
            }
        }
 
        bool getLatchedReady_(sparta::Scheduler::Tick cur_tick) const {
            bool internal_ready = true;
            if ((set_ready_tick_ == cur_tick && !prev_is_ready_) || // 'ready' was set this cycle and the previous value of 'ready' was false OR
                (set_ready_tick_ < cur_tick && !cur_is_ready_))     // 'ready' was set in a previous cycle and the current value of 'ready' is false
            {
                internal_ready = false;
            }
            return internal_ready;
        }
 
        /*
         * Return the actual, non-latched ready value.
         */
        bool getRawReady_() const {
            sparta_assert(getPortDelay() == 0,
                          "Only expected raw-ready to be returned for 0-cycle connections");
            return cur_is_ready_;
        }
 
        void forwardData_(const DataT & dat)
        {
            sparta::Scheduler::Tick cur_tick =
                scheduler_->getCurrentTick();
 
            sparta_assert(set_ready_tick_ <= cur_tick, "Assert in port: " << getLocation());
 
            // If the owner of the Inport isn't ready, then re-send the
            // data to ourself until the owner is ready to accept.
            sparta_assert(num_in_flight_ > 0);
            num_in_flight_--;
            if (getLatchedReady_(cur_tick) == false) {
                if (SPARTA_EXPECT_FALSE(info_logger_)) {
                    info_logger_ << "RESENDING @" << receiver_clock_->currentCycle()
                                 << "(" << num_in_flight_ << ") "
                                 << " # " << dat;
                }
                bool allow_slide = false;
                bool is_fwd_progress = false;
 
                send_(dat, receiver_clock_, sparta::Clock::Cycle(1), allow_slide, is_fwd_progress);
                sparta_assert(num_in_flight_ > 0);
            }
 
            // Else, forward the data to the original handler
            else {
                DataContainer<DataT>::setData_(dat);
 
                // Always call the consumer_handler BEFORE scheduling
                // listeners.
                if(SPARTA_EXPECT_TRUE(explicit_consumer_handler_)) {
                    explicit_consumer_handler_((const void*)&dat);
                }
 
                // Show the data that has arrived on this OutPort that
                // the receiver now sees
                if(SPARTA_EXPECT_FALSE(collector_ != nullptr)) {
                    collector_->collectWithDuration(dat, 1);
                }
 
                if (SPARTA_EXPECT_FALSE(info_logger_)) {
                    info_logger_ << "DELIVERING @" << receiver_clock_->currentCycle()
                                 << "(" << num_in_flight_ << ") "
                                 << " # " << dat;
                }
            }
        }
 
        sparta::EventSet sync_port_events_;
 
        std::unique_ptr<PhasedPayloadEvent<DataT>> forward_event_;
 
        friend void SyncOutPort<DataT>::bind(Port * in);
 
        friend uint64_t SyncOutPort<DataT>::send(const DataT &, uint64_t, const bool);
 
        friend bool SyncOutPort<DataT>::isReady() const;
 
        friend bool SyncOutPort<DataT>::isReadyPS() const;
 
        friend uint64_t SyncOutPort<DataT>::computeNextAvailableCycleForSend(sparta::Clock::Cycle, const uint32_t);
 
        bool couldAccept_(const Clock *send_clk) const {
 
            Scheduler::Tick num_delay_ticks =
                calculateClockCrossingDelay(send_clk->currentTick(), send_clk, receive_delay_ticks_, receiver_clock_);
 
            sparta::Scheduler::Tick cur_tick =
                scheduler_->getCurrentTick();
 
            sparta::Scheduler::Tick abs_scheduled_tick =
                num_delay_ticks + cur_tick;
 
            bool retval = (abs_scheduled_tick > prev_data_arrival_tick_ || prev_data_arrival_tick_ == PREV_DATA_ARRIVAL_TICK_INIT);
 
            sparta_assert(cur_tick >= set_ready_tick_, "Someone drove setReady() in the future in" << getLocation());
 
            // Check for sync-port ready/valid backpressure and override
            // 'ready' if downstream indicated it couldn't take data
            if (getLatchedReady_(cur_tick) == false) {
                sparta_assert(send_clk->getFrequencyMhz() == receiver_clock_->getFrequencyMhz(), "Error in port:" << getLocation());
                sparta_assert(getPortDelay() == 1 || getPortDelay() == 0,
                              "Ready/Valid only tested for zero and one cycle delays (not "<< getPortDelay() <<"); relax this assert once more testing is done; location=" << getLocation());
 
                // Can't accept anything if there's already one request
                // waiting to be delivered OR if this is a 0-cycle delay
                // (since we need to deliver on the same cycle its sent)
                if (num_in_flight_ > 0 || getPortDelay() == 0) {
                    retval = false;
                }
            }
 
            return retval;
        }
 
        Scheduler::Tick getPrevDataArrivalTick_() const
        {
            return prev_data_arrival_tick_;
        }
 
        uint64_t computeSendToReceiveTickDelay_(const Clock *send_clk,
                                                const double send_delay_cycles,
                                                const bool allow_slide,
                                                const Scheduler::Tick prev_data_arrival_tick) const
        {
            Scheduler::Tick num_delay_ticks =
                calculateClockCrossingDelay(send_clk->getTick(send_delay_cycles), send_clk, receive_delay_ticks_, receiver_clock_);
 
            sparta::Scheduler::Tick current_tick       = scheduler_->getCurrentTick();
            sparta::Scheduler::Tick abs_scheduled_tick = num_delay_ticks + current_tick;
 
            // Slide pushes this send out past the previous arrival,
            // rather than faulting as user error on sending too early
            if (allow_slide &&
                ( abs_scheduled_tick <= prev_data_arrival_tick &&
                  prev_data_arrival_tick != PREV_DATA_ARRIVAL_TICK_INIT)) {
                abs_scheduled_tick = prev_data_arrival_tick + receiver_clock_->getPeriod();
            }
 
            num_delay_ticks = abs_scheduled_tick - current_tick;
 
            // Underlying assumption is that all destinations get their
            // event at the same time.
            sparta_assert((abs_scheduled_tick % receiver_clock_->getPeriod()) == 0, "Failed posedge check in:" << getLocation()); // posedge check
 
            return num_delay_ticks;
        }
 
 
        uint64_t computeReverseSendToReceiveTickDelay_(const Clock *send_clk,
                                                       const double send_delay_cycles,
                                                       const Scheduler::Tick data_arrival_tick) const
        {
            sparta::Scheduler::Tick raw_dst_clk_posedge_tick =
                data_arrival_tick / receiver_clock_->getPeriod() * receiver_clock_->getPeriod();
            sparta_assert(data_arrival_tick == raw_dst_clk_posedge_tick);
 
            Scheduler::Tick num_delay_ticks =
                calculateReverseClockCrossingDelay(data_arrival_tick,
                                                   send_clk->getTick(send_delay_cycles),
                                                   send_clk, receive_delay_ticks_, receiver_clock_);
 
            sparta_assert( data_arrival_tick >= num_delay_ticks );
            return num_delay_ticks;
        }
 
        void bind_(SyncOutPort<DataT> * inp) {
            bound_ports_.push_back(inp);
        }
 
        uint64_t send_(const DataT & dat, const Clock *send_clk, Clock::Cycle send_delay_cycles,
                       const bool allow_slide, bool is_fwd_progress)
        {
            return send_(dat, send_clk, static_cast<double>(send_delay_cycles), allow_slide, is_fwd_progress);
        }
 
        uint64_t send_(const DataT & dat, const Clock *send_clk, double send_delay_cycles,
                       const bool allow_slide, bool is_fwd_progress)
        {
            Scheduler::Tick num_delay_ticks =
                computeSendToReceiveTickDelay_(send_clk, send_delay_cycles, allow_slide, prev_data_arrival_tick_);
 
            sparta::Scheduler::Tick current_tick = scheduler_->getCurrentTick();
            sparta::Scheduler::Tick abs_scheduled_tick = num_delay_ticks + current_tick;
 
            // Underlying assumption is that all destinations get their
            // event at the same time.
            sparta_assert((abs_scheduled_tick % receiver_clock_->getPeriod()) == 0, "Failed posedge check in:" << getLocation()); // posedge check
 
            if (SPARTA_EXPECT_FALSE(info_logger_)) {
                info_logger_ << "RECEIVE SCHEDULED @" << receiver_clock_->getCycle(abs_scheduled_tick)
                             << "(" << num_in_flight_ << ") "
                             << " # " << dat;
            }
 
            // Only one item can be received per cycle
            sparta_assert(prev_data_arrival_tick_ < abs_scheduled_tick || prev_data_arrival_tick_ == PREV_DATA_ARRIVAL_TICK_INIT,
                          getLocation() << ": attempt to schedule send for tick "
                          << abs_scheduled_tick
                          << ", which is not later than the previous data at tick "
                          << prev_data_arrival_tick_
                          << "; SyncInPorts should only get data once per cycle; data was: " << dat);
 
            prev_data_arrival_tick_ = abs_scheduled_tick;
 
            if(num_delay_ticks == 0) {
                checkSchedulerPhaseForZeroCycleDelivery_(forward_event_->getSchedulingPhase());
            }
            forward_event_->preparePayload(dat)->scheduleRelativeTick(num_delay_ticks, scheduler_);
            num_in_flight_++;
 
            if (SPARTA_EXPECT_TRUE(is_fwd_progress && is_continuing_)) {
                scheduler_->kickTheDog();
            }
 
            return num_delay_ticks;
        }
        
 
    private:
 
        std::string handler_name_;
 
        bool delay_was_set_                = false;
        Clock::Cycle receive_delay_cycles_ = 0;
        sparta::Scheduler::Tick receive_delay_ticks_ = 0;
 
        const Scheduler::Tick PREV_DATA_ARRIVAL_TICK_INIT = 0xffffffffffffffff;
        Scheduler::Tick prev_data_arrival_tick_ = PREV_DATA_ARRIVAL_TICK_INIT;
 
        bool cur_is_ready_ = true;
        bool prev_is_ready_ = true;
 
        bool is_continuing_ = true;
 
        uint32_t num_in_flight_ = 0;
 
        Scheduler::Tick set_ready_tick_ = 0;
 
        std::unique_ptr<CollectorType> collector_;
 
        sparta::log::MessageSource info_logger_;
    };
}