dunedaq::datahandlinglibs::IterableQueueModel< T > Struct Template Reference

#include <IterableQueueModel.hpp>

Inheritance diagram for dunedaq::datahandlinglibs::IterableQueueModel< T >:

Collaboration diagram for dunedaq::datahandlinglibs::IterableQueueModel< T >:

Classes
struct	Iterator

Public Types
typedef T	value_type
Public Types inherited from dunedaq::opmonlib::MonitorableObject
using	NodePtr = std::weak_ptr<MonitorableObject>
using	NewNodePtr = std::shared_ptr<MonitorableObject>
using	ElementId = std::string

Public Member Functions
	IterableQueueModel (const IterableQueueModel &)=delete
IterableQueueModel &	operator= (const IterableQueueModel &)=delete
	IterableQueueModel ()
	IterableQueueModel (std::size_t size)
	IterableQueueModel (std::size_t size, bool numa_aware=false, uint8_t numa_node=0, bool intrinsic_allocator=false, std::size_t alignment_size=0)
	~IterableQueueModel ()
void	free_memory ()
void	allocate_memory (std::size_t size, bool numa_aware, uint8_t numa_node=0, bool intrinsic_allocator=false, std::size_t alignment_size=0)
void	allocate_memory (std::size_t size) override
void	prefill_task ()
void	force_pagefault ()
bool	write (T &&record) override
	Move referenced object into LB.
bool	read (T &record) override
	Move object from LB to referenced.
void	popFront ()
void	pop (std::size_t x)
	Pop specified amount of elements from LB.
bool	isEmpty () const
bool	isFull () const
std::size_t	occupancy () const override
	Occupancy of LB.
std::size_t	size () const
std::size_t	capacity () const
const T *	front () override
	Write referenced object into LB without moving it.
const T *	back () override
	Get pointer to the back of the LB.
T *	start_of_buffer ()
T *	end_of_buffer ()
void	conf (const appmodel::LatencyBuffer *cfg) override
	Configure the LB.
void	scrap (const nlohmann::json &) override
	Unconfigure the LB.
void	flush () override
	Flush all elements from the latency buffer.
std::size_t	get_alignment_size ()
Iterator	begin ()
Iterator	end ()
Public Member Functions inherited from dunedaq::datahandlinglibs::LatencyBufferConcept< T >
	LatencyBufferConcept ()
virtual	~LatencyBufferConcept ()
	LatencyBufferConcept (const LatencyBufferConcept &)=delete
	LatencyBufferConcept is not copy-constructible.
LatencyBufferConcept &	operator= (const LatencyBufferConcept &)=delete
	LatencyBufferConcept is not copy-assginable.
	LatencyBufferConcept (LatencyBufferConcept &&)=delete
	LatencyBufferConcept is not move-constructible.
LatencyBufferConcept &	operator= (LatencyBufferConcept &&)=delete
	LatencyBufferConcept is not move-assignable.
virtual void	allocate_memory (size_t)=0
	Whether or not the buffer is allocatable. false by default.
Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
	MonitorableObject (const MonitorableObject &)=delete
MonitorableObject &	operator= (const MonitorableObject &)=delete
	MonitorableObject (MonitorableObject &&)=delete
MonitorableObject &	operator= (MonitorableObject &&)=delete
virtual	~MonitorableObject ()=default
auto	get_opmon_id () const noexcept
auto	get_opmon_level () const noexcept

Protected Member Functions
virtual void	generate_opmon_data () override
template<class... Args>
bool	write_ (Args &&... recordArgs)
Protected Member Functions inherited from dunedaq::opmonlib::MonitorableObject
	MonitorableObject ()=default
void	register_node (ElementId name, NewNodePtr)
void	publish (google::protobuf::Message &&, CustomOrigin &&co={}, OpMonLevel l=to_level(EntryOpMonLevel::kDefault)) const noexcept

Protected Attributes
std::atomic< int >	overflow_ctr { 0 }
bool	numa_aware_
uint8_t	numa_node_
bool	intrinsic_allocator_
std::size_t	alignment_size_
bool	invalid_configuration_requested_
std::string	prefiller_name_ {"lbpfn"}
std::mutex	prefill_mutex_
std::condition_variable	prefill_cv_
bool	prefill_ready_
bool	prefill_done_
std::thread	ptrlogger
char	pad0_ [folly::hardware_destructive_interference_size]
uint32_t	size_
T *	records_
std::atomic< unsigned int >	readIndex_
std::atomic< unsigned int >	writeIndex_
char	pad1_ [folly::hardware_destructive_interference_size - sizeof(writeIndex_)]

