OpenDDS::DCPS::ReliableSession Class Reference

#include <ReliableSession.h>

Inheritance diagram for OpenDDS::DCPS::ReliableSession:

Collaboration diagram for OpenDDS::DCPS::ReliableSession:

Public Member Functions
	ReliableSession (RcHandle< ReactorInterceptor > interceptor, MulticastDataLink *link, MulticastPeer remote_peer)
	~ReliableSession ()
virtual bool	check_header (const TransportHeader &header)
virtual void	record_header_received (const TransportHeader &header)
virtual bool	ready_to_deliver (const TransportHeader &header, const ReceivedDataSample &data)
void	deliver_held_data ()
virtual void	release_remote (const GUID_t &remote)
virtual bool	control_received (char submessage_id, const Message_Block_Ptr &control)
void	expire_naks ()
void	send_naks ()
void	nak_received (const Message_Block_Ptr &control)
void	send_naks (DisjointSequence &found)
void	nakack_received (const Message_Block_Ptr &control)
virtual void	send_nakack (SequenceNumber low)
virtual bool	start (bool active, bool acked)
virtual void	stop ()
virtual bool	is_reliable ()
virtual void	syn_hook (const SequenceNumber &seq)
Public Member Functions inherited from OpenDDS::DCPS::MulticastSession
virtual	~MulticastSession ()
MulticastDataLink *	link ()
MulticastPeer	remote_peer () const
bool	acked ()
void	set_acked ()
void	syn_received (const Message_Block_Ptr &control)
void	send_all_syn (const MonotonicTimePoint &now)
void	send_syn (const GUID_t &local_writer, const GUID_t &remote_reader)
void	synack_received (const Message_Block_Ptr &control)
void	send_synack (const GUID_t &local_reader, const GUID_t &remote_writer)
bool	reassemble (ReceivedDataSample &data, const TransportHeader &header)
void	add_remote (const GUID_t &local)
void	add_remote (const GUID_t &local, const GUID_t &remote)
void	remove_remote (const GUID_t &local, const GUID_t &remote)
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 Types
typedef PmfSporadicTask< ReliableSession >	Sporadic
typedef SequenceNumber	TransportHeaderSN

Private Member Functions
TimeDuration	nak_delay ()
void	process_naks (const MonotonicTimePoint &)
typedef	OPENDDS_MAP (MonotonicTimePoint, SequenceNumber) NakRequestMap
	OPENDDS_MULTIMAP (TransportHeaderSN, ReceivedDataSample) held_
typedef	OPENDDS_SET (SequenceRange) NakPeerSet

Private Attributes
RcHandle< Sporadic >	nak_watchdog_
DisjointSequence	nak_sequence_
NakRequestMap	nak_requests_
ACE_Thread_Mutex	held_lock_
NakPeerSet	nak_peers_

Additional Inherited Members
Protected Types inherited from OpenDDS::DCPS::MulticastSession
typedef ACE_Reverse_Lock< ACE_Thread_Mutex >	Reverse_Lock_t
Protected Member Functions inherited from OpenDDS::DCPS::MulticastSession
	MulticastSession (RcHandle< ReactorInterceptor > interceptor, MulticastDataLink *link, MulticastPeer remote_peer)
void	send_control (char submessage_id, Message_Block_Ptr data)
void	start_syn ()
typedef	OPENDDS_MAP_CMP (GUID_t, RepoIdSet, GUID_tKeyLessThan) PendingRemoteMap
Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()
Protected Attributes inherited from OpenDDS::DCPS::MulticastSession
MulticastDataLink *	link_
MulticastPeer	remote_peer_
ACE_Thread_Mutex	start_lock_
Reverse_Lock_t	reverse_start_lock_
bool	started_
bool	active_
TransportReassembly	reassembly_
bool	acked_
PendingRemoteMap	pending_remote_map_

Detailed Description

Definition at line 27 of file ReliableSession.h.

Member Typedef Documentation

Sporadic

typedef PmfSporadicTask<ReliableSession> OpenDDS::DCPS::ReliableSession::Sporadic

private

Definition at line 63 of file ReliableSession.h.

TransportHeaderSN

typedef SequenceNumber OpenDDS::DCPS::ReliableSession::TransportHeaderSN

private

Definition at line 74 of file ReliableSession.h.

Constructor & Destructor Documentation

ReliableSession()

OpenDDS::DCPS::ReliableSession::ReliableSession	(	RcHandle< ReactorInterceptor >	interceptor,
		MulticastDataLink *	link,
		MulticastPeer	remote_peer
	)

Definition at line 33 of file ReliableSession.cpp.

  : MulticastSession(interceptor, link, remote_peer)
  , nak_watchdog_(make_rch<Sporadic>(TheServiceParticipant->time_source(),
                                     interceptor,
                                     rchandle_from(this),
                                     &ReliableSession::process_naks))
{}

~ReliableSession()

OpenDDS::DCPS::ReliableSession::~ReliableSession

(

)

Definition at line 43 of file ReliableSession.cpp.

References nak_watchdog_.

{
  nak_watchdog_->cancel();
}

Member Function Documentation

check_header()

bool OpenDDS::DCPS::ReliableSession::check_header ( const TransportHeader &  header )

