AxiStreamDma.cpp

Go to the documentation of this file.

 
#include "rogue/Directives.h"
 
#include "rogue/hardware/axi/AxiStreamDma.h"
 
#include <inttypes.h>
#include <poll.h>
 
#include <cstdio>
#include <cstdlib>
#include <map>
#include <memory>
#include <mutex>
#include <string>
 
#include "rogue/GeneralError.h"
#include "rogue/GilRelease.h"
#include "rogue/Helpers.h"
#include "rogue/hardware/drivers/AxisDriver.h"
#include "rogue/interfaces/stream/Buffer.h"
#include "rogue/interfaces/stream/Frame.h"
#include "rogue/interfaces/stream/FrameLock.h"
 
namespace rha = rogue::hardware::axi;
namespace ris = rogue::interfaces::stream;
 
#ifndef NO_PYTHON
    #include <boost/python.hpp>
namespace bp = boost::python;
#endif
 
std::map<std::string, std::shared_ptr<rha::AxiStreamDmaShared> > rha::AxiStreamDma::sharedBuffers_;
 
// Protect sharedBuffers_ against concurrent AxiStreamDma construction.
static std::mutex sharedBuffersMtx;
 
rha::AxiStreamDmaShared::AxiStreamDmaShared(std::string path) {
    this->fd        = -1;
    this->path      = path;
    // Start at 0; openShared() is the single point that increments openCount
    // for every caller (including the !zCopyEn early-return path), so a freshly
    // constructed record must contribute zero to the refcount.  Initializing to
    // 1 here previously double-counted entries seeded by zeroCopyDisable() and
    // was the root cause of openCount drifting negative on close.
    this->openCount = 0;
    this->rawBuff   = NULL;
    this->bCount    = 0;
    this->bSize     = 0;
    this->zCopyEn   = true;
}
 
