QueueReadWritePolicy.cci

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 QueueReadWritePolicy inline non-template implementation */

#include "QueueReadWritePolicy.ih"

// Custom includes
#include <sys/types.h>
#include <sys/socket.h>
#include <senf/Utils/Logger.hh>
#include <senf/Utils/senflikely.hh>
#include <senf/Scheduler/Scheduler.hh>

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

prefix_ senf::detail::QueueInfo * senf::detail::QueuePolicyBase::qinfo(FileHandle & handle)
{
    SENF_ASSERT( extraPtr(handle), "invalid queue file handle (no QueueInfo found)");
    return static_cast<detail::QueueInfo *>(extraPtr(handle));
}

prefix_ void senf::detail::QueueInfo::init(unsigned rxqlen, unsigned txqlen)
{
    if (rxqlen > 0) {
        rx.begin = rx.head = rx.tail = map;
        rx.end = rx.begin + frameSize * rxqlen;
        rx.idle = true;
        rx.qlen = rxqlen;
    }

    if (txqlen > 0) {
        tx.begin = tx.head = tx.tail = ((rxqlen == 0 ) ? map : rx.end);
        tx.end = tx.begin + frameSize * txqlen;
        tx.idle = true;
        tx.qlen = txqlen;
    }

    unsigned slice (rx.qlen / NUM_SAMPLES);
    for (unsigned n = 0; n < NUM_SAMPLES; n++)
        rx.samples[n] = &(reinterpret_cast<struct ::tpacket2_hdr *>(rx.begin + (n * slice * frameSize))->tp_status);
    
    slice = tx.qlen / NUM_SAMPLES;
    for (unsigned n = 0; n < NUM_SAMPLES; n++)
        tx.samples[n] = &(reinterpret_cast<struct ::tpacket2_hdr *>(tx.begin + (n * slice * frameSize))->tp_status);
}

prefix_ void senf::detail::QueueInfo::RxStats::dump(std::ostream & os)
    const
{
    os << "received "  << received << ", "
       << "red "       << red << ", "
       << "ignored "   << ignored  << ", "
       << "extMemory " << extMemory << ", "
       << "noop "      << noop  << ".";
}

prefix_ void senf::detail::QueueInfo::TxStats::dump(std::ostream & os)
    const
{
    os << "sent "        << sent << ", "
       << "wrongFormat " << wrongFormat << ", "
       << "red "         << red << ", "
       << "overrun "     << overrun << ", "
       << "dropped "     << dropped << ". ";
}

prefix_ unsigned senf::detail::QueueInfo::usageRx()
    const
{
    unsigned sum (0);
    for (unsigned n = 0; n < NUM_SAMPLES; n++)
        sum += (*(rx.samples[n]) != TP_STATUS_KERNEL);

    return sum;
}

prefix_ unsigned senf::detail::QueueInfo::usageTx()
    const
{
    unsigned sum (0);
    for (unsigned n = 0; n < NUM_SAMPLES; n++)
        sum += (*(tx.samples[n]) != TP_STATUS_AVAILABLE);

    return sum;
}

prefix_ void senf::detail::QueueInfo::inc(unsigned char * & ptr, Queue const & q)
{
    ptr += frameSize;
    if (SENF_UNLIKELY(ptr == q.end))
        ptr = q.begin;
}

prefix_ bool senf::detail::QueueInfo::interfaceDead()
    const
{
    // a simple check, if the underlying interface is 'dead' or not responding (i.e. claims to have a 'link' but is not sending/receiving - ath9k bug ?)
    return (rxStats.received == 0) and (txStats.sent == 0 and (txStats.overrun > 0 or txStats.red > 0 or txStats.dropped > 0));
}

///////////////////////////////////////////////////////////////////////////
// senf::SocketQueueBuffer

prefix_ senf::SocketQueueBuffer::SocketQueueBuffer()
    : b_ (0), e_ (0), hdrlen_ (0)
{}

prefix_ senf::SocketQueueBuffer::SocketQueueBuffer(unsigned char * b, unsigned char * e,
                                                   unsigned hdrlen)
    : b_ (b), e_ (e), hdrlen_ (hdrlen)
{}

prefix_ struct ::tpacket2_hdr & senf::SocketQueueBuffer::hdr()
{
    return * reinterpret_cast<struct ::tpacket2_hdr *>(b_);
}

prefix_ struct ::tpacket2_hdr const & senf::SocketQueueBuffer::hdr()
    const
{
    return * reinterpret_cast<struct ::tpacket2_hdr const *>(b_);
}

prefix_ senf::SocketQueueBuffer::iterator senf::SocketQueueBuffer::begin()
{
    return b_ + hdr().tp_mac;
}

prefix_ senf::SocketQueueBuffer::iterator senf::SocketQueueBuffer::end()
{
    return begin() + hdr().tp_len;
}