virtual

Implements OpenDDS::DCPS::MulticastSession.

Definition at line 49 of file ReliableSession.cpp.

References OpenDDS::DCPS::MulticastSession::active_, OpenDDS::DCPS::DisjointSequence::insert(), nak_sequence_, OpenDDS::DCPS::MulticastSession::remote_peer_, OpenDDS::DCPS::TransportHeader::sequence_, and OpenDDS::DCPS::TransportHeader::source_.

{
  // Not from the remote peer for this session.
  if (this->remote_peer_ != header.source_) return false;

  // Active sessions don't need to track nak_sequence_
  if (this->active_) return true;

  // Update last seen sequence for remote peer; return false if we
  // have already seen this datagram to prevent duplicate delivery
  // Note: SN 2 is first SN recorded - fill in up to 2 when rcvd
  return this->nak_sequence_.insert(SequenceRange(
      header.sequence_ == 2 ? SequenceNumber() : header.sequence_,
      header.sequence_));
}

control_received()

bool OpenDDS::DCPS::ReliableSession::control_received ( char  submessage_id, const Message_Block_Ptr &  control  )

virtual

Reimplemented from OpenDDS::DCPS::MulticastSession.

Definition at line 183 of file ReliableSession.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::MulticastSession::control_received(), LM_WARNING, OpenDDS::DCPS::MULTICAST_NAK, OpenDDS::DCPS::MULTICAST_NAKACK, nak_received(), and nakack_received().

{
  if (MulticastSession::control_received(submessage_id, control)) {
    return true; // base class handled message
  }

  switch (submessage_id) {
  case MULTICAST_NAK:
    nak_received(control);
    break;

  case MULTICAST_NAKACK:
    nakack_received(control);
    break;

  default:
    ACE_ERROR((LM_WARNING,
               ACE_TEXT("(%P|%t) WARNING: ")
               ACE_TEXT("ReliableSession::control_received: ")
               ACE_TEXT("unknown TRANSPORT_CONTROL submessage: 0x%x!\n"),
               submessage_id));
    break;
  }
  return true;
}

deliver_held_data()

void OpenDDS::DCPS::ReliableSession::deliver_held_data

(

)

Definition at line 135 of file ReliableSession.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::DisjointSequence::cumulative_ack(), OpenDDS::DCPS::DataLink::data_received(), OpenDDS::DCPS::DisjointSequence::empty(), held_lock_, OpenDDS::DCPS::MulticastSession::link_, LM_DEBUG, OpenDDS::DCPS::DisjointSequence::low(), nak_sequence_, OPENDDS_MULTIMAP(), OpenDDS::DCPS::OPENDDS_VECTOR(), and OpenDDS::DCPS::Transport_debug_level.

Referenced by expire_naks(), nakack_received(), ready_to_deliver(), and record_header_received().

{
  if (nak_sequence_.empty() || nak_sequence_.low() > 1) return;

  OPENDDS_VECTOR(ReceivedDataSample) to_deliver;
  const SequenceNumber ca = nak_sequence_.cumulative_ack();

  {
    ACE_GUARD(ACE_Thread_Mutex, guard, held_lock_);

    typedef OPENDDS_MULTIMAP(SequenceNumber, ReceivedDataSample)::iterator iter;
    const iter end = this->held_.upper_bound(ca);
    for (iter it = this->held_.begin(); it != end; /*increment in loop body*/) {
      if (Transport_debug_level > 5) {
        LogGuid writer(it->second.header_.publication_id_);
        ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) MulticastDataLink::deliver_held_data -")
                             ACE_TEXT(" deliver tseq: %q dseq: %q from %C\n"),
                             it->first.getValue(),
                             it->second.header_.sequence_.getValue(),
                             writer.c_str()));
      }
      to_deliver.push_back(it->second);
      this->held_.erase(it++);
    }
  }
  for (size_t i = 0; i < to_deliver.size(); ++i) {
    this->link_->data_received(to_deliver.at(i));
  }
}

expire_naks()

void OpenDDS::DCPS::ReliableSession::expire_naks

(

)

Definition at line 224 of file ReliableSession.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::MulticastDataLink::config(), OpenDDS::DCPS::TransportReassembly::data_unavailable(), OpenDDS::DCPS::MulticastInst::DEFAULT_NAK_TIMEOUT, deliver_held_data(), OpenDDS::DCPS::TimeDuration::from_msec(), OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::DisjointSequence::insert(), OpenDDS::DCPS::MulticastSession::link_, LM_WARNING, OpenDDS::DCPS::DisjointSequence::low(), nak_requests_, nak_sequence_, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OpenDDS::DCPS::MulticastSession::reassembly_, and OpenDDS::DCPS::MulticastSession::remote_peer_.

Referenced by process_naks().

