OpenDDS::DCPS::TransportReassembly Class Reference

#include <TransportReassembly.h>

Inheritance diagram for OpenDDS::DCPS::TransportReassembly:

Collaboration diagram for OpenDDS::DCPS::TransportReassembly:

Classes
struct	FragInfo
struct	FragSample

Public Member Functions
	TransportReassembly (const TimeDuration &timeout=TimeDuration(300))
bool	reassemble (const SequenceNumber &transportSeq, bool firstFrag, ReceivedDataSample &data, ACE_UINT32 total_frags=0)
bool	reassemble (const FragmentRange &fragRange, ReceivedDataSample &data, ACE_UINT32 total_frags=0)
void	data_unavailable (const FragmentRange &transportSeqDropped)
void	data_unavailable (const SequenceNumber &dataSampleSeq, const GUID_t &pub_id)
void	clear_completed (const GUID_t &pub_id)
bool	has_frags (const SequenceNumber &seq, const GUID_t &pub_id) const
bool	has_frags (const SequenceNumber &seq, const GUID_t &pub_id, ACE_UINT32 &total_frags) const
CORBA::ULong	get_gaps (const SequenceNumber &msg_seq, const GUID_t &pub_id, CORBA::Long bitmap[], CORBA::ULong length, CORBA::ULong &numBits) const
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 std::pair< MonotonicTimePoint, FragKey >	ElementType

Private Member Functions
bool	reassemble_i (const FragmentRange &fragRange, bool firstFrag, ReceivedDataSample &data, ACE_UINT32 total_frags)
typedef	OPENDDS_MAP (FragKey, FragInfo) FragInfoMap
typedef	OPENDDS_LIST (ElementType) ExpirationQueue
typedef	OPENDDS_MAP_CMP (GUID_t, DisjointSequence, GUID_tKeyLessThan) CompletedMap
void	check_expirations (const MonotonicTimePoint &now)

Private Attributes
ACE_Thread_Mutex	mutex_
FragInfoMap	fragments_
ExpirationQueue	expiration_queue_
CompletedMap	completed_
TimeDuration	timeout_

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

Detailed Description

Definition at line 65 of file TransportReassembly.h.

Member Typedef Documentation

ElementType

typedef std::pair<MonotonicTimePoint, FragKey> OpenDDS::DCPS::TransportReassembly::ElementType

private

Definition at line 157 of file TransportReassembly.h.

Constructor & Destructor Documentation

TransportReassembly()

OpenDDS::DCPS::TransportReassembly::TransportReassembly ( const TimeDuration &  timeout = TimeDuration(300) )

explicit

Definition at line 36 of file TransportReassembly.cpp.

  : timeout_(timeout)
{
}

Member Function Documentation

check_expirations()

void OpenDDS::DCPS::TransportReassembly::check_expirations ( const MonotonicTimePoint &  now )

private

Definition at line 291 of file TransportReassembly.cpp.

References ACE_DEBUG, expiration_queue_, fragments_, LM_DEBUG, OpenDDS::DCPS::TransportDebug::log_fragment_storage, OpenDDS::DCPS::transport_debug, and OpenDDS::DCPS::Transport_debug_level.

Referenced by reassemble_i().

{
  while (!expiration_queue_.empty() && expiration_queue_.front().first <= now) {
    FragInfoMap::iterator iter = fragments_.find(expiration_queue_.front().second);
    if (iter != fragments_.end()) {
      // FragInfo::expiration_ may have changed after insertion into expiration_queue_
      if (iter->second.expiration_ <= now) {
        fragments_.erase(iter);
        if (Transport_debug_level > 5 || transport_debug.log_fragment_storage) {
          ACE_DEBUG((LM_DEBUG, "(%P|%t) TransportReassembly::check_expirations: "
                     "purge expired leaving %B fragments\n", fragments_.size()));
        }
      } else {
        expiration_queue_.push_back(std::make_pair(iter->second.expiration_, iter->first));
      }
    }
    expiration_queue_.pop_front();
  }
}

clear_completed()

