Inheritance diagram for OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter:

Collaboration diagram for OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter:

[legend]

Public Member Functions
	RtpsWriter (const TransportClient_rch &client, const RtpsUdpDataLink_rch &link, const GUID_t &id, bool durable, SequenceNumber max_sn, CORBA::Long heartbeat_count, size_t capacity)

virtual	~RtpsWriter ()

SequenceNumber	max_data_seq (const SingleSendBuffer::Proxy &proxy, const ReaderInfo_rch &) const

SequenceNumber	update_max_sn (const GUID_t &reader, SequenceNumber seq)

void	add_elem_awaiting_ack (TransportQueueElement *element)

RemoveResult	remove_sample (const DataSampleElement *sample)

void	remove_all_msgs ()

bool	add_reader (const ReaderInfo_rch &reader)

bool	has_reader (const GUID_t &id) const

bool	is_leading (const GUID_t &id) const

bool	remove_reader (const GUID_t &id)

size_t	reader_count () const

CORBA::Long	inc_heartbeat_count ()

void	pre_stop_helper (TqeVector &to_drop, bool true_stop)

TransportQueueElement *	customize_queue_element_helper (TransportQueueElement *element, bool requires_inline_qos, MetaSubmessageVec &meta_submessages, bool &deliver_after_send)

void	process_acknack (const RTPS::AckNackSubmessage &acknack, const GUID_t &src, MetaSubmessageVec &meta_submessages)

void	process_nackfrag (const RTPS::NackFragSubmessage &nackfrag, const GUID_t &src, MetaSubmessageVec &meta_submessages)

void	process_acked_by_all ()

void	gather_nack_replies_i (MetaSubmessageVec &meta_submessages)

void	gather_heartbeats_i (MetaSubmessageVec &meta_submessages)

void	gather_heartbeats (RcHandle< ConstSharedRepoIdSet > additional_guids, MetaSubmessageVec &meta_submessages)

void	update_required_acknack_count (const GUID_t &id, CORBA::Long current)

RcHandle< SingleSendBuffer >	get_send_buff ()

RcHandle< ConstSharedRepoIdSet >	get_remote_reader_guids ()

Public Member Functions inherited from OpenDDS::DCPS::RcObject
virtual	~RcObject ()

virtual void	_add_ref ()

virtual void	_remove_ref ()

long	ref_count () const

WeakObject *	_get_weak_object () const

Private Member Functions
typedef	OPENDDS_MULTISET (OpenDDS::DCPS::SequenceNumber) SequenceNumberMultiset

typedef	OPENDDS_SET_CMP (TransportQueueElement *, TransportQueueElement::OrderBySequenceNumber) TqeSet

typedef	OPENDDS_MULTIMAP (SequenceNumber, TransportQueueElement *) SnToTqeMap

void	send_heartbeats (const MonotonicTimePoint &now)

void	send_nack_responses (const MonotonicTimePoint &now)

void	add_gap_submsg_i (RTPS::SubmessageSeq &msg, SequenceNumber gap_start)

void	end_historic_samples_i (const DataSampleHeader &header, ACE_Message_Block *body, MetaSubmessageVec &meta_submessages)

void	request_ack_i (const DataSampleHeader &header, ACE_Message_Block *body, MetaSubmessageVec &meta_submessages)

void	send_heartbeats_manual_i (MetaSubmessageVec &meta_submessages)

void	gather_gaps_i (const ReaderInfo_rch &reader, const DisjointSequence &gaps, MetaSubmessageVec &meta_submessages)

void	acked_by_all_helper_i (TqeSet &to_deliver)

SequenceNumber	expected_max_sn (const ReaderInfo_rch &reader) const

void	make_leader_lagger (const GUID_t &reader, SequenceNumber previous_max_sn)

void	make_lagger_leader (const ReaderInfo_rch &reader, const SequenceNumber previous_acked_sn)

bool	is_lagging (const ReaderInfo_rch &reader) const

bool	is_leading (const ReaderInfo_rch &reader) const

void	check_leader_lagger () const

void	record_directed (const GUID_t &reader, SequenceNumber seq)

void	update_remote_guids_cache_i (bool add, const GUID_t &guid)

bool	is_pvs_writer () const

SequenceNumber	non_durable_first_sn (const SingleSendBuffer::Proxy &proxy) const

void	remove_preassociation_reader (const ReaderInfo_rch &reader)

void	initialize_heartbeat (const SingleSendBuffer::Proxy &proxy, MetaSubmessage &meta_submessage)

void	gather_directed_heartbeat_i (const SingleSendBuffer::Proxy &proxy, MetaSubmessageVec &meta_submessages, MetaSubmessage &meta_submessage, const ReaderInfo_rch &reader)

void	log_remote_counts (const char *funcname)

Static Private Member Functions
static void	snris_insert (RtpsUdpDataLink::SNRIS &snris, const ReaderInfo_rch &reader)

static void	snris_erase (RtpsUdpDataLink::SNRIS &snris, const SequenceNumber sn, const ReaderInfo_rch &reader)

Private Attributes
ReaderInfoMap	remote_readers_

RcHandle< ConstSharedRepoIdSet >	remote_reader_guids_

ReaderInfoSet	preassociation_readers_
	Preassociation readers require a non-final heartbeat. More...

SequenceNumberMultiset	preassociation_reader_start_sns_

SNRIS	lagging_readers_

SNRIS	leading_readers_
	These reader have acked everything they are supposed to have acked. More...

ReaderInfoSet	readers_expecting_data_
	These readers have sent a nack and are expecting data. More...

ReaderInfoSet	readers_expecting_heartbeat_
	These readers have sent a non-final ack are are expecting a heartbeat. More...

RcHandle< SingleSendBuffer >	send_buff_

SequenceNumber	max_sn_

SnToTqeMap	elems_not_acked_

WeakRcHandle< TransportClient >	client_

WeakRcHandle< RtpsUdpDataLink >	link_

const GUID_t	id_

const bool	durable_

bool	stopping_

CORBA::Long	heartbeat_count_

const bool	is_pvs_writer_
	Participant Volatile Secure writer. More...

const bool	is_ps_writer_
	Partcicipant Secure (Reliable SPDP) writer. More...

ACE_Thread_Mutex	mutex_

ACE_Thread_Mutex	remote_reader_guids_mutex_

ACE_Thread_Mutex	elems_not_acked_mutex_

RcHandle< SporadicEvent >	heartbeat_

RcHandle< SporadicEvent >	nack_response_

const TimeDuration	initial_fallback_

FibonacciSequence< TimeDuration >	fallback_

Additional Inherited Members
Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()

Detailed Description

Definition at line 424 of file RtpsUdpDataLink.h.

Constructor & Destructor Documentation

◆ RtpsWriter()

OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::RtpsWriter	(	const TransportClient_rch &	client,
		const RtpsUdpDataLink_rch &	link,
		const GUID_t &	id,
		bool	durable,
		SequenceNumber	max_sn,
		CORBA::Long	heartbeat_count,
		size_t	capacity
	)

Definition at line 4445 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::RcHandle< T >::in(), send_buff_, and OpenDDS::DCPS::RtpsUdpDataLink::send_strategy().

  : send_buff_(make_rch<SingleSendBuffer>(capacity, ONE_SAMPLE_PER_PACKET))
  , max_sn_(max_sn == SequenceNumber::SEQUENCENUMBER_UNKNOWN() ? SequenceNumber::ZERO() : max_sn)
  , client_(client)
  , link_(link)
  , id_(id)
  , durable_(durable)
  , stopping_(false)
  , heartbeat_count_(heartbeat_count)
 #ifdef OPENDDS_SECURITY
  , is_pvs_writer_(id_.entityId == RTPS::ENTITYID_P2P_BUILTIN_PARTICIPANT_VOLATILE_SECURE_WRITER)
  , is_ps_writer_(id_.entityId == RTPS::ENTITYID_SPDP_RELIABLE_BUILTIN_PARTICIPANT_SECURE_WRITER)
 #endif
  , heartbeat_(make_rch<SporadicEvent>(link->event_dispatcher(), make_rch<PmfNowEvent<RtpsWriter> >(rchandle_from(this), &RtpsWriter::send_heartbeats)))
  , nack_response_(make_rch<SporadicEvent>(link->event_dispatcher(), make_rch<PmfNowEvent<RtpsWriter> >(rchandle_from(this), &RtpsWriter::send_nack_responses)))
  , initial_fallback_(link->config()->heartbeat_period_)
  , fallback_(initial_fallback_)
 {
   send_buff_->bind(link->send_strategy().in());
 }

◆ ~RtpsWriter()

OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::~RtpsWriter ( )

virtual

Definition at line 4468 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, ACE_TEXT(), elems_not_acked_, and LM_WARNING.

 {
   if (!elems_not_acked_.empty()) {
     ACE_DEBUG((LM_WARNING, ACE_TEXT("(%P|%t) WARNING: RtpsWriter::~RtpsWriter - ")
       ACE_TEXT("deleting with %d elements left not fully acknowledged\n"),
       elems_not_acked_.size()));
   }
 }

Member Function Documentation

◆ acked_by_all_helper_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::acked_by_all_helper_i ( TqeSet & to_deliver )

private

Definition at line 3969 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::link_, OpenDDS::DCPS::SequenceNumber::MAX_VALUE, OPENDDS_MULTIMAP, and OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_VECTOR().

 {
   using namespace OpenDDS::RTPS;
   typedef OPENDDS_MULTIMAP(SequenceNumber, TransportQueueElement*)::iterator iter_t;
   OPENDDS_VECTOR(GUID_t) to_check;
 
   RtpsUdpDataLink_rch link = link_.lock();
 
   if (!link) {
     return;
   }
 
   //start with the max sequence number writer knows about and decrease
   //by what the min over all readers is
   SequenceNumber all_readers_ack = SequenceNumber::MAX_VALUE;
   if (!preassociation_readers_.empty()) {
     all_readers_ack = std::min(all_readers_ack, *preassociation_reader_start_sns_.begin());
   }
   if (!lagging_readers_.empty()) {
     all_readers_ack = std::min(all_readers_ack, lagging_readers_.begin()->first + 1);
   }
   if (!leading_readers_.empty()) {
     // When is_pvs_writer_ is true, the leading_readers_ will all be
     // at different sequence numbers.  The minimum could actually be
     // before the preassociation readers or lagging readers.  Use the
     // largest sequence number to avoid holding onto samples.
     all_readers_ack = std::min(all_readers_ack, leading_readers_.rbegin()->first + 1);
   }
 
   if (all_readers_ack == SequenceNumber::MAX_VALUE) {
     return;
   }
 
   ACE_GUARD(ACE_Thread_Mutex, g2, elems_not_acked_mutex_);
 
   if (!elems_not_acked_.empty()) {
     for (iter_t it = elems_not_acked_.begin(), limit = elems_not_acked_.end();
          it != limit && it->first < all_readers_ack;) {
       send_buff_->release_acked(it->first);
       to_deliver.insert(it->second);
       elems_not_acked_.erase(it++);
     }
   }
 }

◆ add_elem_awaiting_ack()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::add_elem_awaiting_ack ( TransportQueueElement * element )

Definition at line 4505 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, elems_not_acked_, elems_not_acked_mutex_, and OpenDDS::DCPS::TransportQueueElement::sequence().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, elems_not_acked_mutex_);
   elems_not_acked_.insert(SnToTqeMap::value_type(element->sequence(), element));
 }

◆ add_gap_submsg_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::add_gap_submsg_i	(	RTPS::SubmessageSeq &	msg,
		SequenceNumber	gap_start
	)

private

