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 ()
	Concrete implementation of the synchronization function to block until all buffers have been sent, all replies received and all data validated.
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 107 of file ProtocolFlx.hpp.

Member Typedef Documentation

InnerProtocol

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

private

Definition at line 197 of file ProtocolFlx.hpp.

IPCMutex_t

ipc::RobustMutex uhallibs::Flx::IPCMutex_t

private

Definition at line 178 of file ProtocolFlx.hpp.

IPCScopedLock_t

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

private

Definition at line 179 of file ProtocolFlx.hpp.

SteadyClock_t

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

private

Definition at line 199 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 289 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 321 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 387 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 394 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 431 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 354 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 340 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 378 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 369 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 280 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 327 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 469 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 439 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 248 of file ProtocolFlx.hpp.

mConnected

bool uhallibs::Flx::mConnected

private

Definition at line 228 of file ProtocolFlx.hpp.

mDeviceFile

Card uhallibs::Flx::mDeviceFile

private

Definition at line 230 of file ProtocolFlx.hpp.

mIndexNextPage

uint32_t uhallibs::Flx::mIndexNextPage

private

Definition at line 239 of file ProtocolFlx.hpp.

mIPCExternalSessionActive

bool uhallibs::Flx::mIPCExternalSessionActive

private

Definition at line 233 of file ProtocolFlx.hpp.

mIPCMutex

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

private

Definition at line 232 of file ProtocolFlx.hpp.

mIPCSessionCount

uint64_t uhallibs::Flx::mIPCSessionCount

private

Definition at line 234 of file ProtocolFlx.hpp.

mNumberOfPages

uint32_t uhallibs::Flx::mNumberOfPages

private

Definition at line 239 of file ProtocolFlx.hpp.

mPageSize

uint32_t uhallibs::Flx::mPageSize

private

Definition at line 239 of file ProtocolFlx.hpp.

mPublishedReplyPageCount

uint32_t uhallibs::Flx::mPublishedReplyPageCount

private

Definition at line 239 of file ProtocolFlx.hpp.

mReadReplyPageCount

uint32_t uhallibs::Flx::mReadReplyPageCount

private

Definition at line 239 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 242 of file ProtocolFlx.hpp.

mSleepDuration

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

private

Definition at line 237 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