Additional Inherited Members
Static Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
static bool	publishable_metric (OpMonLevel entry, OpMonLevel system) noexcept

Detailed Description

template<class T>
struct dunedaq::datahandlinglibs::IterableQueueModel< T >

IterableQueueModel is a one producer and one consumer queue without locks. Modified version of the folly::ProducerConsumerQueue via adding a readPtr function. Requires well defined and followed constraints on the consumer side.

Also, note that the number of usable slots in the queue at any given time is actually (size-1), so if you start with an empty queue, isFull() will return true after size-1 insertions.

Definition at line 71 of file IterableQueueModel.hpp.

Member Typedef Documentation

value_type

template<class T >

T dunedaq::datahandlinglibs::IterableQueueModel< T >::value_type

Definition at line 73 of file IterableQueueModel.hpp.

Constructor & Destructor Documentation

IterableQueueModel() [1/4]

template<class T >

dunedaq::datahandlinglibs::IterableQueueModel< T >::IterableQueueModel ( const IterableQueueModel< T > & )

delete

IterableQueueModel() [2/4]

template<class T >

dunedaq::datahandlinglibs::IterableQueueModel< T >::IterableQueueModel ( )

inline

Definition at line 79 of file IterableQueueModel.hpp.

    : LatencyBufferConcept<T>()
    , numa_aware_(false)
    , numa_node_(0)
    , intrinsic_allocator_(false)
    , alignment_size_(0)
    , invalid_configuration_requested_(false)
    , prefill_ready_(false)
    , prefill_done_(false)
    , size_(2)
    , records_(static_cast<T*>(std::malloc(sizeof(T) * 2)))
    , readIndex_(0)
    , writeIndex_(0)
  {}

IterableQueueModel() [3/4]

template<class T >

dunedaq::datahandlinglibs::IterableQueueModel< T >::IterableQueueModel ( std::size_t size )

inlineexplicit

Definition at line 95 of file IterableQueueModel.hpp.

    : LatencyBufferConcept<T>() // NOLINT(build/unsigned)
    , numa_aware_(false)
    , numa_node_(0)
    , intrinsic_allocator_(false)
    , alignment_size_(0)
    , invalid_configuration_requested_(false)
    , prefill_ready_(false)
    , prefill_done_(false)
    , size_(size)
    , records_(static_cast<T*>(std::malloc(sizeof(T) * size)))
    , readIndex_(0)
    , writeIndex_(0)
  {
    assert(size >= 2);
    if (!records_) {
      throw std::bad_alloc();
    }
#if 0
    ptrlogger = std::thread([&](){
      while(true) {
        auto const currentRead = readIndex_.load(std::memory_order_relaxed);
        auto const currentWrite = writeIndex_.load(std::memory_order_relaxed);
        TLOG() << "BEG:" << std::hex << &records_[0] << " END:" << &records_[size] << std::dec
                  << " R:" << currentRead << " - W:" << currentWrite
                  << " OFLOW:" << overflow_ctr;
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
      }
    });
#endif
  }

IterableQueueModel() [4/4]

template<class T >

dunedaq::datahandlinglibs::IterableQueueModel< T >::IterableQueueModel ( std::size_t size, bool numa_aware = false, uint8_t numa_node = 0, bool intrinsic_allocator = false, std::size_t alignment_size = 0 )

inline

Definition at line 128 of file IterableQueueModel.hpp.

    : LatencyBufferConcept<T>() // NOLINT(build/unsigned)
    , numa_aware_(numa_aware)
    , numa_node_(numa_node)
    , intrinsic_allocator_(intrinsic_allocator)
    , alignment_size_(alignment_size)
    , invalid_configuration_requested_(false)
    , prefill_ready_(false)
    , prefill_done_(false)
    , size_(size)
    , readIndex_(0)
    , writeIndex_(0)
  {
    assert(size >= 2);
    allocate_memory(size, numa_aware, numa_node, intrinsic_allocator, alignment_size);
 
    if (!records_) {
      throw std::bad_alloc();
    }
#if 0
    ptrlogger = std::thread([&](){
      while(true) {
        auto const currentRead = readIndex_.load(std::memory_order_relaxed);
        auto const currentWrite = writeIndex_.load(std::memory_order_relaxed);
        TLOG() << "BEG:" << std::hex << &records_[0] << " END:" << &records_[size] << std::dec
                  << " R:" << currentRead << " - W:" << currentWrite
                  << " OFLOW:" << overflow_ctr;
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
      }
    });
#endif
  }

