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

#include <NetworkSenderModel.hpp>

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

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

Public Member Functions
	NetworkSenderModel (ConnectionId const &conn_id)
	NetworkSenderModel (NetworkSenderModel &&other)
void	send (Datatype &&data, Sender::timeout_t timeout) override
bool	try_send (Datatype &&data, Sender::timeout_t timeout) override
void	send_with_topic (Datatype &&data, Sender::timeout_t timeout, std::string topic) override
bool	is_ready_for_sending (Sender::timeout_t timeout) override
Public Member Functions inherited from dunedaq::iomanager::SenderConcept< Datatype >
	SenderConcept (ConnectionId const &conn_id)
Public Member Functions inherited from dunedaq::iomanager::Sender
	Sender (ConnectionId const &this_conn)
virtual	~Sender ()=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_sender (Sender::timeout_t const &timeout)
template<typename MessageType >
std::enable_if< dunedaq::serialization::is_serializable< MessageType >::value, void >::type	write_network (MessageType &message, Sender::timeout_t const &timeout)
template<typename MessageType >
std::enable_if<!dunedaq::serialization::is_serializable< MessageType >::value, void >::type	write_network (MessageType &, Sender::timeout_t const &)
template<typename MessageType >
std::enable_if< dunedaq::serialization::is_serializable< MessageType >::value, bool >::type	try_write_network (MessageType &message, Sender::timeout_t const &timeout)
template<typename MessageType >
std::enable_if<!dunedaq::serialization::is_serializable< MessageType >::value, bool >::type	try_write_network (MessageType &, Sender::timeout_t const &)
template<typename MessageType >
std::enable_if< dunedaq::serialization::is_serializable< MessageType >::value, void >::type	write_network_with_topic (MessageType &message, Sender::timeout_t const &timeout, std::string topic)
template<typename MessageType >
std::enable_if<!dunedaq::serialization::is_serializable< MessageType >::value, void >::type	write_network_with_topic (MessageType &, Sender::timeout_t const &, std::string)
Sender::timeout_t	extend_first_timeout (Sender::timeout_t timeout)

Private Attributes
std::shared_ptr< ipm::Sender >	m_network_sender_ptr
std::mutex	m_send_mutex
std::string	m_topic { "" }
std::atomic< bool >	m_first { true }

Additional Inherited Members
Public Types inherited from dunedaq::iomanager::Sender
using	timeout_t = std::chrono::milliseconds
Static Public Attributes inherited from dunedaq::iomanager::Sender
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::Sender
ConnectionId	m_conn

Detailed Description

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

Definition at line 25 of file NetworkSenderModel.hpp.

Constructor & Destructor Documentation

NetworkSenderModel() [1/2]

template<typename Datatype >

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

inlineexplicit

Definition at line 17 of file NetworkSenderModel.hxx.

  : SenderConcept<Datatype>(conn_id)
{
  TLOG("NetworkSenderModel") << "NetworkSenderModel created with DT! Addr: " << static_cast<void*>(this)
                             << ", uid=" << conn_id.uid << ", data_type=" << conn_id.data_type;
  get_sender(std::chrono::milliseconds(1000));
  if (m_network_sender_ptr == nullptr) {
    TLOG("NetworkSenderModel") << "Initial connection attempt failed for uid=" << conn_id.uid
                               << ", data_type=" << conn_id.data_type;
  }
}

NetworkSenderModel() [2/2]

template<typename Datatype >

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

inline

Definition at line 30 of file NetworkSenderModel.hxx.

  : SenderConcept<Datatype>(other.m_conn.uid)
  , m_network_sender_ptr(std::move(other.m_network_sender_ptr))
  , m_topic(std::move(other.m_topic))
{
}

Member Function Documentation

extend_first_timeout()

template<typename Datatype >

Sender::timeout_t dunedaq::iomanager::NetworkSenderModel< Datatype >::extend_first_timeout ( Sender::timeout_t timeout )

inlineprivate

Definition at line 204 of file NetworkSenderModel.hxx.

{
  if (m_first) {
    m_first = false;
    if (timeout > 1000ms) {
      return timeout;
    }
    return 1000ms;
  }
 
  return timeout;
}