rha::AxiStreamDmaSharedPtr rha::AxiStreamDma::openShared(std::string path, rogue::LoggingPtr log) {
    rogue::GilRelease noGil;
    std::lock_guard<std::mutex> lock(sharedBuffersMtx);
    std::map<std::string, rha::AxiStreamDmaSharedPtr>::iterator it;
    rha::AxiStreamDmaSharedPtr ret;
 
    // Entry already exists
    if ((it = sharedBuffers_.find(path)) != sharedBuffers_.end()) {
        ret = it->second;
        log->debug("Reusing existing shared file descriptor for %s", path.c_str());
 
        // Already opened by an earlier caller; just count the new user.
        if (ret->fd != -1) {
            ret->openCount++;
            log->debug("Shared file descriptor already opened for %s", path.c_str());
            return ret;
        }
 
        // Entry was seeded by zeroCopyDisable() (or fully closed and is being
        // reopened with zCopyEn=false): no kernel fd will be held, but every
        // AxiStreamDma sharing this path is still a refcount user.  Without
        // this branch, closeShared() would decrement an entry that never got
        // counted, driving openCount negative.
        if (!ret->zCopyEn) {
            ret->openCount++;
            log->debug("Zero copy is disabled. Not Mapping Buffers for %s", path.c_str());
            return ret;
        }
 
        // Fall through to the open+map path; increment after success so a
        // throw before insertion cannot leak openCount.
 
        // Create new record
    } else {
        ret = std::make_shared<rha::AxiStreamDmaShared>(path);
        log->debug("Opening new shared file descriptor for %s", path.c_str());
    }
 
    // We need to open device and create shared buffers
    if ((ret->fd = ::open(path.c_str(), O_RDWR)) < 0)
        throw(rogue::GeneralError::create("AxiStreamDma::openShared", "Failed to open device file: %s", path.c_str()));
 
    // Check driver version ( ApiVersion 0x05 (or less) is the 32-bit address version)
    //
    // closeShared() leaves entries in sharedBuffers_ after openCount drops to
    // 0 (only fd/bCount/bSize/rawBuff are reset), so an entry can be re-entered
    // here via the "fd == -1, zCopyEn == true" fall-through.  If we close the
    // fd on error but leave ret->fd != -1, the persistent map entry will still
    // hold the (now closed) descriptor, and the next openShared() caller would
    // hit the "ret->fd != -1" fast-path and reuse a stale fd.  Reset the
    // shared record fd (ret->fd) to -1 (and clear any partial mapping state)
    // on every error path before throwing.  Note: ret->fd here is the shared
    // AxiStreamDmaShared record field, distinct from the AxiStreamDma::fd_
    // per-instance member.
    if (dmaGetApiVersion(ret->fd) < 0x06) {
        ::close(ret->fd);
        ret->fd = -1;
        throw(rogue::GeneralError("AxiStreamDma::openShared",
                                  R"(Bad kernel driver version detected. Please re-compile kernel driver.
          Note that aes-stream-driver (v5.15.2 or earlier) and rogue (v5.11.1 or earlier) are compatible with the 32-bit address API.
          To use later versions (64-bit address API),, you will need to upgrade both rogue and aes-stream-driver at the same time to:
          \t\taes-stream-driver = v5.16.0 (or later)
          \t\trogue = v5.13.0 (or later))"));
    }
 
    // Check for mismatch in the rogue/loaded_driver API versions
    if (dmaCheckVersion(ret->fd) < 0) {
        ::close(ret->fd);
        ret->fd = -1;
        throw(rogue::GeneralError(
            "AxiStreamDma::openShared",
            "Rogue DmaDriver.h API Version (DMA_VERSION) does not match the aes-stream-driver API version"));
    }
 
    // Result may be that rawBuff = NULL
    // Should this just be a warning?
    ret->rawBuff = dmaMapDma(ret->fd, &(ret->bCount), &(ret->bSize));
    if (ret->rawBuff == NULL) {
        // dmaMapDma may have written to bCount/bSize even on failure; query
        // the buffer count one more time for the user-facing error message,
        // then scrub the shared record so a future openShared() on a
        // re-entered entry starts clean.
        uint32_t mapSize = dmaGetBuffCount(ret->fd) + 8192;
 
        ::close(ret->fd);
        ret->fd      = -1;
        ret->bCount  = 0;
        ret->bSize   = 0;
        ret->rawBuff = NULL;
        throw(rogue::GeneralError::create(
            "AxiStreamDma::openShared",
            "Failed to map dma buffers. Increase vm map limit to : sudo sysctl -w vm.max_map_count=%i",
            mapSize));
    }
    log->debug("Mapped buffers. bCount = %i, bSize=%i for %s", ret->bCount, ret->bSize, path.c_str());
 
    // Open + map succeeded; count this caller and publish in the map.
    ret->openCount++;
    sharedBuffers_.insert(std::pair<std::string, rha::AxiStreamDmaSharedPtr>(path, ret));
    return ret;
}
 
void rha::AxiStreamDma::closeShared(rha::AxiStreamDmaSharedPtr desc) {
    rogue::GilRelease noGil;
    std::lock_guard<std::mutex> lock(sharedBuffersMtx);
    desc->openCount--;
 
    if (desc->openCount == 0) {
        if (desc->rawBuff != NULL) {
            dmaUnMapDma(desc->fd, desc->rawBuff);
        }
 
        // Guard against close(-1).  In the zeroCopyDisable() flow desc->fd is
        // never opened, so calling ::close(desc->fd) here would issue a real
        // syscall with fd=-1 and return EBADF.  Skip the close when the fd
        // was never held by this descriptor.
        if (desc->fd != -1) {
            ::close(desc->fd);
        }
        desc->fd      = -1;
        desc->bCount  = 0;
        desc->bSize   = 0;
        desc->rawBuff = NULL;
    }
}
 
rha::AxiStreamDmaPtr rha::AxiStreamDma::create(std::string path, uint32_t dest, bool ssiEnable) {
    rha::AxiStreamDmaPtr r = std::make_shared<rha::AxiStreamDma>(path, dest, ssiEnable);
    return (r);
}
 
void rha::AxiStreamDma::zeroCopyDisable(std::string path) {
    rogue::GilRelease noGil;
    std::lock_guard<std::mutex> lock(sharedBuffersMtx);
    std::map<std::string, rha::AxiStreamDmaSharedPtr>::iterator it;
 
    // Entry already exists
    if ((it = sharedBuffers_.find(path)) != sharedBuffers_.end())
        throw(rogue::GeneralError("AxiStreamDma::zeroCopyDisable",
                                  "zeroCopyDisable can't be called twice or after a device has been opened"));
 
    // Create new record
    rha::AxiStreamDmaSharedPtr ret = std::make_shared<rha::AxiStreamDmaShared>(path);
    ret->zCopyEn                   = false;
 
    sharedBuffers_.insert(std::pair<std::string, rha::AxiStreamDmaSharedPtr>(path, ret));
}
 
rha::AxiStreamDma::AxiStreamDma(std::string path, uint32_t dest, bool ssiEnable) {
    uint8_t mask[DMA_MASK_SIZE];
 
    dest_     = dest;
    enSsi_    = ssiEnable;
    retThold_ = 1;
 
    // Create a shared pointer to use as a lock for runThread()
    std::shared_ptr<int> scopePtr = std::make_shared<int>(0);
 
    rogue::defaultTimeout(timeout_);
 
    log_ = rogue::Logging::create("axi.AxiStreamDma");
 
    rogue::GilRelease noGil;
 
    // Attempt to open shared structure
    desc_ = openShared(path, log_);
 
    // Open non shared file descriptor
    if ((fd_ = ::open(path.c_str(), O_RDWR)) < 0) {
        closeShared(desc_);
        throw(
            rogue::GeneralError::create("AxiStreamDma::AxiStreamDma", "Failed to open device file: %s", path.c_str()));
    }
 
    // desc_->bCount and desc_->bSize live in the shared AxiStreamDmaShared
    // instance returned by openShared(), which is also looked up under
    // sharedBuffersMtx by other concurrent constructions on the same path.
    // Two ctors racing here would otherwise both write the same fields
    // without synchronisation - a data race even when the values written
    // are identical.  Take the same mutex that openShared / closeShared /
    // zeroCopyDisable use, and BOTH read and write bCount under it (the
    // earlier "read at function entry, write inside the conditional" form
    // left the read racing with another ctor's write).
    {
        std::lock_guard<std::mutex> lock(sharedBuffersMtx);
        if (desc_->rawBuff == NULL) {
            desc_->bCount = dmaGetTxBuffCount(fd_) + dmaGetRxBuffCount(fd_);
            desc_->bSize  = dmaGetBuffSize(fd_);
        }
        if (desc_->bCount >= 1000) retThold_ = 80;
    }
 
    dmaInitMaskBytes(mask);
    dmaAddMaskBytes(mask, dest_);
 
    if (dmaSetMaskBytes(fd_, mask) < 0) {
        closeShared(desc_);
        ::close(fd_);
        fd_ = -1;
        throw(rogue::GeneralError::create("AxiStreamDma::AxiStreamDma",
                                          "Failed to open device file %s with dest 0x%" PRIx32
                                          "! Another process may already have it open!",
                                          path.c_str(),
                                          dest));
    }
 
    threadEn_ = true;
    try {
        thread_ = std::make_unique<std::thread>(&rha::AxiStreamDma::runThread, this, std::weak_ptr<int>(scopePtr));
    } catch (...) {
        threadEn_ = false;
        closeShared(desc_);
        ::close(fd_);
        fd_ = -1;
        throw;
    }
 
    // Set a thread name
#ifndef __MACH__
    pthread_setname_np(thread_->native_handle(), "AxiStreamDma");
#endif
}
 
rha::AxiStreamDma::~AxiStreamDma() {
    this->stop();
}
 
void rha::AxiStreamDma::stop() {
    rogue::GilRelease noGil;
 
    // Signal the worker to exit; idempotent so stop() can be called multiple
    // times (user-initiated stop() followed by ~AxiStreamDma()).
    threadEn_ = false;
 
    // Join based on thread state, not threadEn_.  runThread() can now exit on
    // its own (e.g. the in-loop fd_ guard) by setting threadEn_ = false and
    // returning, so gating cleanup on threadEn_ here would skip the join and
    // ~unique_ptr<std::thread> on a still-joinable thread would call
    // std::terminate().  Joining via joinable() handles both that early-exit
    // path and the normal stop() path with the same code.
    if (thread_ && thread_->joinable()) {
        thread_->join();
    }
    thread_.reset();
 
    // Release the shared descriptor exactly once even if stop() is called
    // again (or runThread() self-exited and the user called stop() before
    // ~AxiStreamDma()): clearing desc_ after closeShared() prevents a
    // double-decrement of openCount.
    if (desc_) {
        closeShared(desc_);
        desc_.reset();
    }
 
    // Close the per-instance fd exactly once.
    if (fd_ >= 0) {
        ::close(fd_);
        fd_ = -1;
    }
}
 
void rha::AxiStreamDma::setTimeout(uint32_t timeout) {
    if (timeout > 0) {
        div_t divResult  = div(timeout, 1000000);
        timeout_.tv_sec  = divResult.quot;
        timeout_.tv_usec = divResult.rem;
    }
}
 
void rha::AxiStreamDma::setDriverDebug(uint32_t level) {
    dmaSetDebug(fd_, level);
}
 
void rha::AxiStreamDma::dmaAck() {
    if (fd_ >= 0) axisReadAck(fd_);
}
 
ris::FramePtr rha::AxiStreamDma::acceptReq(uint32_t size, bool zeroCopyEn) {
    int32_t res;
    struct pollfd pfd;
    uint32_t alloc;
    ris::BufferPtr buff;
    ris::FramePtr frame;
    uint32_t buffSize;
 
    // Reject use after stop()/teardown.  stop() resets the shared-descriptor
    // shared_ptr to nullptr and closes the per-instance fd, so a post-stop
    // call would otherwise segfault on the buffer-size read below instead
    // of throwing a clean GeneralError.
    if (!desc_ || fd_ < 0)
        throw rogue::GeneralError("AxiStreamDma::acceptReq",
                                  "instance has been stopped or did not finish construction");
 
    if (size > desc_->bSize)
        buffSize = desc_->bSize;
    else
        buffSize = size;
 
    // Zero copy is disabled. Allocate from memory.
    if (zeroCopyEn == false || desc_->rawBuff == NULL) {
        frame = reqLocalFrame(size, false);
 
        // Allocate zero copy buffers from driver
    } else {
        rogue::GilRelease noGil;
 
        // Create empty frame
        frame = ris::Frame::create();
        alloc = 0;
 
        // Request may be serviced with multiple buffers
        while (alloc < size) {
            // Keep trying since poll call can fire
            // but getIndex fails because we did not win the buffer lock
            do {
                // Re-check fd validity: a concurrent stop() can toggle fd_ to -1.
                // poll() imposes no FD_SETSIZE ceiling, so large fd values are fine.
                if (fd_ < 0)
                    throw rogue::GeneralError::create(
                        "AxiStreamDma::acceptReq",
                        "fd_=%d invalid; instance was stop()'d or never finished construction",
                        fd_);
                pfd.fd      = fd_;
                pfd.events  = POLLOUT;
                pfd.revents = 0;
 
                // Round up µs->ms so sub-ms timeouts do not collapse to a non-blocking poll().
                int rc = poll(&pfd, 1, (timeout_.tv_sec * 1000) + ((timeout_.tv_usec + 999) / 1000));
                // POLLERR/POLLHUP/POLLNVAL can be reported with POLLOUT also set on a fd in
                // error state; surface that as a clear error instead of feeding a known-bad
                // fd into dmaGetIndex() and looping on its return value.
                if (rc > 0 && (pfd.revents & (POLLERR | POLLHUP | POLLNVAL)))
                    throw rogue::GeneralError::create(
                        "AxiStreamDma::acceptReq",
                        "poll() reported fd error revents=0x%x; fd may be closed or in error state",
                        pfd.revents);
                if (rc <= 0 || !(pfd.revents & POLLOUT)) {
                    log_->critical("AxiStreamDma::acceptReq: Timeout waiting for outbound buffer after %" PRIuLEAST32
                                   ".%" PRIuLEAST32 " seconds! May be caused by outbound back pressure.",
                                   timeout_.tv_sec,
                                   timeout_.tv_usec);
                    res = -1;
                } else {
                    // Attempt to get index.
                    // return of less than 0 is a failure to get a buffer
                    res = dmaGetIndex(fd_);
                }
            } while (res < 0);
 
            // Mark zero copy meta with bit 31 set, lower bits are index
            buff = createBuffer(desc_->rawBuff[res], 0x80000000 | res, buffSize, desc_->bSize);
            frame->appendBuffer(buff);
            alloc += buffSize;
        }
    }
    return (frame);
}
 
void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) {
    int32_t res;
    struct pollfd pfd;
    uint32_t meta;
    uint32_t fuser;
    uint32_t luser;
    uint32_t cont;
    bool emptyFrame;
 
    // Reject use after stop()/teardown.  stop() releases the shared descriptor
    // and closes fd_, so the inner-loop fd_ guard would still catch this, but
    // that fires only after frame->lock() and buffer iteration have already
    // run.  Fail-fast at entry for a cleaner error path, mirroring acceptReq().
    if (!desc_ || fd_ < 0)
        throw rogue::GeneralError("AxiStreamDma::acceptFrame",
                                  "instance has been stopped or did not finish construction");
 
    rogue::GilRelease noGil;
    ris::FrameLockPtr lock = frame->lock();
    emptyFrame             = false;
 
    // Drop errored frames
    if (frame->getError()) {
        log_->warning("Dumping errored frame");
        return;
    }
 
    // Go through each (*it)er in the frame
    ris::Frame::BufferIterator it;
    for (it = frame->beginBuffer(); it != frame->endBuffer(); ++it) {
        (*it)->zeroHeader();
 
        // First buffer
        if (it == frame->beginBuffer()) {
            fuser = frame->getFirstUser();
            if (enSsi_) fuser |= 0x2;
        } else {
            fuser = 0;
        }
 
        // Last Buffer
        if (it == (frame->endBuffer() - 1)) {
            cont  = 0;
            luser = frame->getLastUser();
 
            // Continue flag is set if this is not the last (*it)er
        } else {
            cont  = 1;
            luser = 0;
        }
 
        // Get (*it)er meta field
        meta = (*it)->getMeta();
 
        // Meta is zero copy as indicated by bit 31
        if ((meta & 0x80000000) != 0) {
            emptyFrame = true;
 
            // Buffer is not already stale as indicates by bit 30
            if ((meta & 0x40000000) == 0) {
                // Write by passing (*it)er index to driver
                if (dmaWriteIndex(fd_,
                                  meta & 0x3FFFFFFF,
                                  (*it)->getPayload(),
                                  axisSetFlags(fuser, luser, cont),
                                  dest_) <= 0) {
                    throw(rogue::GeneralError("AxiStreamDma::acceptFrame", "AXIS Write Call Failed"));
                }
 
                // Mark (*it)er as stale
                meta |= 0x40000000;
                (*it)->setMeta(meta);
            }
 
            // Write to pgp with (*it)er copy in driver
        } else {
            // Keep trying since poll call can fire
            // but write fails because we did not win the (*it)er lock
            do {
                // Re-check fd validity: a concurrent stop() can toggle fd_ to -1.
                // poll() imposes no FD_SETSIZE ceiling, so large fd values are fine.
                if (fd_ < 0)
                    throw rogue::GeneralError::create(
                        "AxiStreamDma::acceptFrame",
                        "fd_=%d invalid; instance was stop()'d or never finished construction",
                        fd_);
                pfd.fd      = fd_;
                pfd.events  = POLLOUT;
                pfd.revents = 0;
 
                // Round up µs->ms so sub-ms timeouts do not collapse to a non-blocking poll().
                int rc = poll(&pfd, 1, (timeout_.tv_sec * 1000) + ((timeout_.tv_usec + 999) / 1000));
                // POLLERR/POLLHUP/POLLNVAL can be reported with POLLOUT also set on a fd in
                // error state; surface that as a clear error instead of feeding a known-bad
                // fd into dmaWrite() and reporting a generic "AXIS Write Call Failed".
                if (rc > 0 && (pfd.revents & (POLLERR | POLLHUP | POLLNVAL)))
                    throw rogue::GeneralError::create(
                        "AxiStreamDma::acceptFrame",
                        "poll() reported fd error revents=0x%x; fd may be closed or in error state",
                        pfd.revents);
                if (rc <= 0 || !(pfd.revents & POLLOUT)) {
                    log_->critical("AxiStreamDma::acceptFrame: Timeout waiting for outbound write after %" PRIuLEAST32
                                   ".%" PRIuLEAST32 " seconds! May be caused by outbound back pressure.",
                                   timeout_.tv_sec,
                                   timeout_.tv_usec);
                    res = 0;
                } else {
                    // Write with (*it)er copy
                    if ((res =
                             dmaWrite(fd_, (*it)->begin(), (*it)->getPayload(), axisSetFlags(fuser, luser, 0), dest_)) <
                        0) {
                        throw(rogue::GeneralError("AxiStreamDma::acceptFrame", "AXIS Write Call Failed!!!!"));
                    }
                }
            } while (res == 0);  // Exit out if return flag was set false
        }
    }
 
    if (emptyFrame) frame->clear();
}
 