prefix_ senf::SocketQueueBuffer::const_iterator senf::SocketQueueBuffer::begin()
    const
{
    return b_ + hdr().tp_mac;
}

prefix_ senf::SocketQueueBuffer::const_iterator senf::SocketQueueBuffer::end()
    const
{
    return begin() + hdr().tp_len;
}

prefix_ senf::SocketQueueBuffer::size_type senf::SocketQueueBuffer::size()
    const
{
    return hdr().tp_len;
}

prefix_ bool senf::SocketQueueBuffer::empty()
    const
{
    return hdr().tp_len == 0;
}

prefix_ senf::SocketQueueBuffer::iterator senf::SocketQueueBuffer::frameBegin()
{
    return b_ + hdrlen_;
}

prefix_ senf::SocketQueueBuffer::iterator senf::SocketQueueBuffer::frameEnd()
{
    return e_;
}

prefix_ senf::SocketQueueBuffer::size_type senf::SocketQueueBuffer::frameSize()
    const
{
    return frameEnd() - frameBegin();
}

prefix_ senf::SocketQueueBuffer::const_iterator senf::SocketQueueBuffer::frameBegin()
    const
{
    return b_ + hdrlen_;
}

prefix_ senf::SocketQueueBuffer::const_iterator senf::SocketQueueBuffer::frameEnd()
    const
{
    return e_;
}

prefix_ unsigned senf::SocketQueueBuffer::dataOffset()
    const
{
    return begin() - frameBegin();
}

prefix_ senf::ClockService::clock_type senf::SocketQueueBuffer::timestamp()
    const
{
    return ClockService::from_time_t(hdr().tp_sec) + ClockService::nanoseconds(hdr().tp_nsec);
}

prefix_ std::uint32_t const * senf::SocketQueueBuffer::timestampPtr()
    const
{
    return &(hdr().tp_sec);
}

prefix_ senf::LLSocketAddress senf::SocketQueueBuffer::address()
    const
{
    SENF_ASSERT( hdr().tp_mac > hdrlen_,
            "frame has no address field" );
    senf::LLSocketAddress res;
    ::memcpy(res.sockaddr_p(),
             reinterpret_cast<struct ::sockaddr_ll *>(
                 b_ + hdrlen_),
             sizeof(struct  ::sockaddr_ll));
    return res;
}

prefix_ boost::optional<unsigned> senf::SocketQueueBuffer::vlan()
    const
{
    return hdr().tp_status & TP_STATUS_VLAN_VALID
        ? boost::optional<unsigned>(hdr().tp_vlan_tci) : boost::none;
}

prefix_ unsigned senf::SocketQueueBuffer::tpid()
    const
{
    return hdr().tp_status & TP_STATUS_VLAN_TPID_VALID ? hdr().tp_vlan_tpid : 0x8100;
}


prefix_ void senf::SocketQueueBuffer::resize(size_type sz
#ifdef SENF_ENABLE_TPACKET_OFFSET
                                             , int offset
#endif
)
{
#ifdef SENF_ENABLE_TPACKET_OFFSET
    if (offset == -1)
        offset = dataOffset();
#else
    unsigned offset (dataOffset());
#endif
    SENF_ASSERT( frameBegin() + offset + sz <= frameEnd(), "frame size exceeded" );
    hdr().tp_mac = hdr().tp_net = frameBegin() + offset - b_;
    hdr().tp_len = sz;
}

///////////////////////////////////////////////////////////////////////////
// senf::QueueReadPolicy

prefix_ boost::optional<senf::QueueReadPolicy::Buffer>
senf::QueueReadPolicy::dequeue(FileHandle & handle)
{
    detail::QueueInfo & qi (* qinfo(handle));
    for (unsigned count (0); count < 2*qi.rx.qlen; ++count) {
        struct ::tpacket2_hdr & pk (* reinterpret_cast<struct ::tpacket2_hdr *>(qi.rx.head));
        if (SENF_LIKELY((qi.rx.idle || qi.rx.head != qi.rx.tail) && pk.tp_status != TP_STATUS_KERNEL)) {
            qi.rxStats.received++;
            Buffer bf (qi.rx.head, qi.rx.head + qi.frameSize, qi.hdrlen);
            struct ::sockaddr_ll & sa (* reinterpret_cast<struct ::sockaddr_ll *>(qi.rx.head + qi.hdrlen));
            qi.inc(qi.rx.head, qi.rx);
            if (SENF_UNLIKELY(sa.sll_pkttype >= PACKET_OUTGOING || bf.end() > bf.frameEnd() || bf.empty())) {
                qi.rxStats.ignored++;
                pk.tp_status = TP_STATUS_KERNEL;
                if (qi.rx.idle)
                    qi.rx.tail = qi.rx.head;
            } else {
                qi.rx.idle = false;
                return bf;
            }
        }
    }
    return boost::none;
}

