uhallibs::Flx Class Reference

Transport protocol to transfer an IPbus buffer via device file, using mmap. More...

#include <ProtocolFlx.hpp>

Inheritance diagram for uhallibs::Flx:

Collaboration diagram for uhallibs::Flx:

Classes
class	Card
struct	HexTo

Public Member Functions
	Flx (const std::string &aId, const uhal::URI &aUri)
virtual	~Flx ()
	Destructor.

Private Types
typedef ipc::RobustMutex	IPCMutex_t
typedef std::unique_lock< IPCMutex_t >	IPCScopedLock_t
typedef IPbus< 2, 0 >	InnerProtocol
typedef std::chrono::steady_clock	SteadyClock_t

Private Member Functions
	Flx (const Flx &aFlx)
Flx &	operator= (const Flx &aFlx)
void	implementDispatch (std::shared_ptr< uhal::Buffers > aBuffers)
virtual void	Flush ()
virtual void	dispatchExceptionHandler ()
	Function which tidies up this protocol layer in the event of an exception.
uint32_t	getMaxSendSize ()
uint32_t	getMaxReplySize ()
void	connect ()
	Set up the connection to the device.
void	connect (IPCScopedLock_t &)
	Set up the connection to the device.
void	disconnect ()
	Close the connection to the device.
void	write (const std::shared_ptr< uhal::Buffers > &aBuffers)
	Write request packet to next page in host-to-FPGA device file.
void	read ()
	Read next pending reply packet from appropriate page of FPGA-to-host device file, and validate contents.

Static Private Member Functions
static std::string	getSharedMemName (const std::string &aPath)

Private Attributes
bool	mConnected
Card	mDeviceFile
ipc::SharedMemObject< IPCMutex_t >	mIPCMutex
bool	mIPCExternalSessionActive
uint64_t	mIPCSessionCount
std::chrono::microseconds	mSleepDuration
uint32_t	mNumberOfPages
uint32_t	mPageSize
uint32_t	mIndexNextPage
uint32_t	mPublishedReplyPageCount
uint32_t	mReadReplyPageCount
std::deque< std::shared_ptr< uhal::Buffers > >	mReplyQueue
	The list of buffers still awaiting a reply.
uhal::exception::exception *	mAsynchronousException

Detailed Description

Transport protocol to transfer an IPbus buffer via device file, using mmap.

Definition at line 97 of file ProtocolFlx.hpp.

Member Typedef Documentation

InnerProtocol

typedef IPbus<2, 0> uhallibs::Flx::InnerProtocol

private

Definition at line 186 of file ProtocolFlx.hpp.

IPCMutex_t

typedef ipc::RobustMutex uhallibs::Flx::IPCMutex_t

private

Definition at line 166 of file ProtocolFlx.hpp.

IPCScopedLock_t

typedef std::unique_lock<IPCMutex_t> uhallibs::Flx::IPCScopedLock_t

private

Definition at line 167 of file ProtocolFlx.hpp.

SteadyClock_t

typedef std::chrono::steady_clock uhallibs::Flx::SteadyClock_t

private

Definition at line 188 of file ProtocolFlx.hpp.

Constructor & Destructor Documentation

Flx() [1/2]

uhallibs::Flx::Flx ( const Flx & aFlx )

private

Flx() [2/2]

uhallibs::Flx::Flx	(	const std::string &	aId,
		const uhal::URI &	aUri )

Constructor

Parameters

aId	the uinique identifier that the client will be given.
aUri	a struct containing the full URI of the target.

Definition at line 318 of file ProtocolFlx.cpp.

  : IPbus<2, 0>(aId, aUri)
  , mConnected(false)
  , mDeviceFile(aUri.mHostname, LOCK_NONE)
  , mIPCMutex(getSharedMemName(aUri.mHostname))
  , mNumberOfPages(0)
  , mPageSize(0)
  , mIndexNextPage(0)
  , mPublishedReplyPageCount(0)
  , mReadReplyPageCount(0)
  , mAsynchronousException(NULL)
{
  mSleepDuration = std::chrono::microseconds(50);
 
  for (uhal::NameValuePairVectorType::const_iterator lIt = aUri.mArguments.begin(); lIt != aUri.mArguments.end();
       lIt++) {
    if (lIt->first == "sleep") {
      mSleepDuration = std::chrono::microseconds(std::stoul(lIt->second));
      log(uhal::Notice(),
          "flx client with URI ",
          uhal::Quote(uri()),
          " : Inter-poll-/-interrupt sleep duration set to ",
          std::stoul(lIt->second),
          " us by URI 'sleep' attribute");
    }
    // else if (lIt->first == "offset") {
    //   const bool lIsHex = (lIt->second.find("0x") == 0) or (lIt->second.find("0X") == 0);
    //   const size_t lOffset = (lIsHex ? std::lexical_cast<HexTo<size_t> >(lIt->second) : std::stoul(lIt->second));
    //   mDeviceFile.setOffset(lOffset);
    //   log (uhal::Notice(), "flx client with URI ", uhal::Quote (uri()), " : Address offset set to ",
    //   uhal::Integer(lOffset, IntFmt<hex>()));
    // }
    else {
      log(uhal::Warning(), "Unknown attribute ", uhal::Quote(lIt->first), " used in URI ", uhal::Quote(uri()));
    }
  }
}

~Flx()

uhallibs::Flx::~Flx ( )

virtual

Destructor.

Definition at line 356 of file ProtocolFlx.cpp.

{
  disconnect();
}

Member Function Documentation

connect() [1/2]

void uhallibs::Flx::connect ( )

private

Set up the connection to the device.

Definition at line 422 of file ProtocolFlx.cpp.

{
  IPCScopedLock_t lLockGuard(*mIPCMutex);
  connect(lLockGuard);
}

connect() [2/2]

void uhallibs::Flx::connect ( IPCScopedLock_t & aGuard )

private

Set up the connection to the device.

Definition at line 429 of file ProtocolFlx.cpp.

{
  // Read current value of session counter when reading status info from FPGA
  // (So that can check whether this info is up-to-date later on, when sending next request packet)
  mIPCExternalSessionActive = mIPCMutex->isActive() and (not mDeviceFile.haveLock());
  mIPCSessionCount = mIPCMutex->getCounter();
 
  log(uhal::Debug(), "flx client is opening device file ", uhal::Quote(mDeviceFile.getPath()));
  std::vector<uint32_t> lValues;
  mDeviceFile.read(0x0, 4, lValues);
  log(uhal::Debug(),
      "Read status info from addr 0 (",
      uhal::Integer(lValues.at(0)),
      ", ",
      uhal::Integer(lValues.at(1)),
      ", ",
      uhal::Integer(lValues.at(2)),
      ", ",
      uhal::Integer(lValues.at(3)),
      "): ",
      PacketFmt((const uint8_t*)lValues.data(), 4 * lValues.size()));
  aGuard.unlock();
 
  mNumberOfPages = lValues.at(0);
  // mPageSize = std::min(uint32_t(4096), lValues.at(1));
  mPageSize = lValues.at(1);
  mIndexNextPage = lValues.at(2);
  mPublishedReplyPageCount = lValues.at(3);
  mReadReplyPageCount = mPublishedReplyPageCount;
 
  if (lValues.at(1) > 0xFFFF) {
    exception::FlxInitialisationError lExc;
    log(lExc,
        "Invalid page size, ",
        uhal::Integer(lValues.at(1)),
        ", reported in device file ",
        uhal::Quote(mDeviceFile.getPath()));
    throw lExc;
  }
 
  if (mIndexNextPage >= mNumberOfPages) {
    exception::FlxInitialisationError lExc;
    log(lExc,
        "Next page index, ",
        uhal::Integer(mIndexNextPage),
        ", reported in device file ",
        uhal::Quote(mDeviceFile.getPath()),
        " is inconsistent with number of pages, ",
        uhal::Integer(mNumberOfPages));
    throw lExc;
  }
 
  mConnected = true;
  log(uhal::Info(),
      "flx client connected to device at ",
      uhal::Quote(mDeviceFile.getPath()),
      "; FPGA has ",
      uhal::Integer(mNumberOfPages),
      " pages, each of size ",
      uhal::Integer(mPageSize),
      " words, index ",
      uhal::Integer(mIndexNextPage),
      " should be filled next");
}

disconnect()

void uhallibs::Flx::disconnect ( )

private

Close the connection to the device.

Definition at line 495 of file ProtocolFlx.cpp.

{
  log(uhal::Debug(), "flx client is closing device file ", uhal::Quote(mDeviceFile.getPath()));
  mDeviceFile.close();
  mConnected = false;
}

dispatchExceptionHandler()

void uhallibs::Flx::dispatchExceptionHandler ( )

privatevirtual

Function which tidies up this protocol layer in the event of an exception.

Definition at line 388 of file ProtocolFlx.cpp.

