OpenDDS::DCPS::TransportQueueElement

OpenDDS::DCPS::TransportQueueElement Class Reference

Base wrapper class around a data/control sample to be sent. More...

#include <TransportQueueElement.h>

Inheritance diagram for OpenDDS::DCPS::TransportQueueElement:

Inheritance graph
[legend]
Collaboration diagram for OpenDDS::DCPS::TransportQueueElement:

Collaboration graph
[legend]
List of all members.

Public Member Functions

virtual ~TransportQueueElement ()
bool data_dropped (bool dropped_by_transport=false)
bool data_delivered ()
virtual bool requires_exclusive_packet () const
 Does the sample require an exclusive transport packet?
virtual RepoId publication_id () const =0
 Accessor for the publication id that sent the sample.
virtual RepoId subscription_id () const
 Accessor for the subscription id, if sent the sample is sent to 1 sub.
virtual SequenceNumber sequence () const
virtual const ACE_Message_Block * msg () const =0
 The marshalled sample (sample header + sample data).
virtual const ACE_Message_Block * msg_payload () const =0
 The marshalled payload only (sample data).
virtual bool is_control (RepoId pub_id) const
 Is the element a "control" sample from the specified pub_id?
bool released () const
 Is the listener get called ?
void released (bool flag)
virtual bool owned_by_transport ()=0
 Is the sample created by the transport?
virtual ElementPair fragment (size_t size)
virtual bool is_fragment () const
 Is this QueueElement the result of fragmentation?

Static Public Member Functions

static ACE_Message_Block * clone_mb (const ACE_Message_Block *msg, MessageBlockAllocator *mb_allocator, DataBlockAllocator *db_allocator)

Protected Member Functions

 TransportQueueElement (unsigned long initial_count)
virtual void release_element (bool dropped_by_transport)=0
 Invoked when the counter reaches 0.
bool was_dropped () const

Private Member Functions

bool decision_made (bool dropped_by_transport)
 Common logic for data_dropped() and data_delivered().

Private Attributes

ACE_Atomic_Op< ACE_Thread_Mutex,
unsigned long > 		sub_loan_count_
 Counts the number of outstanding sub-loans.
bool dropped_
 Flag flipped to true if any DataLink dropped the sample.
bool released_
 If the callback to DW is made.

Friends

class TransportCustomizedElement

Classes

class  MatchCriteria
class  MatchOnDataPayload
class  MatchOnPubId

Detailed Description

Base wrapper class around a data/control sample to be sent.

This class serves as the base class for different types of samples that can be sent. For example, there are data samples and control samples. A subclass of TransportQueueElement exists for each of these types of samples.

This class maintains a counter that, when decremented to 0, will trigger some logic (defined in the subclass) that will "return the loan" of the sample. The sample is "loaned" to the transport via a send() or send_control() call on the TransportClient. This wrapper object will "return the loan" when all DataLinks have "returned" their sub-loans.

Definition at line 48 of file TransportQueueElement.h.

Constructor & Destructor Documentation

OpenDDS::DCPS::TransportQueueElement::~TransportQueueElement (  )  [virtual]

Definition at line 21 of file TransportQueueElement.cpp.

References DBG_ENTRY_LVL.

00022 {
00023   DBG_ENTRY_LVL("TransportQueueElement", "~TransportQueueElement", 6);
00024 }

ACE_INLINE OpenDDS::DCPS::TransportQueueElement::TransportQueueElement ( unsigned long  initial_count  )  [explicit, protected]

Ctor. The initial_count is the number of DataLinks to which this TransportQueueElement will be sent.

Definition at line 15 of file TransportQueueElement.inl.

References DBG_ENTRY_LVL.

00016   : sub_loan_count_(initial_count),
00017     dropped_(false),
00018     released_(false)
00019 {
00020   DBG_ENTRY_LVL("TransportQueueElement", "TransportQueueElement", 6);
00021 }

Member Function Documentation

ACE_Message_Block * OpenDDS::DCPS::TransportQueueElement::clone_mb ( const ACE_Message_Block *  msg,
MessageBlockAllocator *  mb_allocator,
DataBlockAllocator *  db_allocator 
) [static]