Definition at line 1501 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::GAP, OpenDDS::DCPS::grow(), OpenDDS::DCPS::DataLink::id_, OpenDDS::DCPS::Encoding::KIND_XCDR1, OPENDDS_ASSERT, OpenDDS::DCPS::serialized_size(), OpenDDS::RTPS::SMHDR_SZ, OpenDDS::RTPS::GapSubmessage::smHeader, OpenDDS::RTPS::SubmessageHeader::submessageLength, and OpenDDS::RTPS::to_rtps_seqnum().

 {
   // These are the GAP submessages that we'll send directly in-line with the
   // DATA when we notice that the DataWriter has deliberately skipped seq #s.
   // There are other GAP submessages generated in meta_submessage to reader ACKNACKS,
   // see send_nack_replies().
   using namespace OpenDDS::RTPS;
 
   const LongSeq8 bitmap;
 
   // RTPS v2.1 8.3.7.4: the Gap sequence numbers are those in the range
   // [gapStart, gapListBase) and those in the SNSet.
   GapSubmessage gap = {
     {GAP, FLAG_E, 0 /*length determined below*/},
     ENTITYID_UNKNOWN, // readerId: applies to all matched readers
     id_.entityId,
     to_rtps_seqnum(gap_start),
     {to_rtps_seqnum(max_sn_), 0, bitmap}
   };
   OPENDDS_ASSERT(gap_start < max_sn_);
 
   const size_t size = serialized_size(Encoding(Encoding::KIND_XCDR1), gap);
   gap.smHeader.submessageLength =
       static_cast<CORBA::UShort>(size) - SMHDR_SZ;
 
   const CORBA::ULong idx = grow(msg) - 1;
   msg[idx].gap_sm(gap);
 }

◆ add_reader()

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::add_reader ( const ReaderInfo_rch & reader )

Definition at line 2032 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::RtpsUdpDataLink::heartbeat_, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::id_, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::link_, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::log_remote_counts(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::mutex_, OPENDDS_ASSERT, OpenDDS::DCPS::RtpsUdpDataLink::queue_submessages(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::stopping_.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, false);
   OPENDDS_ASSERT(!reader->durable_ || durable_);
 
   if (stopping_) {
     return false;
   }
 
   ReaderInfoMap::const_iterator iter = remote_readers_.find(reader->id_);
   if (iter == remote_readers_.end()) {
 #ifdef OPENDDS_SECURITY
     if (is_pvs_writer_) {
       reader->max_pvs_sn_ = max_sn_;
     }
 #endif
     remote_readers_.insert(ReaderInfoMap::value_type(reader->id_, reader));
     update_remote_guids_cache_i(true, reader->id_);
     preassociation_readers_.insert(reader);
     preassociation_reader_start_sns_.insert(reader->start_sn_);
     log_remote_counts("add_reader");
 
     RtpsUdpDataLink_rch link = link_.lock();
     if (!link) {
       return false;
     }
 
     fallback_.set(initial_fallback_);
     heartbeat_->schedule(fallback_.get());
     // Durable readers will get their heartbeat from end historic samples.
     if (!reader->durable_) {
       MetaSubmessageVec meta_submessages;
       MetaSubmessage meta_submessage(id_, GUID_UNKNOWN);
       const SingleSendBuffer::Proxy proxy(*send_buff_);
       initialize_heartbeat(proxy, meta_submessage);
       gather_directed_heartbeat_i(proxy, meta_submessages, meta_submessage, reader);
       g.release();
       link->queue_submessages(meta_submessages);
     }
 
     return true;
   }
   return false;
 }

◆ check_leader_lagger()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::check_leader_lagger ( ) const

private

Definition at line 3896 of file RtpsUdpDataLink.cpp.

References OPENDDS_ASSERT, OpenDDS::DCPS::SequenceNumber::previous(), and OpenDDS::DCPS::SequenceNumber::ZERO().

Referenced by update_max_sn().

 {
 #ifndef OPENDDS_SAFETY_PROFILE
 #ifndef NDEBUG
   static const SequenceNumber negative_one = SequenceNumber::ZERO().previous();
   for (SNRIS::const_iterator pos1 = lagging_readers_.begin(), limit = lagging_readers_.end();
        pos1 != limit; ++pos1) {
     const SequenceNumber& sn = pos1->first;
     const ReaderInfoSetHolder_rch& readers = pos1->second;
     for (ReaderInfoSet::const_iterator pos2 = readers->readers.begin(), limit = readers->readers.end();
          pos2 != limit; ++pos2) {
       const ReaderInfo_rch& reader = *pos2;
       OPENDDS_ASSERT(reader->acked_sn() == sn);
       const SequenceNumber expect_max_sn = expected_max_sn(reader);
       OPENDDS_ASSERT(sn == negative_one || sn < expect_max_sn);
       OPENDDS_ASSERT(preassociation_readers_.count(reader) == 0);
     }
   }
 
   for (SNRIS::const_iterator pos1 = leading_readers_.begin(), limit = leading_readers_.end();
        pos1 != limit; ++pos1) {
     const SequenceNumber& sn = pos1->first;
     const ReaderInfoSetHolder_rch& readers = pos1->second;
     for (ReaderInfoSet::const_iterator pos2 = readers->readers.begin(), limit = readers->readers.end();
          pos2 != limit; ++pos2) {
       const ReaderInfo_rch& reader = *pos2;
       OPENDDS_ASSERT(reader->acked_sn() == sn);
       const SequenceNumber expect_max_sn = expected_max_sn(reader);
       OPENDDS_ASSERT(sn == expect_max_sn);
       OPENDDS_ASSERT(preassociation_readers_.count(reader) == 0);
     }
   }
 #endif
 #endif
 }

◆ customize_queue_element_helper()

TransportQueueElement * OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::customize_queue_element_helper	(	TransportQueueElement *	element,
		bool	requires_inline_qos,
		MetaSubmessageVec &	meta_submessages,
		bool &	deliver_after_send
	)

Definition at line 1077 of file RtpsUdpDataLink.cpp.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, 0);
 
   RtpsUdpDataLink_rch link = link_.lock();
   if (stopping_ || !link) {
     g.release();
     element->data_dropped(true);
     return 0;
   }
 
   OPENDDS_ASSERT(element->publication_id() == id_);
 
   const SequenceNumber previous_max_sn = max_sn_;
   RTPS::SubmessageSeq subm;
 
   const SequenceNumber seq = element->sequence();
   if (seq != SequenceNumber::SEQUENCENUMBER_UNKNOWN()) {
     if (!element->is_fragment() || element->is_last_fragment()) {
       max_sn_ = std::max(max_sn_, seq);
     }
     if (!durable_ && !is_pvs_writer() &&
         element->subscription_id() == GUID_UNKNOWN &&
         previous_max_sn < max_sn_.previous()) {
       add_gap_submsg_i(subm, previous_max_sn + 1);
     }
   }
 
   make_leader_lagger(element->subscription_id(), previous_max_sn);
   check_leader_lagger();
 
   TransportSendElement* tse = dynamic_cast<TransportSendElement*>(element);
   TransportCustomizedElement* tce =
     dynamic_cast<TransportCustomizedElement*>(element);
   TransportSendControlElement* tsce =
     dynamic_cast<TransportSendControlElement*>(element);
 
   Message_Block_Ptr data;
   bool durable = false;
 
   const ACE_Message_Block* msg = element->msg();
   const GUID_t pub_id = element->publication_id();
 
   // Based on the type of 'element', find and duplicate the data payload
   // continuation block.
   if (tsce) {        // Control message
     if (RtpsSampleHeader::control_message_supported(tsce->header().message_id_)) {
       data.reset(msg->cont()->duplicate());
       // Create RTPS Submessage(s) in place of the OpenDDS DataSampleHeader
       RtpsSampleHeader::populate_data_control_submessages(
         subm, *tsce, requires_inline_qos);
       record_directed(element->subscription_id(), seq);
     } else if (tsce->header().message_id_ == END_HISTORIC_SAMPLES) {
       end_historic_samples_i(tsce->header(), msg->cont(), meta_submessages);
       g.release();
       element->data_delivered();
       return 0;
     } else if (tsce->header().message_id_ == REQUEST_ACK) {
       request_ack_i(tsce->header(), msg->cont(), meta_submessages);
       deliver_after_send = true;
       return 0;
     } else if (tsce->header().message_id_ == DATAWRITER_LIVELINESS) {
       send_heartbeats_manual_i(meta_submessages);
       deliver_after_send = true;
       return 0;
     } else {
       g.release();
       element->data_dropped(true /*dropped_by_transport*/);
       return 0;
     }
 
   } else if (tse) {  // Basic data message
     // {DataSampleHeader} -> {Data Payload}
     data.reset(msg->cont()->duplicate());
     const DataSampleElement* dsle = tse->sample();
     // Create RTPS Submessage(s) in place of the OpenDDS DataSampleHeader
     RtpsSampleHeader::populate_data_sample_submessages(
       subm, *dsle, requires_inline_qos);
     record_directed(element->subscription_id(), seq);
     durable = dsle->get_header().historic_sample_;
 
   } else if (tce) {  // Customized data message
     // {DataSampleHeader} -> {Content Filtering GUIDs} -> {Data Payload}
     data.reset(msg->cont()->cont()->duplicate());
     const DataSampleElement* dsle = tce->original_send_element()->sample();
     // Create RTPS Submessage(s) in place of the OpenDDS DataSampleHeader
     RtpsSampleHeader::populate_data_sample_submessages(
       subm, *dsle, requires_inline_qos);
     record_directed(element->subscription_id(), seq);
     durable = dsle->get_header().historic_sample_;
 
   } else {
     send_buff_->pre_insert(seq);
     return element;
   }
 
 #ifdef OPENDDS_SECURITY
   {
     GuardType guard(link->strategy_lock_);
     if (link->send_strategy_) {
       link->send_strategy()->encode_payload(pub_id, data, subm);
     }
   }
 #endif
 
   if (stopping_) {
     g.release();
     element->data_dropped(true);
     return 0;
   }
 
   if (transport_debug.log_messages) {
     link->send_strategy()->append_submessages(subm);
   }
 
   Message_Block_Ptr hdr(link->submsgs_to_msgblock(subm));
   hdr->cont(data.release());
   RtpsCustomizedElement* rtps =
     new RtpsCustomizedElement(element, move(hdr));
 
   // Handle durability resends
   if (durable) {
     const GUID_t sub = element->subscription_id();
     if (sub != GUID_UNKNOWN) {
       ReaderInfoMap::iterator ri = remote_readers_.find(sub);
       if (ri != remote_readers_.end()) {
         ri->second->durable_data_[rtps->sequence()] = rtps;
         ri->second->durable_timestamp_.set_to_now();
         if (Transport_debug_level > 3) {
           const LogGuid conv(pub_id), sub_conv(sub);
           ACE_DEBUG((LM_DEBUG,
             "(%P|%t) RtpsUdpDataLink::customize_queue_element() - "
             "storing durable data for local %C remote %C seq %q\n",
             conv.c_str(), sub_conv.c_str(),
             rtps->sequence().getValue()));
         }
         return 0;
       }
     }
   }
 
   send_buff_->pre_insert(seq);
   return rtps;
 }

◆ end_historic_samples_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::end_historic_samples_i	(	const DataSampleHeader &	header,
		ACE_Message_Block *	body,
		MetaSubmessageVec &	meta_submessages
	)

private

Definition at line 1343 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::DataLink::id_, LM_DEBUG, OpenDDS::DCPS::log_progress(), OpenDDS::DCPS::TransportDebug::log_progress, OpenDDS::DCPS::DataSampleHeader::message_length_, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), ACE_Message_Block::rd_ptr(), OpenDDS::DCPS::transport_debug, and OpenDDS::DCPS::Transport_debug_level.

 {
   // Set the ReaderInfo::durable_timestamp_ for the case where no
   // durable samples exist in the DataWriter.
   if (durable_) {
     const MonotonicTimePoint now = MonotonicTimePoint::now();
     GUID_t sub = GUID_UNKNOWN;
     if (body && header.message_length_ >= sizeof(sub)) {
       std::memcpy(&sub, body->rd_ptr(), sizeof(sub));
     }
     typedef ReaderInfoMap::iterator iter_t;
     if (sub == GUID_UNKNOWN) {
       if (Transport_debug_level > 3) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::end_historic_samples "
                    "local %C all readers\n", LogGuid(id_).c_str()));
       }
       for (iter_t iter = remote_readers_.begin();
            iter != remote_readers_.end(); ++iter) {
         if (iter->second->durable_) {
           iter->second->durable_timestamp_ = now;
           if (transport_debug.log_progress) {
             log_progress("durable data queued", id_, iter->first, iter->second->participant_discovered_at_);
           }
         }
       }
     } else {
       iter_t iter = remote_readers_.find(sub);
       if (iter != remote_readers_.end()) {
         if (iter->second->durable_) {
           iter->second->durable_timestamp_ = now;
           if (transport_debug.log_progress) {
             log_progress("durable data queued", id_, iter->first, iter->second->participant_discovered_at_);
           }
           const SingleSendBuffer::Proxy proxy(*send_buff_);
           MetaSubmessage meta_submessage(id_, GUID_UNKNOWN);
           initialize_heartbeat(proxy, meta_submessage);
           gather_directed_heartbeat_i(proxy, meta_submessages, meta_submessage, iter->second);
           if (Transport_debug_level > 3) {
             ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::end_historic_samples"
                        " local %C remote %C\n", LogGuid(id_).c_str(), LogGuid(sub).c_str()));
           }
         }
       }
     }
   }
 }