{
  if (this->nak_requests_.empty()) return; // nothing to expire

  MulticastInst_rch cfg = link_->config();
  const TimeDuration timeout = cfg ? cfg->nak_timeout_: TimeDuration::from_msec(MulticastInst::DEFAULT_NAK_TIMEOUT);
  const MonotonicTimePoint deadline(MonotonicTimePoint::now() - timeout);
  NakRequestMap::iterator first(this->nak_requests_.begin());
  NakRequestMap::iterator last(this->nak_requests_.upper_bound(deadline));

  if (first == last) return; // nothing to expire

  // Skip unrecoverable datagrams to
  // re-establish a baseline to detect future reception gaps.
  SequenceNumber lastSeq = (last == this->nak_requests_.end())
                         ? this->nak_requests_.rbegin()->second
                         : last->second;

  std::vector<SequenceRange> dropped;
  if (this->nak_sequence_.insert(SequenceRange(this->nak_sequence_.low(),
                                               lastSeq), dropped)) {

    for (size_t i = 0; i < dropped.size(); ++i) {
      const SequenceRange& sr = dropped[i];
      reassembly_.data_unavailable(FragmentRange(sr.first.getValue(), sr.second.getValue()));
    }

    ACE_ERROR((LM_WARNING, ACE_TEXT("(%P|%t) WARNING: ReliableSession::expire_naks: ")
      ACE_TEXT("timed out waiting on remote peer %#08x%08x to send missing samples: %q - %q!\n"),
      (unsigned int)(this->remote_peer_ >> 32),
      (unsigned int) this->remote_peer_,
      this->nak_sequence_.low().getValue(),
      lastSeq.getValue()));
  }

  // Clear expired repair requests:
  this->nak_requests_.erase(first, last);
  deliver_held_data();
}

is_reliable()

virtual bool OpenDDS::DCPS::ReliableSession::is_reliable ( )

inlinevirtual

Reimplemented from OpenDDS::DCPS::MulticastSession.

Definition at line 58 of file ReliableSession.h.

{ return true;}

nak_delay()

TimeDuration OpenDDS::DCPS::ReliableSession::nak_delay ( )

private

Definition at line 686 of file ReliableSession.cpp.

References OpenDDS::DCPS::MulticastDataLink::config(), OpenDDS::DCPS::MulticastInst::DEFAULT_NAK_INTERVAL, OpenDDS::DCPS::TimeDuration::from_msec(), and OpenDDS::DCPS::MulticastSession::link_.

Referenced by process_naks(), and start().

{
  MulticastInst_rch cfg = link_->config();
  TimeDuration interval = cfg ? cfg->nak_interval_ : TimeDuration::from_msec(MulticastInst::DEFAULT_NAK_INTERVAL);

  // Apply random backoff to minimize potential collisions:
  interval *= static_cast<double>(std::rand()) /
    static_cast<double>(RAND_MAX) + 1.0;

  return interval;
}

nak_received()

void OpenDDS::DCPS::ReliableSession::nak_received

(

const Message_Block_Ptr &

control

)

Definition at line 482 of file ReliableSession.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::MulticastSession::active_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::SingleSendBuffer::Proxy::empty(), OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), OpenDDS::DCPS::SequenceNumber::getValue(), header, OpenDDS::DCPS::MulticastSession::link(), OpenDDS::DCPS::MulticastSession::link_, LM_DEBUG, OpenDDS::DCPS::MulticastDataLink::local_peer(), OpenDDS::DCPS::SingleSendBuffer::Proxy::low(), OpenDDS::DCPS::MulticastDataLink::receive_strategy(), OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >::received_header(), OpenDDS::DCPS::MulticastSession::remote_peer_, OpenDDS::DCPS::SingleSendBuffer::resend(), OpenDDS::DCPS::MulticastDataLink::send_buffer(), send_nakack(), OpenDDS::DCPS::SequenceNumber::SEQUENCENUMBER_UNKNOWN(), OpenDDS::DCPS::TransportHeader::source_, and OpenDDS::DCPS::TransportHeader::swap_bytes().

Referenced by control_received().

{
  if (!this->active_) return; // sub send naks, then doesn't receive them.

  const TransportHeader& header =
    this->link_->receive_strategy()->received_header();

  Serializer serializer(control.get(), encoding_kind, header.swap_bytes());

  MulticastPeer local_peer;
  CORBA::ULong size = 0;
  serializer >> local_peer; // sent as remote_peer
  serializer >> size;

  std::vector<SequenceRange> ranges;

  for (CORBA::ULong i = 0; i < size; ++i) {
    SequenceRange range;
    serializer >> range.first;
    serializer >> range.second;
    ranges.push_back(range);
  }

  // Ignore sample if not destined for us:
  if ((local_peer != this->link_->local_peer())        // Not to us.
    || (this->remote_peer_ != header.source_)) return; // Not from the remote peer for this session.

  SingleSendBuffer* send_buffer = this->link_->send_buffer();
  // Broadcast a MULTICAST_NAKACK control sample before resending to suppress
  // repair requests for unrecoverable samples by providing a
  // new low-water mark for affected peers:
  SequenceNumber sn = SequenceNumber::SEQUENCENUMBER_UNKNOWN();
  {
    const SingleSendBuffer::Proxy proxy(*send_buffer);
    if (!proxy.empty() && proxy.low() > ranges.begin()->first) {
      if (OpenDDS::DCPS::DCPS_debug_level > 0) {
        ACE_DEBUG ((LM_DEBUG,
                    ACE_TEXT ("(%P|%t) ReliableSession::nak_received")
                    ACE_TEXT (" local %#08x%08x remote %#08x%08x sending nakack for lowest available: %q\n"),
                    (unsigned int)(this->link()->local_peer() >> 32),
                    (unsigned int) this->link()->local_peer(),
                    (unsigned int)(this->remote_peer_ >> 32),
                    (unsigned int) this->remote_peer_,
                    proxy.low().getValue()));
      }
      sn = proxy.low();
    }
  }
  if (sn != SequenceNumber::SEQUENCENUMBER_UNKNOWN()) {
    send_nakack(sn);
  }

  for (CORBA::ULong i = 0; i < size; ++i) {
    bool ret = send_buffer->resend(ranges[i]);
    if (OpenDDS::DCPS::DCPS_debug_level > 0) {
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("(%P|%t) ReliableSession::nak_received")
                  ACE_TEXT (" local %#08x%08x remote %#08x%08x [%q - %q] resend result %C\n"),
                  (unsigned int)(this->link()->local_peer() >> 32),
                  (unsigned int) this->link()->local_peer(),
                  (unsigned int)(this->remote_peer_ >> 32),
                  (unsigned int) this->remote_peer_,
                  ranges[i].first.getValue(), ranges[i].second.getValue(),
                  ret ? "SUCCESS" : "FAILED"));
    }
  }
}