~IterableQueueModel()

template<class T >

dunedaq::datahandlinglibs::IterableQueueModel< T >::~IterableQueueModel ( )

inline

Definition at line 166 of file IterableQueueModel.hpp.

{ free_memory(); }

Member Function Documentation

allocate_memory() [1/2]

template<class T >

void dunedaq::datahandlinglibs::IterableQueueModel< T >::allocate_memory ( std::size_t size )

inlineoverride

Definition at line 178 of file IterableQueueModel.hpp.

{ allocate_memory(size,false); }

allocate_memory() [2/2]

template<class T >

void dunedaq::datahandlinglibs::IterableQueueModel< T >::allocate_memory	(	std::size_t	size,
		bool	numa_aware,
		uint8_t	numa_node = 0,
		bool	intrinsic_allocator = false,
		std::size_t	alignment_size = 0 )

Definition at line 39 of file IterableQueueModel.hxx.

{
  assert(size >= 2);
  // TODO: check for valid alignment sizes! | July-21-2021 | Roland Sipos | rsipos@cern.ch
 
  if (numa_aware && numa_node < 8) { // numa allocator from libnuma; we get "numa_node >= 0" for free, given its datatype
#ifdef WITH_LIBNUMA_SUPPORT
    numa_set_preferred((unsigned)numa_node); // https://linux.die.net/man/3/numa_set_preferred
 #ifdef WITH_LIBNUMA_BIND_POLICY
    numa_set_bind_policy(WITH_LIBNUMA_BIND_POLICY); // https://linux.die.net/man/3/numa_set_bind_policy
 #endif
 #ifdef WITH_LIBNUMA_STRICT_POLICY
    numa_set_strict(WITH_LIBNUMA_STRICT_POLICY);    // https://linux.die.net/man/3/numa_set_strict
 #endif
    records_ = static_cast<T*>(numa_alloc_onnode(sizeof(T) * size, numa_node));
#else
    throw GenericConfigurationError(ERS_HERE,
                                    "NUMA allocation was requested but program was built without USE_LIBNUMA");
#endif
  } else if (intrinsic_allocator && alignment_size > 0) { // _mm allocator
    records_ = static_cast<T*>(_mm_malloc(sizeof(T) * size, alignment_size));
  } else if (!intrinsic_allocator && alignment_size > 0) { // std aligned allocator
    records_ = static_cast<T*>(std::aligned_alloc(alignment_size, sizeof(T) * size));
  } else if (!numa_aware && !intrinsic_allocator && alignment_size == 0) {
    // Standard allocator
    records_ = static_cast<T*>(std::malloc(sizeof(T) * size));
 
  } else {
    // Let it fail, as expected combination might be invalid
    // records_ = static_cast<T*>(std::malloc(sizeof(T) * size_);
  }
 
  size_ = size;
  numa_aware_ = numa_aware;
  numa_node_ = numa_node;
  intrinsic_allocator_ = intrinsic_allocator;
  alignment_size_ = alignment_size;
}

back()

template<class T >

const T * dunedaq::datahandlinglibs::IterableQueueModel< T >::back ( )

overridevirtual

Get pointer to the back of the LB.

Implements dunedaq::datahandlinglibs::LatencyBufferConcept< T >.

Definition at line 279 of file IterableQueueModel.hxx.

{
  auto const currentWrite = writeIndex_.load(std::memory_order_relaxed);
  if (currentWrite == readIndex_.load(std::memory_order_acquire)) {
    return nullptr;
  }
  int currentLast = currentWrite;
  if (currentLast == 0) {
    currentLast = size_ - 1;
  } else {
    currentLast--;
  }
  return &records_[currentLast];
}

begin()

template<class T >

Iterator dunedaq::datahandlinglibs::IterableQueueModel< T >::begin ( )

inline

Definition at line 299 of file IterableQueueModel.hpp.

  {
    auto const currentRead = readIndex_.load(std::memory_order_relaxed);
    if (currentRead == writeIndex_.load(std::memory_order_acquire)) {
      // queue is empty
      return end();
    }
    return Iterator(*this, currentRead);
  }

