OpenDDS::DCPS::TransportReassembly Class Reference

#include <TransportReassembly.h>

List of all members.

Public Member Functions
bool	reassemble (const SequenceNumber &transportSeq, bool firstFrag, ReceivedDataSample &data)
bool	reassemble (const SequenceRange &seqRange, ReceivedDataSample &data)
void	data_unavailable (const SequenceRange &transportSeqDropped)
void	data_unavailable (const SequenceNumber &dataSampleSeq, const RepoId &pub_id)
bool	has_frags (const SequenceNumber &seq, const RepoId &pub_id) const
CORBA::ULong	get_gaps (const SequenceNumber &msg_seq, const RepoId &pub_id, CORBA::Long bitmap[], CORBA::ULong length, CORBA::ULong &numBits) const
Private Member Functions
bool	reassemble_i (const SequenceRange &seqRange, bool firstFrag, ReceivedDataSample &data)
typedef	OPENDDS_MAP (FragKey, OPENDDS_LIST(FragRange)) FragMap
	OPENDDS_SET (FragKey) have_first_
Static Private Member Functions
static bool	insert (OPENDDS_LIST(FragRange)&flist, const SequenceRange &seqRange, ReceivedDataSample &data)
Private Attributes
FragMap	fragments_
Classes
struct	FragKey
struct	FragRange

---

Detailed Description

Definition at line 19 of file TransportReassembly.h.

---

Member Function Documentation

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

Definition at line 337 of file TransportReassembly.cpp.

References fragments_.

{
  const FragKey key(pub_id, dataSampleSeq);
  fragments_.erase(key);
  have_first_.erase(key);
}

void OpenDDS::DCPS::TransportReassembly::data_unavailable

