Fifo.cpp

Go to the documentation of this file.

 
#include "rogue/Directives.h"
 
#include "rogue/interfaces/stream/Fifo.h"
 
#include <stdint.h>
 
#include <memory>
#include <thread>
 
#include "rogue/GilRelease.h"
#include "rogue/Logging.h"
#include "rogue/interfaces/stream/Buffer.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 ris = rogue::interfaces::stream;
 
#ifndef NO_PYTHON
    #include <boost/python.hpp>
namespace bp = boost::python;
#endif
 
ris::FifoPtr ris::Fifo::create(uint32_t maxDepth, uint32_t trimSize, bool noCopy) {
    ris::FifoPtr p = std::make_shared<ris::Fifo>(maxDepth, trimSize, noCopy);
    return (p);
}
 
void ris::Fifo::setup_python() {
#ifndef NO_PYTHON
    bp::class_<ris::Fifo, ris::FifoPtr, bp::bases<ris::Master, ris::Slave>, boost::noncopyable>(
        "Fifo",
        bp::init<uint32_t, uint32_t, bool>())
        .def("size", &Fifo::size)
        .def("dropCnt", &Fifo::dropCnt)
        .def("clearCnt", &Fifo::clearCnt);
#endif
}
 
ris::Fifo::Fifo(uint32_t maxDepth, uint32_t trimSize, bool noCopy)
    : ris::Master(),
      ris::Slave(),
      log_(rogue::Logging::create("stream.Fifo")),
      maxDepth_(maxDepth),
      trimSize_(trimSize),
      noCopy_(noCopy),
      dropFrameCnt_(0),
      threadEn_(true),
      thread_(new std::thread(&ris::Fifo::runThread, this)) {
    queue_.setThold(maxDepth);
 
    // Set a thread name
#ifndef __MACH__
    pthread_setname_np(thread_->native_handle(), "Fifo");
#endif
}
 
ris::Fifo::~Fifo() {
    // No-op if the ctor threw before allocating thread_.
    if (thread_ != nullptr) {
        threadEn_ = false;
        rogue::GilRelease noGil;
        queue_.stop();
        thread_->join();
        delete thread_;
        thread_ = nullptr;
    }
}
 
std::size_t ris::Fifo::size() {
    return queue_.size();
}
 
std::size_t ris::Fifo::dropCnt() const {
    return dropFrameCnt_;
}
 
void ris::Fifo::clearCnt() {
    dropFrameCnt_ = 0;
}
 
void ris::Fifo::acceptFrame(ris::FramePtr frame) {
    uint32_t size;
    ris::FramePtr nFrame;
    ris::FrameIterator src;
    ris::FrameIterator dst;
 
    // FIFO is full, drop frame
    if (queue_.busy()) {
        ++dropFrameCnt_;
        return;
    }
 
    rogue::GilRelease noGil;
    ris::FrameLockPtr lock = frame->lock();
 
    // Do we copy the frame?
    if (noCopy_) {
        nFrame = frame;
    } else {
        // Get size, adjust if trim is enabled
        size = frame->getPayload();
        if (trimSize_ != 0 && trimSize_ < size) size = trimSize_;
 
        // Request a new frame to hold the data
        nFrame = reqFrame(size, true);
        nFrame->setPayload(size);
 
        // Get destination pointer
        src = frame->begin();
        dst = nFrame->begin();
 
        // Copy the frame
        ris::copyFrame(src, size, dst);
        nFrame->setError(frame->getError());
        nFrame->setChannel(frame->getChannel());
        nFrame->setFlags(frame->getFlags());
    }
 
    // Append to buffer
    queue_.push(nFrame);
}
 
void ris::Fifo::runThread() {
    ris::FramePtr frame;
    log_->logThreadId();
 
    while (threadEn_) {
        if ((frame = queue_.pop()) != NULL) sendFrame(frame);
    }
}