void OpenDDS::DCPS::TransportReassembly::clear_completed

(

const GUID_t &

pub_id

)

Clears out "completed" sequence numbers in order to allow resends for new associations to the same (given) publication id

Definition at line 56 of file TransportReassembly.cpp.

References completed_, and mutex_.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::clear_completed_fragments().

{
  ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
  completed_.erase(pub_id);
}

data_unavailable() [1/2]

void OpenDDS::DCPS::TransportReassembly::data_unavailable

(

const FragmentRange &

transportSeqDropped

)

Called by TransportReceiveStrategy to indicate that we can stop tracking partially-reassembled messages when we know the remaining fragments are not expected to arrive.

Definition at line 228 of file TransportReassembly.cpp.

References OpenDDS::DCPS::FragKey::data_sample_seq_, OpenDDS::DCPS::TransportReassembly::FragSample::frag_range_, fragments_, OpenDDS::DCPS::TransportReassembly::FragInfo::have_first_, OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::TransportReassembly::FragInfo::insert(), LM_DEBUG, OpenDDS::DCPS::DataSampleHeader::more_fragments_, mutex_, OPENDDS_LIST(), OpenDDS::DCPS::TransportReassembly::FragInfo::sample_list_, OpenDDS::DCPS::DataSampleHeader::sequence_, and VDBG.

