QueueSocketSourceSink.ct

Go to the documentation of this file.

//
// Copyright (c) 2020 Fraunhofer Institute for Applied Information Technology (FIT)
//                    Network Research Group (NET)
//                    Schloss Birlinghoven, 53754 Sankt Augustin, GERMANY
//                    Contact: support@wiback.org
//
// This file is part of the SENF code tree.
// It is licensed under the 3-clause BSD License (aka New BSD License).
// See LICENSE.txt in the top level directory for details or visit
// https://opensource.org/licenses/BSD-3-Clause
//


/** \file
    \brief QueueSocketSourceSink non-inline template implementation */

#include "QueueSocketSourceSink.ih"

// Custom includes
#include <senf/Utils/membind.hh>
#include <senf/Packets/PacketInfo.hh>
#include <senf/Utils/senflikely.hh>
#include "SocketSource.hh"
#include <senf/Socket/QueueReadWritePolicy.ih>
#include <senf/Socket/Protocols/Raw/MMapPacketSocketHandle.hh>

#define prefix_
//-/////////////////////////////////////////////////////////////////////////////////////////////////

//-/////////////////////////////////////////////////////////////////////////////////////////////////
// senf::ppi::module::ActiveQueueSocketSource<Packet>

template <class Packet, class Connector>
prefix_
senf::ppi::module::ActiveQueueSocketSource<Packet,Connector>::ActiveQueueSocketSource(unsigned burst)
    : handle_ ()
    , event_ ()
    , redFilterCallback_(redFilterDrop)
    , maxBurst_ (burst)
    , burst_(0)
    , flushPending_(false)
    , red_(0)
#ifdef SENF_DEBUG
    , burstMax_ (0)
#endif
{
    registerEvent(event_, &ActiveQueueSocketSource::read);
    route(event_, output);
}

template <class Packet, class Connector>
prefix_ senf::ppi::module::ActiveQueueSocketSource<Packet,Connector>::
ActiveQueueSocketSource(Handle const & handle, unsigned burst)
    : handle_ (handle)
    , event_ (handle_, IOEvent::Read)
    , redFilterCallback_(redFilterDrop)
    , maxBurst_ (burst)
    , burst_(0)
    , flushPending_(false)
    , red_(0)
#ifdef SENF_DEBUG
    , burstMax_ (0)
#endif
{
    registerEvent(event_, &ActiveQueueSocketSource::read);
    noroute(output);
    senf::ppi::detail::RouteConnector<Connector>::route(*this, event_, output);
}

template <class Packet, class Connector>
prefix_ typename senf::ppi::module::ActiveQueueSocketSource<Packet,Connector>::Handle
senf::ppi::module::ActiveQueueSocketSource<Packet,Connector>::handle()
    const
{
    return handle_;
}

template <class Packet, class Connector>
prefix_ void senf::ppi::module::ActiveQueueSocketSource<Packet,Connector>::handle(Handle const & handle)
{
    burst_ = 0;
    flushPending_ = false;
    red_ = 0;
    handle_ = handle;
    event_.set(handle_, IOEvent::Read);
    if (handle)
        event_.enabled(true);
}

#ifdef SENF_DEBUG

template <class Packet, class Connector>
prefix_ unsigned senf::ppi::module::ActiveQueueSocketSource<Packet,Connector>::burstMax()
{
    unsigned res (burstMax_);
    burstMax_ = 0;
    return res;
}

#endif

template <class Packet, class Connector>
prefix_ void senf::ppi::module::ActiveQueueSocketSource<Packet,Connector>::setREDFilterCallback(REDFilterCallback const & cb)
{
    redFilterCallback_ = cb;
}