prefix_ void senf::QueueReadPolicy::release(FileHandle & handle)
{
    detail::QueueInfo & qi (* qinfo(handle));
    while (SENF_LIKELY(!qi.rx.idle || qi.rx.tail != qi.rx.head)) {  // we assume bursts => likely
        struct ::tpacket2_hdr & pk (* reinterpret_cast<struct ::tpacket2_hdr*>(qi.rx.tail));
        pk.tp_status = TP_STATUS_KERNEL;
        qi.inc(qi.rx.tail, qi.rx);
        // We set qi.rx.idle prematurely but this is safe: when the while loop is done,
        // the queue is indeed idle. After the first loop iteration, it is impossible for
        // the queue to be still completely filled (that is head == tail) so the while
        // condition above is now correct with idle = true
        qi.rx.idle = true;
    }
}

prefix_ unsigned senf::QueueReadPolicy::read(FileHandle & handle, char * buffer, unsigned size)
{
    if (SENF_UNLIKELY(!handle))
        return 0;

    SENF_ASSERT( qinfo(handle)->rx.idle,
            "Inconsistent mixing of read() and dequeue() on mmap socket" );
    bool blocking (false);
    for (;;) {
        boost::optional<Buffer> buf (dequeue(handle));
        if (! buf) {
            if (blocking || handle.blocking()) {
                blocking = true;
                handle.waitReadable();
                continue;
            }
            return 0;
        }
        if (unsigned(buf->size()) < size)
            size = buf->size();
        ::memcpy(buffer, buf->begin(), size);
        release(handle);
        return size;
    }
}

///////////////////////////////////////////////////////////////////////////
// senf::QueueWritePolicy

prefix_ boost::optional<senf::QueueWritePolicy::Buffer>
senf::QueueWritePolicy::enqueue(FileHandle & handle)
{
    detail::QueueInfo & qi (* qinfo(handle));    
    struct ::tpacket2_hdr & pk (* reinterpret_cast<struct ::tpacket2_hdr*>(qi.tx.head));
    if (SENF_LIKELY((qi.tx.idle || qi.tx.head != qi.tx.tail) && (pk.tp_status == TP_STATUS_AVAILABLE))) {
        ::memset(&pk, 0, qi.hdrlen);
        pk.tp_mac = pk.tp_net = qi.hdrlen
#ifdef SENF_ENABLE_TPACKET_OFFSET
            + qi.reserve
#endif
            ;
        Buffer bf (qi.tx.head, qi.tx.head + qi.frameSize, qi.hdrlen);
        qi.inc(qi.tx.head, qi.tx);
        qi.tx.idle = false;
        return bf;
    }

    // our TX MMAP ring is full
    qi.txStats.overrun++;
    
    // let's trigger a send() to push out the packets
    // this should be handled via the writable event, but it can't hurt...
    if (handle.writeable())
        send(handle);

    return boost::none;
}

prefix_ void senf::QueueWritePolicy::flush(FileHandle & handle)
{
    if (SENF_UNLIKELY(!handle))
        return;

    detail::QueueInfo & qi (* qinfo(handle));

    //
    // just forget about any frames in the ring which have not yet been pushed into the kernel
    //
    while (SENF_LIKELY(!qi.tx.idle || qi.tx.tail != qi.tx.head)) {  // we assume bursts >> likely
        struct ::tpacket2_hdr & pk (* reinterpret_cast<struct ::tpacket2_hdr*>(qi.tx.tail));
        pk.tp_status = TP_STATUS_AVAILABLE;
        qi.inc(qi.tx.tail, qi.tx);
        // We set qi.tx.idle prematurely but this is safe: when the while loop is done,
        // the queue is indeed idle. After the first loop iteration, it is impossible for
        // the queue to be still completely filled (that is head == tail) so the while
        // condition above is now correct with idle = true
        qi.tx.idle = true;
    }

    // check if there are any more SEND_REQUEST marked frames that we can just discard/flush
    for (unsigned char * buf = qi.tx.begin; buf < qi.tx.end; buf += qi.frameSize) { 
        struct ::tpacket2_hdr & pk (* reinterpret_cast<struct ::tpacket2_hdr*>(buf));
        if (pk.tp_status == TP_STATUS_SEND_REQUEST) {
            pk.tp_status = TP_STATUS_AVAILABLE;
        }
    }
}