Referenced by OpenDDS::DCPS::BestEffortSession::check_header(), OpenDDS::DCPS::ReliableSession::expire_naks(), OpenDDS::DCPS::ReliableSession::nakack_received(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::remove_fragments().

{
  ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
  VDBG((LM_DEBUG, "(%P|%t) TransportReassembly::data_unavailable(): "
    "dropped %q-%q\n", dropped.first, dropped.second));
  typedef OPENDDS_LIST(FragSample)::iterator list_iterator;

  for (FragInfoMap::iterator iter = fragments_.begin(); iter != fragments_.end();
       ++iter) {
    const FragKey& key = iter->first;
    FragInfo& finfo = iter->second;
    FragInfo::FragSampleList& flist = finfo.sample_list_;

    ReceivedDataSample dummy;
    dummy.header_.sequence_ = key.data_sample_seq_;

    // check if we should expand the front element (only if !have_first)
    const FragmentNumber prev = flist.front().frag_range_.first - 1;
    if (dropped.second == prev && !finfo.have_first_) {
      finfo.have_first_ = true;
      dummy.header_.more_fragments_ = true;
      finfo.insert(dropped, dummy);
      continue;
    }

    // find a gap between list elements where "dropped" fits
    for (list_iterator it = flist.begin(); it != flist.end(); ++it) {
      list_iterator it_next = it;
      ++it_next;
      if (it_next == flist.end()) {
        break;
      }
      FragSample& fr1 = *it;
      FragSample& fr2 = *it_next;
      if (dropped.first > fr1.frag_range_.second
          && dropped.second < fr2.frag_range_.first) {
        dummy.header_.more_fragments_ = true;
        finfo.insert(dropped, dummy);
        break;
      }
    }

    // check if we should expand the last element
    const FragmentNumber next = flist.back().frag_range_.second + 1;
    if (dropped.first == next) {
      flist.back().rec_ds_.header_.more_fragments_ = true;
      finfo.insert(dropped, dummy);
    }
  }
}

data_unavailable() [2/2]

void OpenDDS::DCPS::TransportReassembly::data_unavailable	(	const SequenceNumber &	dataSampleSeq,
		const GUID_t &	pub_id
	)

Definition at line 280 of file TransportReassembly.cpp.

References ACE_DEBUG, fragments_, LM_DEBUG, OpenDDS::DCPS::TransportDebug::log_fragment_storage, mutex_, OpenDDS::DCPS::transport_debug, and OpenDDS::DCPS::Transport_debug_level.

{
  ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
  if (fragments_.erase(FragKey(pub_id, dataSampleSeq)) &&
      (Transport_debug_level > 5 || transport_debug.log_fragment_storage)) {
      ACE_DEBUG((LM_DEBUG, "(%P|%t) TransportReassembly::data_unavailable: "
                  "removed leaving %B fragments\n", fragments_.size()));
  }
}

get_gaps()

CORBA::ULong OpenDDS::DCPS::TransportReassembly::get_gaps	(	const SequenceNumber &	msg_seq,
		const GUID_t &	pub_id,
		CORBA::Long	bitmap[],
		CORBA::ULong	length,
		CORBA::ULong &	numBits
	)		const

Populates bitmap for missing fragment sequence numbers and set numBits for the given message sequence and publisher ID.

Returns

the base fragment sequence number for bit zero in the bitmap

Definition at line 63 of file TransportReassembly.cpp.

References OpenDDS::DCPS::DisjointSequence::fill_bitmap_range(), fragments_, mutex_, and OPENDDS_LIST().

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::has_fragments().

{
  ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
  // length is number of (allocated) words in bitmap, max of 8
  // numBits is number of valid bits in the bitmap, <= length * 32, to account for partial words
  if (length == 0) {
    return 0;
  }

  const FragInfoMap::const_iterator iter = fragments_.find(FragKey(pub_id, seq));
  if (iter == fragments_.end()) {
    // Nothing missing
    return 0;
  }

  // RTPS's FragmentNumbers are 32-bit values, so we'll only be using the
  // low 32 bits of the 64-bit generalized sequence numbers in
  // FragSample::frag_range_.

  const OPENDDS_LIST(FragSample)& flist = iter->second.sample_list_;
  const FragmentNumber first = flist.front().frag_range_.first;
  const CORBA::ULong base = static_cast<CORBA::ULong>((first == 1)
    ? flist.front().frag_range_.second + 1
    : 1);

  if (first != 1) {
    // Represent the "gap" before the first list element.
    // base == 1 and the first 2 args to fill_bitmap_range() are deltas of base
    ACE_CDR::ULong bits_added = 0;
    DisjointSequence::fill_bitmap_range(0, static_cast<CORBA::ULong>(first - 2),
                                        bitmap, length, numBits, bits_added);
  } else if (flist.size() == 1) {
    // No gaps, but we know there are (at least 1) more_fragments
    if (iter->second.total_frags_ == 0) {
      ACE_CDR::ULong bits_added = 0;
      DisjointSequence::fill_bitmap_range(0, 0, bitmap, length, numBits, bits_added);
    } else {
      const size_t rlimit = static_cast<size_t>(flist.back().frag_range_.second - 1);
      const CORBA::ULong ulimit = static_cast<CORBA::ULong>(iter->second.total_frags_ - (base < rlimit ? rlimit : base));
      ACE_CDR::ULong bits_added = 0;
      DisjointSequence::fill_bitmap_range(0,
                                          ulimit,
                                          bitmap, length, numBits, bits_added);
    }
    // NOTE: this could send a nack for fragments that are in flight
    // need to defer setting bitmap till heartbeat extending logic
    // in RtpsUdpDataLink::generate_nack_frags
    return base;
  }

  typedef OPENDDS_LIST(FragSample)::const_iterator list_iterator;
  for (list_iterator it = flist.begin(); it != flist.end(); ++it) {
    const list_iterator it_next = ++list_iterator(it);
    if (it_next == flist.end()) {
      break;
    }
    const CORBA::ULong low = static_cast<CORBA::ULong>(it->frag_range_.second + 1 - base),
                       high = static_cast<CORBA::ULong>(it_next->frag_range_.first - 1 - base);
    ACE_CDR::ULong bits_added = 0;
    DisjointSequence::fill_bitmap_range(low, high, bitmap, length, numBits, bits_added);
  }

  return base;
}

has_frags() [1/2]

bool OpenDDS::DCPS::TransportReassembly::has_frags ( const SequenceNumber &  seq, const GUID_t &  pub_id  ) const

inline

Returns true if this object is storing fragments for the given DataSampleHeader sequence number from the given publication.

Definition at line 91 of file TransportReassembly.h.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::has_fragments(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::remove_frags_from_bitmap().

  {
    ACE_UINT32 total_frags;
    return has_frags(seq, pub_id, total_frags);
  }

has_frags() [2/2]

bool OpenDDS::DCPS::TransportReassembly::has_frags	(	const SequenceNumber &	seq,
		const GUID_t &	pub_id,
		ACE_UINT32 &	total_frags
	)		const

Returns true if this object is storing fragments for the given DataSampleHeader sequence number from the given publication.

Definition at line 42 of file TransportReassembly.cpp.

References fragments_, and mutex_.

{
  ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
  const FragInfoMap::const_iterator iter = fragments_.find(FragKey(pub_id, seq));
  if (iter != fragments_.end()) {
    total_frags = iter->second.total_frags_;
    return true;
  }
  return false;
}

OPENDDS_LIST()

typedef OpenDDS::DCPS::TransportReassembly::OPENDDS_LIST ( ElementType  )

private

Referenced by data_unavailable(), and get_gaps().

OPENDDS_MAP()

typedef OpenDDS::DCPS::TransportReassembly::OPENDDS_MAP ( FragKey  , FragInfo    )

private

OPENDDS_MAP_CMP()

typedef OpenDDS::DCPS::TransportReassembly::OPENDDS_MAP_CMP ( GUID_t  , DisjointSequence  , GUID_tKeyLessThan    )

private

reassemble() [1/2]

bool OpenDDS::DCPS::TransportReassembly::reassemble	(	const SequenceNumber &	transportSeq,
		bool	firstFrag,
		ReceivedDataSample &	data,
		ACE_UINT32	total_frags = 0
	)

Called by TransportReceiveStrategy if the fragmentation header flag is set. Returns true/false to indicate if data should be delivered to the datalink. The 'data' argument may be modified by this method.

Definition at line 140 of file TransportReassembly.cpp.

References OpenDDS::DCPS::SequenceNumber::getValue(), mutex_, and reassemble_i().

Referenced by OpenDDS::DCPS::MulticastSession::reassemble(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::reassemble_i().

{
  ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
  return reassemble_i(FragmentRange(transportSeq.getValue(), transportSeq.getValue()),
                      firstFrag, data, total_frags);
}

reassemble() [2/2]

bool OpenDDS::DCPS::TransportReassembly::reassemble	(	const FragmentRange &	fragRange,
		ReceivedDataSample &	data,
		ACE_UINT32	total_frags = 0
	)

Definition at line 131 of file TransportReassembly.cpp.

References mutex_, and reassemble_i().

{
  ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
  return reassemble_i(fragRange, fragRange.first == 1, data, total_frags);
}

reassemble_i()

bool OpenDDS::DCPS::TransportReassembly::reassemble_i ( const FragmentRange &  fragRange, bool  firstFrag, ReceivedDataSample &  data, ACE_UINT32  total_frags  )

private

Definition at line 151 of file TransportReassembly.cpp.

References ACE_DEBUG, OpenDDS::DCPS::LogGuid::c_str(), check_expirations(), OpenDDS::DCPS::ReceivedDataSample::clear(), completed_, OpenDDS::DCPS::FragKey::data_sample_seq_, expiration_queue_, fragments_, OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::ReceivedDataSample::has_data(), OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::TransportReassembly::FragInfo::insert(), LM_DEBUG, OpenDDS::DCPS::TransportDebug::log_fragment_storage, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OpenDDS::DCPS::FragKey::publication_, OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::DataSampleHeader::sequence_, OpenDDS::DCPS::swap(), timeout_, OpenDDS::DCPS::transport_debug, OpenDDS::DCPS::Transport_debug_level, and VDBG.

Referenced by reassemble().

{
  if (Transport_debug_level > 5) {
    LogGuid logger(data.header_.publication_id_);
    ACE_DEBUG((LM_DEBUG, "(%P|%t) TransportReassembly::reassemble_i: "
      "tseq %q-%q first %d dseq %q pub %C\n", fragRange.first,
      fragRange.second, firstFrag ? 1 : 0,
      data.header_.sequence_.getValue(), logger.c_str()));
  }

  const MonotonicTimePoint now = MonotonicTimePoint::now();
  check_expirations(now);

  const FragKey key(data.header_.publication_id_, data.header_.sequence_);
  FragInfoMap::iterator iter = fragments_.find(key);
  const MonotonicTimePoint expiration = now + timeout_;

  if (iter == fragments_.end()) {
    FragInfo& finfo = fragments_[key];
    finfo = FragInfo(firstFrag, FragInfo::FragSampleList(), total_frags, expiration);
    finfo.insert(fragRange, data);
    expiration_queue_.push_back(std::make_pair(expiration, key));
    data.clear();
    // since this is the first fragment we've seen, it can't possibly be done
    if (Transport_debug_level > 5 || transport_debug.log_fragment_storage) {
      ACE_DEBUG((LM_DEBUG, "(%P|%t) TransportReassembly::reassemble_i: "
                 "stored first frag, returning false (incomplete) with %B fragments\n",
                 fragments_.size()));
    }
    return false;
  } else {
    const CompletedMap::const_iterator citer = completed_.find(key.publication_);
    if (citer != completed_.end() && citer->second.contains(key.data_sample_seq_)) {
      // already completed, not storing or delivering this message
      return false;
    }
    if (firstFrag) {
      iter->second.have_first_ = true;
    }
    if (iter->second.total_frags_ < total_frags) {
      iter->second.total_frags_ = total_frags;
    }
    iter->second.expiration_ = expiration;
  }

  if (!iter->second.insert(fragRange, data)) {
    // error condition, already logged by insert()
    return false;
  }

  // We can deliver data if all three of these conditions are met:
  // 1. we've seen the "first fragment" flag  [first frag is here]
  // 2. all fragments have been coalesced     [no gaps in the seq numbers]
  // 3. the "more fragments" flag is not set  [last frag is here]
  if (iter->second.have_first_
      && iter->second.sample_list_.size() == 1
      && !iter->second.sample_list_.front().rec_ds_.header_.more_fragments_) {
    std::swap(data, iter->second.sample_list_.front().rec_ds_);
    fragments_.erase(iter);
    completed_[key.publication_].insert(key.data_sample_seq_);
    if (Transport_debug_level > 5 || transport_debug.log_fragment_storage) {
      ACE_DEBUG((LM_DEBUG, "(%P|%t) TransportReassembly::reassemble_i: "
                 "removed frag, returning %C with %B fragments\n",
                 data.has_data() ? "true (complete)" : "false (incomplete)", fragments_.size()));
    }
    return data.has_data(); // could be false if we had data_unavailable()
  }

  VDBG((LM_DEBUG, "(%P|%t) TransportReassembly::reassemble_i: "
    "returning false (incomplete)\n"));
  return false;
}

Member Data Documentation

completed_

CompletedMap OpenDDS::DCPS::TransportReassembly::completed_

private

Definition at line 162 of file TransportReassembly.h.

Referenced by clear_completed(), and reassemble_i().

expiration_queue_

ExpirationQueue OpenDDS::DCPS::TransportReassembly::expiration_queue_

private

Definition at line 159 of file TransportReassembly.h.

Referenced by check_expirations(), and reassemble_i().

fragments_

FragInfoMap OpenDDS::DCPS::TransportReassembly::fragments_

private

Definition at line 155 of file TransportReassembly.h.

Referenced by check_expirations(), data_unavailable(), get_gaps(), has_frags(), and reassemble_i().

mutex_

ACE_Thread_Mutex OpenDDS::DCPS::TransportReassembly::mutex_

mutableprivate

Definition at line 148 of file TransportReassembly.h.

Referenced by clear_completed(), data_unavailable(), get_gaps(), has_frags(), and reassemble().

timeout_

TimeDuration OpenDDS::DCPS::TransportReassembly::timeout_

private

Definition at line 164 of file TransportReassembly.h.

Referenced by reassemble_i().

---

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

• TransportReassembly.h
• TransportReassembly.cpp