template <class Packet, class Connector>
prefix_ void senf::ppi::module::ActiveQueueSocketSource<Packet,Connector>::read()
{
    // this should not be required, but since we are bursting below, it's not that 'expensive'
    if (SENF_UNLIKELY(!handle_))
        return;

    senf::detail::QueueInfo & qi (* senf::detail::QueuePolicyBase::qinfo(handle_));

    flushPending_ = false;
    for (burst_ = 1; SENF_LIKELY(handle_ and (burst_ <= maxBurst_) and !flushPending_); burst_++) {
        boost::optional<senf::QueueReadPolicy::Buffer> buf (handle_.dequeue());
        if (SENF_UNLIKELY(!buf))
            break;
            
        PacketInfo ifo;
        {
            Packet const & pk (Packet::create(
                           buf->frameBegin(),
                           buf->size(),
                           buf->frameSize(),
                           buf->dataOffset()));
            pk.template annotation<senf::ppi::QueueBufferAnnotation>().value = &(*buf);
#ifdef SENF_PPI_READ_TIMESTAMP
            pk.template annotation<senf::ppi::ReadTimestamp>().value = senf::ClockService::now();
#endif
            ifo = PacketInfo(pk);

            unsigned usage (qi.usageRx());
            if (SENF_UNLIKELY( (usage > (senf::detail::QueueInfo::NUM_SAMPLES/4)) and ((std::uint32_t(rand()) % senf::detail::QueueInfo::NUM_SAMPLES) < (usage))
                               and !redFilterCallback_(pk, (usage * 100) / senf::detail::QueueInfo::NUM_SAMPLES))) {
                // drop frame
                qi.rxStats.red++;
                red_++;
            } else {
                // process frame
                output(pk);
            }
        }
        if (SENF_UNLIKELY(ifo.is_shared() and handle_)) {
            qi.rxStats.extMemory++;
            ifo.releaseExternalMemory();
        }
    }

#ifdef SENF_DEBUG
    if (burst_ > burstMax_)
        burstMax_ = burst_;

    // we were triggered, but no valid packet was available
    if (SENF_UNLIKELY(burst_ == 1)) {
        qi.rxStats.noop++;
    }
#endif

    // this tells flush() that we are done with the for() loop
    burst_ = 0;

    if (SENF_LIKELY(handle_)) {
        if (SENF_UNLIKELY(flushPending_))
            flush();
        else
            handle_.release();
    }
}

template <class Packet, class Connector>
prefix_ std::pair<unsigned,unsigned> senf::ppi::module::ActiveQueueSocketSource<Packet,Connector>::dropped()
{
    unsigned dropped (0);
    if (handle_) {
        try {
            auto & h (reinterpret_cast<ConnectedMMapPacketSocketHandle &>(handle_));
            dropped = h.protocol().rxQueueDropped();
        } catch (...) {};
    }
    std::pair<unsigned,unsigned> tmp (std::make_pair(red_, dropped));
    red_ = 0;
    return tmp;
}


//-/////////////////////////////////////////////////////////////////////////////////////////////////
// senf::ppi::module::PassiveQueueSocketSink

template <class Connector>
prefix_ senf::ppi::module::PassiveQueueSocketSink<Connector>::PassiveQueueSocketSink()
    : handle_ ()
    , eventHook_ ("PassiveQueueSocketSink", senf::membind(&PassiveQueueSocketSink::send, this),
                  senf::scheduler::EventHook::POST, false)
    , redFilterCallback_(redFilterDrop)
    , red_(0)
    , dropped_(0)
#ifdef SENF_DEBUG
    , burst_ (0)
    , burstMax_ (0)
#endif
{
    noroute(input);
    input.onRequest(&PassiveQueueSocketSink::write);
    input.throttlingDisc( ThrottlingDiscipline::NONE);

    registerEvent(writeEvent_, &PassiveQueueSocketSink::send);
}

template <class Connector>
prefix_ senf::ppi::module::PassiveQueueSocketSink<Connector>::PassiveQueueSocketSink(Handle const & handle)
    : handle_ (handle)
    , eventHook_ ("PassiveQueueSocketSink", senf::membind(&PassiveQueueSocketSink::send, this),
              senf::scheduler::EventHook::POST, false)
    , writeEvent_(handle_, IOEvent::Write)
    , redFilterCallback_(redFilterDrop)
    , red_(0)
    , dropped_(0)  
#ifdef SENF_DEBUG
    , burst_ (0)
    , burstMax_ (0)
#endif
{
    noroute(input);
    input.onRequest(&PassiveQueueSocketSink::write);
    input.throttlingDisc( ThrottlingDiscipline::NONE);

    registerEvent(writeEvent_, &PassiveQueueSocketSink::send);
}

