BufferPort.hxx

Go to the documentation of this file.

 
#ifndef _UTILS_BUFFERPORT_HXX_
#define _UTILS_BUFFERPORT_HXX_
 
#include "utils/LimitedPool.hxx"
 
// Added by default on GridConnect bridges.
class BufferPort : public HubPort
{
public:
    BufferPort(Service *service, HubPortInterface *downstream,
        unsigned buffer_bytes, long long delay_nsec)
        : HubPort(service)
        , downstream_(downstream)
        , delayNsec_(delay_nsec)
        , sendBuf_(new char[buffer_bytes])
        , bufSize_(buffer_bytes)
        , bufEnd_(0)
        , timerPending_(0)
    {
        HASSERT(sendBuf_);
    }
 
    ~BufferPort()
    {
        delete [] sendBuf_;
    }
 
    bool shutdown() {
        flush_buffer();
        if (timerPending_) {
            return false;
        }
        if (!is_waiting()) {
            return false;
        }
        return true;
    }
 
private:
    Action entry() override
    {
        if (!tgtBuf_)
        {
            return allocate_and_call(downstream_, STATE(buf_alloc_done),
                config_gridconnect_bridge_max_outgoing_packets() <= 1
                    ? nullptr
                    : &outputPool_);
        }
        // Defines whether we should optimize the traffic and flush right now.
        bool opt_flush = false;
        // This code is OpenLCB-specific. It looks for a certain pattern in the
        // output data stream, and if that pattern is found, inserts an extra
        // flush right now, instead of waiting for the timeout to pass. The
        // data sent on is never modified, so this is purely a performance
        // optimization for OpenLCB.
        if (msg().data()[0] == ':' && msg().data()[1] == 'X') {
            if (msg().data()[3] == 'A' || msg().data()[3] == 'D')
            {
                // Found datagram "only" or "end" packet.
                opt_flush = true;
            }
            else if (strncmp(msg().data() + 3, "9A28", 4) == 0)
            {
                // Found datagram acknowledge packet.
                opt_flush = true;
            }
        }
        if (opt_flush && !bufEnd_)
        {
            // nothing accumulated, send off directly.
            downstream_->send(transfer_message(), priority());
            return exit();
        }
        if (msg().size() < (bufSize_ - bufEnd_))
        {
            // Fits into the buffer.
            memcpy(sendBuf_ + bufEnd_, msg().data(), msg().size());
            bufEnd_ += msg().size();
            if (opt_flush)
            {
                flush_buffer();
            }
            else if (!timerPending_)
            {
                timerPending_ = 1;
                bufferTimer_.start(delayNsec_);
            }
            return release_and_exit();
        }
        else
        {
            flush_buffer();
        }
 
        if (msg().size() >= bufSize_)
        {
            // Cannot buffer: send off directly.
            downstream_->send(transfer_message(), priority());
            return exit();
        }
        else
        {
            // After flushing the buffers this will fit.
            return again();
        }
    }
 
    Action buf_alloc_done()
    {
        tgtBuf_ = get_allocation_result(downstream_);
        tgtBuf_->data()->skipMember_ = message()->data()->skipMember_;
        return call_immediately(STATE(entry));
    }
 
    void flush_buffer()
    {
        if (!bufEnd_) return; // nothing to do
        auto *b = tgtBuf_;
        tgtBuf_ = nullptr;
        b->data()->assign(sendBuf_, bufEnd_);
        bufEnd_ = 0;
        if (message())
        {
            b->set_done(message()->new_child());
        }
        downstream_->send(b);
    }
 
    void timeout()
    {
        timerPending_ = 0;
        flush_buffer();
    }
 
    const string &msg()
    {
        return *message()->data();
    }
 
    class BufferTimer : public ::Timer
    {
    public:
        BufferTimer(BufferPort *parent)
            : Timer(parent->service()->executor()->active_timers())
            , parent_(parent)
        {
        }
 
        long long timeout() override
        {
            parent_->timeout();
            return NONE;
        }
 
    private:
        BufferPort *parent_; 
    } bufferTimer_{this}; 
 
    LimitedPool outputPool_ {sizeof(*tgtBuf_),
        (unsigned)config_gridconnect_bridge_max_outgoing_packets()};
    Buffer<HubData> *tgtBuf_{nullptr};
    HubPortInterface* downstream_;
    long long delayNsec_;
    char *sendBuf_;
    unsigned bufSize_;
    unsigned bufEnd_ : 24;
    unsigned timerPending_ : 1;
};
 
#endif // _UTILS_BUFFERPORT_HXX_