(

const SequenceRange &

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 285 of file TransportReassembly.cpp.

References dummy, fragments_, insert(), OpenDDS::DCPS::OPENDDS_LIST(), OpenDDS::DCPS::TransportReassembly::FragRange::transport_seq_, and VDBG.

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

{
  VDBG((LM_DEBUG, "(%P|%t) DBG:   TransportReassembly::data_unavailable() "
    "dropped %q-%q\n", dropped.first.getValue(), dropped.second.getValue()));
  typedef OPENDDS_LIST(FragRange)::iterator list_iterator;

  for (FragMap::iterator iter = fragments_.begin(); iter != fragments_.end();
       ++iter) {
    const FragKey& key = iter->first;
    OPENDDS_LIST(FragRange)& flist = iter->second;

    ReceivedDataSample dummy(0);
    dummy.header_.sequence_ = key.data_sample_seq_;

    // check if we should expand the front element (only if !have_first)
    const SequenceNumber::Value prev =
      flist.front().transport_seq_.first.getValue() - 1;
    if (dropped.second.getValue() == prev && !have_first_.count(key)) {
      have_first_.insert(key);
      dummy.header_.more_fragments_ = true;
      insert(flist, 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;
      }
      FragRange& fr1 = *it;
      FragRange& fr2 = *it_next;
      if (dropped.first > fr1.transport_seq_.second
          && dropped.second < fr2.transport_seq_.first) {
        dummy.header_.more_fragments_ = true;
        insert(flist, dropped, dummy);
        break;
      }
    }

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

CORBA::ULong OpenDDS::DCPS::TransportReassembly::get_gaps	(	const SequenceNumber &	msg_seq,
		const RepoId &	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 164 of file TransportReassembly.cpp.

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

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

{
  // 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
  const FragMap::const_iterator iter = fragments_.find(FragKey(pub_id, seq));
  if (iter == fragments_.end() || length == 0) {
    // 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
  // FragRange::transport_seq_.

  const OPENDDS_LIST(FragRange)& flist = iter->second;
  const SequenceNumber& first = flist.front().transport_seq_.first;
  const CORBA::ULong base = (first == 1)
    ? flist.front().transport_seq_.second.getLow() + 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
    DisjointSequence::fill_bitmap_range(0, first.getLow() - 2,
                                        bitmap, length, numBits);
  } else if (flist.size() == 1) {
    // No gaps, but we know there is (at least 1) more_framents
    DisjointSequence::fill_bitmap_range(0, 0,
                                        bitmap, length, numBits);
    // 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(FragRange)::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 = it->transport_seq_.second.getLow() + 1 - base,
                       high = it_next->transport_seq_.first.getLow() - 1 - base;
    DisjointSequence::fill_bitmap_range(low, high, bitmap, length, numBits);
  }

  return base;
}

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

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

Definition at line 157 of file TransportReassembly.cpp.

References fragments_.

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

{
  return fragments_.count(FragKey(pub_id, seq));
}

bool OpenDDS::DCPS::TransportReassembly::insert	(	OPENDDS_LIST(FragRange)&	flist,
		const SequenceRange &	seqRange,
		ReceivedDataSample &	data
	)			[static, private]

Definition at line 44 of file TransportReassembly.cpp.

References OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::DataSampleHeader::join(), OpenDDS::DCPS::join_err(), OpenDDS::DCPS::OPENDDS_LIST(), OpenDDS::DCPS::ReceivedDataSample::sample_, and VDBG.

Referenced by data_unavailable(), and reassemble_i().

{
  const SequenceNumber::Value prev = seqRange.first.getValue() - 1,
    next = seqRange.second.getValue() + 1;

  for (OPENDDS_LIST(FragRange)::iterator it = flist.begin(); it != flist.end(); ++it) {
    FragRange& fr = *it;
    if (next < fr.transport_seq_.first.getValue()) {
      // insert before 'it'
      flist.insert(it, FragRange(seqRange, data));
      VDBG((LM_DEBUG, "(%P|%t) DBG:   TransportReassembly::insert() "
        "inserted on left\n"));
      return true;

    } else if (next == fr.transport_seq_.first.getValue()) {
      // combine on left of fr
      DataSampleHeader joined;
      if (!DataSampleHeader::join(data.header_, fr.rec_ds_.header_, joined)) {
        join_err("left");
        return false;
      }
      fr.rec_ds_.header_ = joined;
      if (fr.rec_ds_.sample_ && !data.sample_) {
        ACE_Message_Block::release(fr.rec_ds_.sample_);
        fr.rec_ds_.sample_ = 0;
      } else if (!fr.rec_ds_.sample_) {
        ACE_Message_Block::release(data.sample_);
      } else {
        ACE_Message_Block* last;
        for (last = data.sample_; last->cont(); last = last->cont()) ;
        last->cont(fr.rec_ds_.sample_);
        fr.rec_ds_.sample_ = data.sample_;
      }
      data.sample_ = 0;
      fr.transport_seq_.first = seqRange.first;
      VDBG((LM_DEBUG, "(%P|%t) DBG:   TransportReassembly::insert() "
        "combined on left\n"));
      return true;

    } else if (prev == fr.transport_seq_.second.getValue()) {
      // combine on right of fr
      if (!fr.rec_ds_.sample_) {
        fr.rec_ds_.header_.more_fragments_ = true;
      }
      DataSampleHeader joined;
      if (!DataSampleHeader::join(fr.rec_ds_.header_, data.header_, joined)) {
        join_err("right");
        return false;
      }
      fr.rec_ds_.header_ = joined;
      if (fr.rec_ds_.sample_ && !data.sample_) {
        ACE_Message_Block::release(fr.rec_ds_.sample_);
        fr.rec_ds_.sample_ = 0;
      } else if (!fr.rec_ds_.sample_) {
        ACE_Message_Block::release(data.sample_);
      } else {
        ACE_Message_Block* last;
        for (last = fr.rec_ds_.sample_; last->cont(); last = last->cont()) ;
        last->cont(data.sample_);
      }
      data.sample_ = 0;
      fr.transport_seq_.second = seqRange.second;
      VDBG((LM_DEBUG, "(%P|%t) DBG:   TransportReassembly::insert() "
        "combined on right\n"));

      // check if the next FragRange in the list needs to be combined
      if (++it != flist.end()) {
        if (next == it->transport_seq_.first.getValue()) {
          if (!fr.rec_ds_.sample_) {
            fr.rec_ds_.header_.more_fragments_ = true;
          }
          if (!DataSampleHeader::join(fr.rec_ds_.header_, it->rec_ds_.header_,
                                      joined)) {
            join_err("combined next");
            return false;
          }
          fr.rec_ds_.header_ = joined;
          if (!it->rec_ds_.sample_) {
            ACE_Message_Block::release(fr.rec_ds_.sample_);
            fr.rec_ds_.sample_ = 0;
          } else {
            if (!fr.rec_ds_.sample_) {
              ACE_ERROR((LM_ERROR,
                         ACE_TEXT("(%P|%t) ERROR: ")
                         ACE_TEXT("OpenDDS::DCPS::TransportReassembly::insert, ")
                         ACE_TEXT("Cannot dereference null pointer fr.rec_ds_.sample_\n")));
              return false;
            }
            ACE_Message_Block* last;
            for (last = fr.rec_ds_.sample_; last->cont(); last = last->cont()) ;
            last->cont(it->rec_ds_.sample_);
            it->rec_ds_.sample_ = 0;
          }
          fr.transport_seq_.second = it->transport_seq_.second;
          flist.erase(it);
          VDBG((LM_DEBUG, "(%P|%t) DBG:   TransportReassembly::insert() "
            "coalesced on right\n"));
        }
      }
      return true;
    }
  }

  // add to end of list
  flist.push_back(FragRange(seqRange, data));
  VDBG((LM_DEBUG, "(%P|%t) DBG:   TransportReassembly::insert() "
    "inserted at end of list\n"));
  return true;
}

typedef OpenDDS::DCPS::TransportReassembly::OPENDDS_MAP	(	FragKey	,
		OPENDDS_LIST(FragRange)
	)			[private]

OpenDDS::DCPS::TransportReassembly::OPENDDS_SET

(

FragKey

)

[private]

bool OpenDDS::DCPS::TransportReassembly::reassemble	(	const SequenceRange &	seqRange,
		ReceivedDataSample &	data
	)

Definition at line 216 of file TransportReassembly.cpp.

References reassemble_i().

{
  return reassemble_i(seqRange, seqRange.first == 1, data);
}

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

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 223 of file TransportReassembly.cpp.

References reassemble_i().

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

{
  return reassemble_i(SequenceRange(transportSeq, transportSeq),
                      firstFrag, data);
}

bool OpenDDS::DCPS::TransportReassembly::reassemble_i	(	const SequenceRange &	seqRange,
		bool	firstFrag,
		ReceivedDataSample &	data
	)			[private]

Definition at line 232 of file TransportReassembly.cpp.

References fragments_, OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::ReceivedDataSample::header_, insert(), OPENDDS_STRING, OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::push_back(), OpenDDS::DCPS::ReceivedDataSample::sample_, OpenDDS::DCPS::DataSampleHeader::sequence_, OpenDDS::DCPS::swap(), OpenDDS::DCPS::Transport_debug_level, and VDBG.

Referenced by reassemble().

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

  const FragKey key(data.header_.publication_id_, data.header_.sequence_);

  if (firstFrag) {
    have_first_.insert(key);
  }

  FragMap::iterator iter = fragments_.find(key);
  if (iter == fragments_.end()) {
    fragments_[key].push_back(FragRange(seqRange, data));
    // since this is the first fragment we've seen, it can't possibly be done
    VDBG((LM_DEBUG, "(%P|%t) DBG:   TransportReassembly::reassemble() "
      "stored first frag, returning false (incomplete)\n"));
    return false;
  }

  if (!insert(iter->second, seqRange, 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 (have_first_.count(key)
      && iter->second.size() == 1
      && !iter->second.front().rec_ds_.header_.more_fragments_) {
    swap(data, iter->second.front().rec_ds_);
    fragments_.erase(iter);
    have_first_.erase(key);
    VDBG((LM_DEBUG, "(%P|%t) DBG:   TransportReassembly::reassemble() "
      "removed frag, returning %C\n", data.sample_ ? "true" : "false"));
    return data.sample_; // could be false if we had data_unavailable()
  }

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

---

Member Data Documentation

FragMap OpenDDS::DCPS::TransportReassembly::fragments_ [private]

Definition at line 90 of file TransportReassembly.h.

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

---

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

• TransportReassembly.h
• TransportReassembly.cpp

---