template <class Connector>
prefix_ typename senf::ppi::module::PassiveQueueSocketSink<Connector>::Handle
senf::ppi::module::PassiveQueueSocketSink<Connector>::handle()
    const
{
    return handle_;
}

template <class Connector>
prefix_ void senf::ppi::module::PassiveQueueSocketSink<Connector>::handle(Handle const & handle)
{
    handle_ = handle;
    eventHook_.disable();
    writeEvent_.set(handle_, IOEvent::Write);
    writeEvent_.enabled(false);
    red_ = dropped_ = 0;
}

#ifdef SENF_DEBUG
template <class Connector>
prefix_ unsigned senf::ppi::module::PassiveQueueSocketSink<Connector>::burstMax()
{
    unsigned res (burstMax_);
    burstMax_ = 0;
    return res;
}
#endif

template <class Connector>
prefix_ void senf::ppi::module::PassiveQueueSocketSink<Connector>::setREDFilterCallback(REDFilterCallback const & cb)
{
    redFilterCallback_ = cb;
}

template <class Connector>
prefix_ void senf::ppi::module::PassiveQueueSocketSink<Connector>::send()
{
    if (SENF_LIKELY(handle_)) {
        if (SENF_LIKELY(handle_.send())) {
            // we should try again once the socket is writable again
            writeEvent_.enabled(true);
        } else {
            writeEvent_.enabled(false);
        }
        eventHook_.disable();
    }

#ifdef SENF_DEBUG
    if (burst_ > burstMax_)
        burstMax_ = burst_;
    burst_ = 0;
#endif
}

template <class Connector>
prefix_ void senf::ppi::module::PassiveQueueSocketSink<Connector>::write()
{
    Packet const & packet (input());

    if (SENF_UNLIKELY(!handle_)) {
        dropped_++;
        return;
    }

    // simple RED starting at > 33% usage
    senf::detail::QueueInfo & qi (* senf::detail::QueuePolicyBase::qinfo(handle_));
    unsigned usage (qi.usageTx());

    if (SENF_UNLIKELY(usage > (senf::detail::QueueInfo::NUM_SAMPLES/4))) {
        if ((std::uint32_t(rand()) % senf::detail::QueueInfo::NUM_SAMPLES) < (usage)) {
            if (!redFilterCallback_(packet, (usage * 100) / senf::detail::QueueInfo::NUM_SAMPLES)) {
                qi.txStats.red++;
                red_++;
                return;
            }
        }
    }

    boost::optional<QueueWritePolicy::Buffer> buf (handle_.enqueue());
    if (SENF_UNLIKELY(!buf)) {
        dropped_++;
        return;
    }
    
    writePacket(packet, *buf);
}

template <class Connector>
prefix_ void senf::ppi::module::PassiveQueueSocketSink<Connector>::writePacket(senf::Packet const & packet, QueueReadPolicy::Buffer & buf)
{
    SENF_ASSERT( packet.data().size(), "cannot send empty packet" );
    buf.resize(std::min(packet.data().size(), buf.frameSize()));
    std::copy(packet.data().begin(), packet.data().end(), buf.begin());
    eventHook_.enable();
#ifdef SENF_DEBUG
    ++ burst_;
#endif
}

template <class Connector>
prefix_ std::tuple<unsigned,unsigned,unsigned> senf::ppi::module::PassiveQueueSocketSink<Connector>::dropped()
{
    unsigned wrongFormat (0);
    if (handle_) {
        try {
            auto & h (reinterpret_cast<ConnectedMMapPacketSocketHandle &>(handle_));
            wrongFormat = h.protocol().txWrongFormat();
        } catch (...) {};
    }
    std::tuple<unsigned,unsigned,unsigned> tmp (std::make_tuple(red_,dropped_,wrongFormat));
    red_ = dropped_ = 0;
    return tmp;
}


//-/////////////////////////////////////////////////////////////////////////////////////////////////
#undef prefix_


// Local Variables:
// mode: c++
// fill-column: 100
// comment-column: 40
// c-file-style: "senf"
// indent-tabs-mode: nil
// ispell-local-dictionary: "american"
// compile-command: "scons -u test"
// End: