ControllerV1.cpp

Go to the documentation of this file.

 
#include "rogue/protocols/packetizer/ControllerV1.h"
 
#include <inttypes.h>
#include <sys/time.h>
#include <unistd.h>
 
#include <cmath>
#include <cstdlib>
#include <memory>
 
#include "rogue/GeneralError.h"
#include "rogue/GilRelease.h"
#include "rogue/interfaces/stream/Buffer.h"
#include "rogue/interfaces/stream/Frame.h"
#include "rogue/interfaces/stream/FrameLock.h"
#include "rogue/protocols/packetizer/Application.h"
#include "rogue/protocols/packetizer/Transport.h"
 
namespace rpp = rogue::protocols::packetizer;
namespace ris = rogue::interfaces::stream;
 
rpp::ControllerV1Ptr rpp::ControllerV1::create(bool enSsi, rpp::TransportPtr tran, rpp::ApplicationPtr* app) {
    rpp::ControllerV1Ptr r = std::make_shared<rpp::ControllerV1>(enSsi, tran, app);
    return (r);
}
 
rpp::ControllerV1::ControllerV1(bool enSsi, rpp::TransportPtr tran, rpp::ApplicationPtr* app)
    : rpp::Controller::Controller(tran, app, 8, 1, 8, enSsi) {}
 
rpp::ControllerV1::~ControllerV1() {}
 
void rpp::ControllerV1::transportRx(ris::FramePtr frame) {
    ris::BufferPtr buff;
    uint32_t size;
    uint32_t tmpIdx;
    uint32_t tmpCount;
    uint8_t tmpFuser;
    uint8_t tmpLuser;
    uint8_t tmpDest;
    uint8_t tmpId;
    bool tmpEof;
    uint32_t flags;
    uint8_t* data;
 
    if (frame->isEmpty()) log_->warning("Empty frame received on transport input");
 
    rogue::GilRelease noGil;
    ris::FrameLockPtr flock = frame->lock();
    std::lock_guard<std::mutex> lock(tranMtx_);
 
    buff = *(frame->beginBuffer());
    data = buff->begin();
    size = buff->getPayload();
 
    // Drop invalid data
    if (frame->getError() ||            // Check for frame ERROR
        (frame->bufferCount() != 1) ||  // Incoming frame can only have one buffer
        (size < 10) ||                  // Check for min. size (64-bit header + 8-bit min. payload + 8-bit tail)
        ((data[0] & 0xF) != 0)) {       // Check for invalid version only
        log_->warning("Dropping frame due to contents: error=0x%" PRIx8 ", payload=%" PRIu32 ", buffers=%" PRIu32
                      ", Version=0x%" PRIx8,
                      frame->getError(),
                      size,
                      frame->bufferCount(),
                      data[0] & 0xF);
        dropCount_++;
        return;
    }
 
    tmpIdx = static_cast<uint32_t>(data[0]) >> 4;
    tmpIdx |= static_cast<uint32_t>(data[1]) << 4;
 
    tmpCount = static_cast<uint32_t>(data[2]);
    tmpCount |= static_cast<uint32_t>(data[3]) << 8;
    tmpCount |= static_cast<uint32_t>(data[4]) << 16;
 
    tmpDest  = data[5];
    tmpId    = data[6];
    tmpFuser = data[7];
 
    tmpLuser = data[size - 1] & 0x7F;
    tmpEof   = data[size - 1] & 0x80;
 
    log_->debug("transportRx: Raw header: 0x%" PRIx8 ", 0x%" PRIx8 ", 0x%" PRIx8 ", 0x%" PRIx8 ", 0x%" PRIx8
                ", 0x%" PRIx8 ", 0x%" PRIx8 ", 0x%" PRIx8,
                data[0],
                data[1],
                data[2],
                data[3],
                data[4],
                data[5],
                data[6],
                data[7]);
    log_->debug("transportRx: Raw footer: 0x%" PRIx8, data[size - 1]);
    log_->debug("transportRx: Got frame: Fuser=0x%" PRIx8 ", Dest=0x%" PRIx8 ", Id=0x%" PRIx32 ", Count=%" PRIx32
                ", Luser=0x%" PRIx8 ", Eof=%" PRIu8 ", size=%" PRIu32,
                tmpFuser,
                tmpDest,
                tmpIdx,
                tmpCount,
                tmpLuser,
                tmpEof,
                size);
 
    // Shorten message by one byte (before adjusting tail)
    buff->adjustPayload(-1);
 
    // Adjust header and tail reservations
    buff->adjustHeader(8);
    buff->adjustTail(1);
 
    // Drop frame and reset state if mismatch
    if (tmpCount > 0 && (tmpIdx != tranIndex_ || tmpCount != tranCount_[0])) {
        log_->warning("Dropping frame due to state mismatch: expIdx=%" PRIu32 ", gotIdx=%" PRIu32 ", expCount=%" PRIu32
                      ", gotCount=%" PRIu32,
                      tranIndex_,
                      tmpIdx,
                      tranCount_[0],
                      tmpCount);
        dropCount_++;
        tranCount_[0] = 0;
        tranFrame_[0].reset();
        return;
    }
 
    // First frame
    if (tmpCount == 0) {
        if (tranCount_[0] != 0)
            log_->warning("Dropping frame due to new incoming frame: expIdx=%" PRIu32 ", expCount=%" PRIu32,
                          tranIndex_,
                          tranCount_[0]);
 
        tranFrame_[0] = ris::Frame::create();
        tranIndex_    = tmpIdx;
        tranDest_     = tmpDest;
        tranCount_[0] = 0;
 
        tranFrame_[0]->setFirstUser(tmpFuser);
    }
 
    tranFrame_[0]->appendBuffer(buff);
    frame->clear();  // Empty old frame
 
    // Last of transfer
    if (tmpEof) {
        tranFrame_[0]->setLastUser(tmpLuser);
        tranCount_[0] = 0;
 
        // Detect SSI error — must precede pushFrame/reset or the indication is lost.
        if (enSsi_ & (tmpLuser & 0x1)) tranFrame_[0]->setError(0x80);
 
        if (app_[tranDest_]) {
            app_[tranDest_]->pushFrame(tranFrame_[0]);
        }
        tranFrame_[0].reset();
    } else {
        tranCount_[0]++;
    }
}
 