◆ expected_max_sn()

SequenceNumber OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::expected_max_sn ( const ReaderInfo_rch & reader ) const

private

Definition at line 3758 of file RtpsUdpDataLink.cpp.

 {
   ACE_UNUSED_ARG(reader);
 #ifdef OPENDDS_SECURITY
   if (is_pvs_writer_) {
     return reader->max_pvs_sn_;
   } else {
 #endif
     return max_sn_;
 #ifdef OPENDDS_SECURITY
   }
 #endif
 }

◆ gather_directed_heartbeat_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_directed_heartbeat_i	(	const SingleSendBuffer::Proxy &	proxy,
		MetaSubmessageVec &	meta_submessages,
		MetaSubmessage &	meta_submessage,
		const ReaderInfo_rch &	reader
	)

private

Definition at line 4179 of file RtpsUdpDataLink.cpp.

References OpenDDS::RTPS::HeartBeatSubmessage::count, OpenDDS::DCPS::MetaSubmessage::dst_guid_, OpenDDS::RTPS::HeartBeatSubmessage::firstSN, OpenDDS::RTPS::Submessage::heartbeat_sm, OpenDDS::RTPS::HeartBeatSubmessage::lastSN, OPENDDS_ASSERT, OpenDDS::DCPS::SequenceNumber::previous(), OpenDDS::RTPS::HeartBeatSubmessage::readerId, OpenDDS::DCPS::MetaSubmessage::reset_destination(), OpenDDS::DCPS::MetaSubmessage::sm_, OpenDDS::RTPS::to_rtps_seqnum(), and OpenDDS::RTPS::Count_t::value.

 {
   const SequenceNumber first_sn = reader->durable_ ? 1 : std::max(non_durable_first_sn(proxy), reader->start_sn_);
   SequenceNumber last_sn = expected_max_sn(reader);
 #ifdef OPENDDS_SECURITY
   if (is_pvs_writer_ && last_sn < first_sn.previous()) {
     // This can happen if the reader get's reset.
     // Adjust the heartbeat to be valid.
     // Make lagger_leader will eventually correct the problem.
     last_sn = first_sn.previous();
   }
 #endif
   meta_submessage.dst_guid_ = reader->id_;
   meta_submessage.sm_.heartbeat_sm().count.value = ++heartbeat_count_;
   meta_submessage.sm_.heartbeat_sm().readerId = reader->id_.entityId;
   meta_submessage.sm_.heartbeat_sm().firstSN = to_rtps_seqnum(first_sn);
   meta_submessage.sm_.heartbeat_sm().lastSN = to_rtps_seqnum(last_sn);
   OPENDDS_ASSERT(!(first_sn < 1 || last_sn < 0 || last_sn < first_sn.previous()));
   meta_submessages.push_back(meta_submessage);
   meta_submessage.reset_destination();
 }

◆ gather_gaps_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_gaps_i	(	const ReaderInfo_rch &	reader,
		const DisjointSequence &	gaps,
		MetaSubmessageVec &	meta_submessages
	)

private

Definition at line 3099 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, OpenDDS::DCPS::DisjointSequence::bitmap_num_longs(), OpenDDS::RTPS::SequenceNumberSet::bitmapBase, OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::DisjointSequence::cumulative_ack(), OpenDDS::DCPS::DisjointSequence::disjoint(), OpenDDS::DCPS::DisjointSequence::empty(), OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::GAP, OpenDDS::RTPS::Submessage::gap_sm, OpenDDS::RTPS::GapSubmessage::gapList, OpenDDS::RTPS::GapSubmessage::gapStart, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::DisjointSequence::high(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::id_, LM_DEBUG, OpenDDS::DCPS::DisjointSequence::low(), OPENDDS_ASSERT, OpenDDS::DCPS::SequenceNumber::previous(), OpenDDS::DCPS::MetaSubmessage::sm_, OpenDDS::DCPS::DisjointSequence::to_bitmap(), OpenDDS::RTPS::to_opendds_seqnum(), OpenDDS::RTPS::to_rtps_seqnum(), and OpenDDS::DCPS::Transport_debug_level.

 {
   using namespace RTPS;
 
   OPENDDS_ASSERT(reader || !durable_);
 
   if (gaps.empty()) {
     return;
   }
 
   // RTPS v2.1 8.3.7.4: the Gap sequence numbers are those in the range
   // [gapStart, gapListBase) and those in the SNSet.
   const SequenceNumber firstMissing = gaps.low(),
                        base = ++SequenceNumber(gaps.cumulative_ack());
   const SequenceNumber_t gapStart = to_rtps_seqnum(firstMissing);
   const SequenceNumber_t gapListBase = to_rtps_seqnum(base);
   CORBA::ULong num_bits = 0;
   LongSeq8 bitmap;
 
   if (gaps.disjoint()) {
     bitmap.length(DisjointSequence::bitmap_num_longs(base, gaps.high()));
     if (bitmap.length() > 0) {
       ACE_CDR::ULong cumulative_bits_added = 0;
       (void)gaps.to_bitmap(bitmap.get_buffer(), bitmap.length(), num_bits, cumulative_bits_added);
     }
   }
 
   MetaSubmessage meta_submessage(id_, reader ? reader->id_ : GUID_UNKNOWN);
   GapSubmessage gap = {
     {GAP, FLAG_E, 0 /*length determined later*/},
     reader ? reader->id_.entityId : ENTITYID_UNKNOWN,
     id_.entityId,
     gapStart,
     {gapListBase, num_bits, bitmap}
   };
   OPENDDS_ASSERT(firstMissing < base);
   meta_submessage.sm_.gap_sm(gap);
 
   if (Transport_debug_level > 5) {
     const LogGuid conv(id_);
     SequenceRange sr;
     sr.first = to_opendds_seqnum(gap.gapStart);
     const SequenceNumber srbase = to_opendds_seqnum(gap.gapList.bitmapBase);
     sr.second = srbase.previous();
     ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::RtpsWriter::gather_gaps_i "
               "GAP with range [%q, %q] from %C\n",
               sr.first.getValue(), sr.second.getValue(),
               conv.c_str()));
   }
 
   meta_submessages.push_back(meta_submessage);
 }

◆ gather_heartbeats()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_heartbeats	(	RcHandle< ConstSharedRepoIdSet >	additional_guids,
		MetaSubmessageVec &	meta_submessages
	)

Definition at line 4297 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, OpenDDS::RTPS::HeartBeatSubmessage::count, OpenDDS::DCPS::MetaSubmessage::dst_guid_, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::RTPS::Submessage::heartbeat_sm, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::id_, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::link_, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::mutex_, OPENDDS_ASSERT, OpenDDS::DCPS::MetaSubmessage::reset_destination(), OpenDDS::DCPS::MetaSubmessage::sm_, and OpenDDS::RTPS::Count_t::value.

 {
   OPENDDS_ASSERT(!additional_guids->guids_.empty());
 
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
   RtpsUdpDataLink_rch link = link_.lock();
   if (!link) {
     return;
   }
 
   const SingleSendBuffer::Proxy proxy(*send_buff_);
 
   MetaSubmessage meta_submessage(id_, GUID_UNKNOWN);
   initialize_heartbeat(proxy, meta_submessage);
 
   for (RepoIdSet::const_iterator it = additional_guids->guids_.begin(),
     limit = additional_guids->guids_.end(); it != limit; ++it) {
 
     // Semi-directed (INFO_DST but ENTITYID_UNKNOWN, non-final.
     meta_submessage.dst_guid_ = *it;
     meta_submessage.sm_.heartbeat_sm().count.value = ++heartbeat_count_;
     meta_submessages.push_back(meta_submessage);
     meta_submessage.reset_destination();
   }
 }

◆ gather_heartbeats_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_heartbeats_i ( MetaSubmessageVec & meta_submessages )

Definition at line 4231 of file RtpsUdpDataLink.cpp.

References OpenDDS::RTPS::HeartBeatSubmessage::count, OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::RTPS::HeartBeatSubmessage::firstSN, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::RTPS::Submessage::heartbeat_sm, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::id_, OpenDDS::RTPS::HeartBeatSubmessage::lastSN, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::link_, OpenDDS::RTPS::HeartBeatSubmessage::readerId, OpenDDS::DCPS::MetaSubmessage::reset_destination(), OpenDDS::DCPS::MetaSubmessage::sm_, OpenDDS::RTPS::to_rtps_seqnum(), and OpenDDS::RTPS::Count_t::value.

 {
   if (preassociation_readers_.empty() && lagging_readers_.empty()) {
     return;
   }
 
   check_leader_lagger();
 
   RtpsUdpDataLink_rch link = link_.lock();
   if (!link) {
     return;
   }
 
   using namespace OpenDDS::RTPS;
 
   const SingleSendBuffer::Proxy proxy(*send_buff_);
 
   // Assume no samples are available.
   const SequenceNumber nonDurableFirstSN = non_durable_first_sn(proxy);
   const SequenceNumber firstSN = durable_ ? 1 : nonDurableFirstSN;
   const SequenceNumber lastSN = max_sn_;
 
   MetaSubmessage meta_submessage(id_, GUID_UNKNOWN);
   initialize_heartbeat(proxy, meta_submessage);
 
   // Directed, non-final.
   if (!preassociation_readers_.empty()) {
     meta_submessages.reserve(meta_submessages.size() + preassociation_readers_.size());
     for (ReaderInfoSet::const_iterator pos = preassociation_readers_.begin(), limit = preassociation_readers_.end();
          pos != limit; ++pos) {
       const ReaderInfo_rch& reader = *pos;
       gather_directed_heartbeat_i(proxy, meta_submessages, meta_submessage, reader);
     }
   }
 
   if (!lagging_readers_.empty()) {
     if (leading_readers_.empty() && remote_readers_.size() > 1
 #ifdef OPENDDS_SECURITY
         && !is_pvs_writer_
         && !is_ps_writer_
 #endif
         ) {
       // Every reader is lagging and there is more than one.
       meta_submessage.sm_.heartbeat_sm().count.value = ++heartbeat_count_;
       meta_submessage.sm_.heartbeat_sm().readerId = ENTITYID_UNKNOWN;
       meta_submessage.sm_.heartbeat_sm().firstSN = to_rtps_seqnum(firstSN);
       meta_submessage.sm_.heartbeat_sm().lastSN = to_rtps_seqnum(lastSN);
 
       meta_submessages.push_back(meta_submessage);
       meta_submessage.reset_destination();
     } else {
       for (SNRIS::const_iterator snris_pos = lagging_readers_.begin(), snris_limit = lagging_readers_.end();
            snris_pos != snris_limit; ++snris_pos) {
         meta_submessages.reserve(meta_submessages.size() + snris_pos->second->readers.size());
         for (ReaderInfoSet::const_iterator pos = snris_pos->second->readers.begin(),
                limit = snris_pos->second->readers.end();
              pos != limit; ++pos) {
           const ReaderInfo_rch& reader = *pos;
           gather_directed_heartbeat_i(proxy, meta_submessages, meta_submessage, reader);
         }
       }
     }
   }
 }

◆ gather_nack_replies_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i ( MetaSubmessageVec & meta_submessages )

