A Queue Implementation that uses a std::deque as its backend. More...

#include <StdDeQueue.hpp>

Inheritance diagram for dunedaq::iomanager::StdDeQueue< T >:

Collaboration diagram for dunedaq::iomanager::StdDeQueue< T >:

Public Types
using	value_t = T
	Type of data stored in the StdDeQueue.

using	duration_t = typename Queue<T>::duration_t
	Type used for expressing timeouts.

Public Types inherited from dunedaq::iomanager::Queue< T >
using	value_t = T
	Type stored in the Queue.

using	duration_t = std::chrono::milliseconds
	Base duration type for timeouts.

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
	StdDeQueue (const std::string &name, size_t capacity)
	StdDeQueue Constructor.

bool	can_pop () const noexcept override
	Determine whether the Queue may be popped from.

void	pop (value_t &val, const duration_t &) override
	Pop the first value off of the queue.

bool	try_pop (value_t &val, const duration_t &) override

bool	can_push () const noexcept override
	Determine whether the Queue may be pushed onto.

void	push (value_t &&, const duration_t &) override
	Push a value onto the Queue.

bool	try_push (value_t &&, const duration_t &) override

size_t	get_capacity () const override
	Get the capacity (max size) of the queue.

size_t	get_num_elements () const override

	StdDeQueue (const StdDeQueue &)=delete
	StdDeQueue is not copy-constructible.

StdDeQueue &	operator= (const StdDeQueue &)=delete
	StdDeQueue is not copy-assignable.

	StdDeQueue (StdDeQueue &&)=delete
	StdDeQueue is not move-constructible.

StdDeQueue &	operator= (StdDeQueue &&)=delete
	StdDeQueue is not move-assignable.

Public Member Functions inherited from dunedaq::iomanager::Queue< T >
	Queue (const std::string &name)
	Queue Constructor.

Public Member Functions inherited from dunedaq::iomanager::QueueBase
	QueueBase (const std::string &name)
	QueueBase Constructor.

Public Member Functions inherited from dunedaq::utilities::NamedObject
	NamedObject (const std::string &name)
	NamedObject Constructor.

	NamedObject (NamedObject const &)=delete
	NamedObject is not copy-constructible.

	NamedObject (NamedObject &&)=default
	NamedObject is move-constructible.

NamedObject &	operator= (NamedObject const &)=delete
	NamedObject is not copy-assignable.

NamedObject &	operator= (NamedObject &&)=default
	NamedObject is move-assignable.

virtual	~NamedObject ()=default
	Default virtual destructor.

const std::string &	get_name () const final
	Get the name of this NamedObejct.

Public Member Functions inherited from dunedaq::utilities::Named
	Named ()=default
	Named Constructor.

	Named (Named const &)=delete
	Named is not copy-constructible.

	Named (Named &&)=default
	Named is move-constructible.

Named &	operator= (Named const &)=delete
	Named is not copy-assignable.

Named &	operator= (Named &&)=default
	Named is move-assignable.

virtual	~Named ()=default
	Default virtual destructor.

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

Private Member Functions
void	try_lock_for (std::unique_lock< std::mutex > &, const duration_t &)

Private Attributes
std::deque< value_t >	m_deque

size_t	m_capacity

std::atomic< size_t >	m_size = 0

std::mutex	m_mutex

std::condition_variable	m_no_longer_full

std::condition_variable	m_no_longer_empty

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

Protected Member Functions inherited from dunedaq::iomanager::QueueBase
void	generate_opmon_data () override
	Method to retrieve information (occupancy) from queues.

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

Detailed Description

template<class T>
class dunedaq::iomanager::StdDeQueue< T >

A Queue Implementation that uses a std::deque as its backend.

Template Parameters

T	Data Type to be stored in the std::deque

Definition at line 37 of file StdDeQueue.hpp.

Member Typedef Documentation

◆ duration_t

template<class T >

using dunedaq::iomanager::StdDeQueue< T >::duration_t = typename Queue<T>::duration_t

Type used for expressing timeouts.

Definition at line 41 of file StdDeQueue.hpp.

◆ value_t

template<class T >

using dunedaq::iomanager::StdDeQueue< T >::value_t = T

Type of data stored in the StdDeQueue.

Definition at line 40 of file StdDeQueue.hpp.

Constructor & Destructor Documentation

◆ StdDeQueue() [1/3]

template<class T >

dunedaq::iomanager::StdDeQueue< T >::StdDeQueue	(	const std::string &	name,
		size_t	capacity )

explicit

StdDeQueue Constructor.

Parameters

name	Name of this StdDeQueue instance

Definition at line 7 of file StdDeQueue.hxx.

  : Queue<T>(name)
  , m_deque()
  , m_capacity(capacity)
  , m_size(0)
{
  assert(m_deque.max_size() > this->get_capacity());
}

◆ StdDeQueue() [2/3]

template<class T >

dunedaq::iomanager::StdDeQueue< T >::StdDeQueue ( const StdDeQueue< T > & )

delete

StdDeQueue is not copy-constructible.

◆ StdDeQueue() [3/3]

template<class T >

dunedaq::iomanager::StdDeQueue< T >::StdDeQueue ( StdDeQueue< T > && )

delete

StdDeQueue is not move-constructible.

Member Function Documentation

◆ can_pop()

template<class T >

bool dunedaq::iomanager::StdDeQueue< T >::can_pop ( ) const

inlineoverridevirtualnoexcept

Determine whether the Queue may be popped from.

Returns: True if the queue is not empty, false if it is

Reimplemented from dunedaq::iomanager::Queue< T >.

Definition at line 49 of file StdDeQueue.hpp.

49{ return this->get_num_elements() > 0; }

dunedaq::iomanager::StdDeQueue::get_num_elements

size_t get_num_elements() const override

Definition StdDeQueue.hpp:59

◆ can_push()

template<class T >

bool dunedaq::iomanager::StdDeQueue< T >::can_push ( ) const

inlineoverridevirtualnoexcept

Determine whether the Queue may be pushed onto.

Returns: True if the queue is not full, false if it is

Reimplemented from dunedaq::iomanager::Queue< T >.

Definition at line 53 of file StdDeQueue.hpp.

53{ return this->get_num_elements() < this->get_capacity(); }

dunedaq::iomanager::StdDeQueue::get_capacity

size_t get_capacity() const override

Get the capacity (max size) of the queue.

Definition StdDeQueue.hpp:57

◆ get_capacity()

template<class T >

size_t dunedaq::iomanager::StdDeQueue< T >::get_capacity ( ) const

inlineoverridevirtual

Get the capacity (max size) of the queue.

Returns: size_t capacity

Implements dunedaq::iomanager::QueueBase.

Definition at line 57 of file StdDeQueue.hpp.

57{ return m_capacity; }

◆ get_num_elements()

template<class T >

size_t dunedaq::iomanager::StdDeQueue< T >::get_num_elements ( ) const

inlineoverridevirtual

Implements dunedaq::iomanager::QueueBase.

Definition at line 59 of file StdDeQueue.hpp.

59{ return m_size.load(std::memory_order_acquire); }

◆ operator=() [1/2]

template<class T >

StdDeQueue & dunedaq::iomanager::StdDeQueue< T >::operator= ( const StdDeQueue< T > & )

delete

StdDeQueue is not copy-assignable.

◆ operator=() [2/2]

template<class T >

StdDeQueue & dunedaq::iomanager::StdDeQueue< T >::operator= ( StdDeQueue< T > && )

delete

StdDeQueue is not move-assignable.

◆ pop()

template<class T >

void dunedaq::iomanager::StdDeQueue< T >::pop	(	value_t &	val,
		const duration_t &	timeout )

overridevirtual

Pop the first value off of the queue.

Parameters

val	Reference to the value that is popped from the queue
timeout	Timeout for the pop operation

This is a pure virtual function If pop takes longer than the timeout, implementations should throw an exception

Implements dunedaq::iomanager::Queue< T >.

Definition at line 44 of file StdDeQueue.hxx.

{
 
  auto start_time = std::chrono::steady_clock::now();
  std::unique_lock<std::mutex> lk(m_mutex, std::defer_lock);
 
  this->try_lock_for(lk, timeout);
 
  auto time_to_wait_for_data = (start_time + timeout) - std::chrono::steady_clock::now();
 
  if (time_to_wait_for_data.count() > 0) {
    m_no_longer_empty.wait_for(lk, time_to_wait_for_data, [&]() { return this->can_pop(); });
  }
 
  if (this->can_pop()) {
    val = std::move(m_deque.front());
    m_size--;
    m_deque.pop_front();
    m_no_longer_full.notify_one();
  } else {
    throw QueueTimeoutExpired(
      ERS_HERE, this->get_name(), "pop", std::chrono::duration_cast<std::chrono::milliseconds>(timeout).count());
  }
}

◆ push()

template<class T >

void dunedaq::iomanager::StdDeQueue< T >::push	(	value_t &&	val,
		const duration_t &	timeout )

overridevirtual

Push a value onto the Queue.

Parameters

val	Value to push (rvalue)
timeout	Timeout for the push operation.

This is a pure virtual function. If push takes longer than the timeout, implementations should throw an exception.

Implements dunedaq::iomanager::Queue< T >.

Definition at line 18 of file StdDeQueue.hxx.

{
 
  auto start_time = std::chrono::steady_clock::now();
  std::unique_lock<std::mutex> lk(m_mutex, std::defer_lock);
 
  this->try_lock_for(lk, timeout);
 
  auto time_to_wait_for_space = (start_time + timeout) - std::chrono::steady_clock::now();
 
  if (time_to_wait_for_space.count() > 0) {
    m_no_longer_full.wait_for(lk, time_to_wait_for_space, [&]() { return this->can_push(); });
  }
 
  if (this->can_push()) {
    m_deque.push_back(std::move(object_to_push));
    m_size++;
    m_no_longer_empty.notify_one();
  } else {
    throw QueueTimeoutExpired(
      ERS_HERE, this->get_name(), "push", std::chrono::duration_cast<std::chrono::milliseconds>(timeout).count());
  }
}

◆ try_lock_for()

template<class T >

void dunedaq::iomanager::StdDeQueue< T >::try_lock_for	(	std::unique_lock< std::mutex > &	lk,
		const duration_t &	timeout )

private

Definition at line 133 of file StdDeQueue.hxx.

{
  assert(!lk.owns_lock());
 
  auto start_time = std::chrono::steady_clock::now();
  auto ret = lk.try_lock();
 
  if ((!ret || !lk.owns_lock()) && timeout.count() > 0) {
 
    int approximate_number_of_retries = 5;
    duration_t pause_between_tries = duration_t(timeout.count() / approximate_number_of_retries);
 
    while (std::chrono::steady_clock::now() < start_time + timeout) {
      std::this_thread::sleep_for(pause_between_tries);
      ret = lk.try_lock();
      if (ret && lk.owns_lock()) {
        break;
      }
    }
  }
 
  if (!lk.owns_lock()) {
    throw QueueTimeoutExpired(
      ERS_HERE, this->get_name(), "lock mutex", std::chrono::duration_cast<std::chrono::milliseconds>(timeout).count());
  }
}

◆ try_pop()

template<class T >

bool dunedaq::iomanager::StdDeQueue< T >::try_pop	(	value_t &	val,
		const duration_t &	timeout )

overridevirtual

Implements dunedaq::iomanager::Queue< T >.

Definition at line 99 of file StdDeQueue.hxx.

{
 
  auto start_time = std::chrono::steady_clock::now();
  std::unique_lock<std::mutex> lk(m_mutex, std::defer_lock);
 
  this->try_lock_for(lk, timeout);
 
  auto time_to_wait_for_data = (start_time + timeout) - std::chrono::steady_clock::now();
 
  if (time_to_wait_for_data.count() > 0) {
    m_no_longer_empty.wait_for(lk, time_to_wait_for_data, [&]() { return this->can_pop(); });
  }
 
  if (this->can_pop()) {
    val = std::move(m_deque.front());
    m_size--;
    m_deque.pop_front();
    m_no_longer_full.notify_one();
    return true;
  } else {
    ers::error(QueueTimeoutExpired(
      ERS_HERE, this->get_name(), "pop", std::chrono::duration_cast<std::chrono::milliseconds>(timeout).count()));
    return false;
  }
}

◆ try_push()

template<class T >

bool dunedaq::iomanager::StdDeQueue< T >::try_push	(	value_t &&	object_to_push,
		const duration_t &	timeout )

overridevirtual

Implements dunedaq::iomanager::Queue< T >.

Definition at line 71 of file StdDeQueue.hxx.

{
 
  auto start_time = std::chrono::steady_clock::now();
  std::unique_lock<std::mutex> lk(m_mutex, std::defer_lock);
 
  this->try_lock_for(lk, timeout);
 
  auto time_to_wait_for_space = (start_time + timeout) - std::chrono::steady_clock::now();
 
  if (time_to_wait_for_space.count() > 0) {
    m_no_longer_full.wait_for(lk, time_to_wait_for_space, [&]() { return this->can_push(); });
  }
 
  if (this->can_push()) {
    m_deque.push_back(std::move(object_to_push));
    m_size++;
    m_no_longer_empty.notify_one();
    return true;
  } else {
    ers::error(QueueTimeoutExpired(
      ERS_HERE, this->get_name(), "push", std::chrono::duration_cast<std::chrono::milliseconds>(timeout).count()));
    return false;
  }
}

Definition at line 74 of file StdDeQueue.hpp.

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

/github/workspace/dunedaq/sourcecode/iomanager/include/iomanager/queue/StdDeQueue.hpp
/github/workspace/dunedaq/sourcecode/iomanager/include/iomanager/queue/detail/StdDeQueue.hxx

Public Types

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ duration_t

◆ value_t

Constructor & Destructor Documentation

◆ StdDeQueue() [1/3]

◆ StdDeQueue() [2/3]

◆ StdDeQueue() [3/3]

Member Function Documentation

◆ can_pop()

◆ can_push()

◆ get_capacity()

◆ get_num_elements()

◆ operator=() [1/2]

◆ operator=() [2/2]

◆ pop()

◆ push()

◆ try_lock_for()

◆ try_pop()

◆ try_push()

Member Data Documentation

◆ m_capacity

◆ m_deque

◆ m_mutex

◆ m_no_longer_empty

◆ m_no_longer_full

◆ m_size