void rha::AxiStreamDma::retBuffer(uint8_t* data, uint32_t meta, uint32_t size) {
    rogue::GilRelease noGil;
    uint32_t ret[100];
    uint32_t count;
    uint32_t x;
 
    // Buffer is zero copy as indicated by bit 31
    if ((meta & 0x80000000) != 0) {
        // Device is open and buffer is not stale
        // Bit 30 indicates buffer has already been returned to hardware
        if ((fd_ >= 0) && ((meta & 0x40000000) == 0)) {
            if (dmaRetIndex(fd_, meta & 0x3FFFFFFF) < 0)
                throw(rogue::GeneralError("AxiStreamDma::retBuffer", "AXIS Return Buffer Call Failed!!!!"));
        }
        decCounter(size);
 
        // Buffer is allocated from Pool class
    } else {
        Pool::retBuffer(data, meta, size);
    }
}
 
void rha::AxiStreamDma::runThread(std::weak_ptr<int> lockPtr) {
    ris::BufferPtr buff[RxBufferCount];
    uint32_t meta[RxBufferCount];
    uint32_t rxFlags[RxBufferCount];
    uint32_t rxError[RxBufferCount];
    int32_t rxSize[RxBufferCount];
    int32_t rxCount;
    int32_t x;
    ris::FramePtr frame;
    struct pollfd pfd;
    uint8_t error;
    uint32_t fuser;
    uint32_t luser;
    uint32_t cont;
 
    fuser = 0;
    luser = 0;
    cont  = 0;
 
    // Wait until constructor completes
    while (!lockPtr.expired()) continue;
 
    log_->logThreadId();
 
    // Preallocate empty frame
    frame = ris::Frame::create();
 
    while (threadEn_) {
        // runThread() has no top-level catch, so a throw here would call
        // std::terminate.  This in-loop check only fires if stop()/close()
        // races with the worker and toggles fd_ to -1; log and exit the worker
        // cleanly so stop() can complete instead of crashing the process.
        // poll() imposes no FD_SETSIZE ceiling, so large fd values are fine.
        if (fd_ < 0) {
            log_->error("AxiStreamDma::runThread: fd_ value %d invalid; exiting worker", fd_);
            threadEn_ = false;
            break;
        }
        pfd.fd      = fd_;
        pfd.events  = POLLIN;
        pfd.revents = 0;
 
        // poll returns with available buffer (1 ms timeout)
        if (poll(&pfd, 1, 1) > 0) {
            // poll() can report POLLERR/POLLHUP/POLLNVAL with POLLIN unset (e.g. closed-fd race).
            // runThread() has no top-level catch, so falling through to dmaRead on an invalid fd
            // would throw and call std::terminate.  Log and exit the worker cleanly instead.
            if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
                log_->error("AxiStreamDma::runThread: poll reported fd error revents=0x%x; exiting worker",
                            pfd.revents);
                threadEn_ = false;
                break;
            }
            if (!(pfd.revents & POLLIN)) continue;
            // Zero copy buffers were not allocated
            if (desc_->rawBuff == NULL) {
                // Allocate a buffer
                buff[0] = allocBuffer(desc_->bSize, NULL);
 
                // Attempt read, dest is not needed since only one lane/vc is open
                rxSize[0] = dmaRead(fd_, buff[0]->begin(), buff[0]->getAvailable(), rxFlags, rxError, NULL);
                if (rxSize[0] <= 0)
                    rxCount = rxSize[0];
                else
                    rxCount = 1;
 
                // Zero copy read
            } else {
                // Attempt read, dest is not needed since only one lane/vc is open
                rxCount = dmaReadBulkIndex(fd_, RxBufferCount, rxSize, meta, rxFlags, rxError, NULL);
 
                // Allocate a buffer, Mark zero copy meta with bit 31 set, lower bits are index
                for (x = 0; x < rxCount; x++)
                    buff[x] = createBuffer(desc_->rawBuff[meta[x]], 0x80000000 | meta[x], desc_->bSize, desc_->bSize);
            }
 
            // Return of -1 is bad
            if (rxCount < 0) throw(rogue::GeneralError("AxiStreamDma::runThread", "DMA Interface Failure!"));
 
            // Read was successful
            for (x = 0; x < rxCount; x++) {
                fuser = axisGetFuser(rxFlags[x]);
                luser = axisGetLuser(rxFlags[x]);
                cont  = axisGetCont(rxFlags[x]);
 
                buff[x]->setPayload(rxSize[x]);
 
                error = frame->getError();
 
                // Receive error
                error |= (rxError[x] & 0xFF);
 
                // First buffer of frame
                if (frame->isEmpty()) frame->setFirstUser(fuser & 0xFF);
 
                // Last buffer of frame
                if (cont == 0) {
                    frame->setLastUser(luser & 0xFF);
                    if (enSsi_ && ((luser & 0x1) != 0)) error |= 0x80;
                }
 
                frame->setError(error);
                frame->appendBuffer(buff[x]);
                buff[x].reset();
 
                // If continue flag is not set, push frame and get a new empty frame
                if (cont == 0) {
                    sendFrame(frame);
                    frame = ris::Frame::create();
                }
            }
        }
    }
}
 
