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

#include <NetworkReceiverModel.hpp>

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

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

Public Member Functions
	NetworkReceiverModel (ConnectionId const &conn_id)
	~NetworkReceiverModel ()
	NetworkReceiverModel (NetworkReceiverModel &&other)
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 topic) override
void	unsubscribe (std::string topic) 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 Member Functions
void	get_receiver (Receiver::timeout_t timeout)
template<typename MessageType >
std::enable_if< serialization::is_serializable< MessageType >::value, MessageType >::type	read_network (Receiver::timeout_t const &timeout)
template<typename MessageType >
std::enable_if<!serialization::is_serializable< MessageType >::value, MessageType >::type	read_network (Receiver::timeout_t const &)
template<typename MessageType >
std::enable_if< serialization::is_serializable< MessageType >::value, std::optional< MessageType > >::type	try_read_network (Receiver::timeout_t const &timeout)
template<typename MessageType >
std::enable_if<!serialization::is_serializable< MessageType >::value, std::optional< MessageType > >::type	try_read_network (Receiver::timeout_t const &)
template<typename MessageType >
std::enable_if< serialization::is_serializable< MessageType >::value, void >::type	add_callback_impl (std::function< void(MessageType &)> callback)
template<typename MessageType >
std::enable_if<!serialization::is_serializable< MessageType >::value, void >::type	add_callback_impl (std::function< void(MessageType &)>)

Private Attributes
std::function< void(Datatype &)>	m_callback
std::unique_ptr< std::jthread >	m_event_loop_runner
std::shared_ptr< ipm::Receiver >	m_network_receiver_ptr { nullptr }
std::mutex	m_callback_mutex
std::mutex	m_receive_mutex

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::NetworkReceiverModel< Datatype >

Definition at line 23 of file NetworkReceiverModel.hpp.

Constructor & Destructor Documentation

NetworkReceiverModel() [1/2]

template<typename Datatype >

dunedaq::iomanager::NetworkReceiverModel< Datatype >::NetworkReceiverModel ( ConnectionId const & conn_id )

inlineexplicit

Definition at line 21 of file NetworkReceiverModel.hxx.

  : ReceiverConcept<Datatype>(conn_id)
{
  TLOG() << "NetworkReceiverModel created with DT! ID: " << conn_id.uid << " Addr: " << static_cast<void*>(this);
  try {
    get_receiver(std::chrono::milliseconds(1000));
  } catch (ConnectionNotFound const& ex) {
    TLOG() << "Initial connection attempt failed: " << ex;
  }
}

~NetworkReceiverModel()

template<typename Datatype >

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

inline

Definition at line 27 of file NetworkReceiverModel.hpp.

{ remove_callback(); }

NetworkReceiverModel() [2/2]

template<typename Datatype >

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

inline

Definition at line 33 of file NetworkReceiverModel.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_network_receiver_ptr(std::move(other.m_network_receiver_ptr))
{
}

Member Function Documentation

add_callback()

template<typename Datatype >

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

inlineoverridevirtual

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

Definition at line 37 of file NetworkReceiverModel.hpp.

{ add_callback_impl<Datatype>(callback); }

add_callback_impl() [1/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!serialization::is_serializable< MessageType >::value, void >::type dunedaq::iomanager::NetworkReceiverModel< Datatype >::add_callback_impl ( std::function< void(MessageType &)> callback )

inlineprivate

Definition at line 167 of file NetworkReceiverModel.hxx.

{
  remove_callback();
  {
    // This ensures that add_callback_impl and remove_callback are not processing concurrently
    std::lock_guard<std::mutex> lk(m_callback_mutex);
  }
  TLOG() << "Registering callback.";
  m_callback = callback;
  // start event loop (thread that calls when receive happens). remove_callback() is called in the destructor, so this
  // will never go out-of-scope while this is running
  m_event_loop_runner = std::make_unique<std::jthread>([&](std::stop_token token) {
    std::optional<Datatype> message;
    while (!token.stop_requested() || message) {
      try {
        // 0 timeout when we are trying to stop
        message = try_read_network<Datatype>(token.stop_requested() ? std::chrono::milliseconds(0)
                                                                    : std::chrono::milliseconds(20));
        if (message) {
          m_callback(*message);
        }
      } catch (const ers::Issue&) {
        // Intentionally ignoring any ers::Issues that might have been raised
        ;
      }
    }
  });
  auto handle = m_event_loop_runner->native_handle();
  std::string name = "N_" + 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()));
  }
}