Clone method with provided message block allocator and data block allocators.

Definition at line 59 of file TransportQueueElement.cpp.

References OpenDDS::DCPS::Cached_Allocator_With_Overflow< T, ACE_LOCK >::malloc().

Referenced by OpenDDS::DCPS::SingleSendBuffer::insert_buffer(), and OpenDDS::DCPS::TransportRetainedElement::TransportRetainedElement().

00062 {
00063   ACE_Message_Block* cur_block = const_cast<ACE_Message_Block*>(msg);
00064   ACE_Message_Block* head_copy = 0;
00065   ACE_Message_Block* cur_copy  = 0;
00066   ACE_Message_Block* prev_copy = 0;
00067   // deep copy sample data
00068   while (cur_block != 0) {
00069     ACE_NEW_MALLOC_RETURN(cur_copy,
00070                           static_cast<ACE_Message_Block*>(
00071                           mb_allocator->malloc(sizeof(ACE_Message_Block))),
00072                           ACE_Message_Block(cur_block->capacity(),
00073                                             ACE_Message_Block::MB_DATA,
00074                                             0, //cont
00075                                             0, //data
00076                                             0, //alloc_strategy
00077                                             0, //locking_strategy
00078                                             ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
00079                                             ACE_Time_Value::zero,
00080                                             ACE_Time_Value::max_time,
00081                                             db_allocator,
00082                                             mb_allocator),
00083                           0);
00084 
00085     cur_copy->copy(cur_block->base(), cur_block->size());
00086     cur_copy->rd_ptr(cur_copy->base() +
00087                      (cur_block->rd_ptr() - cur_block->base()));
00088     cur_copy->wr_ptr(cur_copy->base() +
00089                      (cur_block->wr_ptr() - cur_block->base()));
00090 
00091     if (head_copy == 0) {
00092       head_copy = cur_copy;
00093     } else {
00094       prev_copy->cont(cur_copy);
00095     }
00096 
00097     prev_copy = cur_copy;
00098 
00099     cur_block = cur_block->cont();
00100   }
00101 
00102   return head_copy;
00103 }

ACE_INLINE bool OpenDDS::DCPS::TransportQueueElement::data_delivered (  ) 

Invoked when the sample has been sent by a DataLink. The return value indicates if this element is released.

Reimplemented in OpenDDS::DCPS::TransportControlElement.

Definition at line 34 of file TransportQueueElement.inl.

References DBG_ENTRY_LVL, and decision_made().

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element(), OpenDDS::DCPS::RtpsUdpDataLink::process_acked_by_all_i(), and OpenDDS::DCPS::TransportSendStrategy::send_delayed_notifications().

00035 {
00036   DBG_ENTRY_LVL("TransportQueueElement", "data_delivered", 6);
00037   // Decision made depend on dropped_ flag. If any link drops
00038   // the sample even other links deliver successfully, the
00039   // data dropped by transport will called back to writer.
00040   return this->decision_made(this->dropped_);
00041 }

ACE_INLINE bool OpenDDS::DCPS::TransportQueueElement::data_dropped ( bool  dropped_by_transport = false  ) 

Invoked when the sample is dropped from a DataLink due to a remove_sample() call. The dropped_by_transport flag true indicates the data dropping is initiated by transport when the transport send strategy is in a MODE_TERMINATED. The dropped_by_transport flag false indicates the dropping is initiated by the remove_sample and data_dropped() is a result of remove_sample(). The return value indicates if this element is released.

Definition at line 25 of file TransportQueueElement.inl.

References DBG_ENTRY_LVL, decision_made(), and dropped_.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element(), OpenDDS::DCPS::TransportSendStrategy::send(), OpenDDS::DCPS::TransportSendStrategy::send_delayed_notifications(), OpenDDS::DCPS::PacketRemoveVisitor::visit_element_ref(), OpenDDS::DCPS::ThreadPerConRemoveVisitor::visit_element_remove(), OpenDDS::DCPS::RemoveAllVisitor::visit_element_remove(), and OpenDDS::DCPS::QueueRemoveVisitor::visit_element_remove().