void rpp::ControllerV1::applicationRx(ris::FramePtr frame, uint8_t tDest) {
    ris::Frame::BufferIterator it;
    uint32_t segment;
    uint8_t* data;
    uint32_t size;
    uint8_t fUser;
    uint8_t lUser;
    struct timeval startTime;
    struct timeval currTime;
    struct timeval endTime;
 
    gettimeofday(&startTime, NULL);
    timeradd(&startTime, &timeout_, &endTime);
 
    if (frame->isEmpty()) log_->warning("Empty frame received on application input");
 
    if (frame->getError()) return;
 
    rogue::GilRelease noGil;
    ris::FrameLockPtr flock = frame->lock();
    std::lock_guard<std::mutex> lock(appMtx_);
 
    // Wait while queue is busy
    while (tranQueue_.busy()) {
        usleep(10);
        gettimeofday(&currTime, NULL);
        if (timercmp(&currTime, &endTime, >)) {
            log_->critical("ControllerV1::applicationRx: Timeout waiting for outbound queue after %" PRIu32 ".%" PRIu32
                           " seconds! May be caused by outbound backpressure.",
                           timeout_.tv_sec,
                           timeout_.tv_usec);
            gettimeofday(&startTime, NULL);
            timeradd(&startTime, &timeout_, &endTime);
        }
    }
 
    // User fields
    fUser = frame->getFirstUser();
    lUser = frame->getLastUser();
 
    // Inject SOF
    if (enSsi_) fUser |= 0x2;
 
    segment = 0;
    for (it = frame->beginBuffer(); it != frame->endBuffer(); ++it) {
        ris::FramePtr tFrame = ris::Frame::create();
 
        // Adjust header and tail, before setting new size
        (*it)->adjustHeader(-8);
        (*it)->adjustTail(-1);
 
        // Make payload one byte longer
        (*it)->adjustPayload(1);
 
        size = (*it)->getPayload();
        data = (*it)->begin();
 
        data[0] = (appIndex_ & 0xF) << 4;  // Note: BIT3:0 = 0x0 (Version)
        data[1] = (appIndex_ >> 4) & 0xFF;
 
        data[2] = (segment >> 0) & 0xFF;
        data[3] = (segment >> 8) & 0xFF;
        data[4] = (segment >> 16) & 0xFF;
 
        data[5] = tDest;
        data[6] = 0;  // TID Unused
        data[7] = fUser;
 
        data[size - 1] = lUser & 0x7F;
 
        if (it == (frame->endBuffer() - 1)) data[size - 1] |= 0x80;
 
        tFrame->appendBuffer(*it);
        tranQueue_.push(tFrame);
        segment++;
    }
    appIndex_++;
    frame->clear();  // Empty old frame
}