std::string rha::AxiStreamDma::getGitVersion() {
    return dmaGetGitVersion(fd_);
}
 
uint32_t rha::AxiStreamDma::getApiVersion() {
    return dmaGetApiVersion(fd_);
}
 
uint32_t rha::AxiStreamDma::getBuffSize() {
    return dmaGetBuffSize(fd_);
}
 
uint32_t rha::AxiStreamDma::getRxBuffCount() {
    return dmaGetRxBuffCount(fd_);
}
 
uint32_t rha::AxiStreamDma::getRxBuffinUserCount() {
    return dmaGetRxBuffinUserCount(fd_);
}
 
uint32_t rha::AxiStreamDma::getRxBuffinHwCount() {
    return dmaGetRxBuffinHwCount(fd_);
}
 
uint32_t rha::AxiStreamDma::getRxBuffinPreHwQCount() {
    return dmaGetRxBuffinPreHwQCount(fd_);
}
 
uint32_t rha::AxiStreamDma::getRxBuffinSwQCount() {
    return dmaGetRxBuffinSwQCount(fd_);
}
 
uint32_t rha::AxiStreamDma::getRxBuffMissCount() {
    return dmaGetRxBuffMissCount(fd_);
}
 
uint32_t rha::AxiStreamDma::getTxBuffCount() {
    return dmaGetTxBuffCount(fd_);
}
 