00026 {
00027   DBG_ENTRY_LVL("TransportQueueElement", "data_dropped", 6);
00028   this->dropped_ = true;
00029   return this->decision_made(dropped_by_transport);
00030 }

ACE_INLINE bool OpenDDS::DCPS::TransportQueueElement::decision_made ( bool  dropped_by_transport  )  [private]

Common logic for data_dropped() and data_delivered().

Definition at line 45 of file TransportQueueElement.inl.

References DBG_ENTRY_LVL, release_element(), and sub_loan_count_.

Referenced by data_delivered(), data_dropped(), and OpenDDS::DCPS::TransportCustomizedElement::release_element().

00046 {
00047   DBG_ENTRY_LVL("TransportQueueElement", "decision_made", 6);
00048 
00049   const unsigned long new_count = --this->sub_loan_count_;
00050   if (new_count == 0) {
00051     // All interested subscriptions have been satisfied.
00052 
00053     // The queue elements are released to its cached allocator
00054     // in release_element() call.
00055     // It's not necessary to set the released_ flag to true
00056     // as this element will be released anyway and not be
00057     // accessible. Note it can not be set after release_element
00058     // call.
00059     // this->released_ = true;
00060     this->release_element(dropped_by_transport);
00061     return true;
00062   }
00063 
00064   // ciju: The sub_loan_count_ has been observed to drop below zero.
00065   // Since it isn't exactly a ref count and the object is created in
00066   // allocater memory (user space) we *probably* can disregard the
00067   // count for now. Ideally we would like to prevent the count from
00068   // falling below 0 and opening up this assert.
00069   // assert (new_count > 0);
00070   return false;
00071 }

ElementPair OpenDDS::DCPS::TransportQueueElement::fragment ( size_t  size  )  [virtual]