capacity()

template<class T >

std::size_t dunedaq::datahandlinglibs::IterableQueueModel< T >::capacity ( ) const

inline

Definition at line 216 of file IterableQueueModel.hpp.

{ return size_ - 1; }

conf()

template<class T >

void dunedaq::datahandlinglibs::IterableQueueModel< T >::conf ( const appmodel::LatencyBuffer * conf )

overridevirtual

Configure the LB.

Implements dunedaq::datahandlinglibs::LatencyBufferConcept< T >.

Definition at line 297 of file IterableQueueModel.hxx.

{
  assert(cfg->get_size() >= 2);
  free_memory();
 
  allocate_memory(cfg->get_size(),
                  cfg->get_numa_aware(),
                  cfg->get_numa_node(),
                  cfg->get_intrinsic_allocator(),
                  cfg->get_alignment_size());
  readIndex_ = 0;
  writeIndex_ = 0;
 
  if (!records_) {
    throw std::bad_alloc();
  }
 
  if (cfg->get_preallocation()) {
    force_pagefault();
  }
}

end()

template<class T >

Iterator dunedaq::datahandlinglibs::IterableQueueModel< T >::end ( )

inline

Definition at line 309 of file IterableQueueModel.hpp.

  {
    return Iterator(*this, std::numeric_limits<uint32_t>::max()); // NOLINT(build/unsigned)
  }

end_of_buffer()

template<class T >

T * dunedaq::datahandlinglibs::IterableQueueModel< T >::end_of_buffer ( )

inline

Definition at line 228 of file IterableQueueModel.hpp.

{ return &records_[size_]; }

flush()

template<class T >

void dunedaq::datahandlinglibs::IterableQueueModel< T >::flush ( )

inlineoverridevirtual

Flush all elements from the latency buffer.

Implements dunedaq::datahandlinglibs::LatencyBufferConcept< T >.

Definition at line 237 of file IterableQueueModel.hpp.

{ pop(occupancy()); }

force_pagefault()

template<class T >

void dunedaq::datahandlinglibs::IterableQueueModel< T >::force_pagefault

(

)

Definition at line 107 of file IterableQueueModel.hxx.

{
  // Local prefiller thread
  std::thread prefill_thread(&IterableQueueModel<T>::prefill_task, this);
 
  // Tweak prefiller thread
  char tname[16];
  snprintf(tname, 16, "%s-%d", prefiller_name_.c_str(), numa_node_);
  auto handle = prefill_thread.native_handle();
  pthread_setname_np(handle, tname);
 
#ifdef WITH_LIBNUMA_SUPPORT
  cpu_set_t affinitymask;
  CPU_ZERO(&affinitymask);
  struct bitmask *nodecpumask = numa_allocate_cpumask();
  int ret = 0;
  // Get NODE CPU mask
  ret = numa_node_to_cpus(numa_node_, nodecpumask);
  assert(ret == 0);
  // Apply corresponding NODE CPUs to affinity mask
  for (int i=0; i< numa_num_configured_cpus(); ++i) {
    if (numa_bitmask_isbitset(nodecpumask, i)) {
      CPU_SET(i, &affinitymask);
    }
  }
  ret = pthread_setaffinity_np(handle, sizeof(cpu_set_t), &affinitymask);
  assert(ret == 0);
  numa_free_cpumask(nodecpumask);
#endif
 
  // Trigger prefiller thread
  {
    std::lock_guard lk(prefill_mutex_);
    prefill_ready_ = true;
  }
  prefill_cv_.notify_one();
  // Wait for prefiller thread to finish
  {
    std::unique_lock lk(prefill_mutex_);
    prefill_cv_.wait(lk, [this]{ return prefill_done_; });
  }
  // Join with prefiller thread
  prefill_thread.join();
}

free_memory()

template<class T >

void dunedaq::datahandlinglibs::IterableQueueModel< T >::free_memory

(

)

Definition at line 9 of file IterableQueueModel.hxx.