add_callback_impl() [2/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!serialization::is_serializable< MessageType >::value, void >::type dunedaq::iomanager::NetworkReceiverModel< Datatype >::add_callback_impl ( std::function< void(MessageType &)> )

private

get_receiver()

template<typename Datatype >

void dunedaq::iomanager::NetworkReceiverModel< Datatype >::get_receiver ( Receiver::timeout_t timeout )

inlineprivate

Definition at line 87 of file NetworkReceiverModel.hxx.

{
  // get network resources
  auto start = std::chrono::steady_clock::now();
  while (m_network_receiver_ptr == nullptr &&
         std::chrono::duration_cast<Receiver::timeout_t>(std::chrono::steady_clock::now() - start) < timeout) {
 
    try {
      m_network_receiver_ptr = NetworkManager::get().get_receiver(this->id());
    } catch (ConnectionNotFound const& ex) {
      m_network_receiver_ptr = nullptr;
      std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
  }
}

read_network() [1/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!serialization::is_serializable< MessageType >::value, MessageType >::type dunedaq::iomanager::NetworkReceiverModel< Datatype >::read_network ( Receiver::timeout_t const & )

private

read_network() [2/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!serialization::is_serializable< MessageType >::value, MessageType >::type dunedaq::iomanager::NetworkReceiverModel< Datatype >::read_network ( Receiver::timeout_t const & timeout )

inlineprivate

Definition at line 106 of file NetworkReceiverModel.hxx.

{
  std::lock_guard<std::mutex> lk(m_receive_mutex);
  get_receiver(timeout);
 
  if (m_network_receiver_ptr == nullptr) {
    throw ConnectionInstanceNotFound(ERS_HERE, this->id().uid);
  }
 
  auto response = m_network_receiver_ptr->receive(timeout);
  if (response.data.size() > 0) {
    return serialization::deserialize<MessageType>(response.data);
  }
 
  throw TimeoutExpired(ERS_HERE, this->id().uid, "network receive", timeout.count());
  return MessageType();
}

receive()

template<typename Datatype >

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

inlineoverridevirtual

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

Definition at line 43 of file NetworkReceiverModel.hxx.

{
  try {
    return read_network<Datatype>(timeout);
  } catch (ipm::ReceiveTimeoutExpired& ex) {
    throw TimeoutExpired(ERS_HERE, this->id().uid, "receive", timeout.count(), ex);
  }
}

remove_callback()

template<typename Datatype >

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

inlineoverridevirtual

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

Definition at line 54 of file NetworkReceiverModel.hxx.

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

subscribe()

template<typename Datatype >

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

inlineoverridevirtual

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

Definition at line 69 of file NetworkReceiverModel.hxx.

{
  if (NetworkManager::get().is_pubsub_connection(this->m_conn)) {
    std::dynamic_pointer_cast<ipm::Subscriber>(m_network_receiver_ptr)->subscribe(topic);
  }
}

try_read_network() [1/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!serialization::is_serializable< MessageType >::value, std::optional< MessageType > >::type dunedaq::iomanager::NetworkReceiverModel< Datatype >::try_read_network ( Receiver::timeout_t const & )

private

try_read_network() [2/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!serialization::is_serializable< MessageType >::value, std::optional< MessageType > >::type dunedaq::iomanager::NetworkReceiverModel< Datatype >::try_read_network ( Receiver::timeout_t const & timeout )

inlineprivate

Definition at line 136 of file NetworkReceiverModel.hxx.

{
  std::lock_guard<std::mutex> lk(m_receive_mutex);
  get_receiver(timeout);
  if (m_network_receiver_ptr == nullptr) {
    TLOG_DEBUG(5) << ConnectionInstanceNotFound(ERS_HERE, this->id().uid);
    return std::nullopt;
  }
 
  ipm::Receiver::Response res;
  res = m_network_receiver_ptr->receive(timeout, ipm::Receiver::s_any_size, true);
 
  if (res.data.size() > 0) {
    return std::make_optional<MessageType>(serialization::deserialize<MessageType>(res.data));
  }
 
  return std::nullopt;
}

try_receive()

template<typename Datatype >

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

inlineoverridevirtual

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

Definition at line 33 of file NetworkReceiverModel.hpp.

  {
    return try_read_network<Datatype>(timeout);
  }

unsubscribe()

template<typename Datatype >

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

inlineoverridevirtual

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

Definition at line 78 of file NetworkReceiverModel.hxx.

{
  if (NetworkManager::get().is_pubsub_connection(this->m_conn)) {
    std::dynamic_pointer_cast<ipm::Subscriber>(m_network_receiver_ptr)->unsubscribe(topic);
  }
}

Member Data Documentation

m_callback

template<typename Datatype >

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

private

Definition at line 71 of file NetworkReceiverModel.hpp.

m_callback_mutex

template<typename Datatype >

std::mutex dunedaq::iomanager::NetworkReceiverModel< Datatype >::m_callback_mutex

private

Definition at line 74 of file NetworkReceiverModel.hpp.

m_event_loop_runner

template<typename Datatype >

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

private

Definition at line 72 of file NetworkReceiverModel.hpp.

m_network_receiver_ptr

template<typename Datatype >

std::shared_ptr<ipm::Receiver> dunedaq::iomanager::NetworkReceiverModel< Datatype >::m_network_receiver_ptr { nullptr }

private

Definition at line 73 of file NetworkReceiverModel.hpp.

{ nullptr };

m_receive_mutex

template<typename Datatype >

std::mutex dunedaq::iomanager::NetworkReceiverModel< Datatype >::m_receive_mutex

private

Definition at line 75 of file NetworkReceiverModel.hpp.

---

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

• /github/workspace/dunedaq/sourcecode/iomanager/include/iomanager/network/NetworkReceiverModel.hpp
• /github/workspace/dunedaq/sourcecode/iomanager/include/iomanager/network/detail/NetworkReceiverModel.hxx