dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend > Class Template Reference

#include <WIBEthTransmitter.hpp>

Collaboration diagram for dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >:

Classes
struct	StreamStats

Public Types
using	post_tx_observer_t = std::function<void(const std::uint8_t*, std::size_t)>

Public Member Functions
	WIBEthTransmitter ()=default
	~WIBEthTransmitter ()
	WIBEthTransmitter (const WIBEthTransmitter &)=delete
WIBEthTransmitter &	operator= (const WIBEthTransmitter &)=delete
	WIBEthTransmitter (WIBEthTransmitter &&)=delete
WIBEthTransmitter &	operator= (WIBEthTransmitter &&)=delete
void	add_stream (const std::string &callback_id, std::uint64_t initial_seq_id)
void	configure (const wibeth::PacketConfig &packet_config, std::uint64_t det_id, const Backend &backend)
void	set_post_tx_observer (post_tx_observer_t observer)
void	set_accepting (bool accepting)
bool	accepting () const
void	tombstone ()
bool	armed () const
template<typename PayloadT, typename Registry, typename ConfPtr>
void	register_streams (Registry &registry, const std::vector< ConfPtr > &callback_confs)
template<typename PayloadT>
void	transmit (const std::string &callback_id, PayloadT &&payload)
std::map< std::string, StreamStats >	snapshot_stats () const
std::uint64_t	unknown_stream_drops () const
Backend &	backend ()

Private Attributes
std::mutex	m_mutex
std::map< std::string, StreamStats >	m_streams
std::shared_ptr< CallbackGate< WIBEthTransmitter > >	m_gate = std::make_shared<CallbackGate<WIBEthTransmitter>>()
bool	m_streams_registered = false
std::atomic< bool >	m_accepting { false }
std::atomic< std::uint64_t >	m_unknown_stream_drops { 0 }
Backend	m_backend {}
wibeth::PacketConfig	m_packet_config {}
std::uint64_t	m_det_id = 0
post_tx_observer_t	m_post_tx_observer
std::array< std::uint8_t, wibeth::kEthernetPacketBytes >	m_observer_scratch {}

Detailed Description

template<typename Backend>
class dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >

Definition at line 52 of file WIBEthTransmitter.hpp.

Member Typedef Documentation

post_tx_observer_t

template<typename Backend>

using dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::post_tx_observer_t = std::function<void(const std::uint8_t*, std::size_t)>

Definition at line 58 of file WIBEthTransmitter.hpp.

Constructor & Destructor Documentation

WIBEthTransmitter() [1/3]

template<typename Backend>

dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::WIBEthTransmitter ( )

default

~WIBEthTransmitter()

template<typename Backend>

dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::~WIBEthTransmitter ( )

inline

Definition at line 72 of file WIBEthTransmitter.hpp.

{ m_gate->tombstone(); }

WIBEthTransmitter() [2/3]

template<typename Backend>

dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::WIBEthTransmitter ( const WIBEthTransmitter< Backend > & )

delete

WIBEthTransmitter() [3/3]

template<typename Backend>

dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::WIBEthTransmitter ( WIBEthTransmitter< Backend > && )

delete

Member Function Documentation

accepting()

template<typename Backend>

bool dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::accepting ( ) const

inline

Definition at line 105 of file WIBEthTransmitter.hpp.

{ return m_accepting.load(); }

add_stream()

template<typename Backend>

void dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::add_stream ( const std::string & callback_id, std::uint64_t initial_seq_id )

inline

Definition at line 81 of file WIBEthTransmitter.hpp.

  {
    std::lock_guard<std::mutex> guard(m_mutex);
    StreamStats stats;
    stats.next_seq = initial_seq_id % 4096;
    m_streams.emplace(callback_id, stats);
  }

armed()

template<typename Backend>

bool dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::armed ( ) const

inline

Definition at line 110 of file WIBEthTransmitter.hpp.

{ return m_gate->armed(); }

backend()

template<typename Backend>

Backend & dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::backend ( )

inline

Definition at line 206 of file WIBEthTransmitter.hpp.

{ return m_backend; }

configure()

template<typename Backend>

void dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::configure ( const wibeth::PacketConfig & packet_config, std::uint64_t det_id, const Backend & backend )

inline

Definition at line 89 of file WIBEthTransmitter.hpp.

  {
    std::lock_guard<std::mutex> guard(m_mutex);
    m_packet_config = packet_config;
    m_det_id = det_id;
    m_backend = backend;
  }

operator=() [1/2]

template<typename Backend>

WIBEthTransmitter & dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::operator= ( const WIBEthTransmitter< Backend > & )

delete

operator=() [2/2]

template<typename Backend>

WIBEthTransmitter & dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::operator= ( WIBEthTransmitter< Backend > && )

delete

register_streams()

template<typename Backend>

template<typename PayloadT, typename Registry, typename ConfPtr>

void dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::register_streams ( Registry & registry, const std::vector< ConfPtr > & callback_confs )

inline

Definition at line 118 of file WIBEthTransmitter.hpp.

  {
    if (m_streams_registered) {
      m_gate->arm(this);
      return;
    }
    for (const auto& conf : callback_confs) {
      const std::string callback_id = conf->UID();
      // The lambda captures the gate, not the transmitter.  If the
      // process-global registry outlives this object, dispatch becomes a drop
      // rather than a call through a dangling pointer.
      auto gate = m_gate;
      registry.template register_callback<PayloadT>(conf, [gate, callback_id](PayloadT&& payload) {
        gate->dispatch([&](WIBEthTransmitter& owner) { owner.transmit(callback_id, std::move(payload)); });
      });
    }
    m_streams_registered = true;
    m_gate->arm(this);
  }