Definition at line 3466 of file RtpsUdpDataLink.cpp.

 {
   RtpsUdpDataLink_rch link = link_.lock();
 
   if (!link) {
     return;
   }
 
   // Process naks requests from each reader replying with either data or a gap.
   // Requests for directed data are answered with a directed reply.
   // Requests for undirected data are answered with an undirected and consolidated reply.
   // Directed data includes things like the PVS writer.
   // The requests_ for a reader should not contain requests for durable data.
 
   typedef OPENDDS_MAP(SequenceNumber, DisjointSequence) FragmentInfo;
 
   // Consolidated non-directed requests and address sets to be sent together at the end, after directed replies
   typedef OPENDDS_MAP(SequenceNumber, AddrSet) RecipientMap;
   typedef OPENDDS_MAP(SequenceNumber, RepoIdSet) ReaderMap;
   DisjointSequence consolidated_requests;
   ReaderMap consolidated_request_readers;
   RecipientMap consolidated_recipients_unicast;
   RecipientMap consolidated_recipients_multicast;
   FragmentInfo consolidated_fragment_requests;
   ReaderMap consolidated_fragment_request_readers;
   RecipientMap consolidated_fragment_recipients_unicast;
   RecipientMap consolidated_fragment_recipients_multicast;
   DisjointSequence consolidated_gaps;
 
   ACE_GUARD(TransportSendBuffer::LockType, guard, send_buff_->strategy_lock());
   SingleSendBuffer::Proxy proxy(*send_buff_);
 
   size_t cumulative_send_count = 0;
 
   for (ReaderInfoSet::const_iterator pos = readers_expecting_data_.begin(), limit = readers_expecting_data_.end();
        pos != limit; ++pos) {
 
     const ReaderInfo_rch& reader = *pos;
     const AddrSet addrs = link->get_addresses(id_, reader->id_);
 
     DisjointSequence gaps;
 
     if (reader->expecting_durable_data()) {
 
       if (!reader->requests_.empty() && !reader->durable_data_.empty()) {
         const SequenceNumber dd_first = reader->durable_data_.begin()->first;
         const SequenceNumber dd_last = reader->durable_data_.rbegin()->first;
 
         if (reader->requests_.high() < dd_first) {
           gaps.insert(SequenceRange(reader->requests_.low(), dd_first.previous()));
           reader->requests_.reset();
         } else {
           const OPENDDS_VECTOR(SequenceRange) psr = reader->requests_.present_sequence_ranges();
           for (OPENDDS_VECTOR(SequenceRange)::const_iterator iter = psr.begin(), limit = psr.end();
                iter != limit && iter->first <= dd_last; ++iter) {
             for (SequenceNumber s = iter->first; s <= iter->second; ++s) {
               if (s <= dd_last) {
                 const OPENDDS_MAP(SequenceNumber, TransportQueueElement*)::iterator dd_iter = reader->durable_data_.find(s);
                 if (dd_iter != reader->durable_data_.end()) {
                   link->durability_resend(dd_iter->second, cumulative_send_count);
                 } else {
                   gaps.insert(s);
                 }
                 reader->requests_.erase(s);
               }
             }
           }
         }
       }
 
       typedef RequestedFragSeqMap::const_iterator rfs_iter;
       for (rfs_iter rfs = reader->requested_frags_.begin(), limit = reader->requested_frags_.end(); rfs != limit; ++rfs) {
         const OPENDDS_MAP(SequenceNumber, TransportQueueElement*)::iterator dd_iter = reader->durable_data_.find(rfs->first);
         if (dd_iter != reader->durable_data_.end()) {
           for (RequestedFragMap::const_iterator rf = rfs->second.begin(); rf != rfs->second.end(); ++rf) {
             link->durability_resend(dd_iter->second, rf->second, cumulative_send_count);
           }
         } else if ((!reader->durable_data_.empty() && rfs->first < reader->durable_data_.begin()->first)) {
           gaps.insert(rfs->first);
         }
       }
 
       gather_gaps_i(reader, gaps, meta_submessages);
 
       // The writer may not be done replaying durable data.
       // Do not send a gap.
       // TODO: If we have all of the durable data, adjust the request so that we can answer the non-durable part.
       reader->requests_.reset();
       reader->requested_frags_.clear();
       continue;
     }
 
     const SequenceNumber first_sn = std::max(non_durable_first_sn(proxy), reader->start_sn_);
     if (!reader->requests_.empty() &&
         reader->requests_.high() < first_sn) {
       // The reader is not going to get any data.
       // Send a gap that is going to to catch them up.
       gaps.insert(SequenceRange(reader->requests_.low(), first_sn.previous()));
       reader->requests_.reset();
     }
 
     const OPENDDS_VECTOR(SequenceRange) ranges = reader->requests_.present_sequence_ranges();
     for (OPENDDS_VECTOR(SequenceRange)::const_iterator iter = ranges.begin(), limit = ranges.end();
          iter != limit; ++iter) {
       for (SequenceNumber seq = iter->first; seq <= iter->second; ++seq) {
         GUID_t destination;
         if (proxy.contains(seq, destination)) {
           if (destination == GUID_UNKNOWN) {
             // Not directed.
             consolidated_requests.insert(seq);
             consolidated_request_readers[seq].insert(reader->id_);
             consolidated_recipients_unicast[seq].insert(addrs.begin(), addrs.end());
             ACE_Guard<ACE_Thread_Mutex> g(link->locators_lock_);
             link->accumulate_addresses(id_, reader->id_, consolidated_recipients_multicast[seq], false);
             continue;
           } else if (destination != reader->id_) {
             // Directed at another reader.
             gaps.insert(seq);
             continue;
           } else {
             // Directed at the reader.
             const RtpsUdpSendStrategy::OverrideToken ot =
               link->send_strategy()->override_destinations(addrs);
             proxy.resend_i(SequenceRange(seq, seq), 0, reader->id_);
             ++cumulative_send_count;
             continue;
           }
         } else if (proxy.pre_contains(seq) || seq > max_sn_) {
           // Can't answer, don't gap.
           continue;
         }
 
         if (durable_ || is_pvs_writer()) {
           // Must send directed gap.
           gaps.insert(seq);
         } else {
           // Non-directed gap.
           consolidated_gaps.insert(seq);
         }
       }
     }
 
     reader->requests_.reset();
 
     typedef RequestedFragSeqMap::iterator rfs_iter;
     const rfs_iter rfs_end = reader->requested_frags_.end();
     for (rfs_iter rfs = reader->requested_frags_.begin(); rfs != rfs_end; ++rfs) {
       const SequenceNumber& seq = rfs->first;
       GUID_t destination;
       if (proxy.contains(seq, destination)) {
         if (destination == GUID_UNKNOWN) {
           for (RequestedFragMap::iterator rf = rfs->second.begin(); rf != rfs->second.end(); ++rf) {
             consolidated_fragment_requests[seq].insert(rf->second.bitmapBase.value, rf->second.numBits,
                                                        rf->second.bitmap.get_buffer());
           }
           consolidated_fragment_request_readers[seq].insert(reader->id_);
           consolidated_fragment_recipients_unicast[seq].insert(addrs.begin(), addrs.end());
           ACE_Guard<ACE_Thread_Mutex> g(link->locators_lock_);
           link->accumulate_addresses(id_, reader->id_, consolidated_fragment_recipients_multicast[seq], false);
           continue;
         } else if (destination != reader->id_) {
           // Directed at another reader.
           gaps.insert(seq);
           continue;
         } else {
           // Directed at the reader.
           const RtpsUdpSendStrategy::OverrideToken ot =
             link->send_strategy()->override_destinations(addrs);
           for (RequestedFragMap::iterator rf = rfs->second.begin(); rf != rfs->second.end(); ++rf) {
             DisjointSequence x;
             x.insert(rf->second.bitmapBase.value, rf->second.numBits,
                      rf->second.bitmap.get_buffer());
             proxy.resend_fragments_i(seq, x, cumulative_send_count);
           }
           continue;
         }
       } else if (proxy.pre_contains(seq) || seq > max_sn_) {
         // Can't answer, don't gap.
         continue;
       }
 
       if (durable_ || is_pvs_writer()) {
         // Must send directed gap.
         gaps.insert(seq);
       } else {
         // Non-directed gap.
         consolidated_gaps.insert(seq);
       }
     }
 
     reader->requested_frags_.clear();
 
     // Gather directed gaps.
     gather_gaps_i(reader, gaps, meta_submessages);
   }
 
   {
     // Send the consolidated requests.
     const OPENDDS_VECTOR(SequenceRange) ranges = consolidated_requests.present_sequence_ranges();
     for (OPENDDS_VECTOR(SequenceRange)::const_iterator pos = ranges.begin(), limit = ranges.end();
          pos != limit; ++pos) {
       if (Transport_debug_level > 5) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i: "
                    "resend data %q-%q\n", pos->first.getValue(),
                    pos->second.getValue()));
       }
       for (SequenceNumber seq = pos->first; seq <= pos->second; ++seq) {
         const AddrSet& uni = consolidated_recipients_unicast[seq];
         const AddrSet& multi = consolidated_recipients_multicast[seq];
         const RepoIdSet& readers = consolidated_request_readers[seq];
 
         if (proxy.has_frags(seq)) {
           if (consolidated_fragment_requests.find(seq) == consolidated_fragment_requests.end()) {
             consolidated_fragment_requests[seq].insert(1);
           }
           consolidated_fragment_recipients_unicast[seq].insert(uni.begin(), uni.end());
           consolidated_fragment_recipients_multicast[seq].insert(multi.begin(), multi.end());
           consolidated_fragment_request_readers[seq].insert(readers.begin(), readers.end());
         } else {
           const RtpsUdpSendStrategy::OverrideToken ot =
             link->send_strategy()->override_destinations(readers.size() * 2 > remote_readers_.size() ? multi : uni);
 
           proxy.resend_i(SequenceRange(seq, seq));
           ++cumulative_send_count;
         }
       }
     }
 
     for (FragmentInfo::const_iterator pos = consolidated_fragment_requests.begin(),
            limit = consolidated_fragment_requests.end(); pos != limit; ++pos) {
       const AddrSet& uni = consolidated_fragment_recipients_unicast[pos->first];
       const AddrSet& multi = consolidated_fragment_recipients_multicast[pos->first];
       const RepoIdSet& readers = consolidated_fragment_request_readers[pos->first];
       const RtpsUdpSendStrategy::OverrideToken ot =
         link->send_strategy()->override_destinations(readers.size() * 2 > remote_readers_.size() ? multi : uni);
 
       proxy.resend_fragments_i(pos->first, pos->second, cumulative_send_count);
     }
   }
 
   if (cumulative_send_count) {
     RtpsUdpInst_rch cfg = link->config();
     if (cfg && cfg->count_messages()) {
       ACE_GUARD(ACE_Thread_Mutex, g, link->transport_statistics_mutex_);
       link->transport_statistics_.writer_resend_count[id_] += static_cast<ACE_CDR::ULong>(cumulative_send_count);
     }
   }
 
   // Gather the consolidated gaps.
   if (!consolidated_gaps.empty()) {
     if (Transport_debug_level > 5) {
       ACE_DEBUG((LM_DEBUG,
         "(%P|%t) RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i: GAPs:\n"));
       consolidated_gaps.dump();
     }
     gather_gaps_i(ReaderInfo_rch(), consolidated_gaps, meta_submessages);
   } else if (Transport_debug_level > 5) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i: "
       "no GAPs to send\n"));
   }
 
   MetaSubmessage meta_submessage(id_, GUID_UNKNOWN);
   initialize_heartbeat(proxy, meta_submessage);
 
   // Directed, non-final.
   for (ReaderInfoSet::const_iterator pos = readers_expecting_data_.begin(), limit = readers_expecting_data_.end();
        pos != limit; ++pos) {
     const ReaderInfo_rch& reader = *pos;
     readers_expecting_heartbeat_.erase(reader);
     if (reader->reflects_heartbeat_count()) {
       meta_submessage.sm_.heartbeat_sm().smHeader.flags |= RTPS::OPENDDS_FLAG_R;
       gather_directed_heartbeat_i(proxy, meta_submessages, meta_submessage, reader);
       reader->required_acknack_count_ = heartbeat_count_;
       meta_submessage.sm_.heartbeat_sm().smHeader.flags &= ~RTPS::OPENDDS_FLAG_R;
     } else {
       gather_directed_heartbeat_i(proxy, meta_submessages, meta_submessage, reader);
     }
   }
   readers_expecting_data_.clear();
 
   // Directed, final.
   meta_submessage.sm_.heartbeat_sm().smHeader.flags |= RTPS::FLAG_F;
   meta_submessages.reserve(meta_submessages.size() + readers_expecting_heartbeat_.size());
   for (ReaderInfoSet::const_iterator pos = readers_expecting_heartbeat_.begin(), limit = readers_expecting_heartbeat_.end();
        pos != limit; ++pos) {
     const ReaderInfo_rch& reader = *pos;
     gather_directed_heartbeat_i(proxy, meta_submessages, meta_submessage, reader);
   }
   readers_expecting_heartbeat_.clear();
 }