{
  // FIXME: Adapt to PCIe implementation
  // log(uhal::Notice(), "flx client ", uhal::Quote(id()), " (URI: ", uhal::Quote(uri()), ") : closing device files
  // since exception detected");
 
  // ClientInterface::returnBufferToPool ( mReplyQueue );
 
  mDeviceFile.unlock();
 
  disconnect();
 
  InnerProtocol::dispatchExceptionHandler();
}

Flush()

void uhallibs::Flx::Flush ( )

privatevirtual

Concrete implementation of the synchronization function to block until all buffers have been sent, all replies received and all data validated

Definition at line 375 of file ProtocolFlx.cpp.

{
  log(uhal::Debug(), "flx client (URI: ", uhal::Quote(uri()), ") : Flush method called");
  while (!mReplyQueue.empty())
    read();
 
  mDeviceFile.unlock();
 
  IPCScopedLock_t lLockGuard(*mIPCMutex);
  mIPCMutex->endSession();
}

getMaxReplySize()

uint32_t uhallibs::Flx::getMaxReplySize ( )

private

Return the maximum size of reply packet based on the buffer size in the target

Returns

the maximum size of reply packet

Definition at line 413 of file ProtocolFlx.cpp.

{
  if (!mConnected)
    connect();
 
  return (mPageSize - 1) * 4;
}

getMaxSendSize()

uint32_t uhallibs::Flx::getMaxSendSize ( )

private

Return the maximum size to be sent based on the buffer size in the target

Returns

the maximum size to be sent

Definition at line 404 of file ProtocolFlx.cpp.

{
  if (!mConnected)
    connect();
 
  return (mPageSize - 1) * 4;
}

getSharedMemName()

std::string uhallibs::Flx::getSharedMemName ( const std::string & aPath )

staticprivate

Definition at line 310 of file ProtocolFlx.cpp.

{
  std::string lSanitizedPath(aPath);
  std::replace(lSanitizedPath.begin(), lSanitizedPath.end(), '/', ':');
 
  return "/uhal::ipbusflx-2.0::" + lSanitizedPath;
}

implementDispatch()

void uhallibs::Flx::implementDispatch ( std::shared_ptr< uhal::Buffers > aBuffers )

private

Send the IPbus buffer to the target, read back the response and call the packing-protocol's validate function

Parameters

aBuffers

the buffer object wrapping the send and recieve buffers that are to be transported If multithreaded, adds buffer to the dispatch queue and returns. If single-threaded, calls the dispatch-worker dispatch function directly and blocks until the response is validated.

Definition at line 362 of file ProtocolFlx.cpp.

{
  log(uhal::Debug(), "flx client (URI: ", uhal::Quote(uri()), ") : implementDispatch method called");
 
  if (!mConnected)
    connect();
 
  if (mReplyQueue.size() == mNumberOfPages)
    read();
  write(aBuffers);
}

operator=()

Flx & uhallibs::Flx::operator= ( const Flx & aFlx )

private

read()

void uhallibs::Flx::read ( )

private

Read next pending reply packet from appropriate page of FPGA-to-host device file, and validate contents.

Definition at line 549 of file ProtocolFlx.cpp.