prefix_ bool senf::QueueWritePolicy::send(FileHandle & handle)
{
    detail::QueueInfo & qi (* qinfo(handle));
    unsigned char * last_tail (qi.tx.tail);
    while (SENF_LIKELY(!qi.tx.idle || qi.tx.tail != qi.tx.head)) {  // we assume bursts >> likely
        struct ::tpacket2_hdr & pk (* reinterpret_cast<struct ::tpacket2_hdr*>(qi.tx.tail));
        SENF_ASSERT(pk.tp_len, "missing call to Buffer::resize()");
        pk.tp_status = TP_STATUS_SEND_REQUEST;
        qi.inc(qi.tx.tail, qi.tx);
        qi.txStats.sent++;
        // We set qi.tx.idle prematurely but this is safe: when the while loop is done,
        // the queue is indeed idle. After the first loop iteration, it is impossible for
        // the queue to be still completely filled (that is head == tail) so the while
        // condition above is now correct with idle = true
        qi.tx.idle = true;
    }

    bool tryAgain (false);

    ssize_t sent;
    for (;;) {
        if (SENF_UNLIKELY((sent = ::send(handle.fd(), NULL, 0, MSG_DONTWAIT)) < 0)) {
            switch (errno) {
            case EINTR:
                // We used to break here as 3.16.x seems to throw tons of those and 'continue' caused a busy-loop
                // This should be fixed by our patch which made it into Linux 4.1
                tryAgain = true;
                break;  // mtk 01-jul-2015: let's break here again for older kernels
            case EAGAIN:
                tryAgain = true;
                break;
            case EMSGSIZE:
                // Note: As of 1-Apr-2019 we use the PACKET_LOSS option and should not see this error anymore...
                // the trigerring frame(s) will be marked by the kernel as WRONG_FORMAT
                for (unsigned char * buf = qi.tx.begin; buf < qi.tx.end; buf += qi.frameSize) {
                    struct ::tpacket2_hdr & pk (* reinterpret_cast<struct ::tpacket2_hdr*>(buf));
                    if (pk.tp_status & TP_STATUS_WRONG_FORMAT) {
                        pk.tp_status = TP_STATUS_SEND_REQUEST;
                        //  This does not seem to work properly...
                        pk.tp_len = 64;  // set the length of this frame 64 bytes - effectively dropping it without having to rearrange the mmap queue.
                        qi.txStats.wrongFormat++;
                        qi.txStats.dropped++;
                        break;
                    }
                }
                continue;
            case ENOBUFS:
                // Just break'ing here is ok for normal sockets. But may cause problems when QDISC_BYPASS is turned on.
                // In this case, the kernel seems to report ENOBUFS instead of ENOTCON or ENETDOWN and 
                // the sndBuf seems to be full, but the socket is still writable...
                break;
            case ENOTCONN:
            case ENETDOWN:
            case ENXIO:
                // Some frames might have been sent, hence we first have to reset and then advance tx.tail accordingly
                qi.tx.tail = last_tail;
                while (qi.tx.tail != qi.tx.head) {
                    struct ::tpacket2_hdr & pk (* reinterpret_cast<struct ::tpacket2_hdr*>(qi.tx.tail));
                    if (pk.tp_status != TP_STATUS_SEND_REQUEST)
                        qi.inc(qi.tx.tail, qi.tx);
                    else
                        break;
                }
                // We flush our queue, to avoid a busy loop, effectively dropping all frames !!!
                for (unsigned char * buf = qi.tx.begin; buf < qi.tx.end; buf += qi.frameSize) {
                    struct ::tpacket2_hdr & pk (* reinterpret_cast<struct ::tpacket2_hdr*>(buf));
                    if (pk.tp_status == TP_STATUS_SEND_REQUEST) {
                        pk.tp_status = TP_STATUS_AVAILABLE;
                        qi.txStats.dropped++;
                    }
                }
                // reset tx.head effectily dropping all unsent frames
                qi.tx.head = qi.tx.tail;
                break;
            default:
                SENF_THROW_SYSTEM_EXCEPTION("::send(MMAP)");
            }
        } else {
            if (sent > 0) {
                // we might have more data to send...
                tryAgain = true;
            }
        }
        break;
    }

    return tryAgain;
}

prefix_ unsigned senf::QueueWritePolicy::write(FileHandle & handle, char const * buffer,
                                               unsigned size)
{
    if (SENF_UNLIKELY(!handle))
        return 0;

    SENF_ASSERT( qinfo(handle)->tx.idle,
            "Inconsistent mixing of write() and enqueue() on mmap socket" );
    for (bool repeat (true);; repeat = false) {
        boost::optional<Buffer> buf (enqueue(handle));
        if (! buf) {
            if (repeat && handle.blocking()) {
                handle.waitWriteable();
                continue;
            }
            return 0;
        }
        if (unsigned(buf->frameSize()) < size)
            size = buf->frameSize();
        buf->resize(size);
        ::memcpy(buf->begin(), buffer, size);
        send(handle);
        return size;
    }
}

//-/////////////////////////////////////////////////////////////////////////////////////////////////
#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: