dunedaq::iomanager::QueueReceiverModel< Datatype > Class Template Reference

#include <QueueReceiverModel.hpp>

Inheritance diagram for dunedaq::iomanager::QueueReceiverModel< Datatype >:

Collaboration diagram for dunedaq::iomanager::QueueReceiverModel< Datatype >:

Public Member Functions
	QueueReceiverModel (ConnectionId const &request)
	QueueReceiverModel (QueueReceiverModel &&other)
	~QueueReceiverModel ()
Datatype	receive (Receiver::timeout_t timeout) override
std::optional< Datatype >	try_receive (Receiver::timeout_t timeout) override
void	add_callback (std::function< void(Datatype &)> callback) override
void	remove_callback () override
void	subscribe (std::string) override
void	unsubscribe (std::string) override
Public Member Functions inherited from dunedaq::iomanager::ReceiverConcept< Datatype >
	ReceiverConcept (ConnectionId const &conn_id)
Public Member Functions inherited from dunedaq::iomanager::Receiver
	Receiver (ConnectionId const &this_conn)
virtual	~Receiver ()=default
ConnectionId	id () const
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.

Private Attributes
std::function< void(Datatype &)>	m_callback
std::unique_ptr< std::jthread >	m_event_loop_runner
std::shared_ptr< Queue< Datatype > >	m_queue

Additional Inherited Members
Public Types inherited from dunedaq::iomanager::Receiver
using	timeout_t = std::chrono::milliseconds
Static Public Attributes inherited from dunedaq::iomanager::Receiver
static constexpr timeout_t	s_block = timeout_t::max()
static constexpr timeout_t	s_no_block = timeout_t::zero()
Protected Attributes inherited from dunedaq::iomanager::Receiver
ConnectionId	m_conn

Detailed Description

template<typename Datatype>
class dunedaq::iomanager::QueueReceiverModel< Datatype >

Definition at line 27 of file QueueReceiverModel.hpp.

Constructor & Destructor Documentation

QueueReceiverModel() [1/2]

template<typename Datatype >

dunedaq::iomanager::QueueReceiverModel< Datatype >::QueueReceiverModel ( ConnectionId const & request )

inlineexplicit

Definition at line 20 of file QueueReceiverModel.hxx.

  : ReceiverConcept<Datatype>(request)
{
  TLOG() << "QueueReceiverModel created with DT! Addr: " << this;
  // get queue ref from queueregistry based on conn_id
  // std::string sink_name = conn_id to sink_name;
  // m_source = std::make_unique<appfwk::DAQSource<Datatype>>(sink_name);
  m_queue = QueueRegistry::get().get_queue<Datatype>(request.uid);
  TLOG() << "QueueReceiverModel m_queue=" << static_cast<void*>(m_queue.get());
}

QueueReceiverModel() [2/2]

template<typename Datatype >

dunedaq::iomanager::QueueReceiverModel< Datatype >::QueueReceiverModel ( QueueReceiverModel< Datatype > && other )

inline

Definition at line 32 of file QueueReceiverModel.hxx.

  : ReceiverConcept<Datatype>(other.m_conn.uid)
  , m_callback(std::move(other.m_callback))
  , m_event_loop_runner(std::move(other.m_event_loop_runner))
  , m_queue(std::move(other.m_queue))
{
}

~QueueReceiverModel()

template<typename Datatype >

dunedaq::iomanager::QueueReceiverModel< Datatype >::~QueueReceiverModel ( )

inline

Definition at line 34 of file QueueReceiverModel.hpp.

{ remove_callback(); }

Member Function Documentation

add_callback()

template<typename Datatype >

void dunedaq::iomanager::QueueReceiverModel< Datatype >::add_callback ( std::function< void(Datatype &)> callback )

inlineoverridevirtual

Implements dunedaq::iomanager::ReceiverConcept< Datatype >.

Definition at line 87 of file QueueReceiverModel.hxx.

{
  remove_callback();
  TLOG() << "Registering callback.";
  m_callback = callback;
  // start event loop (thread that calls when receive happens)
  m_event_loop_runner = std::make_unique<std::jthread>([&](std::stop_token token) {
    Datatype dt;
    bool ret = true;
    while (!token.stop_requested() || ret) {
      // TLOG() << "Take data from q then invoke callback...";
      ret = m_queue->try_pop(dt, token.stop_requested() ? std::chrono::milliseconds(0) 
          : std::chrono::milliseconds(1));
      if (ret) {
        m_callback(dt);
      }
    }
  });
  auto handle = m_event_loop_runner->native_handle();
  std::string name = "Q_" + this->id().uid;
  name.resize(15);
  auto rc = pthread_setname_np(handle, name.c_str());
  if (rc != 0) {
    std::ostringstream s;
    s << "The name " << name << " provided for the thread is too long.";
    ers::warning(utilities::ThreadingIssue(ERS_HERE, s.str()));
  }
}

receive()

template<typename Datatype >

Datatype dunedaq::iomanager::QueueReceiverModel< Datatype >::receive ( Receiver::timeout_t timeout )

inlineoverridevirtual

Implements dunedaq::iomanager::ReceiverConcept< Datatype >.

Definition at line 42 of file QueueReceiverModel.hxx.

{
  if (m_event_loop_runner != nullptr) {
    TLOG() << "QueueReceiver model is equipped with callback! Ignoring receive call.";
    throw ReceiveCallbackConflict(ERS_HERE, this->id().uid);
  }
  if (m_queue == nullptr) {
    throw ConnectionInstanceNotFound(ERS_HERE, this->id().uid);
  }
  // TLOG() << "Hand off data...";
  Datatype dt;
  try {
    m_queue->pop(dt, timeout);
  } catch (QueueTimeoutExpired& ex) {
    throw TimeoutExpired(ERS_HERE, this->id().uid, "pop", timeout.count(), ex);
  }
  return dt;
  // if (m_queue->write(
}

remove_callback()

template<typename Datatype >

void dunedaq::iomanager::QueueReceiverModel< Datatype >::remove_callback ( )

inlineoverridevirtual

Implements dunedaq::iomanager::ReceiverConcept< Datatype >.

Definition at line 118 of file QueueReceiverModel.hxx.

{
  if (m_event_loop_runner != nullptr && m_event_loop_runner->joinable()) {
    m_event_loop_runner->request_stop();
    m_event_loop_runner->join();
    m_event_loop_runner.reset(nullptr);
  } else if (m_event_loop_runner != nullptr) {
    TLOG() << "Event loop can't be closed!";
  }
  // remove function.
}

subscribe()

template<typename Datatype >

void dunedaq::iomanager::QueueReceiverModel< Datatype >::subscribe ( std::string )

inlineoverridevirtual

Implements dunedaq::iomanager::ReceiverConcept< Datatype >.

Definition at line 45 of file QueueReceiverModel.hpp.

{}

try_receive()

template<typename Datatype >

std::optional< Datatype > dunedaq::iomanager::QueueReceiverModel< Datatype >::try_receive ( Receiver::timeout_t timeout )

inlineoverridevirtual

Implements dunedaq::iomanager::ReceiverConcept< Datatype >.

Definition at line 64 of file QueueReceiverModel.hxx.

{
  if (m_event_loop_runner != nullptr) {
    TLOG() << "QueueReceiver model is equipped with callback! Ignoring receive call.";
    ers::error(ReceiveCallbackConflict(ERS_HERE, this->id().uid));
    return std::nullopt;
  }
  if (m_queue == nullptr) {
    ers::error(ConnectionInstanceNotFound(ERS_HERE, this->id().uid));
    return std::nullopt;
  }
  // TLOG() << "Hand off data...";
  Datatype dt;
  auto ret = m_queue->try_pop(dt, timeout);
  if (ret) {
    return std::make_optional(std::move(dt));
  }
  return std::nullopt;
  // if (m_queue->write(
}

unsubscribe()

template<typename Datatype >

void dunedaq::iomanager::QueueReceiverModel< Datatype >::unsubscribe ( std::string )

inlineoverridevirtual

Implements dunedaq::iomanager::ReceiverConcept< Datatype >.

Definition at line 46 of file QueueReceiverModel.hpp.

{}

Member Data Documentation

m_callback

template<typename Datatype >

std::function<void(Datatype&)> dunedaq::iomanager::QueueReceiverModel< Datatype >::m_callback

private

Definition at line 49 of file QueueReceiverModel.hpp.

m_event_loop_runner

template<typename Datatype >

std::unique_ptr<std::jthread> dunedaq::iomanager::QueueReceiverModel< Datatype >::m_event_loop_runner

private

Definition at line 50 of file QueueReceiverModel.hpp.

m_queue

template<typename Datatype >

std::shared_ptr<Queue<Datatype> > dunedaq::iomanager::QueueReceiverModel< Datatype >::m_queue

private

Definition at line 51 of file QueueReceiverModel.hpp.

---

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

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