uint32_t rha::AxiStreamDma::getTxBuffinUserCount() {
    return dmaGetTxBuffinUserCount(fd_);
}
 
uint32_t rha::AxiStreamDma::getTxBuffinHwCount() {
    return dmaGetTxBuffinHwCount(fd_);
}
 
uint32_t rha::AxiStreamDma::getTxBuffinPreHwQCount() {
    return dmaGetTxBuffinPreHwQCount(fd_);
}
 
uint32_t rha::AxiStreamDma::getTxBuffinSwQCount() {
    return dmaGetTxBuffinSwQCount(fd_);
}
 
uint32_t rha::AxiStreamDma::getTxBuffMissCount() {
    return dmaGetTxBuffMissCount(fd_);
}
 
void rha::AxiStreamDma::setup_python() {
#ifndef NO_PYTHON
 
    bp::class_<rha::AxiStreamDma, rha::AxiStreamDmaPtr, bp::bases<ris::Master, ris::Slave>, boost::noncopyable>(
        "AxiStreamDma",
        bp::init<std::string, uint32_t, bool>())
        .def("setDriverDebug", &rha::AxiStreamDma::setDriverDebug)
        .def("dmaAck", &rha::AxiStreamDma::dmaAck)
        .def("setTimeout", &rha::AxiStreamDma::setTimeout)
        .def("getGitVersion", &rha::AxiStreamDma::getGitVersion)
        .def("getApiVersion", &rha::AxiStreamDma::getApiVersion)
        .def("getBuffSize", &rha::AxiStreamDma::getBuffSize)
        .def("getRxBuffCount", &rha::AxiStreamDma::getRxBuffCount)
        .def("getRxBuffinUserCount", &rha::AxiStreamDma::getRxBuffinUserCount)
        .def("getRxBuffinHwCount", &rha::AxiStreamDma::getRxBuffinHwCount)
        .def("getRxBuffinPreHwQCount", &rha::AxiStreamDma::getRxBuffinPreHwQCount)
        .def("getRxBuffinSwQCount", &rha::AxiStreamDma::getRxBuffinSwQCount)
        .def("getRxBuffMissCount", &rha::AxiStreamDma::getRxBuffMissCount)
        .def("getTxBuffCount", &rha::AxiStreamDma::getTxBuffCount)
        .def("getTxBuffinUserCount", &rha::AxiStreamDma::getTxBuffinUserCount)
        .def("getTxBuffinHwCount", &rha::AxiStreamDma::getTxBuffinHwCount)
        .def("getTxBuffinPreHwQCount", &rha::AxiStreamDma::getTxBuffinPreHwQCount)
        .def("getTxBuffinSwQCount", &rha::AxiStreamDma::getTxBuffinSwQCount)
        .def("getTxBuffMissCount", &rha::AxiStreamDma::getTxBuffMissCount)
        .def("zeroCopyDisable", &rha::AxiStreamDma::zeroCopyDisable);
 
    bp::implicitly_convertible<rha::AxiStreamDmaPtr, ris::MasterPtr>();
    bp::implicitly_convertible<rha::AxiStreamDmaPtr, ris::SlavePtr>();
#endif
}