nakack_received()

void OpenDDS::DCPS::ReliableSession::nakack_received

(

const Message_Block_Ptr &

control

)

Definition at line 587 of file ReliableSession.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::MulticastSession::active_, OpenDDS::DCPS::TransportReassembly::data_unavailable(), OpenDDS::DCPS::DCPS_debug_level, deliver_held_data(), OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), header, OpenDDS::DCPS::DisjointSequence::insert(), OpenDDS::DCPS::MulticastSession::link(), OpenDDS::DCPS::MulticastSession::link_, LM_WARNING, nak_sequence_, OpenDDS::DCPS::MulticastSession::reassembly_, OpenDDS::DCPS::MulticastDataLink::receive_strategy(), OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >::received_header(), OpenDDS::DCPS::MulticastSession::remote_peer_, OpenDDS::DCPS::TransportHeader::source_, and OpenDDS::DCPS::TransportHeader::swap_bytes().

Referenced by control_received().

{
  if (this->active_) return; // pub send nakack, doesn't receive them.

  const TransportHeader& header =
    this->link_->receive_strategy()->received_header();

  // Not from the remote peer for this session.
  if (this->remote_peer_ != header.source_) return;

  Serializer serializer(control.get(), encoding_kind, header.swap_bytes());

  SequenceNumber low;
  serializer >> low;

  // MULTICAST_NAKACK control samples indicate data which cannot be
  // repaired by a remote peer; if any values were needed below
  // this value, then the sequence needs to be shifted:
  std::vector<SequenceRange> dropped;
  SequenceNumber range_low =  SequenceNumber();
  SequenceNumber range_high = low == SequenceNumber() ? SequenceNumber() : low.previous();

  if (range_low == SequenceNumber() && range_high == SequenceNumber()) {

    this->nak_sequence_.insert(range_low);

  } else if (this->nak_sequence_.insert(SequenceRange(range_low, range_high), dropped)) {

    for (size_t i = 0; i < dropped.size(); ++i) {
      const SequenceRange& sr = dropped[i];
      reassembly_.data_unavailable(FragmentRange(sr.first.getValue(), sr.second.getValue()));
    }

    if (DCPS_debug_level > 0) {
      ACE_ERROR((LM_WARNING,
                ACE_TEXT("(%P|%t) WARNING: ReliableSession::nakack_received ")
                ACE_TEXT("local %#08x%08x remote %#08x%08x with low [%q] ")
                ACE_TEXT("- some ranges dropped.\n"),
                (unsigned int)(this->link()->local_peer() >> 32),
                (unsigned int) this->link()->local_peer(),
                (unsigned int)(this->remote_peer_ >> 32),
                (unsigned int) this->remote_peer_,
                low.getValue()));
    }
  }
  deliver_held_data();
}

OPENDDS_MAP()

typedef OpenDDS::DCPS::ReliableSession::OPENDDS_MAP ( MonotonicTimePoint  , SequenceNumber    )

private

OPENDDS_MULTIMAP()

OpenDDS::DCPS::ReliableSession::OPENDDS_MULTIMAP ( TransportHeaderSN  , ReceivedDataSample    )

private

Referenced by deliver_held_data(), ready_to_deliver(), and release_remote().

OPENDDS_SET()

typedef OpenDDS::DCPS::ReliableSession::OPENDDS_SET ( SequenceRange  )

private

process_naks()

void OpenDDS::DCPS::ReliableSession::process_naks ( const MonotonicTimePoint &  )

private

Definition at line 699 of file ReliableSession.cpp.

References expire_naks(), nak_delay(), nak_watchdog_, OPENDDS_END_VERSIONED_NAMESPACE_DECL, and send_naks().

{
  // Expire outstanding repair requests that have not yet been
  // fulfilled; this prevents NAK implosions due to remote
  // peers becoming unresponsive:
  expire_naks();

  // Initiate repairs by sending MULTICAST_NAK control samples
  // to remote peers from which we are missing data:
  send_naks();

  nak_watchdog_->schedule(nak_delay());
}