◆ get_remote_reader_guids()

RcHandle<ConstSharedRepoIdSet> OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::get_remote_reader_guids ( )

inline

Definition at line 570 of file RtpsUdpDataLink.h.

     {
       ACE_Guard<ACE_Thread_Mutex> guard(remote_reader_guids_mutex_);
       return remote_reader_guids_;
     }

◆ get_send_buff()

RcHandle<SingleSendBuffer> OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::get_send_buff ( )

inline

Definition at line 569 of file RtpsUdpDataLink.h.

569 { return send_buff_; }

OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::send_buff_

RcHandle< SingleSendBuffer > send_buff_

Definition: RtpsUdpDataLink.h:441

◆ has_reader()

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::has_reader ( const GUID_t & id ) const

Definition at line 2078 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::mutex_.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, false);
   return remote_readers_.count(id) != 0;
 }

◆ inc_heartbeat_count()

CORBA::Long OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::inc_heartbeat_count ( )

Definition at line 4477 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, heartbeat_count_, and mutex_.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, 0);
   return ++heartbeat_count_;
 }

◆ initialize_heartbeat()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::initialize_heartbeat	(	const SingleSendBuffer::Proxy &	proxy,
		MetaSubmessage &	meta_submessage
	)

private

Definition at line 4155 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::HEARTBEAT, OpenDDS::RTPS::Submessage::heartbeat_sm, OpenDDS::RTPS::HEARTBEAT_SZ, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::id_, OpenDDS::DCPS::MetaSubmessage::sm_, and OpenDDS::RTPS::to_rtps_seqnum().

 {
   using namespace OpenDDS::RTPS;
 
   // Assume no samples are available.
   const SequenceNumber nonDurableFirstSN = non_durable_first_sn(proxy);
   const SequenceNumber firstSN = durable_ ? 1 : nonDurableFirstSN;
 
   const HeartBeatSubmessage hb = {
     {HEARTBEAT,
      FLAG_E,
      HEARTBEAT_SZ},
     ENTITYID_UNKNOWN, // any matched reader may be interested in this
     id_.entityId,
     to_rtps_seqnum(firstSN),
     to_rtps_seqnum(max_sn_),
     {0}
   };
 
   meta_submessage.sm_.heartbeat_sm(hb);
 }

◆ is_lagging()

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::is_lagging ( const ReaderInfo_rch & reader ) const

private

Definition at line 3884 of file RtpsUdpDataLink.cpp.

 {
   return reader->acked_sn() != expected_max_sn(reader);
 }

◆ is_leading() [1/2]

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::is_leading ( const ReaderInfo_rch & reader ) const

private

Definition at line 3890 of file RtpsUdpDataLink.cpp.

Referenced by is_leading().

 {
   return reader->acked_sn() == expected_max_sn(reader);
 }

◆ is_leading() [2/2]

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::is_leading ( const GUID_t & id ) const

Definition at line 4512 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, is_leading(), mutex_, and remote_readers_.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, false);
 
   ReaderInfoMap::const_iterator iter = remote_readers_.find(reader_id);
   if (iter != remote_readers_.end()) {
     return is_leading(iter->second);
   }
 
   return false;
 }

◆ is_pvs_writer()

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::is_pvs_writer ( ) const

inlineprivate

Definition at line 495 of file RtpsUdpDataLink.h.

495 { return is_pvs_writer_; }

OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::is_pvs_writer_

const bool is_pvs_writer_

Participant Volatile Secure writer.

Definition: RtpsUdpDataLink.h:454

◆ log_remote_counts()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::log_remote_counts ( const char * funcname )

private

Definition at line 4816 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, id_, LM_DEBUG, OpenDDS::DCPS::TransportDebug::log_remote_counts, preassociation_readers_, remote_readers_, and OpenDDS::DCPS::transport_debug.

 {
   if (transport_debug.log_remote_counts) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_remote_counts} "
       "RtpsUdpDataLink::RtpsWriter::%C: "
       "%C pre: %b assoc: %b\n",
       funcname, LogGuid(id_).c_str(),
       preassociation_readers_.size(), remote_readers_.size()));
   }
 }

◆ make_lagger_leader()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::make_lagger_leader	(	const ReaderInfo_rch &	reader,
		const SequenceNumber	previous_acked_sn
	)

private

Definition at line 3858 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::RtpsUdpDataLink::heartbeat_, and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::link_.

 {
   RtpsUdpDataLink_rch link = link_.lock();
   if (!link) {
     return;
   }
 
   const SequenceNumber acked_sn = reader->acked_sn();
   if (previous_acked_sn == acked_sn) { return; }
   const SequenceNumber previous_max_sn = expected_max_sn(reader);
 #ifdef OPENDDS_SECURITY
   if (is_pvs_writer_ && acked_sn > previous_max_sn) {
     reader->max_pvs_sn_ = acked_sn;
   }
 #endif
   const SequenceNumber max_sn = expected_max_sn(reader);
 
   snris_erase(previous_acked_sn == previous_max_sn ? leading_readers_ : lagging_readers_, previous_acked_sn, reader);
   snris_insert(acked_sn == max_sn ? leading_readers_ : lagging_readers_, reader);
   if (acked_sn != max_sn) {
     heartbeat_->schedule(fallback_.get());
   }
 }

◆ make_leader_lagger()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::make_leader_lagger	(	const GUID_t &	reader,
		SequenceNumber	previous_max_sn
	)

private

Definition at line 3799 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::RtpsUdpDataLink::heartbeat_, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::link_, and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::stopping_.

Referenced by update_max_sn().

 {
   ACE_UNUSED_ARG(reader_id);
 
   if (stopping_) {
     return;
   }
 
   RtpsUdpDataLink_rch link = link_.lock();
   if (!link) {
     return;
   }
 
 #ifdef OPENDDS_SECURITY
     if (!is_pvs_writer_) {
 #endif
       if (previous_max_sn != max_sn_) {
         // All readers that have acked previous_max_sn are now lagging.
         // Move leader_readers_[previous_max_sn] to lagging_readers_[previous_max_sn].
         SNRIS::iterator leading_pos = leading_readers_.find(previous_max_sn);
         SNRIS::iterator lagging_pos = lagging_readers_.find(previous_max_sn);
         if (leading_pos != leading_readers_.end()) {
           if (lagging_pos != lagging_readers_.end()) {
             lagging_pos->second->readers.insert(leading_pos->second->readers.begin(), leading_pos->second->readers.end());
           } else {
             lagging_readers_[previous_max_sn] = leading_pos->second;
           }
           leading_readers_.erase(leading_pos);
           heartbeat_->schedule(fallback_.get());
         }
       }
 #ifdef OPENDDS_SECURITY
     } else {
       // Move a specific reader.
       const ReaderInfoMap::iterator iter = remote_readers_.find(reader_id);
       if (iter == remote_readers_.end()) {
         return;
       }
 
       const ReaderInfo_rch& reader = iter->second;
       previous_max_sn = reader->max_pvs_sn_;
       reader->max_pvs_sn_ = max_sn_;
       if (preassociation_readers_.count(reader)) {
         // Will be inserted once association is complete.
         return;
       }
 
       const SequenceNumber acked_sn = reader->acked_sn();
       if (acked_sn == previous_max_sn && previous_max_sn != max_sn_) {
         snris_erase(leading_readers_, acked_sn, reader);
         snris_insert(lagging_readers_, reader);
         heartbeat_->schedule(fallback_.get());
       }
     }
 #endif
 }

◆ max_data_seq()

SequenceNumber OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::max_data_seq	(	const SingleSendBuffer::Proxy &	proxy,
		const ReaderInfo_rch &	ri
	)		const

Definition at line 4484 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::SingleSendBuffer::Proxy::empty(), OpenDDS::DCPS::SingleSendBuffer::Proxy::high(), OpenDDS::DCPS::SingleSendBuffer::Proxy::pre_empty(), and OpenDDS::DCPS::SingleSendBuffer::Proxy::pre_high().

 {
   const SequenceNumber durable_max = ri->durable_data_.empty() ? 0 : ri->durable_data_.rbegin()->first;
   const SequenceNumber pre_max = proxy.pre_empty() ? 0 : proxy.pre_high();
   const SequenceNumber data_max = proxy.empty() ? 0 : proxy.high();
   return std::max(durable_max, std::max(pre_max, data_max));
 }

◆ non_durable_first_sn()

SequenceNumber OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::non_durable_first_sn ( const SingleSendBuffer::Proxy & proxy ) const

inlineprivate

Definition at line 500 of file RtpsUdpDataLink.h.

References OpenDDS::DCPS::SingleSendBuffer::Proxy::empty(), OpenDDS::DCPS::SingleSendBuffer::Proxy::low(), OpenDDS::DCPS::SingleSendBuffer::Proxy::pre_empty(), and OpenDDS::DCPS::SingleSendBuffer::Proxy::pre_low().

     {
       if (!proxy.empty()) {
         return proxy.low();
       }
       if (!proxy.pre_empty()) {
         return proxy.pre_low();
       }
       return max_sn_ + 1;
     }

◆ OPENDDS_MULTIMAP()

typedef OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::OPENDDS_MULTIMAP	(	SequenceNumber	,
		TransportQueueElement *
	)

private

◆ OPENDDS_MULTISET()

typedef OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::OPENDDS_MULTISET ( OpenDDS::DCPS::SequenceNumber )

private

◆ OPENDDS_SET_CMP()

typedef OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::OPENDDS_SET_CMP	(	TransportQueueElement *	,
		TransportQueueElement::OrderBySequenceNumber
	)

private

◆ pre_stop_helper()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::pre_stop_helper	(	TqeVector &	to_drop,
		bool	true_stop
	)

Definition at line 823 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, OpenDDS::DCPS::RtpsUdpDataLink::heartbeat_, OpenDDS::DCPS::InternalDataReaderListener< NetworkInterfaceAddress >::mutex_, and OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_SET().

 {
   typedef SnToTqeMap::iterator iter_t;
 
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
   ACE_GUARD(ACE_Thread_Mutex, g2, elems_not_acked_mutex_);
 
   stopping_ = true_stop;
 
   if (!elems_not_acked_.empty()) {
     OPENDDS_SET(SequenceNumber) sns_to_release;
     iter_t iter = elems_not_acked_.begin();
     while (iter != elems_not_acked_.end()) {
       to_drop.push_back(iter->second);
       sns_to_release.insert(iter->first);
       elems_not_acked_.erase(iter);
       iter = elems_not_acked_.begin();
     }
     OPENDDS_SET(SequenceNumber)::iterator sns_it = sns_to_release.begin();
     while (sns_it != sns_to_release.end()) {
       send_buff_->release_acked(*sns_it);
       ++sns_it;
     }
   }
 
   send_buff_->pre_clear();
 
   g2.release();
   g.release();
 
   if (stopping_) {
     heartbeat_->cancel();
     nack_response_->cancel();
   }
 }

◆ process_acked_by_all()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_acked_by_all ( )

Definition at line 3953 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::mutex_.

 {
   TqeSet to_deliver;
   {
     ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
     acked_by_all_helper_i(to_deliver);
   }
 
   TqeSet::iterator deliver_iter = to_deliver.begin();
   while (deliver_iter != to_deliver.end()) {
     (*deliver_iter)->data_delivered();
     ++deliver_iter;
   }
 }

◆ process_acknack()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_acknack	(	const RTPS::AckNackSubmessage &	acknack,
		const GUID_t &	src,
		MetaSubmessageVec &	meta_submessages
	)

