From a8c38d0f40b5a6f2625b00c5a65482efcb5fbd8f Mon Sep 17 00:00:00 2001 From: Benjamin Reese Date: Mon, 6 Jul 2026 11:44:44 -0700 Subject: [PATCH 1/9] Add mutex for thread-safe fd_ management and update stop() documentation --- include/rogue/hardware/axi/AxiStreamDma.h | 13 +++++- src/rogue/hardware/axi/AxiStreamDma.cpp | 49 +++++++++++------------ 2 files changed, 35 insertions(+), 27 deletions(-) diff --git a/include/rogue/hardware/axi/AxiStreamDma.h b/include/rogue/hardware/axi/AxiStreamDma.h index 6d3ce71a71..86a7bd58cc 100644 --- a/include/rogue/hardware/axi/AxiStreamDma.h +++ b/include/rogue/hardware/axi/AxiStreamDma.h @@ -23,6 +23,7 @@ #include #include #include +#include #include #include @@ -104,6 +105,9 @@ class AxiStreamDma : public rogue::interfaces::stream::Master, public rogue::int // Process-local descriptor for TX/RX operations and dest mask programming. int32_t fd_; + // Serializes fd_ close against deferred zero-copy buffer returns. + std::mutex fdMtx_; + // Destination selector used when transmitting frames. uint32_t dest_; @@ -204,7 +208,14 @@ class AxiStreamDma : public rogue::interfaces::stream::Master, public rogue::int /** @brief Destroys the interface and stops background activity. */ ~AxiStreamDma(); - /** @brief Stops RX thread and closes DMA file descriptors. */ + /** + * @brief Stops RX thread and closes the per-instance DMA file descriptor. + * + * @details + * The shared zero-copy DMA mapping remains valid until destruction so + * downstream Rogue buffers retained after `stop()` do not reference + * unmapped memory. + */ void stop(); /** diff --git a/src/rogue/hardware/axi/AxiStreamDma.cpp b/src/rogue/hardware/axi/AxiStreamDma.cpp index 437c1f0f25..fe0e6592ff 100644 --- a/src/rogue/hardware/axi/AxiStreamDma.cpp +++ b/src/rogue/hardware/axi/AxiStreamDma.cpp @@ -294,6 +294,11 @@ rha::AxiStreamDma::AxiStreamDma(std::string path, uint32_t dest, bool ssiEnable) //! Close the device rha::AxiStreamDma::~AxiStreamDma() { this->stop(); + + if (desc_) { + closeShared(desc_); + desc_.reset(); + } } void rha::AxiStreamDma::stop() { @@ -314,19 +319,15 @@ void rha::AxiStreamDma::stop() { } 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; + // Close the per-instance fd exactly once. The shared DMA mapping remains + // open until ~AxiStreamDma() because zero-copy Rogue buffers can outlive a + // user-initiated stop() and still point into desc_->rawBuff. + { + std::lock_guard lock(fdMtx_); + if (fd_ >= 0) { + ::close(fd_); + fd_ = -1; + } } } @@ -358,10 +359,8 @@ ris::FramePtr rha::AxiStreamDma::acceptReq(uint32_t size, bool zeroCopyEn) { 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. + // Reject use after stop()/teardown. stop() closes the per-instance fd, + // and destruction releases the shared descriptor. if (!desc_ || fd_ < 0) throw rogue::GeneralError("AxiStreamDma::acceptReq", "instance has been stopped or did not finish construction"); @@ -442,10 +441,10 @@ void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) { 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(). + // Reject use after stop()/teardown. 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"); @@ -557,16 +556,14 @@ void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) { //! Return a buffer 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) + if ((meta & 0x40000000) == 0) { + std::lock_guard lock(fdMtx_); + if (fd_ >= 0 && dmaRetIndex(fd_, meta & 0x3FFFFFFF) < 0) throw(rogue::GeneralError("AxiStreamDma::retBuffer", "AXIS Return Buffer Call Failed!!!!")); } decCounter(size); From f50febc9b8d671549fea2c2d2451d8b8181bd616 Mon Sep 17 00:00:00 2001 From: Benjamin Reese Date: Mon, 6 Jul 2026 12:25:44 -0700 Subject: [PATCH 2/9] Enhance AxiStreamDma with FdGuard for better resource management and thread safety --- include/rogue/hardware/axi/AxiStreamDma.h | 30 ++- src/rogue/hardware/axi/AxiStreamDma.cpp | 274 ++++++++++++++-------- 2 files changed, 200 insertions(+), 104 deletions(-) diff --git a/include/rogue/hardware/axi/AxiStreamDma.h b/include/rogue/hardware/axi/AxiStreamDma.h index 86a7bd58cc..5af8b05122 100644 --- a/include/rogue/hardware/axi/AxiStreamDma.h +++ b/include/rogue/hardware/axi/AxiStreamDma.h @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -105,9 +106,18 @@ class AxiStreamDma : public rogue::interfaces::stream::Master, public rogue::int // Process-local descriptor for TX/RX operations and dest mask programming. int32_t fd_; - // Serializes fd_ close against deferred zero-copy buffer returns. + // Set once stop() starts; public stream operations reject new work after it. + std::atomic stopped_{false}; + + // Serializes fd_ close against active driver calls and deferred zero-copy returns. std::mutex fdMtx_; + // Number of public/API paths currently using fd_. + uint32_t fdUsers_ = 0; + + // Wakes stop() after the last active fd_ user exits. + std::condition_variable fdCv_; + // Destination selector used when transmitting frames. uint32_t dest_; @@ -143,6 +153,24 @@ class AxiStreamDma : public rogue::interfaces::stream::Master, public rogue::int // Closes shared DMA mapping state when last user exits. static void closeShared(std::shared_ptr); + // RAII guard that pins fd_ open while a driver call runs. + class FdGuard { + rogue::hardware::axi::AxiStreamDma* owner_; + int32_t fd_; + + public: + FdGuard(rogue::hardware::axi::AxiStreamDma* owner, + const char* context, + bool throwOnStopped, + bool allowStopped = false); + ~FdGuard(); + FdGuard(const FdGuard&) = delete; + FdGuard& operator=(const FdGuard&) = delete; + + int32_t fd() const { return fd_; } + bool valid() const { return fd_ >= 0; } + }; + public: /** * @brief Creates an AXI Stream DMA bridge instance. diff --git a/src/rogue/hardware/axi/AxiStreamDma.cpp b/src/rogue/hardware/axi/AxiStreamDma.cpp index fe0e6592ff..2ff7a827f1 100644 --- a/src/rogue/hardware/axi/AxiStreamDma.cpp +++ b/src/rogue/hardware/axi/AxiStreamDma.cpp @@ -49,6 +49,40 @@ std::map > rha::AxiStreamD // Protect sharedBuffers_ against concurrent AxiStreamDma construction. static std::mutex sharedBuffersMtx; +static void throwStopped(const char* context) { + throw rogue::GeneralError(context, "instance has been stopped or did not finish construction"); +} + +static uint32_t dmaValueOrZero(ssize_t value) { + return (value < 0) ? 0 : static_cast(value); +} + +rha::AxiStreamDma::FdGuard::FdGuard(rha::AxiStreamDma* owner, + const char* context, + bool throwOnStopped, + bool allowStopped) { + std::lock_guard lock(owner->fdMtx_); + + if ((!allowStopped && owner->stopped_.load()) || owner->fd_ < 0) { + owner_ = NULL; + fd_ = -1; + if (throwOnStopped) throwStopped(context); + return; + } + + owner_ = owner; + fd_ = owner->fd_; + owner->fdUsers_++; +} + +rha::AxiStreamDma::FdGuard::~FdGuard() { + if (owner_ != NULL) { + std::lock_guard lock(owner_->fdMtx_); + owner_->fdUsers_--; + if (owner_->fdUsers_ == 0) owner_->fdCv_.notify_all(); + } +} + rha::AxiStreamDmaShared::AxiStreamDmaShared(std::string path) { this->fd = -1; this->path = path; @@ -304,12 +338,14 @@ rha::AxiStreamDma::~AxiStreamDma() { void rha::AxiStreamDma::stop() { rogue::GilRelease noGil; + stopped_.store(true); + // 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 + // Join based on thread state, not threadEn_. runThread() can exit on its + // own (for example after a poll fd error) by setting threadEn_ = false and // returning, so gating cleanup on threadEn_ here would skip the join and // ~unique_ptr on a still-joinable thread would call // std::terminate(). Joining via joinable() handles both that early-exit @@ -323,7 +359,8 @@ void rha::AxiStreamDma::stop() { // open until ~AxiStreamDma() because zero-copy Rogue buffers can outlive a // user-initiated stop() and still point into desc_->rawBuff. { - std::lock_guard lock(fdMtx_); + std::unique_lock lock(fdMtx_); + fdCv_.wait(lock, [this] { return fdUsers_ == 0; }); if (fd_ >= 0) { ::close(fd_); fd_ = -1; @@ -342,12 +379,14 @@ void rha::AxiStreamDma::setTimeout(uint32_t timeout) { //! Set driver debug level void rha::AxiStreamDma::setDriverDebug(uint32_t level) { - dmaSetDebug(fd_, level); + FdGuard fd(this, "AxiStreamDma::setDriverDebug", false); + if (fd.valid()) dmaSetDebug(fd.fd(), level); } //! Strobe ack line void rha::AxiStreamDma::dmaAck() { - if (fd_ >= 0) axisReadAck(fd_); + FdGuard fd(this, "AxiStreamDma::dmaAck", false); + if (fd.valid()) axisReadAck(fd.fd()); } //! Generate a buffer. Called from master @@ -361,9 +400,7 @@ ris::FramePtr rha::AxiStreamDma::acceptReq(uint32_t size, bool zeroCopyEn) { // Reject use after stop()/teardown. stop() closes the per-instance fd, // and destruction releases the shared descriptor. - if (!desc_ || fd_ < 0) - throw rogue::GeneralError("AxiStreamDma::acceptReq", - "instance has been stopped or did not finish construction"); + if (!desc_ || stopped_.load()) throwStopped("AxiStreamDma::acceptReq"); //! Adjust allocation size if (size > desc_->bSize) @@ -388,37 +425,36 @@ ris::FramePtr rha::AxiStreamDma::acceptReq(uint32_t size, bool zeroCopyEn) { // 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_); + { + FdGuard fd(this, "AxiStreamDma::acceptReq", true); + + pfd.fd = 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)); + if (stopped_.load()) throwStopped("AxiStreamDma::acceptReq"); + + // 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.fd()); + } } } while (res < 0); @@ -441,13 +477,11 @@ void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) { uint32_t cont; bool emptyFrame; - // Reject use after stop()/teardown. The inner-loop fd_ guard would still + // Reject use after stop()/teardown. 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"); + if (!desc_ || stopped_.load()) throwStopped("AxiStreamDma::acceptFrame"); rogue::GilRelease noGil; ris::FrameLockPtr lock = frame->lock(); @@ -492,13 +526,17 @@ void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) { // 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")); + { + FdGuard fd(this, "AxiStreamDma::acceptFrame", true); + + // Write by passing (*it)er index to driver + if (dmaWriteIndex(fd.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 @@ -511,39 +549,40 @@ void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) { // 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!!!!")); + { + FdGuard fd(this, "AxiStreamDma::acceptFrame", true); + + pfd.fd = 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)); + if (stopped_.load()) throwStopped("AxiStreamDma::acceptFrame"); + + // 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.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 @@ -562,8 +601,8 @@ void rha::AxiStreamDma::retBuffer(uint8_t* data, uint32_t meta, uint32_t size) { // Device is open and buffer is not stale // Bit 30 indicates buffer has already been returned to hardware if ((meta & 0x40000000) == 0) { - std::lock_guard lock(fdMtx_); - if (fd_ >= 0 && dmaRetIndex(fd_, meta & 0x3FFFFFFF) < 0) + FdGuard fd(this, "AxiStreamDma::retBuffer", false, true); + if (fd.valid() && dmaRetIndex(fd.fd(), meta & 0x3FFFFFFF) < 0) throw(rogue::GeneralError("AxiStreamDma::retBuffer", "AXIS Return Buffer Call Failed!!!!")); } decCounter(size); @@ -604,9 +643,8 @@ void rha::AxiStreamDma::runThread(std::weak_ptr lockPtr) { 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. + // std::terminate. Log and exit cleanly if the descriptor was never + // opened or has already been closed by an earlier stop() call. // 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_); @@ -692,77 +730,107 @@ void rha::AxiStreamDma::runThread(std::weak_ptr lockPtr) { //! Get the DMA Driver's Git Version std::string rha::AxiStreamDma::getGitVersion() { - return dmaGetGitVersion(fd_); + FdGuard fd(this, "AxiStreamDma::getGitVersion", false); + if (!fd.valid()) return ""; + return dmaGetGitVersion(fd.fd()); } //! Get the DMA Driver's API Version uint32_t rha::AxiStreamDma::getApiVersion() { - return dmaGetApiVersion(fd_); + FdGuard fd(this, "AxiStreamDma::getApiVersion", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetApiVersion(fd.fd())); } //! Get the size of buffers (RX/TX) uint32_t rha::AxiStreamDma::getBuffSize() { - return dmaGetBuffSize(fd_); + FdGuard fd(this, "AxiStreamDma::getBuffSize", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetBuffSize(fd.fd())); } //! Get the number of RX buffers uint32_t rha::AxiStreamDma::getRxBuffCount() { - return dmaGetRxBuffCount(fd_); + FdGuard fd(this, "AxiStreamDma::getRxBuffCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetRxBuffCount(fd.fd())); } //! Get RX buffer in User count uint32_t rha::AxiStreamDma::getRxBuffinUserCount() { - return dmaGetRxBuffinUserCount(fd_); + FdGuard fd(this, "AxiStreamDma::getRxBuffinUserCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetRxBuffinUserCount(fd.fd())); } //! Get RX buffer in HW count uint32_t rha::AxiStreamDma::getRxBuffinHwCount() { - return dmaGetRxBuffinHwCount(fd_); + FdGuard fd(this, "AxiStreamDma::getRxBuffinHwCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetRxBuffinHwCount(fd.fd())); } //! Get RX buffer in Pre-HW Queue count uint32_t rha::AxiStreamDma::getRxBuffinPreHwQCount() { - return dmaGetRxBuffinPreHwQCount(fd_); + FdGuard fd(this, "AxiStreamDma::getRxBuffinPreHwQCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetRxBuffinPreHwQCount(fd.fd())); } //! Get RX buffer in SW Queue count uint32_t rha::AxiStreamDma::getRxBuffinSwQCount() { - return dmaGetRxBuffinSwQCount(fd_); + FdGuard fd(this, "AxiStreamDma::getRxBuffinSwQCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetRxBuffinSwQCount(fd.fd())); } //! Get RX buffer missing count uint32_t rha::AxiStreamDma::getRxBuffMissCount() { - return dmaGetRxBuffMissCount(fd_); + FdGuard fd(this, "AxiStreamDma::getRxBuffMissCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetRxBuffMissCount(fd.fd())); } //! Get the number of TX buffers uint32_t rha::AxiStreamDma::getTxBuffCount() { - return dmaGetTxBuffCount(fd_); + FdGuard fd(this, "AxiStreamDma::getTxBuffCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetTxBuffCount(fd.fd())); } //! Get TX buffer in User count uint32_t rha::AxiStreamDma::getTxBuffinUserCount() { - return dmaGetTxBuffinUserCount(fd_); + FdGuard fd(this, "AxiStreamDma::getTxBuffinUserCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetTxBuffinUserCount(fd.fd())); } //! Get TX buffer in HW count uint32_t rha::AxiStreamDma::getTxBuffinHwCount() { - return dmaGetTxBuffinHwCount(fd_); + FdGuard fd(this, "AxiStreamDma::getTxBuffinHwCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetTxBuffinHwCount(fd.fd())); } //! Get TX buffer in Pre-HW Queue count uint32_t rha::AxiStreamDma::getTxBuffinPreHwQCount() { - return dmaGetTxBuffinPreHwQCount(fd_); + FdGuard fd(this, "AxiStreamDma::getTxBuffinPreHwQCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetTxBuffinPreHwQCount(fd.fd())); } //! Get TX buffer in SW Queue count uint32_t rha::AxiStreamDma::getTxBuffinSwQCount() { - return dmaGetTxBuffinSwQCount(fd_); + FdGuard fd(this, "AxiStreamDma::getTxBuffinSwQCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetTxBuffinSwQCount(fd.fd())); } //! Get TX buffer missing count uint32_t rha::AxiStreamDma::getTxBuffMissCount() { - return dmaGetTxBuffMissCount(fd_); + FdGuard fd(this, "AxiStreamDma::getTxBuffMissCount", false); + if (!fd.valid()) return 0; + return dmaValueOrZero(dmaGetTxBuffMissCount(fd.fd())); } void rha::AxiStreamDma::setup_python() { From ba12136a8973fb863069d9afb2746ecd2f0f343c Mon Sep 17 00:00:00 2001 From: Benjamin Reese Date: Mon, 6 Jul 2026 12:49:41 -0700 Subject: [PATCH 3/9] Add fake AxiStreamDma driver and lifecycle tests for stop/fd races --- tests/cpp/CMakeLists.txt | 1 + tests/cpp/hardware/CMakeLists.txt | 44 ++ tests/cpp/hardware/fake_axi_dma_driver.cpp | 599 ++++++++++++++++++ tests/cpp/hardware/fake_axi_dma_driver.h | 44 ++ .../test_axi_stream_dma_lifecycle.cpp | 257 ++++++++ 5 files changed, 945 insertions(+) create mode 100644 tests/cpp/hardware/CMakeLists.txt create mode 100644 tests/cpp/hardware/fake_axi_dma_driver.cpp create mode 100644 tests/cpp/hardware/fake_axi_dma_driver.h create mode 100644 tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp diff --git a/tests/cpp/CMakeLists.txt b/tests/cpp/CMakeLists.txt index 8aaca17179..2c7c08bd32 100644 --- a/tests/cpp/CMakeLists.txt +++ b/tests/cpp/CMakeLists.txt @@ -53,6 +53,7 @@ function(rogue_add_cpp_test target) endfunction() add_subdirectory(core) +add_subdirectory(hardware) add_subdirectory(memory) add_subdirectory(stream) add_subdirectory(protocols) diff --git a/tests/cpp/hardware/CMakeLists.txt b/tests/cpp/hardware/CMakeLists.txt new file mode 100644 index 0000000000..e2ad152509 --- /dev/null +++ b/tests/cpp/hardware/CMakeLists.txt @@ -0,0 +1,44 @@ +add_library(rogue-cpp-axi-stream-dma-fake-driver SHARED + fake_axi_dma_driver.cpp +) + +target_include_directories(rogue-cpp-axi-stream-dma-fake-driver + PRIVATE + ${CMAKE_SOURCE_DIR}/include + ${CMAKE_CURRENT_LIST_DIR} +) + +target_link_libraries(rogue-cpp-axi-stream-dma-fake-driver + PRIVATE + ${CMAKE_DL_LIBS} +) + +rogue_add_cpp_test(rogue-cpp-axi-stream-dma-lifecycle + SOURCES + test_axi_stream_dma_lifecycle.cpp + LABELS + cpp-core + no-python +) + +add_dependencies(rogue-cpp-axi-stream-dma-lifecycle rogue-cpp-axi-stream-dma-fake-driver) + +set(_rogue_fake_axi_dma_env + "MPLCONFIGDIR=${CMAKE_BINARY_DIR}/tests/cpp/mplconfig" + "ROGUE_FAKE_AXI_DMA_DRIVER_LIB=$" +) + +if (APPLE) + set_tests_properties(rogue-cpp-axi-stream-dma-lifecycle PROPERTIES + DISABLED TRUE + ) +else() + list(APPEND _rogue_fake_axi_dma_env + "LD_PRELOAD=$" + ) +endif() + +set_tests_properties(rogue-cpp-axi-stream-dma-lifecycle PROPERTIES + ENVIRONMENT "${_rogue_fake_axi_dma_env}" + TIMEOUT 10 +) diff --git a/tests/cpp/hardware/fake_axi_dma_driver.cpp b/tests/cpp/hardware/fake_axi_dma_driver.cpp new file mode 100644 index 0000000000..0dc9090690 --- /dev/null +++ b/tests/cpp/hardware/fake_axi_dma_driver.cpp @@ -0,0 +1,599 @@ +/** + * ---------------------------------------------------------------------------- + * Company : SLAC National Accelerator Laboratory + * ---------------------------------------------------------------------------- + * Description: + * Test-only fake AxiStreamDma driver syscall interposer. + * ---------------------------------------------------------------------------- + * This file is part of the rogue software platform. It is subject to + * the license terms in the LICENSE.txt file found in the top-level directory + * of this distribution and at: + * https://confluence.slac.stanford.edu/display/ppareg/LICENSE.html. + * No part of the rogue software platform, including this file, may be + * copied, modified, propagated, or distributed except according to the terms + * contained in the LICENSE.txt file. + * ---------------------------------------------------------------------------- + **/ +#include "fake_axi_dma_driver.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "rogue/hardware/drivers/DmaDriver.h" + +#ifndef MAP_ANON + #define MAP_ANON MAP_ANONYMOUS +#endif + +namespace { + +constexpr const char* FakePathPrefix = "/tmp/rogue-cpp-fake-axi-dma"; +constexpr const char* FakeBackingPrefix = "/tmp/rogue-cpp-fake-axi-dma-backing"; +constexpr uint32_t FakeBuffCount = 4; +constexpr uint32_t FakeBuffSize = 4096; + +struct FakeState { + std::mutex mtx; + std::condition_variable cv; + std::set fakeFds; + std::map fakeMappings; + RogueFakeAxiDmaBlockOp blockNext = ROGUE_FAKE_AXI_DMA_BLOCK_NONE; + bool blocked = false; + bool releaseBlocked = false; + uint32_t activeDriverCalls = 0; + uint32_t closeCount = 0; + uint32_t munmapCount = 0; + uint32_t retIndexCount = 0; + uint32_t nextIndex = 0; + bool closeDuringDriverCall = false; +}; + +FakeState& state() { + static FakeState* fakeState = new FakeState; + return *fakeState; +} + +template +Func realSymbol(const char* name) { + void* sym = dlsym(RTLD_NEXT, name); + return reinterpret_cast(sym); +} + +using OpenFn = int (*)(const char*, int, ...); +using CloseFn = int (*)(int); +using IoctlFn = int (*)(int, unsigned long, ...); +using MmapFn = void* (*)(void*, size_t, int, int, int, off_t); +using MunmapFn = int (*)(void*, size_t); +using PollFn = int (*)(struct pollfd*, nfds_t, int); +using ReadFn = ssize_t (*)(int, void*, size_t); +using WriteFn = ssize_t (*)(int, const void*, size_t); + +OpenFn realOpen() { + static OpenFn fn = realSymbol("open"); + return fn; +} + +#ifndef __APPLE__ +OpenFn realOpen64() { + static OpenFn fn = realSymbol("open64"); + return fn; +} +#endif + +CloseFn realClose() { + static CloseFn fn = realSymbol("close"); + return fn; +} + +IoctlFn realIoctl() { + static IoctlFn fn = realSymbol("ioctl"); + return fn; +} + +MmapFn realMmap() { + static MmapFn fn = realSymbol("mmap"); + return fn; +} + +MunmapFn realMunmap() { + static MunmapFn fn = realSymbol("munmap"); + return fn; +} + +PollFn realPoll() { + static PollFn fn = realSymbol("poll"); + return fn; +} + +ReadFn realRead() { + static ReadFn fn = realSymbol("read"); + return fn; +} + +WriteFn realWrite() { + static WriteFn fn = realSymbol("write"); + return fn; +} + +int callRealOpen(const char* path, int flags, mode_t mode, bool hasMode) { +#ifdef __APPLE__ + return static_cast(syscall(SYS_open, path, flags, hasMode ? mode : 0)); +#else + if (hasMode) return realOpen()(path, flags, mode); + return realOpen()(path, flags); +#endif +} + +int callRealClose(int fd) { +#ifdef __APPLE__ + return static_cast(syscall(SYS_close, fd)); +#else + return realClose()(fd); +#endif +} + +void* callRealMmap(void* addr, size_t length, int prot, int flags, int fd, off_t offset) { +#ifdef __APPLE__ + return reinterpret_cast(syscall(SYS_mmap, addr, length, prot, flags, fd, offset)); +#else + return realMmap()(addr, length, prot, flags, fd, offset); +#endif +} + +int callRealMunmap(void* addr, size_t length) { +#ifdef __APPLE__ + return static_cast(syscall(SYS_munmap, addr, length)); +#else + return realMunmap()(addr, length); +#endif +} + +int callRealPoll(struct pollfd* fds, nfds_t nfds, int timeout) { +#ifdef __APPLE__ + return static_cast(syscall(SYS_poll, fds, nfds, timeout)); +#else + return realPoll()(fds, nfds, timeout); +#endif +} + +ssize_t callRealRead(int fd, void* buf, size_t count) { +#ifdef __APPLE__ + return static_cast(syscall(SYS_read, fd, buf, count)); +#else + return realRead()(fd, buf, count); +#endif +} + +ssize_t callRealWrite(int fd, const void* buf, size_t count) { +#ifdef __APPLE__ + return static_cast(syscall(SYS_write, fd, buf, count)); +#else + return realWrite()(fd, buf, count); +#endif +} + +bool isFakePath(const char* path) { + return (path != nullptr) && (strncmp(path, FakePathPrefix, strlen(FakePathPrefix)) == 0); +} + +bool isFakeFdLocked(int fd) { + return state().fakeFds.find(fd) != state().fakeFds.end(); +} + +bool isFakeFd(int fd) { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + return isFakeFdLocked(fd); +} + +void maybeBlock(RogueFakeAxiDmaBlockOp op) { + FakeState& st = state(); + std::unique_lock lock(st.mtx); + if (st.blockNext != op) return; + + st.blockNext = ROGUE_FAKE_AXI_DMA_BLOCK_NONE; + st.blocked = true; + st.releaseBlocked = false; + st.cv.notify_all(); + st.cv.wait(lock, [&st] { return st.releaseBlocked; }); + st.blocked = false; + st.releaseBlocked = false; + st.cv.notify_all(); +} + +class DriverCall { + public: + explicit DriverCall(bool enabled) : enabled_(enabled) { + if (enabled_) { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + st.activeDriverCalls++; + } + } + + ~DriverCall() { + if (enabled_) { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + st.activeDriverCalls--; + st.cv.notify_all(); + } + } + + private: + bool enabled_; +}; + +int fakeOpenImpl(const char* path, int flags, mode_t mode, bool hasMode) { + if (!isFakePath(path)) { + return callRealOpen(path, flags, mode, hasMode); + } + + char backingPath[256]; + snprintf(backingPath, sizeof(backingPath), "%s-%d-%ld", FakeBackingPrefix, getpid(), random()); + int fd = callRealOpen(backingPath, O_RDWR | O_CREAT | O_TRUNC, 0600, true); + if (fd >= 0) { + static_cast(unlink(backingPath)); + static_cast(ftruncate(fd, FakeBuffCount * FakeBuffSize)); + FakeState& st = state(); + std::lock_guard lock(st.mtx); + st.fakeFds.insert(fd); + } + return fd; +} + +int fakeIoctlImpl(int fd, unsigned long request, va_list args) { + if (!isFakeFd(fd)) { + void* arg = va_arg(args, void*); +#ifdef __APPLE__ + return static_cast(syscall(SYS_ioctl, fd, request, arg)); +#else + return realIoctl()(fd, request, arg); +#endif + } + + DriverCall call(true); + + switch (request) { + case DMA_Get_Version: + return DMA_VERSION; + case DMA_Get_Buff_Count: + return FakeBuffCount; + case DMA_Get_Buff_Size: + maybeBlock(ROGUE_FAKE_AXI_DMA_BLOCK_IOCTL_BUFF_SIZE); + return FakeBuffSize; + case DMA_Get_TxBuff_Count: + case DMA_Get_RxBuff_Count: + return 2; + case DMA_Get_TxBuffinUser_Count: + case DMA_Get_TxBuffinHW_Count: + case DMA_Get_TxBuffinPreHWQ_Count: + case DMA_Get_TxBuffinSWQ_Count: + case DMA_Get_TxBuffMiss_Count: + case DMA_Get_RxBuffinUser_Count: + case DMA_Get_RxBuffinHW_Count: + case DMA_Get_RxBuffinPreHWQ_Count: + case DMA_Get_RxBuffinSWQ_Count: + case DMA_Get_RxBuffMiss_Count: + return 0; + case DMA_Get_Index: { + maybeBlock(ROGUE_FAKE_AXI_DMA_BLOCK_IOCTL_GET_INDEX); + FakeState& st = state(); + std::lock_guard lock(st.mtx); + const uint32_t ret = st.nextIndex; + st.nextIndex = (st.nextIndex + 1) % FakeBuffCount; + return static_cast(ret); + } + case DMA_Ret_Index | 0x10000: + static_cast(va_arg(args, void*)); + maybeBlock(ROGUE_FAKE_AXI_DMA_BLOCK_IOCTL_RET_INDEX); + { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + st.retIndexCount++; + } + return 0; + case DMA_Set_MaskBytes: + static_cast(va_arg(args, void*)); + return 0; + case DMA_Get_GITV: { + char* gitv = va_arg(args, char*); + if (gitv != nullptr) strncpy(gitv, "fake-axi-dma", 31); + return 0; + } + case DMA_Set_Debug: + case DMA_Set_Mask: + return 0; + default: + return 0; + } +} + +void* fakeMmapImpl(void* addr, size_t length, int prot, int flags, int fd, off_t offset) { + if (!isFakeFd(fd)) return callRealMmap(addr, length, prot, flags, fd, offset); + + void* ptr = callRealMmap(nullptr, length, prot, MAP_PRIVATE | MAP_ANON, -1, 0); + if (ptr != MAP_FAILED) { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + st.fakeMappings[ptr] = length; + } + return ptr; +} + +int fakeMunmapImpl(void* addr, size_t length) { + { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + auto it = st.fakeMappings.find(addr); + if (it != st.fakeMappings.end()) { + length = it->second; + st.fakeMappings.erase(it); + st.munmapCount++; + } else if (length == FakeBuffSize) { + st.munmapCount++; + } + } + return callRealMunmap(addr, length); +} + +int fakeCloseImpl(int fd) { + bool fake = false; + { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + auto it = st.fakeFds.find(fd); + fake = (it != st.fakeFds.end()); + if (fake) { + st.fakeFds.erase(it); + st.closeCount++; + if (st.activeDriverCalls != 0) st.closeDuringDriverCall = true; + } + } + return callRealClose(fd); +} + +int fakePollImpl(struct pollfd* fds, nfds_t nfds, int timeout) { + bool allFake = true; + + for (nfds_t i = 0; i < nfds; ++i) { + if (!isFakeFd(fds[i].fd)) { + allFake = false; + break; + } + } + + if (!allFake) return callRealPoll(fds, nfds, timeout); + + int ready = 0; + for (nfds_t i = 0; i < nfds; ++i) { + fds[i].revents = 0; + if ((fds[i].events & POLLOUT) != 0) { + fds[i].revents = POLLOUT; + ready++; + } + } + + if (ready == 0) { + const int sleepMs = (timeout < 0) ? 1 : timeout; + if (sleepMs > 0) std::this_thread::sleep_for(std::chrono::milliseconds(sleepMs)); + } + return ready; +} + +ssize_t fakeReadImpl(int fd, void* buf, size_t count) { + if (!isFakeFd(fd)) return callRealRead(fd, buf, count); + DriverCall call(true); + return 0; +} + +ssize_t fakeWriteImpl(int fd, const void* buf, size_t count) { + if (!isFakeFd(fd)) return callRealWrite(fd, buf, count); + DriverCall call(true); + maybeBlock(ROGUE_FAKE_AXI_DMA_BLOCK_WRITE); + return static_cast(count); +} + +#ifdef __APPLE__ + #define ROGUE_DYLD_INTERPOSE(replacement, replacee) \ + __attribute__((used)) static struct { \ + const void* replacementFunc; \ + const void* replaceeFunc; \ + } _rogue_interpose_##replacee __attribute__((section("__DATA,__interpose"))) = { \ + reinterpret_cast(replacement), reinterpret_cast(replacee)} +#endif + +} // namespace + +extern "C" const char* rogue_fake_axi_dma_path_prefix() { + return FakePathPrefix; +} + +extern "C" void rogue_fake_axi_dma_reset() { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + st.blockNext = ROGUE_FAKE_AXI_DMA_BLOCK_NONE; + st.blocked = false; + st.releaseBlocked = false; + st.activeDriverCalls = 0; + st.closeCount = 0; + st.munmapCount = 0; + st.retIndexCount = 0; + st.nextIndex = 0; + st.closeDuringDriverCall = false; +} + +extern "C" void rogue_fake_axi_dma_block_next(RogueFakeAxiDmaBlockOp op) { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + st.blockNext = op; + st.blocked = false; + st.releaseBlocked = false; +} + +extern "C" bool rogue_fake_axi_dma_wait_blocked(uint32_t timeoutMs) { + FakeState& st = state(); + std::unique_lock lock(st.mtx); + return st.cv.wait_for(lock, std::chrono::milliseconds(timeoutMs), [&st] { return st.blocked; }); +} + +extern "C" void rogue_fake_axi_dma_release_blocked() { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + st.releaseBlocked = true; + st.cv.notify_all(); +} + +extern "C" uint32_t rogue_fake_axi_dma_close_count() { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + return st.closeCount; +} + +extern "C" uint32_t rogue_fake_axi_dma_munmap_count() { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + return st.munmapCount; +} + +extern "C" uint32_t rogue_fake_axi_dma_ret_index_count() { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + return st.retIndexCount; +} + +extern "C" bool rogue_fake_axi_dma_close_during_driver_call() { + FakeState& st = state(); + std::lock_guard lock(st.mtx); + return st.closeDuringDriverCall; +} + +extern "C" int rogue_fake_open(const char* path, int flags, ...) { + mode_t mode = 0; + bool hasMode = (flags & O_CREAT) != 0; + if (hasMode) { + va_list args; + va_start(args, flags); + mode = static_cast(va_arg(args, int)); + va_end(args); + } + return fakeOpenImpl(path, flags, mode, hasMode); +} + +extern "C" int rogue_fake_close(int fd) { + return fakeCloseImpl(fd); +} + +extern "C" int rogue_fake_ioctl(int fd, unsigned long request, ...) { + va_list args; + va_start(args, request); + int ret = fakeIoctlImpl(fd, request, args); + va_end(args); + return ret; +} + +extern "C" void* rogue_fake_mmap(void* addr, size_t length, int prot, int flags, int fd, off_t offset) { + return fakeMmapImpl(addr, length, prot, flags, fd, offset); +} + +extern "C" int rogue_fake_munmap(void* addr, size_t length) { + return fakeMunmapImpl(addr, length); +} + +extern "C" int rogue_fake_poll(struct pollfd* fds, nfds_t nfds, int timeout) { + return fakePollImpl(fds, nfds, timeout); +} + +extern "C" ssize_t rogue_fake_read(int fd, void* buf, size_t count) { + return fakeReadImpl(fd, buf, count); +} + +extern "C" ssize_t rogue_fake_write(int fd, const void* buf, size_t count) { + return fakeWriteImpl(fd, buf, count); +} + +#ifdef __APPLE__ +ROGUE_DYLD_INTERPOSE(rogue_fake_open, open); +ROGUE_DYLD_INTERPOSE(rogue_fake_close, close); +ROGUE_DYLD_INTERPOSE(rogue_fake_ioctl, ioctl); +ROGUE_DYLD_INTERPOSE(rogue_fake_munmap, munmap); +ROGUE_DYLD_INTERPOSE(rogue_fake_poll, poll); +#else +extern "C" int open(const char* path, int flags, ...) { + mode_t mode = 0; + bool hasMode = (flags & O_CREAT) != 0; + if (hasMode) { + va_list args; + va_start(args, flags); + mode = static_cast(va_arg(args, int)); + va_end(args); + } + return fakeOpenImpl(path, flags, mode, hasMode); +} + +extern "C" int open64(const char* path, int flags, ...) { + mode_t mode = 0; + bool hasMode = (flags & O_CREAT) != 0; + if (hasMode) { + va_list args; + va_start(args, flags); + mode = static_cast(va_arg(args, int)); + va_end(args); + } + + if (isFakePath(path)) return fakeOpenImpl(path, flags, mode, hasMode); + if (hasMode) return realOpen64()(path, flags, mode); + return realOpen64()(path, flags); +} + +extern "C" int close(int fd) { + return fakeCloseImpl(fd); +} + +extern "C" int ioctl(int fd, unsigned long request, ...) { + va_list args; + va_start(args, request); + int ret = fakeIoctlImpl(fd, request, args); + va_end(args); + return ret; +} + +extern "C" void* mmap(void* addr, size_t length, int prot, int flags, int fd, off_t offset) { + return fakeMmapImpl(addr, length, prot, flags, fd, offset); +} + +extern "C" int munmap(void* addr, size_t length) { + return fakeMunmapImpl(addr, length); +} + +extern "C" int poll(struct pollfd* fds, nfds_t nfds, int timeout) { + return fakePollImpl(fds, nfds, timeout); +} + +extern "C" ssize_t read(int fd, void* buf, size_t count) { + return fakeReadImpl(fd, buf, count); +} + +extern "C" ssize_t write(int fd, const void* buf, size_t count) { + return fakeWriteImpl(fd, buf, count); +} +#endif diff --git a/tests/cpp/hardware/fake_axi_dma_driver.h b/tests/cpp/hardware/fake_axi_dma_driver.h new file mode 100644 index 0000000000..f46585909f --- /dev/null +++ b/tests/cpp/hardware/fake_axi_dma_driver.h @@ -0,0 +1,44 @@ +/** + * ---------------------------------------------------------------------------- + * Company : SLAC National Accelerator Laboratory + * ---------------------------------------------------------------------------- + * Description: + * Test-only fake AxiStreamDma driver syscall interposer. + * ---------------------------------------------------------------------------- + * This file is part of the rogue software platform. It is subject to + * the license terms in the LICENSE.txt file found in the top-level directory + * of this distribution and at: + * https://confluence.slac.stanford.edu/display/ppareg/LICENSE.html. + * No part of the rogue software platform, including this file, may be + * copied, modified, propagated, or distributed except according to the terms + * contained in the LICENSE.txt file. + * ---------------------------------------------------------------------------- + **/ +#ifndef ROGUE_TEST_CPP_HARDWARE_FAKE_AXI_DMA_DRIVER_H +#define ROGUE_TEST_CPP_HARDWARE_FAKE_AXI_DMA_DRIVER_H + +#include + +extern "C" { + +enum RogueFakeAxiDmaBlockOp { + ROGUE_FAKE_AXI_DMA_BLOCK_NONE = 0, + ROGUE_FAKE_AXI_DMA_BLOCK_IOCTL_GET_INDEX = 1, + ROGUE_FAKE_AXI_DMA_BLOCK_IOCTL_RET_INDEX = 2, + ROGUE_FAKE_AXI_DMA_BLOCK_IOCTL_BUFF_SIZE = 3, + ROGUE_FAKE_AXI_DMA_BLOCK_WRITE = 4, +}; + +const char* rogue_fake_axi_dma_path_prefix(); +void rogue_fake_axi_dma_reset(); +void rogue_fake_axi_dma_block_next(RogueFakeAxiDmaBlockOp op); +bool rogue_fake_axi_dma_wait_blocked(uint32_t timeoutMs); +void rogue_fake_axi_dma_release_blocked(); +uint32_t rogue_fake_axi_dma_close_count(); +uint32_t rogue_fake_axi_dma_munmap_count(); +uint32_t rogue_fake_axi_dma_ret_index_count(); +bool rogue_fake_axi_dma_close_during_driver_call(); + +} // extern "C" + +#endif diff --git a/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp b/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp new file mode 100644 index 0000000000..e735827339 --- /dev/null +++ b/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp @@ -0,0 +1,257 @@ +/** + * ---------------------------------------------------------------------------- + * Company : SLAC National Accelerator Laboratory + * ---------------------------------------------------------------------------- + * Description: + * Native C++ lifecycle tests for AxiStreamDma stop/fd races using a fake DMA + * syscall backend. + * ---------------------------------------------------------------------------- + * This file is part of the rogue software platform. It is subject to + * the license terms in the LICENSE.txt file found in the top-level directory + * of this distribution and at: + * https://confluence.slac.stanford.edu/display/ppareg/LICENSE.html. + * No part of the rogue software platform, including this file, may be + * copied, modified, propagated, or distributed except according to the terms + * contained in the LICENSE.txt file. + * ---------------------------------------------------------------------------- + **/ +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "doctest/doctest.h" +#include "fake_axi_dma_driver.h" +#include "rogue/GeneralError.h" +#include "rogue/hardware/axi/AxiStreamDma.h" +#include "rogue/interfaces/stream/Frame.h" +#include "support/test_helpers.h" + +namespace rha = rogue::hardware::axi; +namespace ris = rogue::interfaces::stream; + +namespace { + +class FakeAxiDmaDriver { + public: + using PathPrefixFn = const char* (*)(); + using ResetFn = void (*)(); + using BlockNextFn = void (*)(RogueFakeAxiDmaBlockOp); + using WaitBlockedFn = bool (*)(uint32_t); + using ReleaseBlockedFn = void (*)(); + using CountFn = uint32_t (*)(); + using CloseDuringCallFn = bool (*)(); + + FakeAxiDmaDriver() { + const char* path = std::getenv("ROGUE_FAKE_AXI_DMA_DRIVER_LIB"); + if (path == nullptr) throw std::runtime_error("ROGUE_FAKE_AXI_DMA_DRIVER_LIB is not set"); + + handle_ = dlopen(path, RTLD_NOW | RTLD_GLOBAL); + if (handle_ == nullptr) throw std::runtime_error(dlerror()); + + pathPrefix_ = load("rogue_fake_axi_dma_path_prefix"); + reset_ = load("rogue_fake_axi_dma_reset"); + blockNext_ = load("rogue_fake_axi_dma_block_next"); + waitBlocked_ = load("rogue_fake_axi_dma_wait_blocked"); + releaseBlocked_ = load("rogue_fake_axi_dma_release_blocked"); + closeCount_ = load("rogue_fake_axi_dma_close_count"); + munmapCount_ = load("rogue_fake_axi_dma_munmap_count"); + retIndexCount_ = load("rogue_fake_axi_dma_ret_index_count"); + closeDuringCall_ = load("rogue_fake_axi_dma_close_during_driver_call"); + } + + const char* pathPrefix() const { return pathPrefix_(); } + void reset() const { reset_(); } + void blockNext(RogueFakeAxiDmaBlockOp op) const { blockNext_(op); } + bool waitBlocked(uint32_t timeoutMs) const { return waitBlocked_(timeoutMs); } + void releaseBlocked() const { releaseBlocked_(); } + uint32_t closeCount() const { return closeCount_(); } + uint32_t munmapCount() const { return munmapCount_(); } + uint32_t retIndexCount() const { return retIndexCount_(); } + bool closeDuringDriverCall() const { return closeDuringCall_(); } + + private: + template + Func load(const char* name) { + void* sym = dlsym(handle_, name); + if (sym == nullptr) throw std::runtime_error(dlerror()); + return reinterpret_cast(sym); + } + + void* handle_; + PathPrefixFn pathPrefix_; + ResetFn reset_; + BlockNextFn blockNext_; + WaitBlockedFn waitBlocked_; + ReleaseBlockedFn releaseBlocked_; + CountFn closeCount_; + CountFn munmapCount_; + CountFn retIndexCount_; + CloseDuringCallFn closeDuringCall_; +}; + +FakeAxiDmaDriver& fakeDriver() { + static FakeAxiDmaDriver driver; + return driver; +} + +std::string fakePath(const char* name) { + static std::atomic counter{0}; + return std::string(fakeDriver().pathPrefix()) + "-" + name + "-" + std::to_string(getpid()) + "-" + + std::to_string(counter.fetch_add(1)); +} + +rha::AxiStreamDmaPtr makeDma(const char* name) { + return rha::AxiStreamDma::create(fakePath(name), 0, false); +} + +void requireStopWaitsWhileBlocked(const std::function& stopCall, const std::atomic& stopDone) { + std::this_thread::sleep_for(std::chrono::milliseconds(25)); + CHECK_FALSE(stopDone.load()); + CHECK_FALSE(fakeDriver().closeDuringDriverCall()); + stopCall(); +} + +void checkStopWaitsForBlockedDriverCall(const rha::AxiStreamDmaPtr& dma, + RogueFakeAxiDmaBlockOp blockOp, + const std::function& call) { + fakeDriver().blockNext(blockOp); + + std::atomic callDone{false}; + std::thread caller([&] { + call(); + callDone.store(true); + }); + + REQUIRE(fakeDriver().waitBlocked(1000)); + CHECK_FALSE(callDone.load()); + + std::atomic stopDone{false}; + std::thread stopper([&] { + dma->stop(); + stopDone.store(true); + }); + + requireStopWaitsWhileBlocked([&] { fakeDriver().releaseBlocked(); }, stopDone); + + caller.join(); + stopper.join(); + + CHECK(callDone.load()); + CHECK(stopDone.load()); + CHECK_FALSE(fakeDriver().closeDuringDriverCall()); +} + +} // namespace + +TEST_CASE("AxiStreamDma stop keeps shared zero-copy mappings alive until destruction") { + fakeDriver().reset(); + + auto dma = makeDma("mapping-lifetime"); + auto frame = dma->acceptReq(64, true); + + REQUIRE_EQ(frame->bufferCount(), 1U); + CHECK_EQ(fakeDriver().munmapCount(), 0U); + + dma->stop(); + + // Before this branch, stop() called closeShared(), which unmapped the DMA + // pages even though downstream Rogue frames could still hold buffers that + // point into those pages. + CHECK_EQ(fakeDriver().munmapCount(), 0U); + + frame->clear(); + CHECK_EQ(fakeDriver().munmapCount(), 0U); + + dma.reset(); + CHECK_EQ(fakeDriver().munmapCount(), 4U); +} + +TEST_CASE("AxiStreamDma rejects new stream work after stop while retaining shared descriptor") { + fakeDriver().reset(); + + auto dma = makeDma("post-stop-reject"); + auto pool = rogue_test::makePool(); + auto tx = rogue_test::makeFrame(pool, {1, 2, 3, 4}); + + dma->stop(); + + CHECK_THROWS_AS(dma->acceptReq(64, false), rogue::GeneralError); + CHECK_THROWS_AS(dma->acceptReq(64, true), rogue::GeneralError); + CHECK_THROWS_AS(dma->acceptFrame(tx), rogue::GeneralError); + CHECK_EQ(dma->getBuffSize(), 0U); + + dma.reset(); +} + +TEST_CASE("AxiStreamDma stop waits for an in-flight driver getter") { + fakeDriver().reset(); + + auto dma = makeDma("getter-race"); + + uint32_t value = 0; + checkStopWaitsForBlockedDriverCall(dma, ROGUE_FAKE_AXI_DMA_BLOCK_IOCTL_BUFF_SIZE, [&] { + value = dma->getBuffSize(); + }); + + CHECK_EQ(value, 4096U); + dma.reset(); +} + +TEST_CASE("AxiStreamDma stop waits for in-flight zero-copy allocation and return") { + fakeDriver().reset(); + + auto dma = makeDma("alloc-return-race"); + + ris::FramePtr frame; + checkStopWaitsForBlockedDriverCall(dma, ROGUE_FAKE_AXI_DMA_BLOCK_IOCTL_GET_INDEX, [&] { + frame = dma->acceptReq(64, true); + }); + + REQUIRE(frame); + REQUIRE_EQ(frame->bufferCount(), 1U); + frame->clear(); + frame.reset(); + + // Build a fresh instance so the retBuffer path can race stop() while fd_ + // is still open; the allocation half above has already stopped its DMA. + dma.reset(); + fakeDriver().reset(); + dma = makeDma("ret-buffer-race"); + frame = dma->acceptReq(64, true); + REQUIRE(frame); + + checkStopWaitsForBlockedDriverCall(dma, ROGUE_FAKE_AXI_DMA_BLOCK_IOCTL_RET_INDEX, [&] { + frame->clear(); + }); + + CHECK_EQ(fakeDriver().retIndexCount(), 1U); + dma.reset(); +} + +TEST_CASE("AxiStreamDma stop waits for an in-flight transmit write") { +#ifdef __APPLE__ + MESSAGE("Skipping write interposition check on macOS"); +#else + fakeDriver().reset(); + + auto dma = makeDma("write-race"); + auto pool = rogue_test::makePool(); + auto frame = rogue_test::makeFrame(pool, std::vector{0x10, 0x20, 0x30, 0x40}); + + checkStopWaitsForBlockedDriverCall(dma, ROGUE_FAKE_AXI_DMA_BLOCK_WRITE, [&] { + dma->acceptFrame(frame); + }); + + dma.reset(); +#endif +} From e03e9d386a4c78f4536e09a94c943c0e1d1a927b Mon Sep 17 00:00:00 2001 From: Larry Ruckman Date: Mon, 6 Jul 2026 12:55:41 -0700 Subject: [PATCH 4/9] test(axi): add regression test for AxiStreamDma zero-copy teardown lifetime Add a Linux-only C++ doctest that constructs a real AxiStreamDma against an LD_PRELOAD shim faking the aes-stream-driver, retains a zero-copy frame across stop(), and asserts the shared DMA mapping stays mapped after stop() and is released only at destruction. Moving closeShared() back into stop() unmaps the buffers while the frame is still held and fails the post-stop() assertion, so the test guards against regressing the teardown-lifetime fix. The fake-driver shim is a test-only LD_PRELOAD module built only under ROGUE_BUILD_TESTS and never linked into rogue-core. The test is labeled no-python so it runs in both CI ctest jobs, and is skipped on non-Linux where LD_PRELOAD interposition is unavailable. --- tests/cpp/CMakeLists.txt | 1 + tests/cpp/hardware/CMakeLists.txt | 26 ++ tests/cpp/hardware/support/fake_dma_preload.c | 324 ++++++++++++++++++ .../test_axi_stream_dma_zerocopy_lifetime.cpp | 86 +++++ 4 files changed, 437 insertions(+) create mode 100644 tests/cpp/hardware/CMakeLists.txt create mode 100644 tests/cpp/hardware/support/fake_dma_preload.c create mode 100644 tests/cpp/hardware/test_axi_stream_dma_zerocopy_lifetime.cpp diff --git a/tests/cpp/CMakeLists.txt b/tests/cpp/CMakeLists.txt index 8aaca17179..e76c5606e8 100644 --- a/tests/cpp/CMakeLists.txt +++ b/tests/cpp/CMakeLists.txt @@ -56,6 +56,7 @@ add_subdirectory(core) add_subdirectory(memory) add_subdirectory(stream) add_subdirectory(protocols) +add_subdirectory(hardware) if (NOT NO_PYTHON) add_subdirectory(smoke) diff --git a/tests/cpp/hardware/CMakeLists.txt b/tests/cpp/hardware/CMakeLists.txt new file mode 100644 index 0000000000..8e2b5c65e6 --- /dev/null +++ b/tests/cpp/hardware/CMakeLists.txt @@ -0,0 +1,26 @@ +# Hardware-class tests. The AxiStreamDma zero-copy lifetime test drives a real +# AxiStreamDma against an LD_PRELOAD shim that fakes the aes-stream-driver +# device, so it is Linux-only (ELF symbol interposition). Other platforms skip +# it; rogue-core still builds there. +if (NOT CMAKE_SYSTEM_NAME STREQUAL "Linux") + return() +endif() + +# Test-only fake-driver shim, injected via LD_PRELOAD (never linked into rogue). +add_library(rogue-cpp-fake-dma-preload MODULE support/fake_dma_preload.c) +target_link_libraries(rogue-cpp-fake-dma-preload PRIVATE ${CMAKE_DL_LIBS} pthread) + +rogue_add_cpp_test(rogue-cpp-hardware-axi-stream-dma-zerocopy-lifetime + SOURCES test_axi_stream_dma_zerocopy_lifetime.cpp + LABELS cpp-hardware no-python + LIBRARIES ${CMAKE_DL_LIBS}) + +# Ensure the shim is built whenever the test is built. +add_dependencies(rogue-cpp-hardware-axi-stream-dma-zerocopy-lifetime + rogue-cpp-fake-dma-preload) + +# rogue_add_cpp_test() sets ENVIRONMENT=MPLCONFIGDIR; re-set it here to also +# inject the fake-driver shim. The generator expression resolves to the built +# shim's absolute path. +set_tests_properties(rogue-cpp-hardware-axi-stream-dma-zerocopy-lifetime PROPERTIES + ENVIRONMENT "MPLCONFIGDIR=${CMAKE_BINARY_DIR}/tests/cpp/mplconfig;LD_PRELOAD=$") diff --git a/tests/cpp/hardware/support/fake_dma_preload.c b/tests/cpp/hardware/support/fake_dma_preload.c new file mode 100644 index 0000000000..5e821cb251 --- /dev/null +++ b/tests/cpp/hardware/support/fake_dma_preload.c @@ -0,0 +1,324 @@ +/** + * ---------------------------------------------------------------------------- + * Company : SLAC National Accelerator Laboratory + * ---------------------------------------------------------------------------- + * Description: + * Test-only LD_PRELOAD shim that emulates just enough of the aes-stream-driver + * character device for AxiStreamDma to construct, hand out a zero-copy buffer, + * and tear down on a machine with no DMA hardware. It intercepts + * open/close/ioctl/mmap/munmap/poll for a single sentinel device path and + * backs the "DMA" buffers with ordinary anonymous mappings, tracking how many + * are currently mapped. The test queries that count via fakedma_mapped_count() + * to prove that AxiStreamDma keeps the shared mapping alive across stop() and + * releases it only at destruction. + * + * This object is built only under -DROGUE_BUILD_TESTS=ON and is never linked + * into rogue-core or shipped. + * ---------------------------------------------------------------------------- + * This file is part of the rogue software platform. It is subject to + * the license terms in the LICENSE.txt file found in the top-level directory + * of this distribution and at: + * https://confluence.slac.stanford.edu/display/ppareg/LICENSE.html. + * No part of the rogue software platform, including this file, may be + * copied, modified, propagated, or distributed except according to the terms + * contained in the LICENSE.txt file. + * ---------------------------------------------------------------------------- + **/ +#define _GNU_SOURCE + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/* + * DMA ioctl request codes and version. These mirror the stable kernel ABI in + * include/rogue/hardware/drivers/DmaDriver.h (which is C++-only and cannot be + * included here). Keep in sync with that header. + */ +#define FAKE_DMA_Get_Buff_Count 0x1001 +#define FAKE_DMA_Get_Buff_Size 0x1002 +#define FAKE_DMA_Ret_Index 0x1005 +#define FAKE_DMA_Get_Index 0x1006 +#define FAKE_DMA_Set_MaskBytes 0x1008 +#define FAKE_DMA_Get_Version 0x1009 +#define FAKE_DMA_Get_RxBuff_Count 0x100C +#define FAKE_DMA_Get_TxBuff_Count 0x100D +#define FAKE_DMA_VERSION 0x06 + +/* Sentinel device path intercepted by this shim. */ +#define FAKE_PATH "/tmp/rogue-fake-datadev" +#define FAKE_BUFF_SIZE 4096u +#define FAKE_BUFF_COUNT 8u + +#define MAX_FDS 16 +#define MAX_REGIONS 64 + +static pthread_mutex_t g_lock = PTHREAD_MUTEX_INITIALIZER; + +static int g_fds[MAX_FDS]; +static int g_fd_n = 0; + +static void* g_region_addr[MAX_REGIONS]; +static size_t g_region_len[MAX_REGIONS]; +static int g_region_n = 0; + +static int g_mapped_count = 0; +static uint32_t g_next_index = 0; + +typedef int (*open_fn)(const char*, int, ...); +typedef void* (*mmap_fn)(void*, size_t, int, int, int, off_t); +typedef int (*munmap_fn)(void*, size_t); +typedef int (*close_fn)(int); +typedef int (*ioctl_fn)(int, unsigned long, ...); +typedef int (*poll_fn)(struct pollfd*, nfds_t, int); + +static open_fn real_open; +static open_fn real_open64; +static mmap_fn real_mmap; +static mmap_fn real_mmap64; +static munmap_fn real_munmap; +static close_fn real_close; +static ioctl_fn real_ioctl; +static poll_fn real_poll; + +/* ------- bookkeeping helpers (caller holds g_lock) ------- */ + +static int is_our_fd_locked(int fd) { + int i; + for (i = 0; i < g_fd_n; i++) + if (g_fds[i] == fd) return 1; + return 0; +} + +static void add_fd_locked(int fd) { + if (g_fd_n < MAX_FDS) g_fds[g_fd_n++] = fd; +} + +static void del_fd_locked(int fd) { + int i; + for (i = 0; i < g_fd_n; i++) { + if (g_fds[i] == fd) { + g_fds[i] = g_fds[--g_fd_n]; + return; + } + } +} + +static void add_region_locked(void* addr, size_t len) { + if (g_region_n < MAX_REGIONS) { + g_region_addr[g_region_n] = addr; + g_region_len[g_region_n] = len; + g_region_n++; + g_mapped_count++; + } +} + +static int del_region_locked(void* addr) { + int i; + for (i = 0; i < g_region_n; i++) { + if (g_region_addr[i] == addr) { + g_region_addr[i] = g_region_addr[g_region_n - 1]; + g_region_len[i] = g_region_len[g_region_n - 1]; + g_region_n--; + g_mapped_count--; + return 1; + } + } + return 0; +} + +/* Return an anonymous mapping standing in for a DMA buffer, and record it. */ +static void* make_fake_buffer(size_t len, int prot) { + void* p = real_mmap(NULL, len, prot, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0); + if (p != MAP_FAILED) { + pthread_mutex_lock(&g_lock); + add_region_locked(p, len); + pthread_mutex_unlock(&g_lock); + } + return p; +} + +/* ------- intercepted entry points ------- */ + +int open(const char* path, int flags, ...) { + mode_t mode = 0; + if (!real_open) real_open = (open_fn)dlsym(RTLD_NEXT, "open"); + + if (flags & O_CREAT) { + va_list ap; + va_start(ap, flags); + mode = (mode_t)va_arg(ap, int); + va_end(ap); + } + + if (path != NULL && strcmp(path, FAKE_PATH) == 0) { + int fd = real_open("/dev/null", O_RDWR); + if (fd >= 0) { + pthread_mutex_lock(&g_lock); + add_fd_locked(fd); + pthread_mutex_unlock(&g_lock); + } + return fd; + } + return real_open(path, flags, mode); +} + +int open64(const char* path, int flags, ...) { + mode_t mode = 0; + if (!real_open64) real_open64 = (open_fn)dlsym(RTLD_NEXT, "open64"); + + if (flags & O_CREAT) { + va_list ap; + va_start(ap, flags); + mode = (mode_t)va_arg(ap, int); + va_end(ap); + } + + if (path != NULL && strcmp(path, FAKE_PATH) == 0) { + int fd = real_open64("/dev/null", O_RDWR); + if (fd >= 0) { + pthread_mutex_lock(&g_lock); + add_fd_locked(fd); + pthread_mutex_unlock(&g_lock); + } + return fd; + } + return real_open64(path, flags, mode); +} + +int close(int fd) { + if (!real_close) real_close = (close_fn)dlsym(RTLD_NEXT, "close"); + pthread_mutex_lock(&g_lock); + del_fd_locked(fd); + pthread_mutex_unlock(&g_lock); + return real_close(fd); +} + +void* mmap(void* addr, size_t len, int prot, int flags, int fd, off_t off) { + int ours; + if (!real_mmap) { + static __thread int resolving = 0; + if (resolving) return (void*)syscall(SYS_mmap, addr, len, prot, flags, fd, off); + resolving = 1; + real_mmap = (mmap_fn)dlsym(RTLD_NEXT, "mmap"); + resolving = 0; + } + pthread_mutex_lock(&g_lock); + ours = is_our_fd_locked(fd); + pthread_mutex_unlock(&g_lock); + + if (ours) return make_fake_buffer(len, prot); + return real_mmap(addr, len, prot, flags, fd, off); +} + +void* mmap64(void* addr, size_t len, int prot, int flags, int fd, off_t off) { + int ours; + if (!real_mmap64) { + static __thread int resolving = 0; + if (resolving) return (void*)syscall(SYS_mmap, addr, len, prot, flags, fd, off); + resolving = 1; + real_mmap64 = (mmap_fn)dlsym(RTLD_NEXT, "mmap64"); + resolving = 0; + } + if (!real_mmap) real_mmap = (mmap_fn)dlsym(RTLD_NEXT, "mmap"); + + pthread_mutex_lock(&g_lock); + ours = is_our_fd_locked(fd); + pthread_mutex_unlock(&g_lock); + + if (ours) return make_fake_buffer(len, prot); + return real_mmap64(addr, len, prot, flags, fd, off); +} + +int munmap(void* addr, size_t len) { + if (!real_munmap) { + static __thread int resolving = 0; + if (resolving) return (int)syscall(SYS_munmap, addr, len); + resolving = 1; + real_munmap = (munmap_fn)dlsym(RTLD_NEXT, "munmap"); + resolving = 0; + } + pthread_mutex_lock(&g_lock); + del_region_locked(addr); + pthread_mutex_unlock(&g_lock); + return real_munmap(addr, len); +} + +int ioctl(int fd, unsigned long request, ...) { + void* arg; + va_list ap; + int ours; + if (!real_ioctl) real_ioctl = (ioctl_fn)dlsym(RTLD_NEXT, "ioctl"); + + va_start(ap, request); + arg = va_arg(ap, void*); + va_end(ap); + + pthread_mutex_lock(&g_lock); + ours = is_our_fd_locked(fd); + pthread_mutex_unlock(&g_lock); + + if (!ours) return real_ioctl(fd, request, arg); + + switch (request & 0xFFFF) { + case FAKE_DMA_Get_Version: return FAKE_DMA_VERSION; + case FAKE_DMA_Get_Buff_Size: return (int)FAKE_BUFF_SIZE; + case FAKE_DMA_Get_Buff_Count: return (int)FAKE_BUFF_COUNT; + case FAKE_DMA_Get_RxBuff_Count: return (int)(FAKE_BUFF_COUNT / 2); + case FAKE_DMA_Get_TxBuff_Count: return (int)(FAKE_BUFF_COUNT / 2); + case FAKE_DMA_Set_MaskBytes: return 0; + case FAKE_DMA_Get_Index: { + int idx; + pthread_mutex_lock(&g_lock); + idx = (int)(g_next_index++ % FAKE_BUFF_COUNT); + pthread_mutex_unlock(&g_lock); + return idx; + } + case FAKE_DMA_Ret_Index: return 0; + default: return 0; + } +} + +int poll(struct pollfd* fds, nfds_t nfds, int timeout) { + if (!real_poll) real_poll = (poll_fn)dlsym(RTLD_NEXT, "poll"); + + if (nfds == 1 && fds != NULL) { + int ours; + pthread_mutex_lock(&g_lock); + ours = is_our_fd_locked(fds[0].fd); + pthread_mutex_unlock(&g_lock); + + if (ours) { + /* TX/alloc path waits for POLLOUT: report ready so acceptReq/ + * acceptFrame proceed to dmaGetIndex/dmaWrite. */ + if (fds[0].events & POLLOUT) { + fds[0].revents = POLLOUT; + return 1; + } + /* RX path waits for POLLIN: stay idle for the requested timeout and + * report nothing so the worker thread never issues a read. */ + if (timeout > 0) real_poll(NULL, 0, timeout); + fds[0].revents = 0; + return 0; + } + } + return real_poll(fds, nfds, timeout); +} + +/* Query hook for the test: number of fake DMA buffers currently mapped. */ +int fakedma_mapped_count(void) { + int n; + pthread_mutex_lock(&g_lock); + n = g_mapped_count; + pthread_mutex_unlock(&g_lock); + return n; +} diff --git a/tests/cpp/hardware/test_axi_stream_dma_zerocopy_lifetime.cpp b/tests/cpp/hardware/test_axi_stream_dma_zerocopy_lifetime.cpp new file mode 100644 index 0000000000..8f247915f2 --- /dev/null +++ b/tests/cpp/hardware/test_axi_stream_dma_zerocopy_lifetime.cpp @@ -0,0 +1,86 @@ +/** + * ---------------------------------------------------------------------------- + * Company : SLAC National Accelerator Laboratory + * ---------------------------------------------------------------------------- + * Description: + * Regression test for the AxiStreamDma zero-copy teardown lifetime fix. + * + * A zero-copy Rogue Buffer points directly into the shared DMA mapping + * (desc_->rawBuff, established by dmaMapDma()). Releasing that mapping inside + * stop() left any frame retained past stop() pointing at unmapped memory. The + * fix moves the shared-descriptor teardown out of stop() and into + * ~AxiStreamDma(), so the mapping outlives stop() and is freed only at + * destruction. + * + * The aes-stream-driver character device does not exist on CI runners, so this + * test runs against the fake_dma_preload LD_PRELOAD shim (wired in by the + * accompanying CMakeLists.txt). The shim reports how many DMA buffers are + * currently mapped via fakedma_mapped_count(); the test asserts the mapping is + * still present after stop() and released exactly at destruction. Reverting the + * fix (moving closeShared() back into stop()) drops the count to zero at stop() + * and fails the post-stop() check. + * ---------------------------------------------------------------------------- + * This file is part of the rogue software platform. It is subject to + * the license terms in the LICENSE.txt file found in the top-level directory + * of this distribution and at: + * https://confluence.slac.stanford.edu/display/ppareg/LICENSE.html. + * No part of the rogue software platform, including this file, may be + * copied, modified, propagated, or distributed except according to the terms + * contained in the LICENSE.txt file. + * ---------------------------------------------------------------------------- + **/ +#include + +#include "doctest/doctest.h" +#include "rogue/hardware/axi/AxiStreamDma.h" +#include "rogue/interfaces/stream/Frame.h" + +namespace rha = rogue::hardware::axi; + +namespace { + +// Sentinel path recognized by the fake_dma_preload LD_PRELOAD shim. +const char* const kFakePath = "/tmp/rogue-fake-datadev"; + +typedef int (*MappedCountFn)(void); + +// Resolve the shim's query hook. Non-null only when the shim is preloaded. +MappedCountFn mappedCountHook() { + return reinterpret_cast(dlsym(RTLD_DEFAULT, "fakedma_mapped_count")); +} + +} // namespace + +TEST_CASE("AxiStreamDma keeps the shared DMA mapping alive across stop()") { + MappedCountFn mappedCount = mappedCountHook(); + REQUIRE(mappedCount != nullptr); // fake_dma_preload shim must be LD_PRELOAD'd + REQUIRE(mappedCount() == 0); + + auto dma = rha::AxiStreamDma::create(kFakePath, 0, false); + + // dmaMapDma() maps all shared zero-copy buffers at construction. + const int mappedAtCtor = mappedCount(); + REQUIRE(mappedAtCtor > 0); + + // Obtain a zero-copy frame and retain it past stop(); its buffer points + // into the shared mapping. + auto frame = dma->acceptReq(4096, true); + REQUIRE(static_cast(frame)); + REQUIRE(frame->bufferCount() > 0); + + dma->stop(); + + // The fix keeps the shared mapping valid after stop() so the retained frame + // does not dangle. Reverted code releases it inside stop(), dropping the + // count to zero here. + CHECK_EQ(mappedCount(), mappedAtCtor); + + // Returning the buffer post-stop() is a no-op on the driver (fd_ == -1); it + // must not touch the still-mapped memory or throw. + frame.reset(); + CHECK_EQ(mappedCount(), mappedAtCtor); + + // Destruction is the single point that releases the shared mapping. + dma.reset(); + CHECK_EQ(mappedCount(), 0); +} From 31536443686b28ec32560c4a679c9f0f45196868 Mon Sep 17 00:00:00 2001 From: Benjamin Reese Date: Mon, 6 Jul 2026 13:38:19 -0700 Subject: [PATCH 5/9] test(axi): fix REQUIRE(frame) stringification in lifecycle test doctest cannot stringify a shared_ptr, so a bare REQUIRE(frame) fails to compile. Use static_cast(frame), matching the workaround already used in the sibling zerocopy lifetime test. Co-Authored-By: Claude Opus 4.7 --- tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp b/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp index ac1f424b34..4164befde0 100644 --- a/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp +++ b/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp @@ -197,7 +197,7 @@ TEST_CASE("AxiStreamDma stop waits for in-flight zero-copy allocation and return frame = dma->acceptReq(64, true); }); - REQUIRE(frame); + REQUIRE(static_cast(frame)); REQUIRE_EQ(frame->bufferCount(), 1U); frame->clear(); frame.reset(); @@ -207,7 +207,7 @@ TEST_CASE("AxiStreamDma stop waits for in-flight zero-copy allocation and return fakeDma().reset(); dma = makeDma(); frame = dma->acceptReq(64, true); - REQUIRE(frame); + REQUIRE(static_cast(frame)); checkStopWaitsForBlockedDriverCall(dma, FakeDmaBlockRetIndex, [&] { frame->clear(); From 2341441da95027e02e8cf130c42807da2f98b368 Mon Sep 17 00:00:00 2001 From: Benjamin Reese Date: Mon, 6 Jul 2026 13:53:01 -0700 Subject: [PATCH 6/9] Replace mutex with shared_timed_mutex for fd management and simplify FdGuard usage --- include/rogue/hardware/axi/AxiStreamDma.h | 39 ++++-- src/rogue/hardware/axi/AxiStreamDma.cpp | 145 ++++++++-------------- 2 files changed, 78 insertions(+), 106 deletions(-) diff --git a/include/rogue/hardware/axi/AxiStreamDma.h b/include/rogue/hardware/axi/AxiStreamDma.h index 5af8b05122..c8b0a58746 100644 --- a/include/rogue/hardware/axi/AxiStreamDma.h +++ b/include/rogue/hardware/axi/AxiStreamDma.h @@ -21,10 +21,10 @@ #include #include -#include #include #include #include +#include #include #include @@ -110,13 +110,7 @@ class AxiStreamDma : public rogue::interfaces::stream::Master, public rogue::int std::atomic stopped_{false}; // Serializes fd_ close against active driver calls and deferred zero-copy returns. - std::mutex fdMtx_; - - // Number of public/API paths currently using fd_. - uint32_t fdUsers_ = 0; - - // Wakes stop() after the last active fd_ user exits. - std::condition_variable fdCv_; + std::shared_timed_mutex fdMtx_; // Destination selector used when transmitting frames. uint32_t dest_; @@ -155,15 +149,15 @@ class AxiStreamDma : public rogue::interfaces::stream::Master, public rogue::int // RAII guard that pins fd_ open while a driver call runs. class FdGuard { - rogue::hardware::axi::AxiStreamDma* owner_; - int32_t fd_; + std::shared_lock lock_; + int32_t fd_ = -1; public: FdGuard(rogue::hardware::axi::AxiStreamDma* owner, const char* context, bool throwOnStopped, bool allowStopped = false); - ~FdGuard(); + ~FdGuard() = default; FdGuard(const FdGuard&) = delete; FdGuard& operator=(const FdGuard&) = delete; @@ -171,6 +165,29 @@ class AxiStreamDma : public rogue::interfaces::stream::Master, public rogue::int bool valid() const { return fd_ >= 0; } }; + // Runs a void driver operation while fd_ is pinned open. + template + void withFd(const char* context, bool throwOnStopped, Func func, bool allowStopped = false) { + FdGuard fd(this, context, throwOnStopped, allowStopped); + if (fd.valid()) func(fd.fd()); + } + + // Runs a value-returning driver operation while fd_ is pinned open. + template + Result withFd(const char* context, bool throwOnStopped, Result stoppedValue, Func func, bool allowStopped = false) { + FdGuard fd(this, context, throwOnStopped, allowStopped); + if (!fd.valid()) return stoppedValue; + return func(fd.fd()); + } + + template + uint32_t readFdValue(const char* context, Func func) { + return withFd(context, false, 0, [func](int32_t fd) { + auto value = func(fd); + return (value < 0) ? 0 : static_cast(value); + }); + } + public: /** * @brief Creates an AXI Stream DMA bridge instance. diff --git a/src/rogue/hardware/axi/AxiStreamDma.cpp b/src/rogue/hardware/axi/AxiStreamDma.cpp index 2ff7a827f1..c6c1191694 100644 --- a/src/rogue/hardware/axi/AxiStreamDma.cpp +++ b/src/rogue/hardware/axi/AxiStreamDma.cpp @@ -53,34 +53,24 @@ static void throwStopped(const char* context) { throw rogue::GeneralError(context, "instance has been stopped or did not finish construction"); } -static uint32_t dmaValueOrZero(ssize_t value) { - return (value < 0) ? 0 : static_cast(value); -} - rha::AxiStreamDma::FdGuard::FdGuard(rha::AxiStreamDma* owner, const char* context, bool throwOnStopped, bool allowStopped) { - std::lock_guard lock(owner->fdMtx_); - - if ((!allowStopped && owner->stopped_.load()) || owner->fd_ < 0) { - owner_ = NULL; - fd_ = -1; + if (!allowStopped && owner->stopped_.load()) { if (throwOnStopped) throwStopped(context); return; } - owner_ = owner; - fd_ = owner->fd_; - owner->fdUsers_++; -} + lock_ = std::shared_lock(owner->fdMtx_); -rha::AxiStreamDma::FdGuard::~FdGuard() { - if (owner_ != NULL) { - std::lock_guard lock(owner_->fdMtx_); - owner_->fdUsers_--; - if (owner_->fdUsers_ == 0) owner_->fdCv_.notify_all(); + if ((!allowStopped && owner->stopped_.load()) || owner->fd_ < 0) { + lock_.unlock(); + if (throwOnStopped) throwStopped(context); + return; } + + fd_ = owner->fd_; } rha::AxiStreamDmaShared::AxiStreamDmaShared(std::string path) { @@ -359,8 +349,7 @@ void rha::AxiStreamDma::stop() { // open until ~AxiStreamDma() because zero-copy Rogue buffers can outlive a // user-initiated stop() and still point into desc_->rawBuff. { - std::unique_lock lock(fdMtx_); - fdCv_.wait(lock, [this] { return fdUsers_ == 0; }); + std::unique_lock lock(fdMtx_); if (fd_ >= 0) { ::close(fd_); fd_ = -1; @@ -379,14 +368,12 @@ void rha::AxiStreamDma::setTimeout(uint32_t timeout) { //! Set driver debug level void rha::AxiStreamDma::setDriverDebug(uint32_t level) { - FdGuard fd(this, "AxiStreamDma::setDriverDebug", false); - if (fd.valid()) dmaSetDebug(fd.fd(), level); + withFd("AxiStreamDma::setDriverDebug", false, [level](int32_t fd) { dmaSetDebug(fd, level); }); } //! Strobe ack line void rha::AxiStreamDma::dmaAck() { - FdGuard fd(this, "AxiStreamDma::dmaAck", false); - if (fd.valid()) axisReadAck(fd.fd()); + withFd("AxiStreamDma::dmaAck", false, [](int32_t fd) { axisReadAck(fd); }); } //! Generate a buffer. Called from master @@ -425,10 +412,8 @@ ris::FramePtr rha::AxiStreamDma::acceptReq(uint32_t size, bool zeroCopyEn) { // Keep trying since poll call can fire // but getIndex fails because we did not win the buffer lock do { - { - FdGuard fd(this, "AxiStreamDma::acceptReq", true); - - pfd.fd = fd.fd(); + res = withFd("AxiStreamDma::acceptReq", true, -1, [&](int32_t fd) { + pfd.fd = fd; pfd.events = POLLOUT; pfd.revents = 0; @@ -449,13 +434,13 @@ ris::FramePtr rha::AxiStreamDma::acceptReq(uint32_t size, bool zeroCopyEn) { ".%" PRIuLEAST32 " seconds! May be caused by outbound back pressure.", timeout_.tv_sec, timeout_.tv_usec); - res = -1; + return -1; } else { // Attempt to get index. // return of less than 0 is a failure to get a buffer - res = dmaGetIndex(fd.fd()); + return dmaGetIndex(fd); } - } + }); } while (res < 0); // Mark zero copy meta with bit 31 set, lower bits are index @@ -526,18 +511,16 @@ void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) { // Buffer is not already stale as indicates by bit 30 if ((meta & 0x40000000) == 0) { - { - FdGuard fd(this, "AxiStreamDma::acceptFrame", true); - + withFd("AxiStreamDma::acceptFrame", true, [&](int32_t fd) { // Write by passing (*it)er index to driver - if (dmaWriteIndex(fd.fd(), + 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; @@ -549,10 +532,8 @@ void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) { // Keep trying since poll call can fire // but write fails because we did not win the (*it)er lock do { - { - FdGuard fd(this, "AxiStreamDma::acceptFrame", true); - - pfd.fd = fd.fd(); + res = withFd("AxiStreamDma::acceptFrame", true, 0, [&](int32_t fd) { + pfd.fd = fd; pfd.events = POLLOUT; pfd.revents = 0; @@ -573,18 +554,17 @@ void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) { ".%" PRIuLEAST32 " seconds! May be caused by outbound back pressure.", timeout_.tv_sec, timeout_.tv_usec); - res = 0; + return 0; } else { // Write with (*it)er copy - if ((res = dmaWrite(fd.fd(), - (*it)->begin(), - (*it)->getPayload(), - axisSetFlags(fuser, luser, 0), - dest_)) < 0) { + ssize_t writeSize = dmaWrite( + fd, (*it)->begin(), (*it)->getPayload(), axisSetFlags(fuser, luser, 0), dest_); + if (writeSize < 0) { throw(rogue::GeneralError("AxiStreamDma::acceptFrame", "AXIS Write Call Failed!!!!")); } + return static_cast(writeSize); } - } + }); } while (res == 0); // Exit out if return flag was set false } } @@ -601,9 +581,14 @@ void rha::AxiStreamDma::retBuffer(uint8_t* data, uint32_t meta, uint32_t size) { // Device is open and buffer is not stale // Bit 30 indicates buffer has already been returned to hardware if ((meta & 0x40000000) == 0) { - FdGuard fd(this, "AxiStreamDma::retBuffer", false, true); - if (fd.valid() && dmaRetIndex(fd.fd(), meta & 0x3FFFFFFF) < 0) - throw(rogue::GeneralError("AxiStreamDma::retBuffer", "AXIS Return Buffer Call Failed!!!!")); + withFd( + "AxiStreamDma::retBuffer", + false, + [meta](int32_t fd) { + if (dmaRetIndex(fd, meta & 0x3FFFFFFF) < 0) + throw(rogue::GeneralError("AxiStreamDma::retBuffer", "AXIS Return Buffer Call Failed!!!!")); + }, + true); } decCounter(size); @@ -730,107 +715,77 @@ void rha::AxiStreamDma::runThread(std::weak_ptr lockPtr) { //! Get the DMA Driver's Git Version std::string rha::AxiStreamDma::getGitVersion() { - FdGuard fd(this, "AxiStreamDma::getGitVersion", false); - if (!fd.valid()) return ""; - return dmaGetGitVersion(fd.fd()); + return withFd("AxiStreamDma::getGitVersion", false, std::string(), dmaGetGitVersion); } //! Get the DMA Driver's API Version uint32_t rha::AxiStreamDma::getApiVersion() { - FdGuard fd(this, "AxiStreamDma::getApiVersion", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetApiVersion(fd.fd())); + return readFdValue("AxiStreamDma::getApiVersion", dmaGetApiVersion); } //! Get the size of buffers (RX/TX) uint32_t rha::AxiStreamDma::getBuffSize() { - FdGuard fd(this, "AxiStreamDma::getBuffSize", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetBuffSize(fd.fd())); + return readFdValue("AxiStreamDma::getBuffSize", dmaGetBuffSize); } //! Get the number of RX buffers uint32_t rha::AxiStreamDma::getRxBuffCount() { - FdGuard fd(this, "AxiStreamDma::getRxBuffCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetRxBuffCount(fd.fd())); + return readFdValue("AxiStreamDma::getRxBuffCount", dmaGetRxBuffCount); } //! Get RX buffer in User count uint32_t rha::AxiStreamDma::getRxBuffinUserCount() { - FdGuard fd(this, "AxiStreamDma::getRxBuffinUserCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetRxBuffinUserCount(fd.fd())); + return readFdValue("AxiStreamDma::getRxBuffinUserCount", dmaGetRxBuffinUserCount); } //! Get RX buffer in HW count uint32_t rha::AxiStreamDma::getRxBuffinHwCount() { - FdGuard fd(this, "AxiStreamDma::getRxBuffinHwCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetRxBuffinHwCount(fd.fd())); + return readFdValue("AxiStreamDma::getRxBuffinHwCount", dmaGetRxBuffinHwCount); } //! Get RX buffer in Pre-HW Queue count uint32_t rha::AxiStreamDma::getRxBuffinPreHwQCount() { - FdGuard fd(this, "AxiStreamDma::getRxBuffinPreHwQCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetRxBuffinPreHwQCount(fd.fd())); + return readFdValue("AxiStreamDma::getRxBuffinPreHwQCount", dmaGetRxBuffinPreHwQCount); } //! Get RX buffer in SW Queue count uint32_t rha::AxiStreamDma::getRxBuffinSwQCount() { - FdGuard fd(this, "AxiStreamDma::getRxBuffinSwQCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetRxBuffinSwQCount(fd.fd())); + return readFdValue("AxiStreamDma::getRxBuffinSwQCount", dmaGetRxBuffinSwQCount); } //! Get RX buffer missing count uint32_t rha::AxiStreamDma::getRxBuffMissCount() { - FdGuard fd(this, "AxiStreamDma::getRxBuffMissCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetRxBuffMissCount(fd.fd())); + return readFdValue("AxiStreamDma::getRxBuffMissCount", dmaGetRxBuffMissCount); } //! Get the number of TX buffers uint32_t rha::AxiStreamDma::getTxBuffCount() { - FdGuard fd(this, "AxiStreamDma::getTxBuffCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetTxBuffCount(fd.fd())); + return readFdValue("AxiStreamDma::getTxBuffCount", dmaGetTxBuffCount); } //! Get TX buffer in User count uint32_t rha::AxiStreamDma::getTxBuffinUserCount() { - FdGuard fd(this, "AxiStreamDma::getTxBuffinUserCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetTxBuffinUserCount(fd.fd())); + return readFdValue("AxiStreamDma::getTxBuffinUserCount", dmaGetTxBuffinUserCount); } //! Get TX buffer in HW count uint32_t rha::AxiStreamDma::getTxBuffinHwCount() { - FdGuard fd(this, "AxiStreamDma::getTxBuffinHwCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetTxBuffinHwCount(fd.fd())); + return readFdValue("AxiStreamDma::getTxBuffinHwCount", dmaGetTxBuffinHwCount); } //! Get TX buffer in Pre-HW Queue count uint32_t rha::AxiStreamDma::getTxBuffinPreHwQCount() { - FdGuard fd(this, "AxiStreamDma::getTxBuffinPreHwQCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetTxBuffinPreHwQCount(fd.fd())); + return readFdValue("AxiStreamDma::getTxBuffinPreHwQCount", dmaGetTxBuffinPreHwQCount); } //! Get TX buffer in SW Queue count uint32_t rha::AxiStreamDma::getTxBuffinSwQCount() { - FdGuard fd(this, "AxiStreamDma::getTxBuffinSwQCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetTxBuffinSwQCount(fd.fd())); + return readFdValue("AxiStreamDma::getTxBuffinSwQCount", dmaGetTxBuffinSwQCount); } //! Get TX buffer missing count uint32_t rha::AxiStreamDma::getTxBuffMissCount() { - FdGuard fd(this, "AxiStreamDma::getTxBuffMissCount", false); - if (!fd.valid()) return 0; - return dmaValueOrZero(dmaGetTxBuffMissCount(fd.fd())); + return readFdValue("AxiStreamDma::getTxBuffMissCount", dmaGetTxBuffMissCount); } void rha::AxiStreamDma::setup_python() { From df4cf3caf125d804de4edca598a5df909fd21513 Mon Sep 17 00:00:00 2001 From: Benjamin Reese Date: Mon, 6 Jul 2026 15:10:47 -0700 Subject: [PATCH 7/9] Add fake DMA hooks and enhance lifecycle tests for AxiStreamDma --- tests/cpp/hardware/support/fake_dma_hooks.h | 107 ++++++++++++++ tests/cpp/hardware/support/fake_dma_preload.c | 50 ++++++- .../test_axi_stream_dma_lifecycle.cpp | 130 ++++++------------ .../test_axi_stream_dma_zerocopy_lifetime.cpp | 43 +++--- 4 files changed, 226 insertions(+), 104 deletions(-) create mode 100644 tests/cpp/hardware/support/fake_dma_hooks.h diff --git a/tests/cpp/hardware/support/fake_dma_hooks.h b/tests/cpp/hardware/support/fake_dma_hooks.h new file mode 100644 index 0000000000..6379b59a31 --- /dev/null +++ b/tests/cpp/hardware/support/fake_dma_hooks.h @@ -0,0 +1,107 @@ +/** + * ---------------------------------------------------------------------------- + * Company : SLAC National Accelerator Laboratory + * ---------------------------------------------------------------------------- + * Description: + * Shared C++ accessors for the fake_dma_preload LD_PRELOAD test shim. + * ---------------------------------------------------------------------------- + * This file is part of the rogue software platform. It is subject to + * the license terms in the LICENSE.txt file found in the top-level directory + * of this distribution and at: + * https://confluence.slac.stanford.edu/display/ppareg/LICENSE.html. + * No part of the rogue software platform, including this file, may be + * copied, modified, propagated, or distributed except according to the terms + * contained in the LICENSE.txt file. + * ---------------------------------------------------------------------------- + **/ +#ifndef ROGUE_TEST_CPP_HARDWARE_SUPPORT_FAKE_DMA_HOOKS_H +#define ROGUE_TEST_CPP_HARDWARE_SUPPORT_FAKE_DMA_HOOKS_H + +#include +#include + +#include + +namespace rogue_test { + +enum FakeDmaBlockOp { + FakeDmaBlockNone = 0, + FakeDmaBlockGetIndex = 1, + FakeDmaBlockRetIndex = 2, + FakeDmaBlockBuffSize = 3, + FakeDmaBlockWrite = 4, +}; + +class FakeDma { + public: + using PathFn = const char* (*)(); + using ResetFn = void (*)(); + using BlockNextFn = void (*)(int); + using WaitBlockedFn = int (*)(uint32_t); + using ReleaseFn = void (*)(); + using CountFn = int (*)(); + + FakeDma() { + path_ = load("fakedma_path"); + reset_ = load("fakedma_reset"); + blockNext_ = load("fakedma_block_next"); + waitBlocked_ = load("fakedma_wait_blocked"); + release_ = load("fakedma_release_blocked"); + mappedCount_ = load("fakedma_mapped_count"); + fdCount_ = load("fakedma_fd_count"); + regionCount_ = load("fakedma_region_count"); + activeCalls_ = load("fakedma_active_call_count"); + overflowCount_ = load("fakedma_tracking_overflow_count"); + isClean_ = load("fakedma_is_clean"); + retIndexCount_ = load("fakedma_ret_index_count"); + closeDuringCall_ = load("fakedma_close_during_driver_call"); + } + + const char* path() const { return path_(); } + void reset() const { reset_(); } + void blockNext(FakeDmaBlockOp op) const { blockNext_(static_cast(op)); } + bool waitBlocked(uint32_t timeoutMs) const { return waitBlocked_(timeoutMs) != 0; } + void releaseBlocked() const { release_(); } + int mappedCount() const { return mappedCount_(); } + int fdCount() const { return fdCount_(); } + int regionCount() const { return regionCount_(); } + int activeCallCount() const { return activeCalls_(); } + int overflowCount() const { return overflowCount_(); } + bool isClean() const { return isClean_() != 0; } + int retIndexCount() const { return retIndexCount_(); } + bool closeDuringDriverCall() const { return closeDuringCall_() != 0; } + + private: + template + Func load(const char* name) { + void* sym = dlsym(RTLD_DEFAULT, name); + if (sym == nullptr) { + const char* err = dlerror(); + throw std::runtime_error(err != nullptr ? err : name); + } + return reinterpret_cast(sym); + } + + PathFn path_; + ResetFn reset_; + BlockNextFn blockNext_; + WaitBlockedFn waitBlocked_; + ReleaseFn release_; + CountFn mappedCount_; + CountFn fdCount_; + CountFn regionCount_; + CountFn activeCalls_; + CountFn overflowCount_; + CountFn isClean_; + CountFn retIndexCount_; + CountFn closeDuringCall_; +}; + +inline FakeDma& fakeDma() { + static FakeDma fake; + return fake; +} + +} // namespace rogue_test + +#endif diff --git a/tests/cpp/hardware/support/fake_dma_preload.c b/tests/cpp/hardware/support/fake_dma_preload.c index 0f055b10be..8e27269b7b 100644 --- a/tests/cpp/hardware/support/fake_dma_preload.c +++ b/tests/cpp/hardware/support/fake_dma_preload.c @@ -87,6 +87,7 @@ static int g_release = 0; static int g_active_calls = 0; static int g_close_during_driver_call = 0; static int g_ret_index_count = 0; +static int g_tracking_overflow_count = 0; typedef int (*open_fn)(const char*, int, ...); typedef void* (*mmap_fn)(void*, size_t, int, int, int, off_t); @@ -116,7 +117,11 @@ static int is_our_fd_locked(int fd) { } static void add_fd_locked(int fd) { - if (g_fd_n < MAX_FDS) g_fds[g_fd_n++] = fd; + if (g_fd_n < MAX_FDS) { + g_fds[g_fd_n++] = fd; + } else { + g_tracking_overflow_count++; + } } static void del_fd_locked(int fd) { @@ -143,6 +148,8 @@ static void add_region_locked(void* addr, size_t len) { g_region_len[g_region_n] = len; g_region_n++; g_mapped_count++; + } else { + g_tracking_overflow_count++; } } @@ -412,6 +419,47 @@ int fakedma_mapped_count(void) { return n; } +int fakedma_fd_count(void) { + int n; + pthread_mutex_lock(&g_lock); + n = g_fd_n; + pthread_mutex_unlock(&g_lock); + return n; +} + +int fakedma_region_count(void) { + int n; + pthread_mutex_lock(&g_lock); + n = g_region_n; + pthread_mutex_unlock(&g_lock); + return n; +} + +int fakedma_active_call_count(void) { + int n; + pthread_mutex_lock(&g_lock); + n = g_active_calls; + pthread_mutex_unlock(&g_lock); + return n; +} + +int fakedma_tracking_overflow_count(void) { + int n; + pthread_mutex_lock(&g_lock); + n = g_tracking_overflow_count; + pthread_mutex_unlock(&g_lock); + return n; +} + +int fakedma_is_clean(void) { + int clean; + pthread_mutex_lock(&g_lock); + clean = (g_fd_n == 0) && (g_region_n == 0) && (g_active_calls == 0) && + (g_tracking_overflow_count == 0); + pthread_mutex_unlock(&g_lock); + return clean; +} + const char* fakedma_path(void) { return FAKE_PATH; } diff --git a/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp b/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp index 4164befde0..6dd8f486e5 100644 --- a/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp +++ b/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp @@ -15,7 +15,6 @@ * contained in the LICENSE.txt file. * ---------------------------------------------------------------------------- **/ -#include #include #include @@ -23,8 +22,6 @@ #include #include #include -#include -#include #include #include @@ -32,6 +29,7 @@ #include "rogue/GeneralError.h" #include "rogue/hardware/axi/AxiStreamDma.h" #include "rogue/interfaces/stream/Frame.h" +#include "support/fake_dma_hooks.h" #include "support/test_helpers.h" namespace rha = rogue::hardware::axi; @@ -39,74 +37,34 @@ namespace ris = rogue::interfaces::stream; namespace { -enum FakeDmaBlockOp { - FakeDmaBlockNone = 0, - FakeDmaBlockGetIndex = 1, - FakeDmaBlockRetIndex = 2, - FakeDmaBlockBuffSize = 3, - FakeDmaBlockWrite = 4, -}; - -class FakeDma { - public: - using PathFn = const char* (*)(); - using ResetFn = void (*)(); - using BlockNextFn = void (*)(int); - using WaitBlockedFn = int (*)(uint32_t); - using ReleaseBlockedFn = void (*)(); - using CountFn = int (*)(); - - FakeDma() { - path_ = load("fakedma_path"); - reset_ = load("fakedma_reset"); - blockNext_ = load("fakedma_block_next"); - waitBlocked_ = load("fakedma_wait_blocked"); - releaseBlocked_ = load("fakedma_release_blocked"); - mappedCount_ = load("fakedma_mapped_count"); - retIndexCount_ = load("fakedma_ret_index_count"); - closeDuringCall_ = load("fakedma_close_during_driver_call"); - } - - const char* path() const { return path_(); } - void reset() const { reset_(); } - void blockNext(FakeDmaBlockOp op) const { blockNext_(static_cast(op)); } - bool waitBlocked(uint32_t timeoutMs) const { return waitBlocked_(timeoutMs) != 0; } - void releaseBlocked() const { releaseBlocked_(); } - int mappedCount() const { return mappedCount_(); } - int retIndexCount() const { return retIndexCount_(); } - bool closeDuringDriverCall() const { return closeDuringCall_() != 0; } - - private: - template - Func load(const char* name) { - void* sym = dlsym(RTLD_DEFAULT, name); - if (sym == nullptr) throw std::runtime_error(dlerror()); - return reinterpret_cast(sym); - } +void requireFakeDmaClean() { + auto& fake = rogue_test::fakeDma(); + INFO("fd_count=" << fake.fdCount()); + INFO("region_count=" << fake.regionCount()); + INFO("active_calls=" << fake.activeCallCount()); + INFO("overflow_count=" << fake.overflowCount()); + REQUIRE(fake.isClean()); + REQUIRE_EQ(fake.mappedCount(), 0); +} - PathFn path_; - ResetFn reset_; - BlockNextFn blockNext_; - WaitBlockedFn waitBlocked_; - ReleaseBlockedFn releaseBlocked_; - CountFn mappedCount_; - CountFn retIndexCount_; - CountFn closeDuringCall_; -}; - -FakeDma& fakeDma() { - static FakeDma fake; - return fake; +void checkFakeDmaClean() { + auto& fake = rogue_test::fakeDma(); + INFO("fd_count=" << fake.fdCount()); + INFO("region_count=" << fake.regionCount()); + INFO("active_calls=" << fake.activeCallCount()); + INFO("overflow_count=" << fake.overflowCount()); + CHECK(fake.isClean()); + CHECK_EQ(fake.mappedCount(), 0); } rha::AxiStreamDmaPtr makeDma() { - return rha::AxiStreamDma::create(fakeDma().path(), 0, false); + return rha::AxiStreamDma::create(rogue_test::fakeDma().path(), 0, false); } void checkStopWaitsForBlockedDriverCall(const rha::AxiStreamDmaPtr& dma, - FakeDmaBlockOp blockOp, + rogue_test::FakeDmaBlockOp blockOp, const std::function& call) { - fakeDma().blockNext(blockOp); + rogue_test::fakeDma().blockNext(blockOp); std::atomic callDone{false}; std::exception_ptr callError; @@ -119,7 +77,7 @@ void checkStopWaitsForBlockedDriverCall(const rha::AxiStreamDmaPtr& dma, } }); - const bool blocked = fakeDma().waitBlocked(1000); + const bool blocked = rogue_test::fakeDma().waitBlocked(1000); CHECK(blocked); if (!blocked) { if (caller.joinable()) caller.join(); @@ -136,9 +94,9 @@ void checkStopWaitsForBlockedDriverCall(const rha::AxiStreamDmaPtr& dma, std::this_thread::sleep_for(std::chrono::milliseconds(25)); CHECK_FALSE(stopDone.load()); - CHECK_FALSE(fakeDma().closeDuringDriverCall()); + CHECK_FALSE(rogue_test::fakeDma().closeDuringDriverCall()); - fakeDma().releaseBlocked(); + rogue_test::fakeDma().releaseBlocked(); caller.join(); stopper.join(); @@ -146,14 +104,14 @@ void checkStopWaitsForBlockedDriverCall(const rha::AxiStreamDmaPtr& dma, CHECK(callDone.load()); CHECK(stopDone.load()); - CHECK_FALSE(fakeDma().closeDuringDriverCall()); + CHECK_FALSE(rogue_test::fakeDma().closeDuringDriverCall()); } } // namespace TEST_CASE("AxiStreamDma rejects new stream work after stop while retaining shared descriptor") { - fakeDma().reset(); - REQUIRE_EQ(fakeDma().mappedCount(), 0); + requireFakeDmaClean(); + rogue_test::fakeDma().reset(); auto dma = makeDma(); auto pool = rogue_test::makePool(); @@ -167,33 +125,33 @@ TEST_CASE("AxiStreamDma rejects new stream work after stop while retaining share CHECK_EQ(dma->getBuffSize(), 0U); dma.reset(); - CHECK_EQ(fakeDma().mappedCount(), 0); + checkFakeDmaClean(); } TEST_CASE("AxiStreamDma stop waits for an in-flight driver getter") { - fakeDma().reset(); - REQUIRE_EQ(fakeDma().mappedCount(), 0); + requireFakeDmaClean(); + rogue_test::fakeDma().reset(); auto dma = makeDma(); uint32_t value = 0; - checkStopWaitsForBlockedDriverCall(dma, FakeDmaBlockBuffSize, [&] { + checkStopWaitsForBlockedDriverCall(dma, rogue_test::FakeDmaBlockBuffSize, [&] { value = dma->getBuffSize(); }); CHECK_EQ(value, 4096U); dma.reset(); - CHECK_EQ(fakeDma().mappedCount(), 0); + checkFakeDmaClean(); } TEST_CASE("AxiStreamDma stop waits for in-flight zero-copy allocation and return") { - fakeDma().reset(); - REQUIRE_EQ(fakeDma().mappedCount(), 0); + requireFakeDmaClean(); + rogue_test::fakeDma().reset(); auto dma = makeDma(); ris::FramePtr frame; - checkStopWaitsForBlockedDriverCall(dma, FakeDmaBlockGetIndex, [&] { + checkStopWaitsForBlockedDriverCall(dma, rogue_test::FakeDmaBlockGetIndex, [&] { frame = dma->acceptReq(64, true); }); @@ -202,35 +160,35 @@ TEST_CASE("AxiStreamDma stop waits for in-flight zero-copy allocation and return frame->clear(); frame.reset(); dma.reset(); - CHECK_EQ(fakeDma().mappedCount(), 0); + checkFakeDmaClean(); - fakeDma().reset(); + rogue_test::fakeDma().reset(); dma = makeDma(); frame = dma->acceptReq(64, true); REQUIRE(static_cast(frame)); - checkStopWaitsForBlockedDriverCall(dma, FakeDmaBlockRetIndex, [&] { + checkStopWaitsForBlockedDriverCall(dma, rogue_test::FakeDmaBlockRetIndex, [&] { frame->clear(); }); - CHECK_EQ(fakeDma().retIndexCount(), 1); + CHECK_EQ(rogue_test::fakeDma().retIndexCount(), 1); frame.reset(); dma.reset(); - CHECK_EQ(fakeDma().mappedCount(), 0); + checkFakeDmaClean(); } TEST_CASE("AxiStreamDma stop waits for an in-flight transmit write") { - fakeDma().reset(); - REQUIRE_EQ(fakeDma().mappedCount(), 0); + requireFakeDmaClean(); + rogue_test::fakeDma().reset(); auto dma = makeDma(); auto pool = rogue_test::makePool(); auto frame = rogue_test::makeFrame(pool, std::vector{0x10, 0x20, 0x30, 0x40}); - checkStopWaitsForBlockedDriverCall(dma, FakeDmaBlockWrite, [&] { + checkStopWaitsForBlockedDriverCall(dma, rogue_test::FakeDmaBlockWrite, [&] { dma->acceptFrame(frame); }); dma.reset(); - CHECK_EQ(fakeDma().mappedCount(), 0); + checkFakeDmaClean(); } diff --git a/tests/cpp/hardware/test_axi_stream_dma_zerocopy_lifetime.cpp b/tests/cpp/hardware/test_axi_stream_dma_zerocopy_lifetime.cpp index 8f247915f2..270a5db7eb 100644 --- a/tests/cpp/hardware/test_axi_stream_dma_zerocopy_lifetime.cpp +++ b/tests/cpp/hardware/test_axi_stream_dma_zerocopy_lifetime.cpp @@ -29,37 +29,46 @@ * contained in the LICENSE.txt file. * ---------------------------------------------------------------------------- **/ -#include - #include "doctest/doctest.h" #include "rogue/hardware/axi/AxiStreamDma.h" #include "rogue/interfaces/stream/Frame.h" +#include "support/fake_dma_hooks.h" namespace rha = rogue::hardware::axi; namespace { -// Sentinel path recognized by the fake_dma_preload LD_PRELOAD shim. -const char* const kFakePath = "/tmp/rogue-fake-datadev"; - -typedef int (*MappedCountFn)(void); +void requireFakeDmaClean() { + auto& fake = rogue_test::fakeDma(); + INFO("fd_count=" << fake.fdCount()); + INFO("region_count=" << fake.regionCount()); + INFO("active_calls=" << fake.activeCallCount()); + INFO("overflow_count=" << fake.overflowCount()); + REQUIRE(fake.isClean()); + REQUIRE_EQ(fake.mappedCount(), 0); +} -// Resolve the shim's query hook. Non-null only when the shim is preloaded. -MappedCountFn mappedCountHook() { - return reinterpret_cast(dlsym(RTLD_DEFAULT, "fakedma_mapped_count")); +void checkFakeDmaClean() { + auto& fake = rogue_test::fakeDma(); + INFO("fd_count=" << fake.fdCount()); + INFO("region_count=" << fake.regionCount()); + INFO("active_calls=" << fake.activeCallCount()); + INFO("overflow_count=" << fake.overflowCount()); + CHECK(fake.isClean()); + CHECK_EQ(fake.mappedCount(), 0); } } // namespace TEST_CASE("AxiStreamDma keeps the shared DMA mapping alive across stop()") { - MappedCountFn mappedCount = mappedCountHook(); - REQUIRE(mappedCount != nullptr); // fake_dma_preload shim must be LD_PRELOAD'd - REQUIRE(mappedCount() == 0); + requireFakeDmaClean(); + auto& fake = rogue_test::fakeDma(); + fake.reset(); - auto dma = rha::AxiStreamDma::create(kFakePath, 0, false); + auto dma = rha::AxiStreamDma::create(fake.path(), 0, false); // dmaMapDma() maps all shared zero-copy buffers at construction. - const int mappedAtCtor = mappedCount(); + const int mappedAtCtor = fake.mappedCount(); REQUIRE(mappedAtCtor > 0); // Obtain a zero-copy frame and retain it past stop(); its buffer points @@ -73,14 +82,14 @@ TEST_CASE("AxiStreamDma keeps the shared DMA mapping alive across stop()") { // The fix keeps the shared mapping valid after stop() so the retained frame // does not dangle. Reverted code releases it inside stop(), dropping the // count to zero here. - CHECK_EQ(mappedCount(), mappedAtCtor); + CHECK_EQ(fake.mappedCount(), mappedAtCtor); // Returning the buffer post-stop() is a no-op on the driver (fd_ == -1); it // must not touch the still-mapped memory or throw. frame.reset(); - CHECK_EQ(mappedCount(), mappedAtCtor); + CHECK_EQ(fake.mappedCount(), mappedAtCtor); // Destruction is the single point that releases the shared mapping. dma.reset(); - CHECK_EQ(mappedCount(), 0); + checkFakeDmaClean(); } From e30812114e7e6c338d8ab204d2ec2f61aa1527e8 Mon Sep 17 00:00:00 2001 From: Benjamin Reese Date: Mon, 6 Jul 2026 15:19:32 -0700 Subject: [PATCH 8/9] fix(fake_dma_hooks): clear previous dlerror before loading symbol --- tests/cpp/hardware/support/fake_dma_hooks.h | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/cpp/hardware/support/fake_dma_hooks.h b/tests/cpp/hardware/support/fake_dma_hooks.h index 6379b59a31..7e7ede3e6f 100644 --- a/tests/cpp/hardware/support/fake_dma_hooks.h +++ b/tests/cpp/hardware/support/fake_dma_hooks.h @@ -74,6 +74,7 @@ class FakeDma { private: template Func load(const char* name) { + dlerror(); void* sym = dlsym(RTLD_DEFAULT, name); if (sym == nullptr) { const char* err = dlerror(); From 2f5ce0853d7e76051e710f6a00f1af4bc1fedc0a Mon Sep 17 00:00:00 2001 From: Benjamin Reese Date: Mon, 20 Jul 2026 16:05:45 -0700 Subject: [PATCH 9/9] refactor(axi): simplify fd management and enhance stop lifecycle handling --- include/rogue/hardware/axi/AxiStreamDma.h | 57 +---- src/rogue/hardware/axi/AxiStreamDma.cpp | 241 ++++++++---------- tests/cpp/hardware/support/fake_dma_hooks.h | 36 +-- tests/cpp/hardware/support/fake_dma_preload.c | 73 ++---- .../test_axi_stream_dma_lifecycle.cpp | 146 +++-------- 5 files changed, 189 insertions(+), 364 deletions(-) diff --git a/include/rogue/hardware/axi/AxiStreamDma.h b/include/rogue/hardware/axi/AxiStreamDma.h index c8b0a58746..48d7f63687 100644 --- a/include/rogue/hardware/axi/AxiStreamDma.h +++ b/include/rogue/hardware/axi/AxiStreamDma.h @@ -24,7 +24,6 @@ #include #include #include -#include #include #include @@ -86,6 +85,9 @@ typedef std::shared_ptr AxiStreamDmaSh * - A background RX thread is started in the constructor and runs until `stop()` * or destruction. * - TX operations execute synchronously in caller context of `acceptFrame()`. + * - Callers must quiesce public driver operations before calling `stop()`. + * Deferred zero-copy buffer returns may overlap `stop()` and are serialized + * against descriptor closure internally. * * Zero-copy model: * - Enabled by default per device path. @@ -106,11 +108,8 @@ class AxiStreamDma : public rogue::interfaces::stream::Master, public rogue::int // Process-local descriptor for TX/RX operations and dest mask programming. int32_t fd_; - // Set once stop() starts; public stream operations reject new work after it. - std::atomic stopped_{false}; - - // Serializes fd_ close against active driver calls and deferred zero-copy returns. - std::shared_timed_mutex fdMtx_; + // Serializes fd_ close against deferred zero-copy buffer returns. + std::mutex fdMtx_; // Destination selector used when transmitting frames. uint32_t dest_; @@ -147,47 +146,6 @@ class AxiStreamDma : public rogue::interfaces::stream::Master, public rogue::int // Closes shared DMA mapping state when last user exits. static void closeShared(std::shared_ptr); - // RAII guard that pins fd_ open while a driver call runs. - class FdGuard { - std::shared_lock lock_; - int32_t fd_ = -1; - - public: - FdGuard(rogue::hardware::axi::AxiStreamDma* owner, - const char* context, - bool throwOnStopped, - bool allowStopped = false); - ~FdGuard() = default; - FdGuard(const FdGuard&) = delete; - FdGuard& operator=(const FdGuard&) = delete; - - int32_t fd() const { return fd_; } - bool valid() const { return fd_ >= 0; } - }; - - // Runs a void driver operation while fd_ is pinned open. - template - void withFd(const char* context, bool throwOnStopped, Func func, bool allowStopped = false) { - FdGuard fd(this, context, throwOnStopped, allowStopped); - if (fd.valid()) func(fd.fd()); - } - - // Runs a value-returning driver operation while fd_ is pinned open. - template - Result withFd(const char* context, bool throwOnStopped, Result stoppedValue, Func func, bool allowStopped = false) { - FdGuard fd(this, context, throwOnStopped, allowStopped); - if (!fd.valid()) return stoppedValue; - return func(fd.fd()); - } - - template - uint32_t readFdValue(const char* context, Func func) { - return withFd(context, false, 0, [func](int32_t fd) { - auto value = func(fd); - return (value < 0) ? 0 : static_cast(value); - }); - } - public: /** * @brief Creates an AXI Stream DMA bridge instance. @@ -260,6 +218,11 @@ class AxiStreamDma : public rogue::interfaces::stream::Master, public rogue::int * The shared zero-copy DMA mapping remains valid until destruction so * downstream Rogue buffers retained after `stop()` do not reference * unmapped memory. + * + * Before calling `stop()`, callers must ensure that stream operations and + * driver controls/accessors have completed and that no new ones can begin. + * Deferred returns from previously issued zero-copy buffers may overlap + * `stop()`; descriptor closure is synchronized with those callbacks. */ void stop(); diff --git a/src/rogue/hardware/axi/AxiStreamDma.cpp b/src/rogue/hardware/axi/AxiStreamDma.cpp index c6c1191694..c2d0dc6295 100644 --- a/src/rogue/hardware/axi/AxiStreamDma.cpp +++ b/src/rogue/hardware/axi/AxiStreamDma.cpp @@ -49,30 +49,6 @@ std::map > rha::AxiStreamD // Protect sharedBuffers_ against concurrent AxiStreamDma construction. static std::mutex sharedBuffersMtx; -static void throwStopped(const char* context) { - throw rogue::GeneralError(context, "instance has been stopped or did not finish construction"); -} - -rha::AxiStreamDma::FdGuard::FdGuard(rha::AxiStreamDma* owner, - const char* context, - bool throwOnStopped, - bool allowStopped) { - if (!allowStopped && owner->stopped_.load()) { - if (throwOnStopped) throwStopped(context); - return; - } - - lock_ = std::shared_lock(owner->fdMtx_); - - if ((!allowStopped && owner->stopped_.load()) || owner->fd_ < 0) { - lock_.unlock(); - if (throwOnStopped) throwStopped(context); - return; - } - - fd_ = owner->fd_; -} - rha::AxiStreamDmaShared::AxiStreamDmaShared(std::string path) { this->fd = -1; this->path = path; @@ -328,14 +304,12 @@ rha::AxiStreamDma::~AxiStreamDma() { void rha::AxiStreamDma::stop() { rogue::GilRelease noGil; - stopped_.store(true); - // 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 exit on its - // own (for example after a poll fd error) by setting threadEn_ = false and + // 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 on a still-joinable thread would call // std::terminate(). Joining via joinable() handles both that early-exit @@ -349,7 +323,7 @@ void rha::AxiStreamDma::stop() { // open until ~AxiStreamDma() because zero-copy Rogue buffers can outlive a // user-initiated stop() and still point into desc_->rawBuff. { - std::unique_lock lock(fdMtx_); + std::lock_guard lock(fdMtx_); if (fd_ >= 0) { ::close(fd_); fd_ = -1; @@ -368,12 +342,12 @@ void rha::AxiStreamDma::setTimeout(uint32_t timeout) { //! Set driver debug level void rha::AxiStreamDma::setDriverDebug(uint32_t level) { - withFd("AxiStreamDma::setDriverDebug", false, [level](int32_t fd) { dmaSetDebug(fd, level); }); + dmaSetDebug(fd_, level); } //! Strobe ack line void rha::AxiStreamDma::dmaAck() { - withFd("AxiStreamDma::dmaAck", false, [](int32_t fd) { axisReadAck(fd); }); + if (fd_ >= 0) axisReadAck(fd_); } //! Generate a buffer. Called from master @@ -387,7 +361,9 @@ ris::FramePtr rha::AxiStreamDma::acceptReq(uint32_t size, bool zeroCopyEn) { // Reject use after stop()/teardown. stop() closes the per-instance fd, // and destruction releases the shared descriptor. - if (!desc_ || stopped_.load()) throwStopped("AxiStreamDma::acceptReq"); + if (!desc_ || fd_ < 0) + throw rogue::GeneralError("AxiStreamDma::acceptReq", + "instance has been stopped or did not finish construction"); //! Adjust allocation size if (size > desc_->bSize) @@ -412,35 +388,38 @@ ris::FramePtr rha::AxiStreamDma::acceptReq(uint32_t size, bool zeroCopyEn) { // Keep trying since poll call can fire // but getIndex fails because we did not win the buffer lock do { - res = withFd("AxiStreamDma::acceptReq", true, -1, [&](int32_t 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)); - if (stopped_.load()) throwStopped("AxiStreamDma::acceptReq"); - - // 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); - return -1; - } else { - // Attempt to get index. - // return of less than 0 is a failure to get a buffer - return dmaGetIndex(fd); - } - }); + // Re-check fd validity before using it in poll() and the driver call. + // 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 @@ -462,11 +441,11 @@ void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) { uint32_t cont; bool emptyFrame; - // Reject use after stop()/teardown. 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_ || stopped_.load()) throwStopped("AxiStreamDma::acceptFrame"); + // Reject use after stop()/teardown before locking or iterating the frame, + // 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(); @@ -511,16 +490,14 @@ void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) { // Buffer is not already stale as indicates by bit 30 if ((meta & 0x40000000) == 0) { - withFd("AxiStreamDma::acceptFrame", true, [&](int32_t fd) { - // 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")); - } - }); + // 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; @@ -532,39 +509,41 @@ void rha::AxiStreamDma::acceptFrame(ris::FramePtr frame) { // Keep trying since poll call can fire // but write fails because we did not win the (*it)er lock do { - res = withFd("AxiStreamDma::acceptFrame", true, 0, [&](int32_t 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)); - if (stopped_.load()) throwStopped("AxiStreamDma::acceptFrame"); - - // 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); - return 0; - } else { - // Write with (*it)er copy - ssize_t writeSize = dmaWrite( - fd, (*it)->begin(), (*it)->getPayload(), axisSetFlags(fuser, luser, 0), dest_); - if (writeSize < 0) { - throw(rogue::GeneralError("AxiStreamDma::acceptFrame", "AXIS Write Call Failed!!!!")); - } - return static_cast(writeSize); + // Re-check fd validity before using it in poll() and the driver call. + // 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 } } @@ -581,14 +560,9 @@ void rha::AxiStreamDma::retBuffer(uint8_t* data, uint32_t meta, uint32_t size) { // Device is open and buffer is not stale // Bit 30 indicates buffer has already been returned to hardware if ((meta & 0x40000000) == 0) { - withFd( - "AxiStreamDma::retBuffer", - false, - [meta](int32_t fd) { - if (dmaRetIndex(fd, meta & 0x3FFFFFFF) < 0) - throw(rogue::GeneralError("AxiStreamDma::retBuffer", "AXIS Return Buffer Call Failed!!!!")); - }, - true); + std::lock_guard lock(fdMtx_); + if (fd_ >= 0 && dmaRetIndex(fd_, meta & 0x3FFFFFFF) < 0) + throw(rogue::GeneralError("AxiStreamDma::retBuffer", "AXIS Return Buffer Call Failed!!!!")); } decCounter(size); @@ -628,8 +602,9 @@ void rha::AxiStreamDma::runThread(std::weak_ptr lockPtr) { while (threadEn_) { // runThread() has no top-level catch, so a throw here would call - // std::terminate. Log and exit cleanly if the descriptor was never - // opened or has already been closed by an earlier stop() 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_); @@ -715,77 +690,77 @@ void rha::AxiStreamDma::runThread(std::weak_ptr lockPtr) { //! Get the DMA Driver's Git Version std::string rha::AxiStreamDma::getGitVersion() { - return withFd("AxiStreamDma::getGitVersion", false, std::string(), dmaGetGitVersion); + return dmaGetGitVersion(fd_); } //! Get the DMA Driver's API Version uint32_t rha::AxiStreamDma::getApiVersion() { - return readFdValue("AxiStreamDma::getApiVersion", dmaGetApiVersion); + return dmaGetApiVersion(fd_); } //! Get the size of buffers (RX/TX) uint32_t rha::AxiStreamDma::getBuffSize() { - return readFdValue("AxiStreamDma::getBuffSize", dmaGetBuffSize); + return dmaGetBuffSize(fd_); } //! Get the number of RX buffers uint32_t rha::AxiStreamDma::getRxBuffCount() { - return readFdValue("AxiStreamDma::getRxBuffCount", dmaGetRxBuffCount); + return dmaGetRxBuffCount(fd_); } //! Get RX buffer in User count uint32_t rha::AxiStreamDma::getRxBuffinUserCount() { - return readFdValue("AxiStreamDma::getRxBuffinUserCount", dmaGetRxBuffinUserCount); + return dmaGetRxBuffinUserCount(fd_); } //! Get RX buffer in HW count uint32_t rha::AxiStreamDma::getRxBuffinHwCount() { - return readFdValue("AxiStreamDma::getRxBuffinHwCount", dmaGetRxBuffinHwCount); + return dmaGetRxBuffinHwCount(fd_); } //! Get RX buffer in Pre-HW Queue count uint32_t rha::AxiStreamDma::getRxBuffinPreHwQCount() { - return readFdValue("AxiStreamDma::getRxBuffinPreHwQCount", dmaGetRxBuffinPreHwQCount); + return dmaGetRxBuffinPreHwQCount(fd_); } //! Get RX buffer in SW Queue count uint32_t rha::AxiStreamDma::getRxBuffinSwQCount() { - return readFdValue("AxiStreamDma::getRxBuffinSwQCount", dmaGetRxBuffinSwQCount); + return dmaGetRxBuffinSwQCount(fd_); } //! Get RX buffer missing count uint32_t rha::AxiStreamDma::getRxBuffMissCount() { - return readFdValue("AxiStreamDma::getRxBuffMissCount", dmaGetRxBuffMissCount); + return dmaGetRxBuffMissCount(fd_); } //! Get the number of TX buffers uint32_t rha::AxiStreamDma::getTxBuffCount() { - return readFdValue("AxiStreamDma::getTxBuffCount", dmaGetTxBuffCount); + return dmaGetTxBuffCount(fd_); } //! Get TX buffer in User count uint32_t rha::AxiStreamDma::getTxBuffinUserCount() { - return readFdValue("AxiStreamDma::getTxBuffinUserCount", dmaGetTxBuffinUserCount); + return dmaGetTxBuffinUserCount(fd_); } //! Get TX buffer in HW count uint32_t rha::AxiStreamDma::getTxBuffinHwCount() { - return readFdValue("AxiStreamDma::getTxBuffinHwCount", dmaGetTxBuffinHwCount); + return dmaGetTxBuffinHwCount(fd_); } //! Get TX buffer in Pre-HW Queue count uint32_t rha::AxiStreamDma::getTxBuffinPreHwQCount() { - return readFdValue("AxiStreamDma::getTxBuffinPreHwQCount", dmaGetTxBuffinPreHwQCount); + return dmaGetTxBuffinPreHwQCount(fd_); } //! Get TX buffer in SW Queue count uint32_t rha::AxiStreamDma::getTxBuffinSwQCount() { - return readFdValue("AxiStreamDma::getTxBuffinSwQCount", dmaGetTxBuffinSwQCount); + return dmaGetTxBuffinSwQCount(fd_); } //! Get TX buffer missing count uint32_t rha::AxiStreamDma::getTxBuffMissCount() { - return readFdValue("AxiStreamDma::getTxBuffMissCount", dmaGetTxBuffMissCount); + return dmaGetTxBuffMissCount(fd_); } void rha::AxiStreamDma::setup_python() { diff --git a/tests/cpp/hardware/support/fake_dma_hooks.h b/tests/cpp/hardware/support/fake_dma_hooks.h index 7e7ede3e6f..368afce5ea 100644 --- a/tests/cpp/hardware/support/fake_dma_hooks.h +++ b/tests/cpp/hardware/support/fake_dma_hooks.h @@ -24,42 +24,34 @@ namespace rogue_test { -enum FakeDmaBlockOp { - FakeDmaBlockNone = 0, - FakeDmaBlockGetIndex = 1, - FakeDmaBlockRetIndex = 2, - FakeDmaBlockBuffSize = 3, - FakeDmaBlockWrite = 4, -}; - class FakeDma { public: using PathFn = const char* (*)(); using ResetFn = void (*)(); - using BlockNextFn = void (*)(int); + using BlockNextFn = void (*)(); using WaitBlockedFn = int (*)(uint32_t); using ReleaseFn = void (*)(); using CountFn = int (*)(); FakeDma() { - path_ = load("fakedma_path"); - reset_ = load("fakedma_reset"); - blockNext_ = load("fakedma_block_next"); - waitBlocked_ = load("fakedma_wait_blocked"); - release_ = load("fakedma_release_blocked"); - mappedCount_ = load("fakedma_mapped_count"); - fdCount_ = load("fakedma_fd_count"); - regionCount_ = load("fakedma_region_count"); - activeCalls_ = load("fakedma_active_call_count"); - overflowCount_ = load("fakedma_tracking_overflow_count"); - isClean_ = load("fakedma_is_clean"); + path_ = load("fakedma_path"); + reset_ = load("fakedma_reset"); + blockNext_ = load("fakedma_block_next_ret_index"); + waitBlocked_ = load("fakedma_wait_blocked"); + release_ = load("fakedma_release_blocked"); + mappedCount_ = load("fakedma_mapped_count"); + fdCount_ = load("fakedma_fd_count"); + regionCount_ = load("fakedma_region_count"); + activeCalls_ = load("fakedma_active_call_count"); + overflowCount_ = load("fakedma_tracking_overflow_count"); + isClean_ = load("fakedma_is_clean"); retIndexCount_ = load("fakedma_ret_index_count"); - closeDuringCall_ = load("fakedma_close_during_driver_call"); + closeDuringCall_ = load("fakedma_close_during_driver_call"); } const char* path() const { return path_(); } void reset() const { reset_(); } - void blockNext(FakeDmaBlockOp op) const { blockNext_(static_cast(op)); } + void blockNextRetIndex() const { blockNext_(); } bool waitBlocked(uint32_t timeoutMs) const { return waitBlocked_(timeoutMs) != 0; } void releaseBlocked() const { release_(); } int mappedCount() const { return mappedCount_(); } diff --git a/tests/cpp/hardware/support/fake_dma_preload.c b/tests/cpp/hardware/support/fake_dma_preload.c index 8e27269b7b..e83ee6e885 100644 --- a/tests/cpp/hardware/support/fake_dma_preload.c +++ b/tests/cpp/hardware/support/fake_dma_preload.c @@ -6,10 +6,11 @@ * Test-only LD_PRELOAD shim that emulates just enough of the aes-stream-driver * character device for AxiStreamDma to construct, hand out a zero-copy buffer, * and tear down on a machine with no DMA hardware. It intercepts - * open/close/ioctl/mmap/munmap/poll/write for a single sentinel device path and + * open/close/ioctl/mmap/munmap/poll for a single sentinel device path and * backs the "DMA" buffers with ordinary anonymous mappings, tracking how many * are currently mapped. Tests query that count via fakedma_mapped_count() and - * use the blocking hooks below to exercise stop() races deterministically. + * use the blocking hooks below to exercise deferred buffer returns overlapping + * stop() deterministically. * * This object is built only under -DROGUE_BUILD_TESTS=ON and is never linked * into rogue-core or shipped. @@ -60,12 +61,6 @@ #define FAKE_BUFF_SIZE 4096u #define FAKE_BUFF_COUNT 8u -#define FAKE_BLOCK_NONE 0 -#define FAKE_BLOCK_GET_INDEX 1 -#define FAKE_BLOCK_RET_INDEX 2 -#define FAKE_BLOCK_BUFF_SIZE 3 -#define FAKE_BLOCK_WRITE 4 - #define MAX_FDS 16 #define MAX_REGIONS 64 @@ -79,14 +74,14 @@ static void* g_region_addr[MAX_REGIONS]; static size_t g_region_len[MAX_REGIONS]; static int g_region_n = 0; -static int g_mapped_count = 0; -static uint32_t g_next_index = 0; -static int g_block_next = FAKE_BLOCK_NONE; -static int g_blocked = 0; -static int g_release = 0; -static int g_active_calls = 0; +static int g_mapped_count = 0; +static uint32_t g_next_index = 0; +static int g_block_ret_index = 0; +static int g_blocked = 0; +static int g_release = 0; +static int g_active_calls = 0; static int g_close_during_driver_call = 0; -static int g_ret_index_count = 0; +static int g_ret_index_count = 0; static int g_tracking_overflow_count = 0; typedef int (*open_fn)(const char*, int, ...); @@ -95,7 +90,6 @@ typedef int (*munmap_fn)(void*, size_t); typedef int (*close_fn)(int); typedef int (*ioctl_fn)(int, unsigned long, ...); typedef int (*poll_fn)(struct pollfd*, nfds_t, int); -typedef ssize_t (*write_fn)(int, const void*, size_t); static open_fn real_open; static open_fn real_open64; @@ -105,7 +99,6 @@ static munmap_fn real_munmap; static close_fn real_close; static ioctl_fn real_ioctl; static poll_fn real_poll; -static write_fn real_write; /* ------- bookkeeping helpers (caller holds g_lock) ------- */ @@ -191,12 +184,11 @@ static void driver_call_end(void) { pthread_mutex_unlock(&g_lock); } -static void maybe_block(int op) { +static void maybe_block_ret_index(void) { pthread_mutex_lock(&g_lock); - if (g_block_next == op) { - g_block_next = FAKE_BLOCK_NONE; - g_blocked = 1; - g_release = 0; + if (g_block_ret_index) { + g_block_ret_index = 0; + g_blocked = 1; pthread_cond_broadcast(&g_cond); while (!g_release) pthread_cond_wait(&g_cond, &g_lock); g_blocked = 0; @@ -334,7 +326,6 @@ int ioctl(int fd, unsigned long request, ...) { driver_call_end(); return FAKE_DMA_VERSION; case FAKE_DMA_Get_Buff_Size: - maybe_block(FAKE_BLOCK_BUFF_SIZE); driver_call_end(); return (int)FAKE_BUFF_SIZE; case FAKE_DMA_Get_Buff_Count: @@ -351,7 +342,6 @@ int ioctl(int fd, unsigned long request, ...) { return 0; case FAKE_DMA_Get_Index: { int idx; - maybe_block(FAKE_BLOCK_GET_INDEX); pthread_mutex_lock(&g_lock); idx = (int)(g_next_index++ % FAKE_BUFF_COUNT); pthread_mutex_unlock(&g_lock); @@ -359,7 +349,7 @@ int ioctl(int fd, unsigned long request, ...) { return idx; } case FAKE_DMA_Ret_Index: - maybe_block(FAKE_BLOCK_RET_INDEX); + maybe_block_ret_index(); pthread_mutex_lock(&g_lock); g_ret_index_count++; pthread_mutex_unlock(&g_lock); @@ -397,19 +387,6 @@ int poll(struct pollfd* fds, nfds_t nfds, int timeout) { return real_poll(fds, nfds, timeout); } -ssize_t write(int fd, const void* buf, size_t count) { - int ours; - if (!real_write) real_write = (write_fn)dlsym(RTLD_NEXT, "write"); - - ours = is_our_fd(fd); - if (!ours) return real_write(fd, buf, count); - - driver_call_begin(); - maybe_block(FAKE_BLOCK_WRITE); - driver_call_end(); - return (ssize_t)count; -} - /* Query hook for the test: number of fake DMA buffers currently mapped. */ int fakedma_mapped_count(void) { int n; @@ -466,22 +443,22 @@ const char* fakedma_path(void) { void fakedma_reset(void) { pthread_mutex_lock(&g_lock); - g_block_next = FAKE_BLOCK_NONE; - g_blocked = 0; - g_release = 0; - g_active_calls = 0; + g_block_ret_index = 0; + g_blocked = 0; + g_release = 0; + g_active_calls = 0; g_close_during_driver_call = 0; - g_ret_index_count = 0; - g_next_index = 0; + g_ret_index_count = 0; + g_next_index = 0; pthread_cond_broadcast(&g_cond); pthread_mutex_unlock(&g_lock); } -void fakedma_block_next(int op) { +void fakedma_block_next_ret_index(void) { pthread_mutex_lock(&g_lock); - g_block_next = op; - g_blocked = 0; - g_release = 0; + g_block_ret_index = 1; + g_blocked = 0; + g_release = 0; pthread_mutex_unlock(&g_lock); } diff --git a/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp b/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp index 6dd8f486e5..58a3b897e1 100644 --- a/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp +++ b/tests/cpp/hardware/test_axi_stream_dma_lifecycle.cpp @@ -3,8 +3,8 @@ * Company : SLAC National Accelerator Laboratory * ---------------------------------------------------------------------------- * Description: - * Native C++ lifecycle tests for AxiStreamDma stop/fd races using the - * fake_dma_preload LD_PRELOAD backend. + * Native C++ lifecycle test for deferred AxiStreamDma zero-copy buffer + * returns overlapping stop(), using the fake_dma_preload LD_PRELOAD backend. * ---------------------------------------------------------------------------- * This file is part of the rogue software platform. It is subject to * the license terms in the LICENSE.txt file found in the top-level directory @@ -20,20 +20,14 @@ #include #include #include -#include -#include #include -#include #include "doctest/doctest.h" -#include "rogue/GeneralError.h" #include "rogue/hardware/axi/AxiStreamDma.h" #include "rogue/interfaces/stream/Frame.h" #include "support/fake_dma_hooks.h" -#include "support/test_helpers.h" namespace rha = rogue::hardware::axi; -namespace ris = rogue::interfaces::stream; namespace { @@ -57,34 +51,40 @@ void checkFakeDmaClean() { CHECK_EQ(fake.mappedCount(), 0); } -rha::AxiStreamDmaPtr makeDma() { - return rha::AxiStreamDma::create(rogue_test::fakeDma().path(), 0, false); -} +} // namespace + +TEST_CASE("AxiStreamDma stop waits for an in-flight zero-copy buffer return") { + requireFakeDmaClean(); + auto& fake = rogue_test::fakeDma(); + fake.reset(); + + auto dma = rha::AxiStreamDma::create(fake.path(), 0, false); + auto frame = dma->acceptReq(64, true); + REQUIRE(static_cast(frame)); + REQUIRE_EQ(frame->bufferCount(), 1U); -void checkStopWaitsForBlockedDriverCall(const rha::AxiStreamDmaPtr& dma, - rogue_test::FakeDmaBlockOp blockOp, - const std::function& call) { - rogue_test::fakeDma().blockNext(blockOp); + fake.blockNextRetIndex(); - std::atomic callDone{false}; - std::exception_ptr callError; - std::thread caller([&] { + std::atomic returnDone{false}; + std::exception_ptr returnError; + std::thread returner([&] { try { - call(); - callDone.store(true); + frame->clear(); + returnDone.store(true); } catch (...) { - callError = std::current_exception(); + returnError = std::current_exception(); } }); - const bool blocked = rogue_test::fakeDma().waitBlocked(1000); + const bool blocked = fake.waitBlocked(1000); CHECK(blocked); if (!blocked) { - if (caller.joinable()) caller.join(); - if (callError) std::rethrow_exception(callError); + fake.releaseBlocked(); + returner.join(); + if (returnError) std::rethrow_exception(returnError); return; } - CHECK_FALSE(callDone.load()); + CHECK_FALSE(returnDone.load()); std::atomic stopDone{false}; std::thread stopper([&] { @@ -94,101 +94,19 @@ void checkStopWaitsForBlockedDriverCall(const rha::AxiStreamDmaPtr& dma, std::this_thread::sleep_for(std::chrono::milliseconds(25)); CHECK_FALSE(stopDone.load()); - CHECK_FALSE(rogue_test::fakeDma().closeDuringDriverCall()); - - rogue_test::fakeDma().releaseBlocked(); + CHECK_FALSE(fake.closeDuringDriverCall()); - caller.join(); + fake.releaseBlocked(); + returner.join(); stopper.join(); - if (callError) std::rethrow_exception(callError); + if (returnError) std::rethrow_exception(returnError); - CHECK(callDone.load()); + CHECK(returnDone.load()); CHECK(stopDone.load()); - CHECK_FALSE(rogue_test::fakeDma().closeDuringDriverCall()); -} - -} // namespace - -TEST_CASE("AxiStreamDma rejects new stream work after stop while retaining shared descriptor") { - requireFakeDmaClean(); - rogue_test::fakeDma().reset(); - - auto dma = makeDma(); - auto pool = rogue_test::makePool(); - auto tx = rogue_test::makeFrame(pool, {1, 2, 3, 4}); - - dma->stop(); - - CHECK_THROWS_AS(dma->acceptReq(64, false), rogue::GeneralError); - CHECK_THROWS_AS(dma->acceptReq(64, true), rogue::GeneralError); - CHECK_THROWS_AS(dma->acceptFrame(tx), rogue::GeneralError); - CHECK_EQ(dma->getBuffSize(), 0U); - - dma.reset(); - checkFakeDmaClean(); -} - -TEST_CASE("AxiStreamDma stop waits for an in-flight driver getter") { - requireFakeDmaClean(); - rogue_test::fakeDma().reset(); - - auto dma = makeDma(); - - uint32_t value = 0; - checkStopWaitsForBlockedDriverCall(dma, rogue_test::FakeDmaBlockBuffSize, [&] { - value = dma->getBuffSize(); - }); - - CHECK_EQ(value, 4096U); - dma.reset(); - checkFakeDmaClean(); -} - -TEST_CASE("AxiStreamDma stop waits for in-flight zero-copy allocation and return") { - requireFakeDmaClean(); - rogue_test::fakeDma().reset(); - - auto dma = makeDma(); - - ris::FramePtr frame; - checkStopWaitsForBlockedDriverCall(dma, rogue_test::FakeDmaBlockGetIndex, [&] { - frame = dma->acceptReq(64, true); - }); + CHECK_EQ(fake.retIndexCount(), 1); + CHECK_FALSE(fake.closeDuringDriverCall()); - REQUIRE(static_cast(frame)); - REQUIRE_EQ(frame->bufferCount(), 1U); - frame->clear(); frame.reset(); dma.reset(); checkFakeDmaClean(); - - rogue_test::fakeDma().reset(); - dma = makeDma(); - frame = dma->acceptReq(64, true); - REQUIRE(static_cast(frame)); - - checkStopWaitsForBlockedDriverCall(dma, rogue_test::FakeDmaBlockRetIndex, [&] { - frame->clear(); - }); - - CHECK_EQ(rogue_test::fakeDma().retIndexCount(), 1); - frame.reset(); - dma.reset(); - checkFakeDmaClean(); -} - -TEST_CASE("AxiStreamDma stop waits for an in-flight transmit write") { - requireFakeDmaClean(); - rogue_test::fakeDma().reset(); - - auto dma = makeDma(); - auto pool = rogue_test::makePool(); - auto frame = rogue_test::makeFrame(pool, std::vector{0x10, 0x20, 0x30, 0x40}); - - checkStopWaitsForBlockedDriverCall(dma, rogue_test::FakeDmaBlockWrite, [&] { - dma->acceptFrame(frame); - }); - - dma.reset(); - checkFakeDmaClean(); }