ready_to_deliver()

bool OpenDDS::DCPS::ReliableSession::ready_to_deliver ( const TransportHeader &  header, const ReceivedDataSample &  data  )

virtual

Implements OpenDDS::DCPS::MulticastSession.

Definition at line 81 of file ReliableSession.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::MulticastSession::acked(), OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::DisjointSequence::cumulative_ack(), deliver_held_data(), OpenDDS::DCPS::DisjointSequence::disjoint(), OpenDDS::DCPS::DisjointSequence::empty(), OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::ReceivedDataSample::header_, held_lock_, LM_DEBUG, OpenDDS::DCPS::DisjointSequence::low(), nak_sequence_, OPENDDS_MULTIMAP(), OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::TransportHeader::sequence_, OpenDDS::DCPS::DataSampleHeader::sequence_, and OpenDDS::DCPS::Transport_debug_level.

{
  if (!acked()
      || nak_sequence_.disjoint()
      || (!nak_sequence_.empty() && nak_sequence_.cumulative_ack() != header.sequence_)
      || (!nak_sequence_.empty() && nak_sequence_.low() > 1)
      || (nak_sequence_.empty() && header.sequence_ > 1)) {

    if (Transport_debug_level > 5) {
      LogGuid writer(data.header_.publication_id_);
      ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) ReliableSession::ready_to_deliver -")
                           ACE_TEXT(" tseq: %q data seq: %q from %C being WITHHELD because can't receive yet\n"),
                           header.sequence_.getValue(),
                           data.header_.sequence_.getValue(),
                           writer.c_str()));
    }
    {
      ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, held_lock_, false);

      held_.insert(std::pair<const SequenceNumber, ReceivedDataSample>(header.sequence_, data));

      if (Transport_debug_level > 5) {
        OPENDDS_MULTIMAP(SequenceNumber, ReceivedDataSample)::iterator it = held_.begin();
        ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) ReliableSession::ready_to_deliver -")
                             ACE_TEXT(" held_ data currently contains: %d samples\n"),
                             held_.size()));
        while (it != held_.end()) {
          LogGuid writer(it->second.header_.publication_id_);
          ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) ReliableSession::ready_to_deliver -")
                               ACE_TEXT(" held_ data currently contains: tseq: %q dseq: %q from %C HELD\n"),
                               it->first.getValue(),
                               it->second.header_.sequence_.getValue(),
                               writer.c_str()));
          ++it;
        }
      }
    }
    deliver_held_data();
    return false;
  } else {
    if (Transport_debug_level > 5) {
      LogGuid writer(data.header_.publication_id_);
      ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) ReliableSession::ready_to_deliver -")
                           ACE_TEXT(" tseq: %q data seq: %q from %C OK to deliver\n"),
                           header.sequence_.getValue(),
                           data.header_.sequence_.getValue(),
                           writer.c_str()));
    }
    return true;
  }
}

record_header_received()

void OpenDDS::DCPS::ReliableSession::record_header_received ( const TransportHeader &  header )

virtual

Implements OpenDDS::DCPS::MulticastSession.

Definition at line 66 of file ReliableSession.cpp.

References OpenDDS::DCPS::MulticastSession::active_, deliver_held_data(), OpenDDS::DCPS::DisjointSequence::insert(), nak_sequence_, OpenDDS::DCPS::MulticastSession::remote_peer_, OpenDDS::DCPS::TransportHeader::sequence_, and OpenDDS::DCPS::TransportHeader::source_.

{
  // Not from the remote peer for this session.
  if (this->remote_peer_ != header.source_) return;

  // Active sessions don't need to track nak_sequence_
  if (this->active_) return;

  // Update nak sequence for seen sequence from remote peer
  this->nak_sequence_.insert(header.sequence_);
  deliver_held_data();
}

release_remote()

void OpenDDS::DCPS::ReliableSession::release_remote ( const GUID_t &  remote )

virtual

Reimplemented from OpenDDS::DCPS::MulticastSession.

Definition at line 166 of file ReliableSession.cpp.

References ACE_GUARD, held_lock_, and OPENDDS_MULTIMAP().

{
  ACE_GUARD(ACE_Thread_Mutex, guard, held_lock_);
  if (!held_.empty()) {
    OPENDDS_MULTIMAP(TransportHeaderSN, ReceivedDataSample)::iterator it = held_.begin();
    while (it != held_.end()) {
      if (it->second.header_.publication_id_ == remote) {
        held_.erase(it++);
      } else {
        it++;
      }
    }
  }
}

send_nakack()

void OpenDDS::DCPS::ReliableSession::send_nakack ( SequenceNumber  low )

virtual

Definition at line 636 of file ReliableSession.cpp.

References OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::move(), OpenDDS::DCPS::MULTICAST_NAKACK, and OpenDDS::DCPS::MulticastSession::send_control().

Referenced by nak_received().

{
  size_t len = sizeof(low.getValue());

  Message_Block_Ptr data(new ACE_Message_Block(len));

  Serializer serializer(data.get(), encoding_unaligned_native);

  serializer << low;
  // Broadcast control sample to all peers:
  send_control(MULTICAST_NAKACK, move(data));
}