Create two TransportQueueElements representing the same data payload as the current TransportQueueElement, with the first one (including its DataSampleHeader) fitting in "size" bytes. This method leaves the current TransportQueueElement alone (but can't be made const because the newly-created elements will need to invoke non-const methods on it). Each element in the pair will contain its own serialized modified DataSampleHeader.

Reimplemented in OpenDDS::DCPS::RtpsCustomizedElement.

Definition at line 41 of file TransportQueueElement.cpp.

References OpenDDS::DCPS::TransportCustomizedElement::alloc(), msg(), publication_id(), OpenDDS::DCPS::TransportCustomizedElement::set_msg(), OpenDDS::DCPS::TransportCustomizedElement::set_publication_id(), and OpenDDS::DCPS::DataSampleHeader::split().

Referenced by OpenDDS::DCPS::TransportSendStrategy::send().

00042 {
00043   ACE_Message_Block* head;
00044   ACE_Message_Block* tail;
00045   DataSampleHeader::split(*msg(), size, head, tail);
00046 
00047   TransportCustomizedElement* frag = TransportCustomizedElement::alloc(0, true);
00048   frag->set_publication_id(publication_id());
00049   frag->set_msg(head);
00050 
00051   TransportCustomizedElement* rest =
00052     TransportCustomizedElement::alloc(this, true);
00053   rest->set_msg(tail);
00054 
00055   return ElementPair(frag, rest);
00056 }

bool OpenDDS::DCPS::TransportQueueElement::is_control ( RepoId  pub_id  )  const [virtual]

Is the element a "control" sample from the specified pub_id?

Reimplemented in OpenDDS::DCPS::TransportSendControlElement.

Definition at line 34 of file TransportQueueElement.cpp.

References DBG_ENTRY_LVL.

00035 {
00036   DBG_ENTRY_LVL("TransportQueueElement", "is_control", 6);
00037   return false;
00038 }

virtual bool OpenDDS::DCPS::TransportQueueElement::is_fragment (  )  const [inline, virtual]

Is this QueueElement the result of fragmentation?

Reimplemented in OpenDDS::DCPS::TransportCustomizedElement.

Definition at line 145 of file TransportQueueElement.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::MultiSendBuffer::insert().

00145 { return false; }

virtual const ACE_Message_Block* OpenDDS::DCPS::TransportQueueElement::msg (  )  const [pure virtual]

The marshalled sample (sample header + sample data).

Implemented in OpenDDS::DCPS::TransportControlElement, OpenDDS::DCPS::TransportCustomizedElement, OpenDDS::DCPS::TransportReplacedElement, OpenDDS::DCPS::TransportRetainedElement, OpenDDS::DCPS::TransportSendControlElement, and OpenDDS::DCPS::TransportSendElement.

Referenced by OpenDDS::DCPS::TransportSendStrategy::adjust_packet_after_send(), OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element(), OpenDDS::DCPS::RtpsUdpDataLink::durability_resend(), fragment(), OpenDDS::DCPS::TransportSendStrategy::send(), OpenDDS::DCPS::TransportReplacedElement::TransportReplacedElement(), OpenDDS::DCPS::CopyChainVisitor::visit_element(), OpenDDS::DCPS::BuildChainVisitor::visit_element(), OpenDDS::DCPS::PacketRemoveVisitor::visit_element_ref(), OpenDDS::DCPS::RemoveAllVisitor::visit_element_remove(), and OpenDDS::DCPS::QueueRemoveVisitor::visit_element_remove().

virtual const ACE_Message_Block* OpenDDS::DCPS::TransportQueueElement::msg_payload (  )  const [pure virtual]

The marshalled payload only (sample data).

Implemented in OpenDDS::DCPS::TransportControlElement, OpenDDS::DCPS::TransportCustomizedElement, OpenDDS::DCPS::TransportReplacedElement, OpenDDS::DCPS::TransportRetainedElement, OpenDDS::DCPS::TransportSendControlElement, OpenDDS::DCPS::TransportSendElement, and OpenDDS::DCPS::RtpsCustomizedElement.

Referenced by OpenDDS::DCPS::TransportQueueElement::MatchOnDataPayload::matches(), and OpenDDS::DCPS::TransportCustomizedElement::msg_payload().

virtual bool OpenDDS::DCPS::TransportQueueElement::owned_by_transport (  )  [pure virtual]

Is the sample created by the transport?

Implemented in OpenDDS::DCPS::TransportControlElement, OpenDDS::DCPS::TransportCustomizedElement, OpenDDS::DCPS::TransportReplacedElement, OpenDDS::DCPS::TransportRetainedElement, OpenDDS::DCPS::TransportSendControlElement, and OpenDDS::DCPS::TransportSendElement.

Referenced by OpenDDS::DCPS::TransportSendStrategy::send_delayed_notifications().

virtual RepoId OpenDDS::DCPS::TransportQueueElement::publication_id (  )  const [pure virtual]

Accessor for the publication id that sent the sample.

Implemented in OpenDDS::DCPS::TransportControlElement, OpenDDS::DCPS::TransportCustomizedElement, OpenDDS::DCPS::TransportReplacedElement, OpenDDS::DCPS::TransportRetainedElement, OpenDDS::DCPS::TransportSendControlElement, and OpenDDS::DCPS::TransportSendElement.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::add_delayed_notification(), OpenDDS::DCPS::RtpsUdpDataLink::add_gap_submsg(), OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element(), fragment(), OpenDDS::DCPS::RtpsUdpDataLink::MultiSendBuffer::insert(), OpenDDS::DCPS::TransportQueueElement::MatchOnPubId::matches(), OpenDDS::DCPS::RtpsUdpSendStrategy::send_bytes_i(), OpenDDS::DCPS::TransportReplacedElement::TransportReplacedElement(), and OpenDDS::DCPS::CopyChainVisitor::visit_element().

virtual void OpenDDS::DCPS::TransportQueueElement::release_element ( bool  dropped_by_transport  )  [protected, pure virtual]

Invoked when the counter reaches 0.

Implemented in OpenDDS::DCPS::TransportControlElement, OpenDDS::DCPS::TransportCustomizedElement, OpenDDS::DCPS::TransportReplacedElement, OpenDDS::DCPS::TransportRetainedElement, OpenDDS::DCPS::TransportSendControlElement, and OpenDDS::DCPS::TransportSendElement.

Referenced by decision_made().

ACE_INLINE void OpenDDS::DCPS::TransportQueueElement::released ( bool  flag  ) 

Definition at line 89 of file TransportQueueElement.inl.

References released_.

00090 {
00091   this->released_ = flag;
00092 }

ACE_INLINE bool OpenDDS::DCPS::TransportQueueElement::released (  )  const

Is the listener get called ?

Definition at line 82 of file TransportQueueElement.inl.

References released_.

00083 {
00084   return this->released_;
00085 }

bool OpenDDS::DCPS::TransportQueueElement::requires_exclusive_packet (  )  const [virtual]

Does the sample require an exclusive transport packet?

Reimplemented in OpenDDS::DCPS::TransportControlElement, OpenDDS::DCPS::TransportCustomizedElement, and OpenDDS::DCPS::TransportSendControlElement.

Definition at line 27 of file TransportQueueElement.cpp.

References DBG_ENTRY_LVL.

Referenced by OpenDDS::DCPS::TransportSendStrategy::send().

00028 {
00029   DBG_ENTRY_LVL("TransportQueueElement", "requires_exclusive_packet", 6);
00030   return false;
00031 }

virtual SequenceNumber OpenDDS::DCPS::TransportQueueElement::sequence (  )  const [inline, virtual]

Reimplemented in OpenDDS::DCPS::TransportCustomizedElement, OpenDDS::DCPS::TransportSendControlElement, OpenDDS::DCPS::TransportSendElement, and OpenDDS::DCPS::RtpsCustomizedElement.

Definition at line 109 of file TransportQueueElement.h.

References OpenDDS::RTPS::SEQUENCENUMBER_UNKNOWN.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::add_elem_awaiting_ack(), OpenDDS::DCPS::RtpsUdpDataLink::add_gap_submsg(), OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element(), OpenDDS::DCPS::RtpsUdpDataLink::MultiSendBuffer::insert(), and OpenDDS::DCPS::TransportCustomizedElement::sequence().

00109                                           {
00110     return SequenceNumber::SEQUENCENUMBER_UNKNOWN();
00111   }

virtual RepoId OpenDDS::DCPS::TransportQueueElement::subscription_id (  )  const [inline, virtual]

Accessor for the subscription id, if sent the sample is sent to 1 sub.

Reimplemented in OpenDDS::DCPS::TransportCustomizedElement, and OpenDDS::DCPS::TransportSendElement.

Definition at line 105 of file TransportQueueElement.h.

References OpenDDS::DCPS::GUID_UNKNOWN.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::add_gap_submsg(), OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element(), OpenDDS::DCPS::RtpsUdpDataLink::durability_resend(), and OpenDDS::DCPS::RtpsUdpSendStrategy::send_bytes_i().

00105                                          {
00106     return GUID_UNKNOWN;
00107   }

ACE_INLINE bool OpenDDS::DCPS::TransportQueueElement::was_dropped (  )  const [protected]

May be used by subclass' implementation of release_element() to determine if any DataLinks dropped the data instead of delivering it.

Definition at line 75 of file TransportQueueElement.inl.

References dropped_.

Referenced by OpenDDS::DCPS::TransportSendControlElement::release_element().

00076 {
00077   return this->dropped_;
00078 }

Friends And Related Function Documentation

friend class TransportCustomizedElement [friend]

Definition at line 165 of file TransportQueueElement.h.

Member Data Documentation

bool OpenDDS::DCPS::TransportQueueElement::dropped_ [private]

Flag flipped to true if any DataLink dropped the sample.

Definition at line 171 of file TransportQueueElement.h.

Referenced by data_dropped(), and was_dropped().

bool OpenDDS::DCPS::TransportQueueElement::released_ [private]

If the callback to DW is made.

Definition at line 174 of file TransportQueueElement.h.

Referenced by released().

ACE_Atomic_Op<ACE_Thread_Mutex, unsigned long> OpenDDS::DCPS::TransportQueueElement::sub_loan_count_ [private]

Counts the number of outstanding sub-loans.

Definition at line 168 of file TransportQueueElement.h.

Referenced by decision_made().

The documentation for this class was generated from the following files:
Generated on Fri Feb 12 20:06:31 2016 for OpenDDS by  doxygen 1.4.7