BusMaster.hxx

Go to the documentation of this file.

 
#ifndef _UTILS_BUSMASTER_HXX_
#define _UTILS_BUSMASTER_HXX_
 
#include "executor/StateFlow.hxx"
#include "utils/Buffer.hxx"
#include "utils/LimitedPool.hxx"
#include "utils/ScheduledQueue.hxx"
 
template <class MessageType> class Bus
{
public:
    using Packet = Buffer<MessageType>;
 
    using PacketPtr = BufferPtr<MessageType>;
 
    using PacketSink = FlowInterface<Packet>;
 
    class Activity : public QMember
    {
    public:
        using Packet = Bus::Packet;
        using PacketPtr = Bus::PacketPtr;
 
        virtual void fill_packet(Packet *packet) = 0;
    };
 
    class Master : public StateFlowBase
    {
    public:
        Master(
            Service *s, PacketSink *sink, Activity *idle, unsigned num_enqueued)
            : StateFlowBase(s)
            , idle_(idle)
            , sink_(sink)
            , pool_(sizeof(Packet), num_enqueued)
            , numPacketsInPool_(num_enqueued)
            , needShutdown_(false)
        {
        }
 
#ifdef GTEST        
        void request_shutdown()
        {
            needShutdown_ = true;
        }
 
        void shutdown()
        {
            needShutdown_ = true;
            while (!is_terminated() && (pool_.free_items() < numPacketsInPool_))
            {
                usleep(200);
            }
        }
#endif
        
        void set_policy(unsigned num_prio, const Fixed16 *strides)
        {
            queue_.emplace(num_prio, strides);
            start_flow(STATE(get_buffer));
        }
 
        void schedule_activity(Activity *a, unsigned prio)
        {
            queue_->insert(a, prio);
        }
 
    private:
        Action get_buffer()
        {
            return allocate_and_call(sink_, STATE(fill_pkt), &pool_);
        }
 
        Action fill_pkt()
        {
            auto buf = get_buffer_deleter(get_allocation_result(sink_));
            if (needShutdown_)
            {
                return exit();
            }
            // Picks the next activity to do on the bus.
            Activity *a = (Activity *)queue_->next().item;
            if (!a)
            {
                a = idle_;
            }
            a->fill_packet(buf.get());
            sink_->send(buf.release());
            return call_immediately(STATE(get_buffer));
        }
 
        uninitialized<ScheduledQueue> queue_;
        Activity *idle_;
        PacketSink *sink_;
        LimitedPool pool_;
        uint16_t numPacketsInPool_;
        uint16_t needShutdown_ : 1;
    }; // class Master
 
private:
    Bus();
};
 
#endif // _UTILS_BUSMASTER_HXX_