{
  const size_t lPageIndexToRead = (mIndexNextPage - mReplyQueue.size() + mNumberOfPages) % mNumberOfPages;
  SteadyClock_t::time_point lStartTime = SteadyClock_t::now();
 
  if (mReadReplyPageCount == mPublishedReplyPageCount) {
    uint32_t lHwPublishedPageCount = 0x0;
 
    std::vector<uint32_t> lValues;
    while (true) {
      IPCScopedLock_t lGuard(*mIPCMutex);
      // FIXME : Improve by simply adding dmaWrite method that takes uint32_t ref as argument (or returns uint32_t)
      mDeviceFile.read(0, 4, lValues);
      lHwPublishedPageCount = lValues.at(3);
      // log (uhal::Info(), "Read status info from addr 0 (", uhal::Integer(lValues.at(0)), ", ",
      // uhal::Integer(lValues.at(1)), ", ", uhal::Integer(lValues.at(2)), ", ", uhal::Integer(lValues.at(3)), "): ",
      // PacketFmt((const uint8_t*)lValues.data(), 4 * lValues.size()));
      log(uhal::Debug(),
          "Read status info from addr 0 (",
          uhal::Integer(lValues.at(0)),
          ", ",
          uhal::Integer(lValues.at(1)),
          ", ",
          uhal::Integer(lValues.at(2)),
          ", ",
          uhal::Integer(lValues.at(3)),
          "): ",
          PacketFmt((const uint8_t*)lValues.data(), 4 * lValues.size()));
 
      if (lHwPublishedPageCount != mPublishedReplyPageCount) {
        mPublishedReplyPageCount = lHwPublishedPageCount;
        break;
      }
      // FIXME: Throw if published page count is invalid number
 
      if (SteadyClock_t::now() - lStartTime > std::chrono::microseconds(getBoostTimeoutPeriod().total_microseconds())) {
        exception::FlxTimeout lExc;
        log(lExc,
            "Next page (index ",
            uhal::Integer(lPageIndexToRead),
            " count ",
            uhal::Integer(mPublishedReplyPageCount + 1),
            ") of flx device '" + mDeviceFile.getPath() + "' is not ready after timeout period");
        throw lExc;
      }
 
      log(uhal::Debug(),
          "flx client ",
          uhal::Quote(id()),
          " (URI: ",
          uhal::Quote(uri()),
          ") : Trying to read page index ",
          uhal::Integer(lPageIndexToRead),
          " = count ",
          uhal::Integer(mReadReplyPageCount + 1),
          "; published page count is ",
          uhal::Integer(lHwPublishedPageCount),
          "; sleeping for ",
          mSleepDuration.count(),
          "us");
      if (mSleepDuration > std::chrono::microseconds(0))
        std::this_thread::sleep_for(mSleepDuration);
      lValues.clear();
    }
 
    log(uhal::Info(),
        "flx client ",
        uhal::Quote(id()),
        " (URI: ",
        uhal::Quote(uri()),
        ") : Reading page ",
        uhal::Integer(lPageIndexToRead),
        " (published count ",
        uhal::Integer(lHwPublishedPageCount),
        ", surpasses required, ",
        uhal::Integer(mReadReplyPageCount + 1),
        ")");
  }
  mReadReplyPageCount++;
 
  // PART 1 : Read the page
  std::shared_ptr<uhal::Buffers> lBuffers = mReplyQueue.front();
  mReplyQueue.pop_front();
 
  uint32_t lNrWordsToRead(lBuffers->replyCounter() >> 2);
  lNrWordsToRead += 1;
 
  std::vector<uint32_t> lPageContents;
  IPCScopedLock_t lGuard(*mIPCMutex);
  mDeviceFile.read(4 + lPageIndexToRead * mPageSize, lNrWordsToRead, lPageContents);
  lGuard.unlock();
  log(uhal::Debug(),
      "Read ",
      uhal::Integer(lNrWordsToRead),
      " 32-bit words from address ",
      uhal::Integer(4 + lPageIndexToRead * 4 * mPageSize),
      " ... ",
      PacketFmt((const uint8_t*)lPageContents.data(), 4 * lPageContents.size()));
 
  // PART 2 : Transfer to reply buffer
  const std::deque<std::pair<uint8_t*, uint32_t>>& lReplyBuffers(lBuffers->getReplyBuffer());
  size_t lNrWordsInPacket = (lPageContents.at(0) >> 16) + (lPageContents.at(0) & 0xFFFF);
  if (lNrWordsInPacket != (lBuffers->replyCounter() >> 2))
    log(uhal::Warning(),
        "Expected reply packet to contain ",
        uhal::Integer(lBuffers->replyCounter() >> 2),
        " words, but it actually contains ",
        uhal::Integer(lNrWordsInPacket),
        " words");
 
  size_t lNrBytesCopied = 0;
  for (std::deque<std::pair<uint8_t*, uint32_t>>::const_iterator lIt = lReplyBuffers.begin();
       lIt != lReplyBuffers.end();
       ++lIt) {
    // Don't copy more of page than was written to, for cases when less data received than expected
    if (lNrBytesCopied >= 4 * lNrWordsInPacket)
      break;
 
    size_t lNrBytesToCopy = std::min(lIt->second, uint32_t(4 * lNrWordsInPacket - lNrBytesCopied));
    memcpy(lIt->first, &lPageContents.at(1 + (lNrBytesCopied / 4)), lNrBytesToCopy);
    lNrBytesCopied += lNrBytesToCopy;
  }
 
  // PART 3 : Validate the packet contents
  try {
    if (uhal::exception::exception* lExc =
          ClientInterface::validate(lBuffers)) // Control of the pointer has been passed back to the client interface
    {
      mAsynchronousException = lExc;
    }
  } catch (uhal::exception::exception& aExc) {
    mAsynchronousException = new uhal::exception::ValidationError();
    log(*mAsynchronousException,
        "Exception caught during reply validation for flx device with URI ",
        uhal::Quote(this->uri()),
        "; what returned: ",
        uhal::Quote(aExc.what()));
  }
 
  if (mAsynchronousException) {
    mAsynchronousException->throwAsDerivedType();
  }
}

write()