Definition at line 3155 of file RtpsUdpDataLink.cpp.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::received().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
   if (stopping_) {
     return;
   }
 
   RtpsUdpDataLink_rch link = link_.lock();
 
   if (!link) {
     return;
   }
 
   const SequenceNumber ack = to_opendds_seqnum(acknack.readerSNState.bitmapBase);
 
   if (Transport_debug_level > 5) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::RtpsWriter::process_acknack: %C -> %C base %q bits %u count %d\n",
       LogGuid(src).c_str(), LogGuid(id_).c_str(), ack.getValue(), acknack.readerSNState.numBits, acknack.count.value));
   }
 
   ReaderInfoMap::iterator ri = remote_readers_.find(src);
   if (ri == remote_readers_.end()) {
     if (transport_debug.log_dropped_messages) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsWriter::process_acknack: %C -> %C unknown remote reader\n", LogGuid(src).c_str(), LogGuid(id_).c_str()));
     }
     VDBG((LM_WARNING, "(%P|%t) RtpsUdpDataLink::received(ACKNACK) "
       "WARNING ReaderInfo not found\n"));
     return;
   }
 
   const ReaderInfo_rch& reader = ri->second;
 
   SequenceNumber previous_acked_sn = reader->acked_sn();
   const bool count_is_not_zero = acknack.count.value != 0;
   const CORBA::Long previous_count = reader->acknack_recvd_count_;
   bool dont_schedule_nack_response = false;
 
   if (count_is_not_zero) {
     if (!compare_and_update_counts(acknack.count.value, reader->acknack_recvd_count_) &&
         (!reader->reflects_heartbeat_count() || acknack.count.value != 0 || reader->acknack_recvd_count_ != 0)) {
       if (transport_debug.log_dropped_messages) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsWriter::process_acknack: %C -> %C stale/duplicate message\n", LogGuid(src).c_str(), LogGuid(id_).c_str()));
       }
       VDBG((LM_WARNING, "(%P|%t) RtpsUdpDataLink::received(ACKNACK) "
             "WARNING Count indicates duplicate, dropping\n"));
       return;
     }
 
     if (reader->reflects_heartbeat_count()) {
       if (acknack.count.value < reader->required_acknack_count_) {
         if (transport_debug.log_dropped_messages) {
           ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsWriter::process_acknack: %C -> %C stale message (reflect %d < %d)\n", LogGuid(src).c_str(), LogGuid(id_).c_str(), acknack.count.value, reader->required_acknack_count_));
         }
         dont_schedule_nack_response = true;
       }
     }
   }
 
   fallback_.set(initial_fallback_);
 
   const bool is_final = acknack.smHeader.flags & RTPS::FLAG_F;
   const bool is_postassociation = count_is_not_zero && (is_final || bitmapNonEmpty(acknack.readerSNState) || ack != 1);
 
   if (preassociation_readers_.count(reader)) {
     if (is_postassociation) {
       remove_preassociation_reader(reader);
       if (transport_debug.log_progress) {
         log_progress("RTPS writer/reader association complete", id_, reader->id_, reader->participant_discovered_at_);
       }
       log_remote_counts("process_acknack");
 
       const SequenceNumber max_sn = expected_max_sn(reader);
       const SequenceNumber acked_sn = reader->acked_sn();
       snris_insert(acked_sn == max_sn ? leading_readers_ : lagging_readers_, reader);
       check_leader_lagger();
       // Heartbeat is already scheduled.
     }
   }
 
   OPENDDS_MAP(SequenceNumber, TransportQueueElement*) pendingCallbacks;
 
   if (!is_final && transport_debug.log_nonfinal_messages) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_nonfinal_messages} RtpsUdpDataLink::RtpsWriter::process_acknack: %C -> %C base %q bits %u count %d\n",
       LogGuid(src).c_str(), LogGuid(id_).c_str(), ack.getValue(), acknack.readerSNState.numBits, acknack.count.value));
   }
 
   // Process the ack.
   bool inform_send_listener = false;
   if (ack != SequenceNumber::SEQUENCENUMBER_UNKNOWN()) {
 
     if (ack >= reader->cur_cumulative_ack_) {
       reader->cur_cumulative_ack_ = ack;
       inform_send_listener = true;
     } else if (count_is_not_zero) {
       // Count increased but ack decreased.  Reset.
       ACE_ERROR((LM_WARNING, "(%P|%t) WARNING RtpsUdpDataLink::RtpsWriter::process_acknack: "
                  "%C -> %C reset detected count %d > %d ack %q < %q\n",
                  LogGuid(id_).c_str(), LogGuid(reader->id_).c_str(),
                  acknack.count.value, previous_count, ack.getValue(), reader->cur_cumulative_ack_.getValue()));
       const SequenceNumber max_sn = expected_max_sn(reader);
       snris_erase(previous_acked_sn == max_sn ? leading_readers_ : lagging_readers_, previous_acked_sn, reader);
       reader->cur_cumulative_ack_ = ack;
       const SequenceNumber acked_sn = reader->acked_sn();
       snris_insert(acked_sn == max_sn ? leading_readers_ : lagging_readers_, reader);
       previous_acked_sn = acked_sn;
       check_leader_lagger();
       heartbeat_->schedule(fallback_.get());
 
       if (reader->durable_) {
         if (Transport_debug_level > 5) {
           ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::RtpsWriter::process_acknack: enqueuing ReplayDurableData\n"));
         }
         reader->durable_data_.swap(pendingCallbacks);
         link->event_dispatcher()->dispatch(make_rch<ReplayDurableData>(link_, id_, src));
         reader->durable_timestamp_ = MonotonicTimePoint::zero_value;
       }
     }
 
     if (!reader->durable_data_.empty()) {
       if (Transport_debug_level > 5) {
         const LogGuid local_conv(id_), remote_conv(src);
         ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::RtpsWriter::process_acknack: "
                    "local %C has durable for remote %C\n",
                    local_conv.c_str(),
                    remote_conv.c_str()));
       }
       const SequenceNumber& dd_last = reader->durable_data_.rbegin()->first;
       if (Transport_debug_level > 5) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::RtpsWriter::process_acknack: "
                    "check base %q against last durable %q\n",
                    ack.getValue(), dd_last.getValue()));
       }
       if (ack > dd_last) {
         if (transport_debug.log_progress) {
           log_progress("durable delivered", id_, reader->id_, reader->participant_discovered_at_);
         }
         // Reader acknowledges durable data, we no longer need to store it
         if (Transport_debug_level > 5) {
           ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::RtpsWriter::process_acknack: "
                      "durable data acked\n"));
         }
         reader->durable_data_.swap(pendingCallbacks);
       } else {
         for (OPENDDS_MAP(SequenceNumber, TransportQueueElement*)::iterator pos = reader->durable_data_.begin(),
                limit = reader->durable_data_.end(); pos != limit && pos->first < ack;) {
           pendingCallbacks.insert(*pos);
           reader->durable_data_.erase(pos++);
         }
       }
     }
   } else {
     ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: RtpsUdpDataLink::RtpsWriter::process_acknack: %C -> %C invalid acknack\n", LogGuid(src).c_str(), LogGuid(id_).c_str()));
   }
 
   // Process the nack.
   bool schedule_nack_response = false;
   if (!dont_schedule_nack_response) {
     if (count_is_not_zero) {
       reader->requests_.reset();
       {
         const SingleSendBuffer::Proxy proxy(*send_buff_);
         if ((acknack.readerSNState.numBits == 0 ||
              (acknack.readerSNState.numBits == 1 && !(acknack.readerSNState.bitmap[0] & 1)))
             && ack == max_data_seq(proxy, reader)) {
           // Since there is an entry in requested_changes_, the DR must have
           // sent a non-final AckNack.  If the base value is the high end of
           // the heartbeat range, treat it as a request for that seq#.
           if (reader->durable_data_.count(ack) || proxy.contains(ack) || proxy.pre_contains(ack)) {
             reader->requests_.insert(ack);
           }
         } else {
           reader->requests_.insert(ack, acknack.readerSNState.numBits, acknack.readerSNState.bitmap.get_buffer());
         }
 
         if (!reader->requests_.empty()) {
           readers_expecting_data_.insert(reader);
           schedule_nack_response = true;
         } else if (reader->requested_frags_.empty()) {
           readers_expecting_data_.erase(reader);
         }
       }
     }
 
     if (!is_final) {
       readers_expecting_heartbeat_.insert(reader);
       schedule_nack_response = true;
     }
   }
 
   if (preassociation_readers_.count(reader) == 0) {
     make_lagger_leader(reader, previous_acked_sn);
     check_leader_lagger();
   }
 
   TqeSet to_deliver;
   acked_by_all_helper_i(to_deliver);
 
 #ifdef OPENDDS_SECURITY
   if (is_pvs_writer_ &&
       !reader->pvs_outstanding_.empty() &&
       reader->pvs_outstanding_.low() < reader->cur_cumulative_ack_) {
     const OPENDDS_VECTOR(SequenceRange) psr = reader->pvs_outstanding_.present_sequence_ranges();
     for (OPENDDS_VECTOR(SequenceRange)::const_iterator pos = psr.begin(), limit = psr.end();
          pos != limit && pos->first < reader->cur_cumulative_ack_; ++pos) {
       ACE_GUARD(ACE_Thread_Mutex, g, elems_not_acked_mutex_);
       for (SequenceNumber seq = pos->first; seq <= pos->second && seq < reader->cur_cumulative_ack_; ++seq) {
         reader->pvs_outstanding_.erase(seq);
         OPENDDS_MULTIMAP(SequenceNumber, TransportQueueElement*)::iterator iter = elems_not_acked_.find(seq);
         if (iter != elems_not_acked_.end()) {
           send_buff_->release_acked(iter->first);
           to_deliver.insert(iter->second);
           elems_not_acked_.erase(iter);
         }
       }
     }
   }
 #endif
 
   if (!dont_schedule_nack_response && schedule_nack_response) {
     RtpsUdpInst_rch cfg = link->config();
     nack_response_->schedule(cfg ? cfg->nak_response_delay_ : TimeDuration(0, RtpsUdpInst::DEFAULT_NAK_RESPONSE_DELAY_USEC));
   }
 
   TransportClient_rch client = client_.lock();
 
   g.release();
 
   if (inform_send_listener && client) {
     client->data_acked(src);
   }
 
   typedef OPENDDS_MAP(SequenceNumber, TransportQueueElement*)::iterator iter_t;
   for (iter_t it = pendingCallbacks.begin();
        it != pendingCallbacks.end(); ++it) {
     it->second->data_delivered();
   }
 
   TqeSet::iterator deliver_iter = to_deliver.begin();
   while (deliver_iter != to_deliver.end()) {
     (*deliver_iter)->data_delivered();
     ++deliver_iter;
   }
 }

◆ process_nackfrag()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_nackfrag	(	const RTPS::NackFragSubmessage &	nackfrag,
		const GUID_t &	src,
		MetaSubmessageVec &	meta_submessages
	)

Definition at line 3411 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, ACE_GUARD, OpenDDS::RTPS::FragmentNumberSet::bitmapBase, OpenDDS::DCPS::RtpsUdpDataLink::config(), OpenDDS::RTPS::NackFragSubmessage::count, OpenDDS::DCPS::RtpsUdpInst::DEFAULT_NAK_RESPONSE_DELAY_USEC, OpenDDS::RTPS::NackFragSubmessage::fragmentNumberState, OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::id_, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::link_, LM_DEBUG, OpenDDS::DCPS::TransportDebug::log_dropped_messages, OpenDDS::DCPS::TransportDebug::log_nonfinal_messages, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::mutex_, OpenDDS::RTPS::FragmentNumberSet::numBits, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::stopping_, OpenDDS::RTPS::to_opendds_seqnum(), OpenDDS::DCPS::transport_debug, OpenDDS::DCPS::Transport_debug_level, OpenDDS::RTPS::Count_t::value, OpenDDS::RTPS::FragmentNumber_t::value, and OpenDDS::RTPS::NackFragSubmessage::writerSN.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::received().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
   if (stopping_) {
     return;
   }
 
   RtpsUdpDataLink_rch link = link_.lock();
 
   if (!link) {
     return;
   }
 
   if (Transport_debug_level > 5) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::RtpsWriter::process_nackfrag: %C -> %C\n",
       LogGuid(src).c_str(), LogGuid(id_).c_str()));
   }
 
   const ReaderInfoMap::iterator ri = remote_readers_.find(src);
   if (ri == remote_readers_.end()) {
     if (Transport_debug_level > 5 || transport_debug.log_dropped_messages) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsWriter::process_nackfrag: %C -> %C unknown remote reader\n",
         LogGuid(src).c_str(), LogGuid(id_).c_str()));
     }
     return;
   }
 
   const ReaderInfo_rch& reader = ri->second;
 
   if (!compare_and_update_counts(nackfrag.count.value, reader->nackfrag_recvd_count_)) {
     if (Transport_debug_level > 5 || transport_debug.log_dropped_messages) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsWriter::process_nackfrag: %C -> %C stale/duplicate message\n",
         LogGuid(src).c_str(), LogGuid(id_).c_str()));
     }
     return;
   }
 
   const SequenceNumber seq = to_opendds_seqnum(nackfrag.writerSN);
 
   // All NackFrag messages are technically 'non-final' since they are only used to negatively acknowledge fragments and expect a response
   if (transport_debug.log_nonfinal_messages) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_nonfinal_messages} RtpsUdpDataLink::RtpsWriter::process_nackfrag: %C -> %C seq %q base %u bits %u\n",
       LogGuid(src).c_str(), LogGuid(id_).c_str(), seq.getValue(), nackfrag.fragmentNumberState.bitmapBase.value, nackfrag.fragmentNumberState.numBits));
   }
 
   reader->requested_frags_[seq][nackfrag.fragmentNumberState.bitmapBase.value] = nackfrag.fragmentNumberState;
   readers_expecting_data_.insert(reader);
   RtpsUdpInst_rch cfg = link->config();
   nack_response_->schedule(cfg ? cfg->nak_response_delay_ : TimeDuration(0, RtpsUdpInst::DEFAULT_NAK_RESPONSE_DELAY_USEC));
 }