set_accepting()

template<typename Backend>

void dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::set_accepting ( bool accepting )

inline

Definition at line 104 of file WIBEthTransmitter.hpp.

{ m_accepting.store(accepting); }

set_post_tx_observer()

template<typename Backend>

void dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::set_post_tx_observer ( post_tx_observer_t observer )

inline

Definition at line 98 of file WIBEthTransmitter.hpp.

  {
    std::lock_guard<std::mutex> guard(m_mutex);
    m_post_tx_observer = std::move(observer);
  }

snapshot_stats()

template<typename Backend>

std::map< std::string, StreamStats > dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::snapshot_stats ( ) const

inline

Definition at line 196 of file WIBEthTransmitter.hpp.

  {
    std::lock_guard<std::mutex> guard(m_mutex);
    return m_streams;
  }

tombstone()

template<typename Backend>

void dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::tombstone ( )

inline

Definition at line 109 of file WIBEthTransmitter.hpp.

{ m_gate->tombstone(); }

transmit()

template<typename Backend>

template<typename PayloadT>

void dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::transmit ( const std::string & callback_id, PayloadT && payload )

inline

Definition at line 139 of file WIBEthTransmitter.hpp.

  {
    std::lock_guard<std::mutex> guard(m_mutex);
    auto state_it = m_streams.find(callback_id);
    if (state_it == m_streams.end()) {
      ++m_unknown_stream_drops;
      return;
    }
    auto& state = state_it->second;
    ++state.accepted;
 
    if (!m_accepting.load()) {
      ++state.dropped_not_running;
      return;
    }
 
    auto packet_config = m_packet_config;
    packet_config.packet_id = static_cast<std::uint16_t>(state.sent & 0xffffU);
 
    wibeth::HeaderPatch patch;
    patch.set_det_id = true;
    patch.det_id = m_det_id;
    patch.set_seq_id = true;
    patch.seq_id = state.next_seq;
    patch.force_block_length = true;
 
    const bool want_observer = static_cast<bool>(m_post_tx_observer);
    std::uint8_t* pre_tx_copy = want_observer ? m_observer_scratch.data() : nullptr;
 
    const auto outcome = send_packet(
      m_backend,
      wibeth::kEthernetPacketBytes,
      pre_tx_copy,
      [&](std::uint8_t* dst) { wibeth::construct_packet(dst, payload.data, packet_config, patch); },
      [&]() {
        ++state.sent;
        state.next_seq = (state.next_seq + 1) % 4096;
      });
 
    switch (outcome) {
      case TxOutcome::kSent:
        if (want_observer) {
          m_post_tx_observer(m_observer_scratch.data(), wibeth::kEthernetPacketBytes);
        }
        break;
      case TxOutcome::kAllocFailed:
        ++state.tx_alloc_failures;
        break;
      case TxOutcome::kPrepareFailed:
        ++state.tx_prepare_failures;
        break;
      case TxOutcome::kTxFailed:
        ++state.tx_failures;
        break;
    }
  }

unknown_stream_drops()

template<typename Backend>

std::uint64_t dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::unknown_stream_drops ( ) const

inline

Definition at line 202 of file WIBEthTransmitter.hpp.

{ return m_unknown_stream_drops.load(); }

Member Data Documentation

m_accepting

template<typename Backend>

std::atomic<bool> dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::m_accepting { false }

private

Definition at line 215 of file WIBEthTransmitter.hpp.

m_backend

template<typename Backend>

Backend dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::m_backend {}

private

Definition at line 218 of file WIBEthTransmitter.hpp.

m_det_id

template<typename Backend>

std::uint64_t dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::m_det_id = 0

private

Definition at line 220 of file WIBEthTransmitter.hpp.

m_gate

template<typename Backend>

std::shared_ptr<CallbackGate<WIBEthTransmitter> > dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::m_gate = std::make_shared<CallbackGate<WIBEthTransmitter>>()

private

Definition at line 212 of file WIBEthTransmitter.hpp.

m_mutex

template<typename Backend>

std::mutex dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::m_mutex

mutableprivate

Definition at line 209 of file WIBEthTransmitter.hpp.

m_observer_scratch

template<typename Backend>

std::array<std::uint8_t, wibeth::kEthernetPacketBytes> dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::m_observer_scratch {}

private

Definition at line 225 of file WIBEthTransmitter.hpp.

m_packet_config

template<typename Backend>

wibeth::PacketConfig dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::m_packet_config {}

private

Definition at line 219 of file WIBEthTransmitter.hpp.

m_post_tx_observer

template<typename Backend>

post_tx_observer_t dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::m_post_tx_observer

private

Definition at line 222 of file WIBEthTransmitter.hpp.

m_streams

template<typename Backend>

std::map<std::string, StreamStats> dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::m_streams

private

Definition at line 210 of file WIBEthTransmitter.hpp.

m_streams_registered

template<typename Backend>

bool dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::m_streams_registered = false

private

Definition at line 213 of file WIBEthTransmitter.hpp.

m_unknown_stream_drops

template<typename Backend>

std::atomic<std::uint64_t> dunedaq::dpdklibs::sender::WIBEthTransmitter< Backend >::m_unknown_stream_drops { 0 }

private

Definition at line 216 of file WIBEthTransmitter.hpp.

---

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

• /github/workspace/dunedaq/sourcecode/dpdklibs/include/dpdklibs/sender/WIBEthTransmitter.hpp