send_naks() [1/2]

void OpenDDS::DCPS::ReliableSession::send_naks ( )

virtual

The range first - second will be skipped (no naks sent for it).

Reimplemented from OpenDDS::DCPS::MulticastSession.

Definition at line 265 of file ReliableSession.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::MulticastSession::acked(), OpenDDS::DCPS::MulticastDataLink::config(), OpenDDS::DCPS::DisjointSequence::cumulative_ack(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::MulticastInst::DEFAULT_NAK_DELAY_INTERVALS, OpenDDS::DCPS::MulticastInst::DEFAULT_NAK_MAX, OpenDDS::DCPS::DisjointSequence::disjoint(), OpenDDS::DCPS::DisjointSequence::dump(), OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::DisjointSequence::insert(), OpenDDS::DCPS::MulticastSession::link(), OpenDDS::DCPS::MulticastSession::link_, LM_DEBUG, OpenDDS::DCPS::DisjointSequence::low(), OpenDDS::DCPS::move(), OpenDDS::DCPS::MULTICAST_NAK, nak_peers_, nak_requests_, nak_sequence_, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OpenDDS::DCPS::MulticastSession::remote_peer_, OpenDDS::DCPS::MulticastSession::send_control(), and ACE_Utils::truncate_cast().

Referenced by process_naks().

{
  // Could get data samples before syn control message.
  // No use nak'ing until syn control message is received and session is acked.
  if (!this->acked()) {
    if (DCPS_debug_level > 5) {
      ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) ReliableSession::send_naks local %#08x%08x ")
                           ACE_TEXT("remote %#08x%08x session NOT acked yet, don't send naks\n"),
                           (unsigned int)(this->link()->local_peer() >> 32),
                           (unsigned int) this->link()->local_peer(),
                           (unsigned int)(this->remote_peer_ >> 32),
                           (unsigned int) this->remote_peer_));
    }
    return;
  }

  if (DCPS_debug_level > 5) {
    ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) ReliableSession::send_naks local %#08x%08x ")
                         ACE_TEXT("remote %#08x%08x nak request size %d\n"),
                         (unsigned int)(this->link()->local_peer() >> 32),
                         (unsigned int) this->link()->local_peer(),
                         (unsigned int)(this->remote_peer_ >> 32),
                         (unsigned int) this->remote_peer_,
                         this->nak_requests_.size()));
  }

  if (!(this->nak_sequence_.low() > 1) && !this->nak_sequence_.disjoint()) {
    if (DCPS_debug_level > 5) {
      ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) ReliableSession::send_naks local %#08x%08x ")
                           ACE_TEXT("remote %#08x%08x nak sequence not disjoint, don't send naks\n"),
                           (unsigned int)(this->link()->local_peer() >> 32),
                           (unsigned int) this->link()->local_peer(),
                           (unsigned int)(this->remote_peer_ >> 32),
                           (unsigned int) this->remote_peer_));
    }

    if (DCPS_debug_level > 9) {
      const std::vector<SequenceRange> ranges(this->nak_sequence_.present_sequence_ranges());
      for (std::vector<SequenceRange>::const_iterator iter = ranges.begin();
           iter != ranges.end(); ++iter) {
          ACE_DEBUG((LM_DEBUG, "(%P|%t) ReliableSession::send_naks - local %#08x%08x remote %#08x%08x nak_sequence includes: [%q - %q]\n",
                     (unsigned int)(this->link()->local_peer() >> 32),
                     (unsigned int) this->link()->local_peer(),
                     (unsigned int)(this->remote_peer_ >> 32),
                     (unsigned int) this->remote_peer_,
                     iter->first.getValue(),
                     iter->second.getValue()));
      }
    }
    return;  // nothing to send
  }

  // Record low-water mark for this interval; this value will
  // be used to reset the low-water mark in the event the remote
  // peer becomes unresponsive:
  const MonotonicTimePoint now = MonotonicTimePoint::now();
  if (this->nak_sequence_.low() > 1) {
    this->nak_requests_[now] = SequenceNumber();
  } else {
    this->nak_requests_[now] = this->nak_sequence_.cumulative_ack();
  }

  typedef std::vector<SequenceRange> RangeVector;
  RangeVector ignored;

  /// The range first - second will be skipped (no naks sent for it).
  SequenceNumber first;
  SequenceNumber second;

  NakRequestMap::reverse_iterator itr(this->nak_requests_.rbegin());

  if (this->nak_requests_.size() > 1) {
    // The sequences between rbegin - 1 and rbegin will not be ignored for naking.
    ++itr;

    MulticastInst_rch cfg = link_->config();
    size_t nak_delay_intervals = cfg ? cfg->nak_delay_intervals_ : MulticastInst::DEFAULT_NAK_DELAY_INTERVALS;
    size_t nak_max = cfg ? cfg->nak_max_ : MulticastInst::DEFAULT_NAK_MAX;
    size_t sz = nak_requests_.size();

    // Image i is the index of element in nak_requests_ in reverse order.
    // index 0 sequence is most recent high water mark.
    // e.g index , 20, 19, 18, 17, 16, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0
    //  0 (rbegin) is always skipped because missing sample between 1 and 0 interval
    //  should always be naked.,
    //  if nak_delay_intervals=4, nak_max=3, any sequence between 5 - 1, 10 - 6, 15 - 11
    //  are skipped for naking due to nak_delay_intervals and 20 - 16 are skipped for
    //  naking due to nak_max.
    for (size_t i = 1; i < sz; ++i) {
      if ((i * 1.0) / (nak_delay_intervals + 1) > nak_max) {
        if (first != SequenceNumber()) {
          first = this->nak_requests_.begin()->second;
        }
        else {
          ignored.push_back(std::make_pair(this->nak_requests_.begin()->second, itr->second));
        }
        break;
      }

      if (i % (nak_delay_intervals + 1) == 1) {
        second = itr->second;
      }
      if (second != SequenceNumber()) {
        first = itr->second;
      }

      if (i % (nak_delay_intervals + 1) == 0) {
        first = itr->second;

        if (first != SequenceNumber() && second != SequenceNumber()) {
          ignored.push_back(std::make_pair(first, second));
          first = SequenceNumber();
          second = SequenceNumber();
        }
      }

      ++itr;
    }

    if (first != SequenceNumber() && second != SequenceNumber() && first != second) {
      ignored.push_back(std::make_pair(first, second));
    }
  }

  // Take a copy to facilitate temporary suppression:
  DisjointSequence received(this->nak_sequence_);
  if (DCPS_debug_level > 0) {
    received.dump();
  }

  size_t sz = ignored.size();
  for (size_t i = 0; i < sz; ++i) {

    if (ignored[i].second > received.cumulative_ack()) {
      SequenceNumber high = ignored[i].second;
      SequenceNumber low = ignored[i].first;
      if (low < received.cumulative_ack()) {
        low = received.cumulative_ack();
      }

      if (DCPS_debug_level > 5) {
        ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) ReliableSession::send_naks local %#08x%08x ")
          ACE_TEXT("remote %#08x%08x ignore missing [%q - %q]\n"),
          (unsigned int)(this->link()->local_peer() >> 32),
          (unsigned int) this->link()->local_peer(),
          (unsigned int)(this->remote_peer_ >> 32),
          (unsigned int) this->remote_peer_,
          low.getValue(), high.getValue()));
      }

      // Make contiguous between ignored sequences.
      received.insert(SequenceRange(low, high));
    }
  }

  for (NakPeerSet::iterator it(this->nak_peers_.begin());
       it != this->nak_peers_.end(); ++it) {
    // Update sequence to temporarily suppress repair requests for
    // ranges already requested by other peers for this interval:
    received.insert(*it);
  }
  bool sending_naks = false;
  if (received.low() > 1){
    //Special case: nak from beginning to make sure no missing sequence
    //number below the first received
    sending_naks = true;
    std::vector<SequenceRange> ranges;
    ranges.push_back(SequenceRange(SequenceNumber(), received.low()));

    CORBA::ULong size = ACE_Utils::truncate_cast<CORBA::ULong>(ranges.size());

    size_t len = sizeof(this->remote_peer_)
               + sizeof(size)
               + size * 2 * sizeof(SequenceNumber);

    Message_Block_Ptr data(new ACE_Message_Block(len));

    Serializer serializer(data.get(), encoding_unaligned_native);

    serializer << this->remote_peer_;
    serializer << size;
    for (std::vector<SequenceRange>::const_iterator iter = ranges.begin();
         iter != ranges.end(); ++iter) {
      serializer << iter->first;
      serializer << iter->second;
      if (DCPS_debug_level > 0) {
        ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) ReliableSession::send_naks ")
                              ACE_TEXT (" local %#08x%08x remote %#08x%08x [%q - %q]\n"),
                              (unsigned int)(this->link()->local_peer() >> 32),
                              (unsigned int) this->link()->local_peer(),
                              (unsigned int)(this->remote_peer_ >> 32),
                              (unsigned int) this->remote_peer_,
                              iter->first.getValue(), iter->second.getValue()));
      }
    }
    // Send control sample to remote peer:
    send_control(MULTICAST_NAK, move(data));
  }
  if (received.disjoint()) {
    sending_naks = true;
    send_naks(received);
  }

  if (!sending_naks && DCPS_debug_level > 5){
    ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) ReliableSession::send_naks local %#08x%08x ")
                         ACE_TEXT("remote %#08x%08x received sequence not disjoint, don't send naks\n"),
                         (unsigned int)(this->link()->local_peer() >> 32),
                         (unsigned int) this->link()->local_peer(),
                         (unsigned int)(this->remote_peer_ >> 32),
                         (unsigned int) this->remote_peer_));
  }

  // Clear peer repair requests:
  this->nak_peers_.clear();
}