void uhallibs::Flx::write ( const std::shared_ptr< uhal::Buffers > & aBuffers )

private

Write request packet to next page in host-to-FPGA device file.

Definition at line 503 of file ProtocolFlx.cpp.

{
  if (not mDeviceFile.haveLock()) {
    mDeviceFile.lock();
 
    IPCScopedLock_t lGuard(*mIPCMutex);
    mIPCMutex->startSession();
    mIPCSessionCount++;
 
    if (mIPCExternalSessionActive or (mIPCMutex->getCounter() != mIPCSessionCount)) {
      connect(lGuard);
    }
  }
 
  log(uhal::Info(),
      "flx client ",
      uhal::Quote(id()),
      " (URI: ",
      uhal::Quote(uri()),
      ") : writing ",
      uhal::Integer(aBuffers->sendCounter() / 4),
      "-word packet to page ",
      uhal::Integer(mIndexNextPage),
      " in ",
      uhal::Quote(mDeviceFile.getPath()));
 
  const uint32_t lHeaderWord = (0x10000 | (((aBuffers->sendCounter() / 4) - 1) & 0xFFFF));
  std::vector<std::pair<const uint8_t*, size_t>> lDataToWrite;
  lDataToWrite.push_back(std::make_pair(reinterpret_cast<const uint8_t*>(&lHeaderWord), sizeof lHeaderWord));
  lDataToWrite.push_back(std::make_pair(aBuffers->getSendBuffer(), aBuffers->sendCounter()));
 
  IPCScopedLock_t lGuard(*mIPCMutex);
  mDeviceFile.write(mIndexNextPage * mPageSize, lDataToWrite);
  log(uhal::Debug(),
      "Wrote ",
      uhal::Integer((aBuffers->sendCounter() / 4) + 1),
      " 32-bit words at address ",
      uhal::Integer(mIndexNextPage * 4 * mPageSize),
      " ... ",
      PacketFmt(lDataToWrite));
 
  mIndexNextPage = (mIndexNextPage + 1) % mNumberOfPages;
  mReplyQueue.push_back(aBuffers);
}

Member Data Documentation

mAsynchronousException

uhal::exception::exception* uhallibs::Flx::mAsynchronousException

private

A pointer to an exception object for passing exceptions from the worker thread to the main thread. Exceptions must always be created on the heap (i.e. using new) and deletion will be handled in the main thread

Definition at line 236 of file ProtocolFlx.hpp.

mConnected

bool uhallibs::Flx::mConnected

private

Definition at line 217 of file ProtocolFlx.hpp.

mDeviceFile

Card uhallibs::Flx::mDeviceFile

private

Definition at line 219 of file ProtocolFlx.hpp.

mIndexNextPage

uint32_t uhallibs::Flx::mIndexNextPage

private

Definition at line 227 of file ProtocolFlx.hpp.

mIPCExternalSessionActive

bool uhallibs::Flx::mIPCExternalSessionActive

private

Definition at line 222 of file ProtocolFlx.hpp.

mIPCMutex

ipc::SharedMemObject<IPCMutex_t> uhallibs::Flx::mIPCMutex

private

Definition at line 221 of file ProtocolFlx.hpp.

mIPCSessionCount

uint64_t uhallibs::Flx::mIPCSessionCount

private

Definition at line 223 of file ProtocolFlx.hpp.

mNumberOfPages

uint32_t uhallibs::Flx::mNumberOfPages

private

Definition at line 227 of file ProtocolFlx.hpp.

mPageSize

uint32_t uhallibs::Flx::mPageSize

private

Definition at line 227 of file ProtocolFlx.hpp.

mPublishedReplyPageCount

uint32_t uhallibs::Flx::mPublishedReplyPageCount

private

Definition at line 227 of file ProtocolFlx.hpp.

mReadReplyPageCount

uint32_t uhallibs::Flx::mReadReplyPageCount

private

Definition at line 227 of file ProtocolFlx.hpp.

mReplyQueue

std::deque<std::shared_ptr<uhal::Buffers> > uhallibs::Flx::mReplyQueue

private

The list of buffers still awaiting a reply.

Definition at line 230 of file ProtocolFlx.hpp.

mSleepDuration

std::chrono::microseconds uhallibs::Flx::mSleepDuration

private

Definition at line 225 of file ProtocolFlx.hpp.

---

The documentation for this class was generated from the following files:

• /github/workspace/dunedaq/sourcecode/uhallibs/include/uhallibs/ProtocolFlx.hpp
• /github/workspace/dunedaq/sourcecode/uhallibs/src/ProtocolFlx.cpp