{
  // We need to destruct anything that may still exist in our queue.
  // (No real synchronization needed at destructor time: only one
  // thread can be doing this.)
  if (!std::is_trivially_destructible<T>::value) {
    std::size_t readIndex = readIndex_;
    std::size_t endIndex = writeIndex_;
    while (readIndex != endIndex) {
      records_[readIndex].~T();
      if (++readIndex == size_) { // NOLINT(runtime/increment_decrement)
        readIndex = 0;
      }
    }
  }
  // Different allocators require custom free functions
  if (intrinsic_allocator_) {
    _mm_free(records_);
  } else if (numa_aware_) {
#ifdef WITH_LIBNUMA_SUPPORT
      numa_free(records_, sizeof(T) * size_);
#endif
  } else {
    std::free(records_);
  }
}

front()

template<class T >

const T * dunedaq::datahandlinglibs::IterableQueueModel< T >::front ( )

overridevirtual

Write referenced object into LB without moving it.

Get pointer to the front of the LB

Implements dunedaq::datahandlinglibs::LatencyBufferConcept< T >.

Definition at line 267 of file IterableQueueModel.hxx.

{
  auto const currentRead = readIndex_.load(std::memory_order_relaxed);
  if (currentRead == writeIndex_.load(std::memory_order_acquire)) {
    return nullptr;
  }
  return &records_[currentRead];
}

generate_opmon_data()

template<class T >

void dunedaq::datahandlinglibs::IterableQueueModel< T >::generate_opmon_data ( )

overrideprotectedvirtual

Hook for customisable pubblication. The function can throw, exception will be caught by the monitoring thread

Reimplemented from dunedaq::opmonlib::MonitorableObject.

Definition at line 372 of file IterableQueueModel.hxx.

                                           {
   opmon::LatencyBufferInfo info;
   info.set_num_buffer_elements(this->occupancy());
   this->publish(std::move(info)); 
 
}

get_alignment_size()

template<class T >

std::size_t dunedaq::datahandlinglibs::IterableQueueModel< T >::get_alignment_size ( )

inline

Definition at line 240 of file IterableQueueModel.hpp.

{ return alignment_size_; }

isEmpty()

template<class T >

bool dunedaq::datahandlinglibs::IterableQueueModel< T >::isEmpty

(

)

const

Definition at line 225 of file IterableQueueModel.hxx.

{
  return readIndex_.load(std::memory_order_acquire) == writeIndex_.load(std::memory_order_acquire);
}

isFull()

template<class T >

bool dunedaq::datahandlinglibs::IterableQueueModel< T >::isFull

(

)

const

Definition at line 233 of file IterableQueueModel.hxx.

{
  auto nextRecord = writeIndex_.load(std::memory_order_acquire) + 1;
  if (nextRecord == size_) {
    nextRecord = 0;
  }
  if (nextRecord != readIndex_.load(std::memory_order_acquire)) {
    return false;
  }
  // queue is full
  return true;
}

occupancy()

template<class T >

std::size_t dunedaq::datahandlinglibs::IterableQueueModel< T >::occupancy ( ) const

overridevirtual

Occupancy of LB.

Implements dunedaq::datahandlinglibs::LatencyBufferConcept< T >.

Definition at line 254 of file IterableQueueModel.hxx.

{
  int ret = static_cast<int>(writeIndex_.load(std::memory_order_acquire)) -
            static_cast<int>(readIndex_.load(std::memory_order_acquire));
  if (ret < 0) {
    ret += static_cast<int>(size_);
  }
  return static_cast<std::size_t>(ret);
}

operator=()

template<class T >

IterableQueueModel & dunedaq::datahandlinglibs::IterableQueueModel< T >::operator= ( const IterableQueueModel< T > & )

delete

pop()

template<class T >

void dunedaq::datahandlinglibs::IterableQueueModel< T >::pop ( std::size_t amount )

virtual

Pop specified amount of elements from LB.

Implements dunedaq::datahandlinglibs::LatencyBufferConcept< T >.

Definition at line 215 of file IterableQueueModel.hxx.

{
  for (std::size_t i = 0; i < x; i++) {
    popFront();
  }
}

popFront()

template<class T >

void dunedaq::datahandlinglibs::IterableQueueModel< T >::popFront

(

)

Definition at line 198 of file IterableQueueModel.hxx.

{
  auto const currentRead = readIndex_.load(std::memory_order_relaxed);
  assert(currentRead != writeIndex_.load(std::memory_order_acquire));
 
  auto nextRecord = currentRead + 1;
  if (nextRecord == size_) {
    nextRecord = 0;
  }
 
  records_[currentRead].~T();
  readIndex_.store(nextRecord, std::memory_order_release);
}

prefill_task()

template<class T >

void dunedaq::datahandlinglibs::IterableQueueModel< T >::prefill_task

(

)

Definition at line 84 of file IterableQueueModel.hxx.

{
  // Wait until LB issues ready
  std::unique_lock lk(prefill_mutex_);
  prefill_cv_.wait(lk, [this]{ return prefill_ready_; });
  
  // After wait, we are ready to force page-fault
  for (size_t i = 0; i < size_ - 1; ++i) {
    T element = T();
    write_(std::move(element));
  }
  flush();
  
  // Preallocation done
  prefill_done_ = true;
  
  // Manual unlock is done before notify: avoid waking up the waiting thread only to block again.
  lk.unlock();
  prefill_cv_.notify_one();
}

read()

template<class T >

bool dunedaq::datahandlinglibs::IterableQueueModel< T >::read ( T & element )

overridevirtual

Move object from LB to referenced.

Implements dunedaq::datahandlinglibs::LatencyBufferConcept< T >.

Definition at line 177 of file IterableQueueModel.hxx.

{
  auto const currentRead = readIndex_.load(std::memory_order_relaxed);
  if (currentRead == writeIndex_.load(std::memory_order_acquire)) {
    // queue is empty
    return false;
  }
 
  auto nextRecord = currentRead + 1;
  if (nextRecord == size_) {
    nextRecord = 0;
  }
  record = std::move(records_[currentRead]);
  records_[currentRead].~T();
  readIndex_.store(nextRecord, std::memory_order_release);
  return true;
}

scrap()

template<class T >

void dunedaq::datahandlinglibs::IterableQueueModel< T >::scrap ( const nlohmann::json & cfg )

overridevirtual

Unconfigure the LB.

Implements dunedaq::datahandlinglibs::LatencyBufferConcept< T >.

Definition at line 322 of file IterableQueueModel.hxx.

{
  free_memory();
  numa_aware_ = false;
  numa_node_ = 0;
  intrinsic_allocator_ = false;
  alignment_size_ = 0;
  invalid_configuration_requested_ = false;
  prefill_ready_ = false;
  prefill_done_ = false;
  size_ = 2;
  records_ = static_cast<T*>(std::malloc(sizeof(T) * 2));
  readIndex_ = 0;
  writeIndex_ = 0;
}

size()

template<class T >

std::size_t dunedaq::datahandlinglibs::IterableQueueModel< T >::size ( ) const

inline

Definition at line 213 of file IterableQueueModel.hpp.

{ return size_; }

start_of_buffer()

template<class T >

T * dunedaq::datahandlinglibs::IterableQueueModel< T >::start_of_buffer ( )

inline

Definition at line 225 of file IterableQueueModel.hpp.

{ return &records_[0]; }

write()

template<class T >

bool dunedaq::datahandlinglibs::IterableQueueModel< T >::write ( T && element )

overridevirtual

Move referenced object into LB.

Implements dunedaq::datahandlinglibs::LatencyBufferConcept< T >.

Definition at line 155 of file IterableQueueModel.hxx.

{
  auto const currentWrite = writeIndex_.load(std::memory_order_relaxed);
  auto nextRecord = currentWrite + 1;
  if (nextRecord == size_) {
    nextRecord = 0;
  }
 
  if (nextRecord != readIndex_.load(std::memory_order_acquire)) {
    new (&records_[currentWrite]) T(std::move(record));
    writeIndex_.store(nextRecord, std::memory_order_release);
    return true;
  }
 
  // queue is full
  ++overflow_ctr;
  return false;
}

write_()

template<class T >

template<class... Args>

bool dunedaq::datahandlinglibs::IterableQueueModel< T >::write_ ( Args &&... recordArgs )

protected

Definition at line 342 of file IterableQueueModel.hxx.

{
  // const std::lock_guard<std::mutex> lock(m_mutex);
  auto const currentWrite = writeIndex_.load(std::memory_order_relaxed);
  auto nextRecord = currentWrite + 1;
  if (nextRecord == size_) {
    nextRecord = 0;
  }
  // if (nextRecord == readIndex_.load(std::memory_order_acquire)) {
  // std::cout << "SPSC WARNING -> Queue is full! WRITE PASSES READ!!! \n";
  //}
  //    new (&records_[currentWrite]) T(std::forward<Args>(recordArgs)...);
  // writeIndex_.store(nextRecord, std::memory_order_release);
  // return true;
 
  // ORIGINAL:
  if (nextRecord != readIndex_.load(std::memory_order_acquire)) {
    new (&records_[currentWrite]) T(std::forward<Args>(recordArgs)...);
    writeIndex_.store(nextRecord, std::memory_order_release);
    return true;
  }
  // queue is full
 
  ++overflow_ctr;
 
  return false;
}

Member Data Documentation

alignment_size_

template<class T >

std::size_t dunedaq::datahandlinglibs::IterableQueueModel< T >::alignment_size_

protected

Definition at line 328 of file IterableQueueModel.hpp.

intrinsic_allocator_

template<class T >

bool dunedaq::datahandlinglibs::IterableQueueModel< T >::intrinsic_allocator_

protected

Definition at line 327 of file IterableQueueModel.hpp.

invalid_configuration_requested_

template<class T >

bool dunedaq::datahandlinglibs::IterableQueueModel< T >::invalid_configuration_requested_

protected

Definition at line 329 of file IterableQueueModel.hpp.

numa_aware_

template<class T >

bool dunedaq::datahandlinglibs::IterableQueueModel< T >::numa_aware_

protected

Definition at line 325 of file IterableQueueModel.hpp.

numa_node_

template<class T >

uint8_t dunedaq::datahandlinglibs::IterableQueueModel< T >::numa_node_

protected

Definition at line 326 of file IterableQueueModel.hpp.

overflow_ctr

template<class T >

std::atomic<int> dunedaq::datahandlinglibs::IterableQueueModel< T >::overflow_ctr { 0 }

protected

Definition at line 322 of file IterableQueueModel.hpp.

{ 0 };

pad0_

template<class T >

char dunedaq::datahandlinglibs::IterableQueueModel< T >::pad0_[folly::hardware_destructive_interference_size]

protected

Definition at line 344 of file IterableQueueModel.hpp.

pad1_

template<class T >

char dunedaq::datahandlinglibs::IterableQueueModel< T >::pad1_[folly::hardware_destructive_interference_size - sizeof(writeIndex_)]

protected

Definition at line 351 of file IterableQueueModel.hpp.

prefill_cv_

template<class T >

std::condition_variable dunedaq::datahandlinglibs::IterableQueueModel< T >::prefill_cv_

protected

Definition at line 334 of file IterableQueueModel.hpp.

prefill_done_

template<class T >

bool dunedaq::datahandlinglibs::IterableQueueModel< T >::prefill_done_

protected

Definition at line 336 of file IterableQueueModel.hpp.

prefill_mutex_

template<class T >

std::mutex dunedaq::datahandlinglibs::IterableQueueModel< T >::prefill_mutex_

protected

Definition at line 333 of file IterableQueueModel.hpp.

prefill_ready_

template<class T >

bool dunedaq::datahandlinglibs::IterableQueueModel< T >::prefill_ready_

protected

Definition at line 335 of file IterableQueueModel.hpp.

prefiller_name_

template<class T >

std::string dunedaq::datahandlinglibs::IterableQueueModel< T >::prefiller_name_ {"lbpfn"}

protected

Definition at line 332 of file IterableQueueModel.hpp.

{"lbpfn"};

ptrlogger

template<class T >

std::thread dunedaq::datahandlinglibs::IterableQueueModel< T >::ptrlogger

protected

Definition at line 339 of file IterableQueueModel.hpp.

readIndex_

template<class T >

std::atomic<unsigned int> dunedaq::datahandlinglibs::IterableQueueModel< T >::readIndex_

protected

Definition at line 348 of file IterableQueueModel.hpp.

records_

template<class T >

T* dunedaq::datahandlinglibs::IterableQueueModel< T >::records_

protected

Definition at line 346 of file IterableQueueModel.hpp.

size_

template<class T >

uint32_t dunedaq::datahandlinglibs::IterableQueueModel< T >::size_

protected

Definition at line 345 of file IterableQueueModel.hpp.

writeIndex_

template<class T >

std::atomic<unsigned int> dunedaq::datahandlinglibs::IterableQueueModel< T >::writeIndex_

protected

Definition at line 350 of file IterableQueueModel.hpp.

---

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

• /github/workspace/dunedaq/sourcecode/datahandlinglibs/include/datahandlinglibs/models/IterableQueueModel.hpp
• /github/workspace/dunedaq/sourcecode/datahandlinglibs/include/datahandlinglibs/models/detail/IterableQueueModel.hxx