send_naks() [2/2]

void OpenDDS::DCPS::ReliableSession::send_naks

(

DisjointSequence &

found

)

Definition at line 551 of file ReliableSession.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::MulticastSession::link(), LM_DEBUG, OpenDDS::DCPS::move(), OpenDDS::DCPS::MULTICAST_NAK, OpenDDS::DCPS::MulticastSession::remote_peer_, OpenDDS::DCPS::MulticastSession::send_control(), and ACE_Utils::truncate_cast().

{
  const std::vector<SequenceRange> ranges(received.missing_sequence_ranges());

  CORBA::ULong size = ACE_Utils::truncate_cast<CORBA::ULong>(ranges.size());

  size_t len = sizeof(this->remote_peer_)
             + sizeof(size)
             + size * 2 * sizeof(SequenceNumber);

  Message_Block_Ptr data(new ACE_Message_Block(len));

  Serializer serializer(data.get(), encoding_unaligned_native);

  serializer << this->remote_peer_;
  serializer << size;
  for (std::vector<SequenceRange>::const_iterator iter = ranges.begin();
       iter != ranges.end(); ++iter) {
    serializer << iter->first;
    serializer << iter->second;
    if (OpenDDS::DCPS::DCPS_debug_level > 0) {
      ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) ReliableSession::send_naks (Disjoint) ")
                            ACE_TEXT (" local %#08x%08x remote %#08x%08x [%q - %q]\n"),
                            (unsigned int)(this->link()->local_peer() >> 32),
                            (unsigned int) this->link()->local_peer(),
                            (unsigned int)(this->remote_peer_ >> 32),
                            (unsigned int) this->remote_peer_,
                            iter->first.getValue(), iter->second.getValue()));
    }
  }
  // Send control sample to remote peer:
  send_control(MULTICAST_NAK, move(data));
}

