CombinerV2.cpp

Go to the documentation of this file.

 
#include "rogue/protocols/batcher/CombinerV2.h"
 
#include <inttypes.h>
#include <stdint.h>
 
#include <vector>
#include <string.h>
 
#include <memory>
 
#include "rogue/GilRelease.h"
#include "rogue/interfaces/stream/Frame.h"
#include "rogue/interfaces/stream/FrameIterator.h"
#include "rogue/interfaces/stream/FrameLock.h"
#include "rogue/interfaces/stream/Master.h"
#include "rogue/interfaces/stream/Slave.h"
 
namespace rpb = rogue::protocols::batcher;
namespace ris = rogue::interfaces::stream;
 
#ifndef NO_PYTHON
    #include <boost/python.hpp>
namespace bp = boost::python;
#endif
 
rpb::CombinerV2Ptr rpb::CombinerV2::create() {
    rpb::CombinerV2Ptr p = std::make_shared<rpb::CombinerV2>();
    return (p);
}
 
void rpb::CombinerV2::setup_python() {
#ifndef NO_PYTHON
    bp::class_<rpb::CombinerV2, rpb::CombinerV2Ptr, bp::bases<ris::Master, ris::Slave>, boost::noncopyable>(
        "CombinerV2",
        bp::init<>())
        .def("sendBatch", &rpb::CombinerV2::sendBatch)
        .def("getCount", &rpb::CombinerV2::getCount);
#endif
}
 
rpb::CombinerV2::CombinerV2() : ris::Master(), ris::Slave() {
    log_ = rogue::Logging::create("batcher.CombinerV2");
    seq_ = 0;
}
 
rpb::CombinerV2::~CombinerV2() {}
 
void rpb::CombinerV2::acceptFrame(ris::FramePtr frame) {
    rogue::GilRelease noGil;
    ris::FrameLockPtr lock = frame->lock();
 
    std::lock_guard<std::mutex> guard(mtx_);
    queue_.push_back(frame);
}
 
uint32_t rpb::CombinerV2::getCount() {
    std::lock_guard<std::mutex> guard(mtx_);
    return queue_.size();
}
 
void rpb::CombinerV2::sendBatch() {
    const uint32_t headerSize = 2;
    const uint32_t tailSize   = 7;
 
    std::vector<ris::FramePtr> frames;
 
    {
        std::lock_guard<std::mutex> guard(mtx_);
        if (queue_.empty()) return;
        frames.swap(queue_);
    }
 
    // Compute total super-frame size
    uint32_t totalSize = headerSize;
    for (auto& f : frames) {
        totalSize += f->getPayload() + tailSize;
    }
 
    rogue::GilRelease noGil;
 
    // Allocate the super-frame
    ris::FramePtr sFrame = reqFrame(totalSize, true);
    sFrame->setPayload(totalSize);
 
    ris::FrameIterator it = sFrame->begin();
 
    // Write super header (2 bytes)
    uint8_t byte0 = 0x02;  // version=2, reserved bits=0
    ris::toFrame(it, 1, &byte0);
    ris::toFrame(it, 1, &seq_);
 
    // Write each record: data then tail
    for (auto& f : frames) {
        ris::FrameLockPtr fLock = f->lock();
        uint32_t payloadSize    = f->getPayload();
 
        // Copy payload data
        ris::FrameIterator srcIt = f->begin();
        ris::copyFrame(srcIt, payloadSize, it);
 
        // Write tail (7 bytes)
        // Word 0: size (4 bytes)
        ris::toFrame(it, 4, &payloadSize);
 
        // Word 1: dest, fUser, lUser (3 bytes)
        uint8_t dest  = f->getChannel();
        uint8_t fUser = f->getFirstUser();
        uint8_t lUser = f->getLastUser();
        ris::toFrame(it, 1, &dest);
        ris::toFrame(it, 1, &fUser);
        ris::toFrame(it, 1, &lUser);
    }
 
    seq_++;
 
    sendFrame(sFrame);
}