◆ reader_count()

size_t OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::reader_count ( ) const

Definition at line 2142 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::mutex_.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, 0);
   return remote_readers_.size();
 }

◆ record_directed()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::record_directed	(	const GUID_t &	reader,
		SequenceNumber	seq
	)

private

Definition at line 3933 of file RtpsUdpDataLink.cpp.

 {
   ACE_UNUSED_ARG(reader_id);
   ACE_UNUSED_ARG(seq);
 #ifdef OPENDDS_SECURITY
   if (!is_pvs_writer_) {
     return;
   }
 
   const ReaderInfoMap::iterator iter = remote_readers_.find(reader_id);
   if (iter == remote_readers_.end()) {
     return;
   }
 
   const ReaderInfo_rch& reader = iter->second;
   reader->pvs_outstanding_.insert(seq);
 #endif
 }

◆ remove_all_msgs()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_all_msgs ( )

Definition at line 249 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, OpenDDS::DCPS::DataLink::id_, OpenDDS::DCPS::InternalDataReaderListener< NetworkInterfaceAddress >::mutex_, OpenDDS::DCPS::TransportSendStrategy::remove_all_msgs(), OpenDDS::DCPS::DataLink::send_strategy_, OpenDDS::DCPS::DataLink::strategy_lock_, and OpenDDS::DCPS::SequenceNumber::ZERO().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
   RtpsUdpDataLink_rch link = link_.lock();
   if (!link) {
     return;
   }
 
   send_buff_->retain_all(id_);
 
   {
     ACE_Reverse_Lock<ACE_Thread_Mutex> rev_lock(mutex_);
     ACE_Guard<ACE_Reverse_Lock<ACE_Thread_Mutex> > rg(rev_lock);
     GuardType guard(link->strategy_lock_);
     if (link->send_strategy_) {
       link->send_strategy_->remove_all_msgs(id_);
     }
   }
 
   ACE_GUARD(ACE_Thread_Mutex, g2, elems_not_acked_mutex_);
 
   SnToTqeMap sn_tqe_map;
   sn_tqe_map.swap(elems_not_acked_);
 
   g2.release();
 
   SequenceNumber prev = SequenceNumber::ZERO();
   typedef SnToTqeMap::iterator iter_t;
   for (iter_t it = sn_tqe_map.begin(); it != sn_tqe_map.end(); ++it) {
     if (it->first != prev) {
       send_buff_->release_acked(it->first);
       prev = it->first;
     }
   }
 
   g.release();
 
   for (iter_t it = sn_tqe_map.begin(); it != sn_tqe_map.end(); ++it) {
     it->second->data_dropped(true);
   }
 }

◆ remove_preassociation_reader()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_preassociation_reader ( const ReaderInfo_rch & reader )

inlineprivate

Definition at line 511 of file RtpsUdpDataLink.h.

References OPENDDS_ASSERT.

     {
       if (preassociation_readers_.erase(reader)) {
         SequenceNumberMultiset::iterator pos = preassociation_reader_start_sns_.find(reader->start_sn_);
         OPENDDS_ASSERT(pos != preassociation_reader_start_sns_.end());
         preassociation_reader_start_sns_.erase(pos);
       }
     }

◆ remove_reader()

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_reader ( const GUID_t & id )

Definition at line 2085 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::log_remote_counts(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::mutex_, OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP(), OPENDDS_MULTIMAP, and OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_VECTOR().

 {
   OPENDDS_MAP(SequenceNumber, TransportQueueElement*) dd;
   TqeSet to_drop;
 
   bool result = false;
   {
     ACE_Guard<ACE_Thread_Mutex> g(mutex_);
     ReaderInfoMap::iterator it = remote_readers_.find(id);
     if (it != remote_readers_.end()) {
       const ReaderInfo_rch& reader = it->second;
       reader->swap_durable_data(dd);
       remove_preassociation_reader(reader);
       const SequenceNumber acked_sn = reader->acked_sn();
       const SequenceNumber max_sn = expected_max_sn(reader);
       readers_expecting_data_.erase(reader);
       readers_expecting_heartbeat_.erase(reader);
       snris_erase(acked_sn == max_sn ? leading_readers_ : lagging_readers_, acked_sn, reader);
       check_leader_lagger();
 
 #ifdef OPENDDS_SECURITY
       if (is_pvs_writer_ &&
           !reader->pvs_outstanding_.empty()) {
         const OPENDDS_VECTOR(SequenceRange) psr = reader->pvs_outstanding_.present_sequence_ranges();
         for (OPENDDS_VECTOR(SequenceRange)::const_iterator pos = psr.begin(), limit = psr.end(); pos != limit; ++pos) {
           ACE_GUARD_RETURN(ACE_Thread_Mutex, g, elems_not_acked_mutex_, result);
           for (SequenceNumber seq = pos->first; seq <= pos->second; ++seq) {
             OPENDDS_MULTIMAP(SequenceNumber, TransportQueueElement*)::iterator iter = elems_not_acked_.find(seq);
             if (iter != elems_not_acked_.end()) {
               send_buff_->release_acked(iter->first);
               to_drop.insert(iter->second);
               elems_not_acked_.erase(iter);
             }
           }
         }
       }
 #endif
 
       remote_readers_.erase(it);
       update_remote_guids_cache_i(false, id);
       result = true;
       log_remote_counts("remove_reader");
     }
   }
   typedef OPENDDS_MAP(SequenceNumber, TransportQueueElement*)::iterator iter_t;
   for (iter_t it = dd.begin(); it != dd.end(); ++it) {
     it->second->data_dropped();
   }
 
   for (TqeSet::iterator pos = to_drop.begin(), limit = to_drop.end(); pos != limit; ++pos) {
     (*pos)->data_dropped();
   }
 
   return result;
 }

◆ remove_sample()

RemoveResult OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_sample ( const DataSampleElement * sample )

Definition at line 182 of file RtpsUdpDataLink.cpp.

References ACE_Message_Block::cont(), OpenDDS::DCPS::TransportQueueElement::data_dropped(), OpenDDS::DCPS::DataSampleElement::get_header(), OpenDDS::DCPS::DataSampleElement::get_sample(), OpenDDS::DCPS::TransportQueueElement::MatchOnDataPayload::matches(), OpenDDS::DCPS::InternalDataReaderListener< NetworkInterfaceAddress >::mutex_, ACE_Message_Block::rd_ptr(), ACE_Guard< ACE_LOCK >::release(), OpenDDS::DCPS::REMOVE_FOUND, OpenDDS::DCPS::REMOVE_NOT_FOUND, OpenDDS::DCPS::TransportSendStrategy::remove_sample(), OpenDDS::DCPS::DataLink::send_strategy_, OpenDDS::DCPS::DataSampleHeader::sequence_, and OpenDDS::DCPS::DataLink::strategy_lock_.

 {
   bool found = false;
   SequenceNumber to_release;
   TransportQueueElement* tqe = 0;
 
   const SequenceNumber& seq = sample->get_header().sequence_;
   const char* const payload = sample->get_sample()->cont()->rd_ptr();
   const TransportQueueElement::MatchOnDataPayload modp(payload);
   SingleSendBuffer::BufferVec removed;
 
   ACE_Guard<ACE_Thread_Mutex> g(mutex_);
 
   RtpsUdpDataLink_rch link = link_.lock();
   if (!link) {
     return REMOVE_NOT_FOUND;
   }
 
   RemoveResult result = REMOVE_NOT_FOUND;
   {
     GuardType guard(link->strategy_lock_);
     if (link->send_strategy_) {
       ACE_Reverse_Lock<ACE_Thread_Mutex> rev_lock(mutex_);
       ACE_Guard<ACE_Reverse_Lock<ACE_Thread_Mutex> > rg(rev_lock);
       result = link->send_strategy_->remove_sample(sample);
       guard.release();
     }
   }
 
   ACE_Guard<ACE_Thread_Mutex> g2(elems_not_acked_mutex_);
 
   if (!elems_not_acked_.empty()) {
     typedef SnToTqeMap::iterator iter_t;
     for (std::pair<iter_t, iter_t> er = elems_not_acked_.equal_range(seq); er.first != er.second; ++er.first) {
       if (modp.matches(*er.first->second)) {
         found = true;
         to_release = seq;
         tqe = er.first->second;
         elems_not_acked_.erase(er.first);
         break;
       }
     }
   }
 
   g2.release();
 
   if (found) {
     send_buff_->remove_acked(to_release, removed);
   }
 
   g.release();
 
   if (found) {
     for (size_t i = 0; i < removed.size(); ++i) {
       RemoveAllVisitor visitor;
       removed[i].first->accept_remove_visitor(visitor);
       delete removed[i].first;
       removed[i].second->release();
     }
     removed.clear();
     tqe->data_dropped(true);
     result = REMOVE_FOUND;
   }
   return result;
 }

◆ request_ack_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::request_ack_i	(	const DataSampleHeader &	header,
		ACE_Message_Block *	body,
		MetaSubmessageVec &	meta_submessages
	)

private

Definition at line 1393 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::DataLink::id_, LM_DEBUG, OpenDDS::DCPS::DataSampleHeader::message_length_, ACE_Message_Block::rd_ptr(), and OpenDDS::DCPS::Transport_debug_level.

 {
   // Set the ReaderInfo::durable_timestamp_ for the case where no
   // durable samples exist in the DataWriter.
   GUID_t sub = GUID_UNKNOWN;
   if (body && header.message_length_ >= sizeof(sub)) {
     std::memcpy(&sub, body->rd_ptr(), sizeof(sub));
   }
   typedef ReaderInfoMap::iterator iter_t;
   if (sub == GUID_UNKNOWN) {
     gather_heartbeats_i(meta_submessages);
     if (Transport_debug_level > 3) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::request_ack "
                  "local %C all readers\n", LogGuid(id_).c_str()));
     }
   } else {
     iter_t iter = remote_readers_.find(sub);
     if (iter != remote_readers_.end()) {
       const SingleSendBuffer::Proxy proxy(*send_buff_);
       MetaSubmessage meta_submessage(id_, GUID_UNKNOWN);
       initialize_heartbeat(proxy, meta_submessage);
       gather_directed_heartbeat_i(proxy, meta_submessages, meta_submessage, iter->second);
       if (Transport_debug_level > 3) {
         const LogGuid conv(id_), sub_conv(sub);
         ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::request_ack"
                    " local %C remote %C\n", conv.c_str(), sub_conv.c_str()));
       }
     }
   }
 }

◆ send_heartbeats()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::send_heartbeats ( const MonotonicTimePoint & now )

private

Definition at line 1449 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, OpenDDS::DCPS::RtpsUdpDataLink::heartbeat_, OpenDDS::DCPS::InternalDataReaderListener< NetworkInterfaceAddress >::mutex_, and OpenDDS::DCPS::RtpsUdpDataLink::queue_submessages().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
   if (stopping_) {
     return;
   }
 
   RtpsUdpDataLink_rch link = link_.lock();
 
   if (!link) {
     return;
   }
 
   MetaSubmessageVec meta_submessages;
   gather_heartbeats_i(meta_submessages);
 
   if (!preassociation_readers_.empty() || !lagging_readers_.empty()) {
     heartbeat_->schedule(fallback_.get());
     fallback_.advance();
   } else {
     fallback_.set(initial_fallback_);
   }
 
   g.release();
 
   link->queue_submessages(meta_submessages);
 }