start()

bool OpenDDS::DCPS::ReliableSession::start ( bool  active, bool  acked  )

virtual

Implements OpenDDS::DCPS::MulticastSession.

Definition at line 650 of file ReliableSession.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::MulticastSession::active_, nak_delay(), nak_watchdog_, OpenDDS::DCPS::MulticastSession::reverse_start_lock_, OpenDDS::DCPS::MulticastSession::set_acked(), OpenDDS::DCPS::MulticastSession::start_lock_, and OpenDDS::DCPS::MulticastSession::started_.

{
  ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, this->start_lock_, false);

  if (this->started_) {
    return true;  // already started
  }

  this->active_  = active;
  {
    //can't call accept_datalink while holding lock due to possible reactor deadlock with passive_connection
    ACE_GUARD_RETURN(Reverse_Lock_t, unlock_guard, this->reverse_start_lock_, false);

    // A watchdog timer is scheduled to periodically check for gaps in
    // received data. If a gap is discovered, MULTICAST_NAK control
    // samples will be sent to initiate repairs.
    // Only subscriber send naks so just schedule for sub role.
    if (!active) {
      if (acked) {
        this->set_acked();
      }
      this->nak_watchdog_->schedule(nak_delay());
    }
  } //Reacquire start_lock_ after releasing unlock_guard with release_start_lock_

  return this->started_ = true;
}

stop()

void OpenDDS::DCPS::ReliableSession::stop ( void  )

virtual

Reimplemented from OpenDDS::DCPS::MulticastSession.

Definition at line 679 of file ReliableSession.cpp.

References nak_watchdog_, and OpenDDS::DCPS::MulticastSession::stop().

{
  MulticastSession::stop();
  this->nak_watchdog_->cancel();
}

syn_hook()

void OpenDDS::DCPS::ReliableSession::syn_hook ( const SequenceNumber &  seq )

virtual

Reimplemented from OpenDDS::DCPS::MulticastSession.

Definition at line 211 of file ReliableSession.cpp.

References OpenDDS::DCPS::DisjointSequence::insert(), nak_sequence_, and OpenDDS::DCPS::DisjointSequence::reset().

{
  const std::vector<SequenceRange> ranges(this->nak_sequence_.present_sequence_ranges());
  this->nak_sequence_.reset();
  this->nak_sequence_.insert(seq);

  for (std::vector<SequenceRange>::const_iterator iter = ranges.begin();
       iter != ranges.end(); ++iter) {
    this->nak_sequence_.insert(SequenceRange(iter->first, iter->second));
  }
}

Member Data Documentation

held_lock_

ACE_Thread_Mutex OpenDDS::DCPS::ReliableSession::held_lock_

private

Definition at line 73 of file ReliableSession.h.

Referenced by deliver_held_data(), ready_to_deliver(), and release_remote().

nak_peers_

NakPeerSet OpenDDS::DCPS::ReliableSession::nak_peers_

private

Definition at line 78 of file ReliableSession.h.

Referenced by send_naks().

nak_requests_

NakRequestMap OpenDDS::DCPS::ReliableSession::nak_requests_

private

Definition at line 71 of file ReliableSession.h.

Referenced by expire_naks(), and send_naks().

nak_sequence_

DisjointSequence OpenDDS::DCPS::ReliableSession::nak_sequence_

private

Definition at line 68 of file ReliableSession.h.

Referenced by check_header(), deliver_held_data(), expire_naks(), nakack_received(), ready_to_deliver(), record_header_received(), send_naks(), and syn_hook().

nak_watchdog_

RcHandle<Sporadic> OpenDDS::DCPS::ReliableSession::nak_watchdog_

private

Definition at line 64 of file ReliableSession.h.

Referenced by process_naks(), start(), stop(), and ~ReliableSession().

---

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

• ReliableSession.h
• ReliableSession.cpp