get_sender()

template<typename Datatype >

void dunedaq::iomanager::NetworkSenderModel< Datatype >::get_sender ( Sender::timeout_t const & timeout )

inlineprivate

Definition at line 76 of file NetworkSenderModel.hxx.

{
  auto start = std::chrono::steady_clock::now();
  while (m_network_sender_ptr == nullptr &&
         std::chrono::duration_cast<Sender::timeout_t>(std::chrono::steady_clock::now() - start) <= timeout) {
    // get network resources
    try {
      m_network_sender_ptr = NetworkManager::get().get_sender(this->id());
 
      if (NetworkManager::get().is_pubsub_connection(this->id())) {
        TLOG("NetworkSenderModel") << "Setting topic to " << this->id().data_type;
        m_topic = this->id().data_type;
      }
    } catch (ers::Issue const& ex) {
      m_network_sender_ptr = nullptr;
      std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
  }
}

is_ready_for_sending()

template<typename Datatype >

bool dunedaq::iomanager::NetworkSenderModel< Datatype >::is_ready_for_sending ( Sender::timeout_t timeout )

inlineoverridevirtual

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

Definition at line 68 of file NetworkSenderModel.hxx.

{
  get_sender(timeout);
  return (m_network_sender_ptr != nullptr);
}

send()

template<typename Datatype >

void dunedaq::iomanager::NetworkSenderModel< Datatype >::send ( Datatype && data, Sender::timeout_t timeout )

inlineoverridevirtual

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

Definition at line 39 of file NetworkSenderModel.hxx.

{
  try {
    write_network<Datatype>(data, timeout);
  } catch (ipm::SendTimeoutExpired& ex) {
    throw TimeoutExpired(ERS_HERE, this->id().uid, "send", timeout.count(), ex);
  }
}

send_with_topic()

template<typename Datatype >

void dunedaq::iomanager::NetworkSenderModel< Datatype >::send_with_topic ( Datatype && data, Sender::timeout_t timeout, std::string topic )

inlineoverridevirtual

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

Definition at line 57 of file NetworkSenderModel.hxx.

{
  try {
    write_network_with_topic<Datatype>(data, timeout, topic);
  } catch (ipm::SendTimeoutExpired& ex) {
    throw TimeoutExpired(ERS_HERE, this->id().uid, "send", timeout.count(), ex);
  }
}

try_send()

template<typename Datatype >

bool dunedaq::iomanager::NetworkSenderModel< Datatype >::try_send ( Datatype && data, Sender::timeout_t timeout )

inlineoverridevirtual

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

Definition at line 50 of file NetworkSenderModel.hxx.

{
  return try_write_network<Datatype>(data, timeout);
}

try_write_network() [1/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!dunedaq::serialization::is_serializable< MessageType >::value, bool >::type dunedaq::iomanager::NetworkSenderModel< Datatype >::try_write_network ( MessageType & , Sender::timeout_t const &  )

private

try_write_network() [2/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!dunedaq::serialization::is_serializable< MessageType >::value, bool >::type dunedaq::iomanager::NetworkSenderModel< Datatype >::try_write_network ( MessageType & message, Sender::timeout_t const & timeout )

inlineprivate

Definition at line 134 of file NetworkSenderModel.hxx.

{
  std::lock_guard<std::mutex> lk(m_send_mutex);
  get_sender(timeout);
  if (m_network_sender_ptr == nullptr) {
    TLOG("NetworkSenderModel") << ConnectionInstanceNotFound(ERS_HERE, this->id().uid);
    return false;
  }
 
  auto serialized = dunedaq::serialization::serialize(message, dunedaq::serialization::kMsgPack);
  // TLOG("NetworkSenderModel") << "Serialized message for network sending: " << serialized.size() << ", topic=" <<
  // m_topic <<
  // ", this=" << (void*)this;
 
  auto res =
    m_network_sender_ptr->send(serialized.data(), serialized.size(), extend_first_timeout(timeout), m_topic, true);
  if (!res) {
    TLOG("NetworkSenderModel") << "Timeout detected, removing sender to re-acquire connection";
    NetworkManager::get().remove_sender(this->id());
    m_network_sender_ptr = nullptr;
  }
  return res;
}

write_network() [1/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!dunedaq::serialization::is_serializable< MessageType >::value, void >::type dunedaq::iomanager::NetworkSenderModel< Datatype >::write_network ( MessageType & , Sender::timeout_t const &  )

private

write_network() [2/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!dunedaq::serialization::is_serializable< MessageType >::value, void >::type dunedaq::iomanager::NetworkSenderModel< Datatype >::write_network ( MessageType & message, Sender::timeout_t const & timeout )

inlineprivate

Definition at line 99 of file NetworkSenderModel.hxx.

{
  std::lock_guard<std::mutex> lk(m_send_mutex);
  get_sender(timeout);
  if (m_network_sender_ptr == nullptr) {
    throw TimeoutExpired(
      ERS_HERE, this->id().uid, "send", timeout.count(), ConnectionInstanceNotFound(ERS_HERE, this->id().uid));
  }
 
  auto serialized = dunedaq::serialization::serialize(message, dunedaq::serialization::kMsgPack);
  //  TLOG("NetworkSenderModel") << "Serialized message for network sending: " << serialized.size() << ", topic=" <<
  //  m_topic << ", this="
  //  << (void*)this;
 
  try {
    m_network_sender_ptr->send(serialized.data(), serialized.size(), extend_first_timeout(timeout), m_topic);
  } catch (ipm::SendTimeoutExpired const& ex) {
    TLOG("NetworkSenderModel") << "Timeout detected, removing sender to re-acquire connection";
    NetworkManager::get().remove_sender(this->id());
    m_network_sender_ptr = nullptr;
    throw;
  }
}

write_network_with_topic() [1/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!dunedaq::serialization::is_serializable< MessageType >::value, void >::type dunedaq::iomanager::NetworkSenderModel< Datatype >::write_network_with_topic ( MessageType & , Sender::timeout_t const & , std::string  )

private

write_network_with_topic() [2/2]

template<typename Datatype >

template<typename MessageType >

std::enable_if<!dunedaq::serialization::is_serializable< MessageType >::value, void >::type dunedaq::iomanager::NetworkSenderModel< Datatype >::write_network_with_topic ( MessageType & message, Sender::timeout_t const & timeout, std::string topic )

inlineprivate

Definition at line 170 of file NetworkSenderModel.hxx.

{
  std::lock_guard<std::mutex> lk(m_send_mutex);
  get_sender(timeout);
  if (m_network_sender_ptr == nullptr) {
    throw TimeoutExpired(
      ERS_HERE, this->id().uid, "send", timeout.count(), ConnectionInstanceNotFound(ERS_HERE, this->id().uid));
  }
 
  auto serialized = dunedaq::serialization::serialize(message, dunedaq::serialization::kMsgPack);
  //  TLOG("NetworkSenderModel") << "Serialized message for network sending: " << serialized.size() << ", topic=" <<
  //  m_topic << ", this="
  //  << (void*)this;
 
  try {
    m_network_sender_ptr->send(serialized.data(), serialized.size(), timeout, topic);
  } catch (TimeoutExpired const& ex) {
    m_network_sender_ptr = nullptr;
    throw;
  }
}

Member Data Documentation

m_first

template<typename Datatype >

std::atomic<bool> dunedaq::iomanager::NetworkSenderModel< Datatype >::m_first { true }

private

Definition at line 81 of file NetworkSenderModel.hpp.

{ true };

m_network_sender_ptr

template<typename Datatype >

std::shared_ptr<ipm::Sender> dunedaq::iomanager::NetworkSenderModel< Datatype >::m_network_sender_ptr

private

Definition at line 78 of file NetworkSenderModel.hpp.

m_send_mutex

template<typename Datatype >

std::mutex dunedaq::iomanager::NetworkSenderModel< Datatype >::m_send_mutex

private

Definition at line 79 of file NetworkSenderModel.hpp.

m_topic

template<typename Datatype >

std::string dunedaq::iomanager::NetworkSenderModel< Datatype >::m_topic { "" }

private

Definition at line 80 of file NetworkSenderModel.hpp.

{ "" };

---

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

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