◆ send_heartbeats_manual_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::send_heartbeats_manual_i ( MetaSubmessageVec & meta_submessages )

private

Definition at line 4380 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::FLAG_F, OpenDDS::RTPS::FLAG_L, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::RTPS::HEARTBEAT, OpenDDS::RTPS::Submessage::heartbeat_sm, OpenDDS::RTPS::HEARTBEAT_SZ, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::id_, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::link_, OpenDDS::DCPS::MetaSubmessage::sm_, and OpenDDS::RTPS::to_rtps_seqnum().

 {
   using namespace OpenDDS::RTPS;
 
   RtpsUdpDataLink_rch link = link_.lock();
   if (!link) {
     return;
   }
 
   const SingleSendBuffer::Proxy proxy(*send_buff_);
   const SequenceNumber firstSN = durable_ ? 1 : non_durable_first_sn(proxy);
   const SequenceNumber lastSN = max_sn_;
   const int counter = ++heartbeat_count_;
 
   const HeartBeatSubmessage hb = {
     {HEARTBEAT,
      CORBA::Octet(FLAG_E | FLAG_F | FLAG_L),
      HEARTBEAT_SZ},
     ENTITYID_UNKNOWN, // any matched reader may be interested in this
     id_.entityId,
     to_rtps_seqnum(firstSN),
     to_rtps_seqnum(lastSN),
     {counter}
   };
 
   MetaSubmessage meta_submessage(id_, GUID_UNKNOWN);
   meta_submessage.sm_.heartbeat_sm(hb);
 
   meta_submessages.push_back(meta_submessage);
 }

◆ send_nack_responses()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::send_nack_responses ( const MonotonicTimePoint & now )

private

Definition at line 1479 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, OpenDDS::DCPS::InternalDataReaderListener< NetworkInterfaceAddress >::mutex_, and OpenDDS::DCPS::RtpsUdpDataLink::queue_submessages().

 {
   RtpsUdpDataLink_rch link = link_.lock();
   if (!link) {
     return;
   }
 
   MetaSubmessageVec meta_submessages;
   {
     ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
     if (stopping_) {
       return;
     }
 
     gather_nack_replies_i(meta_submessages);
   }
 
   link->queue_submessages(meta_submessages);
 }

◆ snris_erase()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::snris_erase	(	RtpsUdpDataLink::SNRIS &	snris,
		const SequenceNumber	sn,
		const ReaderInfo_rch &	reader
	)

staticprivate

Definition at line 3785 of file RtpsUdpDataLink.cpp.

 {
   SNRIS::iterator pos = snris.find(sn);
   if (pos != snris.end()) {
     pos->second->readers.erase(reader);
     if (pos->second->readers.empty()) {
       snris.erase(pos);
     }
   }
 }

◆ snris_insert()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::snris_insert	(	RtpsUdpDataLink::SNRIS &	snris,
		const ReaderInfo_rch &	reader
	)

staticprivate

Definition at line 3773 of file RtpsUdpDataLink.cpp.

 {
   const SequenceNumber sn = reader->acked_sn();
   SNRIS::iterator pos = snris.find(sn);
   if (pos == snris.end()) {
     pos = snris.insert(RtpsUdpDataLink::SNRIS::value_type(sn, make_rch<ReaderInfoSetHolder>())).first;
   }
   pos->second->readers.insert(reader);
 }

◆ update_max_sn()

SequenceNumber OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::update_max_sn	(	const GUID_t &	reader,
		SequenceNumber	seq
	)

Definition at line 4494 of file RtpsUdpDataLink.cpp.

References check_leader_lagger(), make_leader_lagger(), max_sn_, and mutex_.

 {
   ACE_Guard<ACE_Thread_Mutex> g(mutex_);
   SequenceNumber previous_max_sn = max_sn_;
   max_sn_ = std::max(max_sn_, seq);
   make_leader_lagger(reader, previous_max_sn);
   check_leader_lagger();
   return max_sn_;
 }

◆ update_remote_guids_cache_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::update_remote_guids_cache_i	(	bool	add,
		const GUID_t &	guid
	)

private

Definition at line 4204 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::RtpsUdpDataLink::bundling_cache_, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::insert(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::link_, and OpenDDS::DCPS::AddressCache< Key >::remove_id().

 {
   RtpsUdpDataLink_rch link = link_.lock();
   if (!link) {
     return;
   }
 
   {
     ACE_Guard<ACE_Thread_Mutex> rrg_guard(remote_reader_guids_mutex_);
 
     // We make a new RcHandle to prevent changing what's being pointed to by existing references in the send queue (i.e. to preserve historic values)
     RcHandle<ConstSharedRepoIdSet> temp = make_rch<ConstSharedRepoIdSet>();
     if (remote_reader_guids_) {
       const_cast<RepoIdSet&>(temp->guids_) = remote_reader_guids_->guids_;
     }
     if (add) {
       const_cast<RepoIdSet&>(temp->guids_).insert(guid);
     } else {
       const_cast<RepoIdSet&>(temp->guids_).erase(guid);
     }
     remote_reader_guids_ = temp;
   }
 
   link->bundling_cache_.remove_id(GUID_UNKNOWN);
 }

◆ update_required_acknack_count()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::update_required_acknack_count	(	const GUID_t &	id,
		CORBA::Long	current
	)

Definition at line 2731 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::mutex_.

 {
   ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
   ReaderInfoMap::iterator ri = remote_readers_.find(id);
   if (ri != remote_readers_.end()) {
     ri->second->required_acknack_count_ = current;
   }
 }

Member Data Documentation

◆ client_

WeakRcHandle<TransportClient> OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::client_

private

Definition at line 446 of file RtpsUdpDataLink.h.

◆ durable_

const bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::durable_

private

Definition at line 449 of file RtpsUdpDataLink.h.

◆ elems_not_acked_

SnToTqeMap OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::elems_not_acked_

private

Definition at line 445 of file RtpsUdpDataLink.h.

Referenced by add_elem_awaiting_ack(), and ~RtpsWriter().

◆ elems_not_acked_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::elems_not_acked_mutex_

mutableprivate

Definition at line 460 of file RtpsUdpDataLink.h.

Referenced by add_elem_awaiting_ack().

◆ fallback_

FibonacciSequence<TimeDuration> OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::fallback_

private

Definition at line 466 of file RtpsUdpDataLink.h.

◆ heartbeat_

RcHandle<SporadicEvent> OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::heartbeat_

private

Definition at line 462 of file RtpsUdpDataLink.h.

◆ heartbeat_count_

CORBA::Long OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::heartbeat_count_

private

Definition at line 451 of file RtpsUdpDataLink.h.

Referenced by inc_heartbeat_count().

◆ id_

const GUID_t OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::id_

private

Definition at line 448 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::deliver_held_data(), log_remote_counts(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::log_remote_counts().

◆ initial_fallback_

const TimeDuration OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::initial_fallback_

private

Definition at line 465 of file RtpsUdpDataLink.h.

◆ is_ps_writer_

const bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::is_ps_writer_

private

Partcicipant Secure (Reliable SPDP) writer.

Definition at line 456 of file RtpsUdpDataLink.h.

◆ is_pvs_writer_

const bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::is_pvs_writer_

private

Participant Volatile Secure writer.

Definition at line 454 of file RtpsUdpDataLink.h.

◆ lagging_readers_

SNRIS OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::lagging_readers_

private

These readers have not acked everything they are supposed to have acked.

Definition at line 434 of file RtpsUdpDataLink.h.

◆ leading_readers_

SNRIS OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::leading_readers_

private

These reader have acked everything they are supposed to have acked.

Definition at line 436 of file RtpsUdpDataLink.h.

◆ link_

WeakRcHandle<RtpsUdpDataLink> OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::link_

private

Definition at line 447 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::deliver_held_data().

◆ max_sn_

SequenceNumber OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::max_sn_

private

Definition at line 442 of file RtpsUdpDataLink.h.

Referenced by update_max_sn().

◆ mutex_

ACE_Thread_Mutex OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::mutex_

mutableprivate

Definition at line 458 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::deliver_held_data(), inc_heartbeat_count(), is_leading(), and update_max_sn().

◆ nack_response_

RcHandle<SporadicEvent> OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::nack_response_

private

Definition at line 463 of file RtpsUdpDataLink.h.

◆ preassociation_reader_start_sns_

SequenceNumberMultiset OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::preassociation_reader_start_sns_

private

Definition at line 431 of file RtpsUdpDataLink.h.

◆ preassociation_readers_

ReaderInfoSet OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::preassociation_readers_

private

Preassociation readers require a non-final heartbeat.

Definition at line 429 of file RtpsUdpDataLink.h.

Referenced by log_remote_counts().

◆ readers_expecting_data_

ReaderInfoSet OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::readers_expecting_data_

private

These readers have sent a nack and are expecting data.

Definition at line 438 of file RtpsUdpDataLink.h.

◆ readers_expecting_heartbeat_

ReaderInfoSet OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::readers_expecting_heartbeat_

private

These readers have sent a non-final ack are are expecting a heartbeat.

Definition at line 440 of file RtpsUdpDataLink.h.

◆ remote_reader_guids_

RcHandle<ConstSharedRepoIdSet> OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remote_reader_guids_

private

Definition at line 427 of file RtpsUdpDataLink.h.

◆ remote_reader_guids_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remote_reader_guids_mutex_

mutableprivate

Definition at line 459 of file RtpsUdpDataLink.h.

◆ remote_readers_

ReaderInfoMap OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remote_readers_

private

Definition at line 426 of file RtpsUdpDataLink.h.

Referenced by is_leading(), and log_remote_counts().

◆ send_buff_

RcHandle<SingleSendBuffer> OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::send_buff_

private

Definition at line 441 of file RtpsUdpDataLink.h.

Referenced by RtpsWriter().

◆ stopping_

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::stopping_

private

Definition at line 450 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::deliver_held_data().

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

Public Member Functions

Private Member Functions

Static Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ RtpsWriter()

◆ ~RtpsWriter()

Member Function Documentation

◆ acked_by_all_helper_i()

◆ add_elem_awaiting_ack()

◆ add_gap_submsg_i()

◆ add_reader()

◆ check_leader_lagger()

◆ customize_queue_element_helper()

◆ end_historic_samples_i()

◆ expected_max_sn()

◆ gather_directed_heartbeat_i()

◆ gather_gaps_i()

◆ gather_heartbeats()

◆ gather_heartbeats_i()

◆ gather_nack_replies_i()

◆ get_remote_reader_guids()

◆ get_send_buff()

◆ has_reader()

◆ inc_heartbeat_count()

◆ initialize_heartbeat()

◆ is_lagging()

◆ is_leading() [1/2]

◆ is_leading() [2/2]

◆ is_pvs_writer()

◆ log_remote_counts()

◆ make_lagger_leader()

◆ make_leader_lagger()

◆ max_data_seq()

◆ non_durable_first_sn()

◆ OPENDDS_MULTIMAP()

◆ OPENDDS_MULTISET()

◆ OPENDDS_SET_CMP()

◆ pre_stop_helper()

◆ process_acked_by_all()

◆ process_acknack()

◆ process_nackfrag()

◆ reader_count()

◆ record_directed()

◆ remove_all_msgs()

◆ remove_preassociation_reader()

◆ remove_reader()

◆ remove_sample()

◆ request_ack_i()

◆ send_heartbeats()

◆ send_heartbeats_manual_i()

◆ send_nack_responses()

◆ snris_erase()

◆ snris_insert()

◆ update_max_sn()

◆ update_remote_guids_cache_i()

◆ update_required_acknack_count()

Member Data Documentation

◆ client_

◆ durable_

◆ elems_not_acked_

◆ elems_not_acked_mutex_

◆ fallback_

◆ heartbeat_

◆ heartbeat_count_

◆ id_

◆ initial_fallback_

◆ is_ps_writer_

◆ is_pvs_writer_

◆ lagging_readers_

◆ leading_readers_

◆ link_

◆ max_sn_

◆ mutex_

◆ nack_response_

◆ preassociation_reader_start_sns_

◆ preassociation_readers_

◆ readers_expecting_data_