OpenDDS::DCPS::TransportSendStrategy Class Reference

#include <TransportSendStrategy.h>

Inheritance diagram for OpenDDS::DCPS::TransportSendStrategy:

Collaboration diagram for OpenDDS::DCPS::TransportSendStrategy:

Public Types
enum	SendMode {   MODE_NOT_SET, MODE_DIRECT, MODE_QUEUE, MODE_SUSPEND,   MODE_TERMINATED }
typedef BasicQueue< TransportQueueElement >	QueueType
Public Types inherited from OpenDDS::DCPS::ThreadSynchWorker
enum	WorkOutcome { WORK_OUTCOME_MORE_TO_DO, WORK_OUTCOME_NO_MORE_TO_DO, WORK_OUTCOME_CLOGGED_RESOURCE, WORK_OUTCOME_BROKEN_RESOURCE }

Public Member Functions
virtual	~TransportSendStrategy ()
void	send_buffer (TransportSendBuffer *send_buffer)
	Assigns an optional send buffer. More...
int	start ()
void	stop ()
void	send_start ()
void	send (TransportQueueElement *element, bool relink=true)
void	send_stop (GUID_t repoId)
RemoveResult	remove_sample (const DataSampleElement *sample)
void	remove_all_msgs (const GUID_t &pub_id)
virtual WorkOutcome	perform_work ()
virtual void	relink (bool do_suspend=true)
void	suspend_send ()
void	resume_send ()
void	terminate_send (bool graceful_disconnecting=false)
	Remove all samples in the backpressure queue and packet queue. More...
virtual void	terminate_send_if_suspended ()
virtual void	stop_i ()=0
	Let the subclass stop. More...
virtual bool	start_i ()
	Let the subclass start. More...
void	link_released (bool flag)
bool	isDirectMode ()
virtual ACE_HANDLE	get_handle ()
void	deliver_ack_request (TransportQueueElement *element)
bool	fragmentation_helper (TransportQueueElement *original_element, TqeVector &elements_to_send)
void	clear (SendMode new_mode, SendMode old_mode=MODE_NOT_SET)
SendMode	mode () const
	Access the current sending mode. More...
Public Member Functions inherited from OpenDDS::DCPS::ThreadSynchWorker
virtual	~ThreadSynchWorker ()
virtual void	schedule_output ()
	Indicate that queued data is available to be sent. More...
std::size_t	id () const
	DataLink reference value for diagnostics. More...
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

Static Public Member Functions
static int	mb_to_iov (const ACE_Message_Block &msg, iovec *iov)

Static Public Attributes
static const size_t	UDP_MAX_MESSAGE_SIZE = 65466

Protected Member Functions
	TransportSendStrategy (std::size_t id, const TransportImpl_rch &transport, ThreadSynchResource *synch_resource, Priority priority, const ThreadSynchStrategy_rch &thread_sync_strategy)
virtual ssize_t	send_bytes (const iovec iov[], int n, int &bp)
virtual ssize_t	non_blocking_send (const iovec iov[], int n, int &bp)
virtual ssize_t	send_bytes_i (const iovec iov[], int n)=0
virtual void	prepare_header_i ()
	Specific implementation processing of prepared packet header. More...
virtual void	prepare_packet_i ()
	Specific implementation processing of prepared packet. More...
TransportQueueElement *	current_packet_first_element () const
virtual size_t	max_message_size () const
void	set_graceful_disconnecting (bool flag)
	Set graceful disconnecting flag. More...
virtual void	add_delayed_notification (TransportQueueElement *element)
bool	send_delayed_notifications (const TransportQueueElement::MatchCriteria *match=0)
virtual Security::SecurityConfig_rch	security_config () const
virtual RemoveResult	do_remove_sample (const GUID_t &pub_id, const TransportQueueElement::MatchCriteria &criteria, bool remove_all=false)
	Implement framework chain visitations to remove a sample. More...
ThreadSynch *	synch () const
void	set_header_source (ACE_INT64 source)
Protected Member Functions inherited from OpenDDS::DCPS::ThreadSynchWorker
	ThreadSynchWorker (std::size_t id=0)
Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()

Private Types
enum	SendPacketOutcome {   OUTCOME_COMPLETE_SEND, OUTCOME_PARTIAL_SEND, OUTCOME_BACKPRESSURE, OUTCOME_PEER_LOST,   OUTCOME_SEND_ERROR }
typedef ACE_SYNCH_MUTEX	LockType
typedef ACE_Guard< LockType >	GuardType
typedef std::pair< TransportQueueElement *, SendMode >	TQESendModePair
	Used for delayed notifications when performing work. More...
typedef Dynamic_Cached_Allocator_With_Overflow< ACE_Thread_Mutex >	DataAllocator
	Allocator for data buffers. More...

Private Member Functions
void	direct_send (bool relink)
void	get_packet_elems_from_queue ()
void	prepare_header ()
void	prepare_packet ()
SendPacketOutcome	send_packet ()
ssize_t	do_send_packet (const ACE_Message_Block *packet, int &bp)
	Form an IOV and call the send_bytes() template method. More...
virtual ACE_Message_Block *	pre_send_packet (const ACE_Message_Block *m)
int	adjust_packet_after_send (ssize_t num_bytes_sent)
size_t	space_available (size_t already_used=0) const
size_t	current_space_available () const
virtual bool	marshal_transport_header (ACE_Message_Block *mb)
	OPENDDS_VECTOR (TQESendModePair) delayed_delivered_notification_queue_

Static Private Member Functions
static const char *	mode_as_str (SendMode mode)
	Helper function to debugging. More...

Private Attributes
size_t	max_samples_
	Configuration - max number of samples per transport packet. More...
ACE_UINT32	optimum_size_
	Configuration - optimum transport packet size (bytes) More...
ACE_UINT32	max_size_
	Configuration - max transport packet size (bytes) More...
QueueType	queue_
size_t	max_header_size_
	Maximum marshalled size of the transport packet header. More...
ACE_Message_Block *	header_block_
	Current transport packet header, marshalled. More...
SequenceNumber	header_sequence_
	Current transport header sequence number. More...
QueueType	elems_
ACE_Message_Block *	pkt_chain_
bool	header_complete_
unsigned	start_counter_
Atomic< SendMode >	mode_
	This mode determines how send() calls will be handled. More...
SendMode	mode_before_suspend_
unique_ptr< TransportMessageBlockAllocator >	header_mb_allocator_
	Allocator for header data block. More...
unique_ptr< TransportDataBlockAllocator >	header_db_allocator_
	Allocator for header message block. More...
unique_ptr< DataBlockLockPool >	header_db_lock_pool_
	DataBlockLockPool. More...
unique_ptr< DataAllocator >	header_data_allocator_
unique_ptr< ThreadSynch >	synch_
	The thread synch object. More...
LockType	lock_
MessageBlockAllocator	replaced_element_mb_allocator_
	Cached allocator for TransportReplaceElement. More...
DataBlockAllocator	replaced_element_db_allocator_
WeakRcHandle< TransportImpl >	transport_
bool	graceful_disconnecting_
bool	link_released_
TransportSendBuffer *	send_buffer_
TransportHeader	header_
	Current transport packet header. More...

Friends
class	TransportSendBuffer

Detailed Description

This class provides methods to fill packets with samples for sending and handles backpressure. It maintains the list of samples in current packets and also the list of samples queued during backpressure. A thread per connection is created to handle the queued samples.

Notes for the object ownership: 1) Owns ThreadSynch object, list of samples in current packet and list of samples in queue.

Definition at line 61 of file TransportSendStrategy.h.

Member Typedef Documentation

DataAllocator

typedef Dynamic_Cached_Allocator_With_Overflow<ACE_Thread_Mutex> OpenDDS::DCPS::TransportSendStrategy::DataAllocator

private

Allocator for data buffers.

Definition at line 411 of file TransportSendStrategy.h.

GuardType

typedef ACE_Guard<LockType> OpenDDS::DCPS::TransportSendStrategy::GuardType

private

Definition at line 302 of file TransportSendStrategy.h.

LockType

typedef ACE_SYNCH_MUTEX OpenDDS::DCPS::TransportSendStrategy::LockType

private

Definition at line 301 of file TransportSendStrategy.h.

QueueType

typedef BasicQueue<TransportQueueElement> OpenDDS::DCPS::TransportSendStrategy::QueueType

Definition at line 142 of file TransportSendStrategy.h.

TQESendModePair

typedef std::pair<TransportQueueElement*, SendMode> OpenDDS::DCPS::TransportSendStrategy::TQESendModePair

private

Used for delayed notifications when performing work.

Definition at line 398 of file TransportSendStrategy.h.

Member Enumeration Documentation

SendMode

enum OpenDDS::DCPS::TransportSendStrategy::SendMode

Enumerator
MODE_NOT_SET
MODE_DIRECT
MODE_QUEUE
MODE_SUSPEND
MODE_TERMINATED

Definition at line 305 of file TransportSendStrategy.h.

                {
    // MODE_NOT_SET is used as the initial value of mode_before_suspend_ so
    // we can check if the resume_send is paired with suspend_send.
    MODE_NOT_SET,
    // Send out the sample with current packet.
    MODE_DIRECT,
    // The samples need be queued because of the backpressure or partial send.
    MODE_QUEUE,
    // The samples need be queued because the connection is lost and we are
    // trying to reconnect.
    MODE_SUSPEND,
    // The samples need be dropped since we lost connection and could not
    // reconnect.
    MODE_TERMINATED
  };

SendPacketOutcome

enum OpenDDS::DCPS::TransportSendStrategy::SendPacketOutcome

private

Enumerator
OUTCOME_COMPLETE_SEND
OUTCOME_PARTIAL_SEND
OUTCOME_BACKPRESSURE
OUTCOME_PEER_LOST
OUTCOME_SEND_ERROR

Definition at line 224 of file TransportSendStrategy.h.

                         {
    OUTCOME_COMPLETE_SEND,
    OUTCOME_PARTIAL_SEND,
    OUTCOME_BACKPRESSURE,
    OUTCOME_PEER_LOST,
    OUTCOME_SEND_ERROR
  };

Constructor & Destructor Documentation

~TransportSendStrategy()

OpenDDS::DCPS::TransportSendStrategy::~TransportSendStrategy ( )

virtual

Definition at line 113 of file TransportSendStrategy.cpp.

References DBG_ENTRY_LVL.

{
  DBG_ENTRY_LVL("TransportSendStrategy","~TransportSendStrategy",6);


  delayed_delivered_notification_queue_.clear();
}

TransportSendStrategy()

OpenDDS::DCPS::TransportSendStrategy::TransportSendStrategy ( std::size_t  id, const TransportImpl_rch &  transport, ThreadSynchResource *  synch_resource, Priority  priority, const ThreadSynchStrategy_rch &  thread_sync_strategy  )

protected

Definition at line 61 of file TransportSendStrategy.cpp.

References ACE_DEFAULT_THREAD_PRIORITY, OpenDDS::DCPS::TransportImpl::config(), DBG_ENTRY_LVL, OpenDDS::DCPS::TransportHeader::get_max_serialized_size(), max_header_size_, OpenDDS::DCPS::TransportInst::max_packet_size_, max_samples_, OpenDDS::DCPS::TransportInst::max_samples_per_packet_, max_size_, OpenDDS::DCPS::TransportInst::optimum_packet_size_, optimum_size_, synch_, TheServiceParticipant, and OpenDDS::DCPS::DirectPriorityMapper::thread_priority().

  : ThreadSynchWorker(id),
    max_samples_(DEFAULT_CONFIG_MAX_SAMPLES_PER_PACKET),
    optimum_size_(DEFAULT_CONFIG_OPTIMUM_PACKET_SIZE),
    max_size_(DEFAULT_CONFIG_MAX_PACKET_SIZE),
    max_header_size_(0),
    header_block_(0),
    pkt_chain_(0),
    header_complete_(false),
    start_counter_(0),
    mode_(MODE_DIRECT),
    mode_before_suspend_(MODE_NOT_SET),
    lock_(),
    replaced_element_mb_allocator_(NUM_REPLACED_ELEMENT_CHUNKS * 2),
    replaced_element_db_allocator_(NUM_REPLACED_ELEMENT_CHUNKS * 2),
    transport_(transport),
    graceful_disconnecting_(false),
    link_released_(true),
    send_buffer_(0)
{
  DBG_ENTRY_LVL("TransportSendStrategy","TransportSendStrategy",6);

  TransportInst_rch cfg = transport->config();
  if (cfg) {
    max_samples_ = cfg->max_samples_per_packet_;
    optimum_size_ = cfg->optimum_packet_size_;
    max_size_ = cfg->max_packet_size_;
  }

  // Create a ThreadSynch object just for us.
  DirectPriorityMapper mapper(priority);
  synch_.reset(thread_sync_strategy->create_synch_object(
                   synch_resource,
#ifdef ACE_WIN32
                   ACE_DEFAULT_THREAD_PRIORITY,
#else
                   mapper.thread_priority(),
#endif
                   TheServiceParticipant->scheduler()));

  // We cache this value in data member since it doesn't change, and we
  // don't want to keep asking for it over and over.
  max_header_size_ = TransportHeader::get_max_serialized_size();

  delayed_delivered_notification_queue_.reserve(max_samples_);
}

Member Function Documentation

add_delayed_notification()

void OpenDDS::DCPS::TransportSendStrategy::add_delayed_notification ( TransportQueueElement *  element )

protectedvirtual

Reimplemented in OpenDDS::DCPS::RtpsUdpSendStrategy, and OpenDDS::DCPS::TcpSendStrategy.

Definition at line 1926 of file TransportSendStrategy.cpp.

References ACE_DEBUG, LM_DEBUG, OpenDDS::DCPS::Atomic< T >::load(), max_samples_, mode_, and OpenDDS::DCPS::Transport_debug_level.

Referenced by OpenDDS::DCPS::TcpSendStrategy::add_delayed_notification(), OpenDDS::DCPS::RtpsUdpSendStrategy::add_delayed_notification(), adjust_packet_after_send(), and send().

{
  if (Transport_debug_level) {
    size_t size = delayed_delivered_notification_queue_.size();
    if ((size > 0) && (size % max_samples_ == 0)) {
      ACE_DEBUG((LM_DEBUG,
                 "(%P|%t) Transport send strategy notification queue threshold, size=%d\n",
                 size));
    }
  }

  delayed_delivered_notification_queue_.push_back(std::make_pair(element, mode_.load()));
}

adjust_packet_after_send()

int OpenDDS::DCPS::TransportSendStrategy::adjust_packet_after_send ( ssize_t  num_bytes_sent )

private

This is called from the send_packet() method after it has sent at least one byte from the current packet. This method will update the current packet appropriately, as well as deal with all of the release()'ing of fully sent ACE_Message_Blocks, and the data_delivered() calls on the fully sent elements. Returns 0 if the entire packet was sent, and returns 1 if the entire packet was not sent.

Definition at line 363 of file TransportSendStrategy.cpp.

References ACE_DEBUG, add_delayed_notification(), ACE_Message_Block::base(), ACE_Message_Block::cont(), DBG_ENTRY_LVL, elems_, OpenDDS::DCPS::BasicQueue< T >::get(), header_, header_complete_, ACE_Message_Block::length(), OpenDDS::DCPS::TransportHeader::length_, LM_DEBUG, LM_INFO, OpenDDS::DCPS::TransportQueueElement::msg(), OpenDDS::DCPS::BasicQueue< T >::peek(), pkt_chain_, ACE_Message_Block::rd_ptr(), ACE_Message_Block::release(), and VDBG.

Referenced by clear(), and send_packet().

{
  DBG_ENTRY_LVL("TransportSendStrategy", "adjust_packet_after_send", 6);

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Adjusting the current packet because %d bytes of the packet "
        "have been sent.\n", num_bytes_sent));

  ssize_t num_bytes_left = num_bytes_sent;
  ssize_t num_non_header_bytes_sent = 0;

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Set num_bytes_left to %d.\n", num_bytes_left));
  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Set num_non_header_bytes_sent to %d.\n",
        num_non_header_bytes_sent));

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Peek at the element at the front of the packet elems_.\n"));

  // This is the element currently at the front of elems_.
  TransportQueueElement* element = elems_.peek();

  if(!element){
    ACE_DEBUG((LM_INFO, "(%P|%t) WARNING: adjust_packet_after_send skipping due to NULL element\n"));
  } else {
    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "Use the element's msg() to find the last block in "
          "the msg() chain.\n"));

    // Get a pointer to the last message block in the element.
    const ACE_Message_Block* elem_tail_block = element->msg();

    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "Start with tail block == element->msg().\n"));

    while (elem_tail_block->cont() != 0) {
      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "Set tail block to its cont() block (next in chain).\n"));
      elem_tail_block = elem_tail_block->cont();
    }

    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "Tail block now set (because tail block's cont() is 0).\n"));

    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "Start the 'while (num_bytes_left > 0)' loop.\n"));

    while (num_bytes_left > 0) {
      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "At top of 'num bytes left' loop.  num_bytes_left == [%d].\n",
            num_bytes_left));

      const int block_length = static_cast<int>(pkt_chain_->length());

      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "Length of block at front of pkt_chain_ is [%d].\n",
            block_length));

      if (block_length <= num_bytes_left) {
        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "The whole block at the front of pkt_chain_ was sent.\n"));

        // The entire message block at the front of the chain has been sent.
        // Detach the head message block from the chain and adjust
        // the pkt_chain_ to point to the next block (if any) in
        // the chain.
        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Extract the fully sent block from the pkt_chain_.\n"));

        ACE_Message_Block* fully_sent_block = pkt_chain_;

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Set pkt_chain_ to pkt_chain_->cont().\n"));

        pkt_chain_ = pkt_chain_->cont();

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Set the fully sent block's cont() to 0.\n"));

        fully_sent_block->cont(0);

        // Update the num_bytes_left to indicate that we have
        // processed the entire length of the block.
        num_bytes_left -= block_length;

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Updated num_bytes_left to account for fully sent "
              "block (block_length == [%d]).\n", block_length));
        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Now, num_bytes_left == [%d].\n", num_bytes_left));

        if (!header_complete_) {
          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "Since the header_complete_ flag is false, it means "
                "that the packet header block was still in the "
                "pkt_chain_.\n"));

          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "Not anymore...  Set the header_complete_ flag "
                "to true.\n"));

          // That was the packet header block.  And now we know that it
          // has been completely sent.
          header_complete_ = true;

          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "Release the fully sent block.\n"));

          // Release the fully_sent_block
          fully_sent_block->release();

        } else {
          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "Since the header_complete_ flag is true, it means "
                "that the packet header block was not in the "
                "pkt_chain_.\n"));
          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "So, the fully sent block was part of an element.\n"));

          // That wasn't the packet header block.  It was from the
          // element currently at the front of the elems_
          // collection.  If it was the last block from the
          // element, then we need to extract the element from the
          // elems_ collection and invoke data_delivered() on it.
          num_non_header_bytes_sent += block_length;

          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "Updated num_non_header_bytes_sent to account for "
                "fully sent block (block_length == [%d]).\n",
                block_length));

          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "Now, num_non_header_bytes_sent == [%d].\n",
                num_non_header_bytes_sent));

          if (fully_sent_block->base() == elem_tail_block->base()) {
            VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                  "Ok.  The fully sent block was a duplicate of "
                  "the tail block of the element that is at the "
                  "front of the packet elems_.\n"));

            VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                  "This means that we have completely sent the "
                  "element at the front of the packet elems_.\n"));

            // This means that we have completely sent the element
            // that is currently at the front of the elems_ collection.

            VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                  "We can release the fully sent block now.\n"));

            // Release the fully_sent_block
            fully_sent_block->release();

            VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                  "We can extract the element from the front of "
                  "the packet elems_ (we were just peeking).\n"));

            // Extract the element from the elems_ collection
            element = elems_.get();

            VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                  "Tell the element that a decision has been made "
                  "regarding its fate - data_delivered().\n"));

            // Inform the element that the data has been delivered.
            add_delayed_notification(element);

            VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                  "Peek at the next element in the packet "
                  "elems_.\n"));

            // Set up for the next element in elems_ by peek()'ing.
            element = elems_.peek();

            if (element != 0) {
              VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                    "The is an element still in the packet "
                    "elems_ (we are peeking at it now).\n"));

              VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                    "We are going to find the tail block for the "
                    "current element (we are peeking at).\n"));

              // There was a "next element".  Determine the
              // elem_tail_block for it.
              elem_tail_block = element->msg();

              VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                    "Start w/tail block == element->msg().\n"));

              while (elem_tail_block->cont() != 0) {
                VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                      "Set tail block to next in chain.\n"));
                elem_tail_block = elem_tail_block->cont();
              }

              VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                    "Done finding tail block.\n"));
            }

          } else {
            VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                  "Ok.  The fully sent block is *not* a "
                  "duplicate of the tail block of the element "
                  "at the front of the packet elems_.\n"));

            VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                  "Thus, we have not completely sent the "
                  "element yet.\n"));

            // We didn't completely send the element - it has more
            // message blocks that haven't been sent (that we know of).

            VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                  "We can release the fully_sent_block now.\n"));

            // Release the fully_sent_block
            fully_sent_block->release();
          }
        }

      } else {
        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Only part of the block at the front of pkt_chain_ "
              "was sent.\n"));

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Advance the rd_ptr() of the front block (of pkt_chain_) "
              "by the num_bytes_left (%d).\n", num_bytes_left));

        // Only part of the current block was sent.
        pkt_chain_->rd_ptr(num_bytes_left);

        if (header_complete_) {
          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "And since the packet header block has already been "
                "completely sent, add num_bytes_left to the "
                "num_non_header_bytes_sent.\n"));

          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "Before, num_non_header_bytes_sent == %d.\n",
                num_non_header_bytes_sent));

          // We know that the current block isn't the packet header
          // block because the packet header block has already been
          // completely sent.  We need to count these bytes in the
          // num_non_header_bytes_sent.
          num_non_header_bytes_sent += num_bytes_left;

          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "After, num_non_header_bytes_sent == %d.\n",
                num_non_header_bytes_sent));
        }

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Set the num_bytes_left to 0 now.\n"));

        num_bytes_left = 0;
      }
    }
  }

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "The 'num_bytes_left' loop has completed.\n"));

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Adjust the header_.length_ to account for the "
        "num_non_header_bytes_sent.\n"));
  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Before, header_.length_ == %d.\n",
        header_.length_));

  // Adjust the packet header_.length_ to indicate how many non header
  // bytes are left to send.
  header_.length_ -= static_cast<ACE_UINT32>(num_non_header_bytes_sent);

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "After, header_.length_ == %d.\n",
        header_.length_));

  // Returns 0 if the entire packet was sent, and returns 1 otherwise.
  int rc = (header_.length_ == 0) ? 0 : 1;

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Adjustments all done.  Returning [%d].  0 means entire packet "
        "has been sent.  1 means otherwise.\n",
        rc));

  return rc;
}

clear()

void OpenDDS::DCPS::TransportSendStrategy::clear	(	SendMode	new_mode,
		SendMode	old_mode = MODE_NOT_SET
	)

Clear queued messages and messages in current packet and set the current mode to new_mod if the current mode equals old_mode or old_mode is MODE_NOT_SET.

Definition at line 779 of file TransportSendStrategy.cpp.

References OpenDDS::DCPS::BasicQueue< T >::accept_remove_visitor(), adjust_packet_after_send(), DBG_ENTRY_LVL, elems_, header_, header_complete_, OpenDDS::DCPS::TransportHeader::length_, LM_DEBUG, lock_, mode_, mode_before_suspend_, MODE_NOT_SET, pkt_chain_, queue_, send_delayed_notifications(), start_counter_, OpenDDS::DCPS::BasicQueue< T >::swap(), ACE_Message_Block::total_length(), and VDBG.

Referenced by OpenDDS::DCPS::TcpSendStrategy::reset(), OpenDDS::DCPS::DataLink::schedule_delayed_release(), terminate_send(), and OpenDDS::DCPS::TcpSendStrategy::terminate_send_if_suspended().

{
  DBG_ENTRY_LVL("TransportSendStrategy","clear",6);

  send_delayed_notifications();
  QueueType elems;
  QueueType queue;
  {
    GuardType guard(lock_);

    if (old_mode != MODE_NOT_SET && mode_ != old_mode)
      return;

    if (header_.length_ > 0 && pkt_chain_) {
      // Clear the messages in the pkt_chain_ that is partially sent.
      // We just reuse these functions for normal partial send except actual sending.
      int num_bytes_left = static_cast<int>(pkt_chain_->total_length());
      int result = adjust_packet_after_send(num_bytes_left);

      if (result == 0) {
        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "The adjustment logic says that the packet is cleared.\n"));

      } else {
        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "The adjustment returned partial sent.\n"));
      }
    }

    elems.swap(elems_);
    queue.swap(queue_);

    header_.length_ = 0;
    pkt_chain_ = 0;
    header_complete_ = false;
    start_counter_ = 0;
    mode_ = new_mode;
    mode_before_suspend_ = MODE_NOT_SET;
  }

  // We need remove the queued elements outside the lock,
  // otherwise we have a deadlock situation when remove visitor
  // calls the data_dropped on each dropped elements.

  // Clear all samples in queue.
  RemoveAllVisitor remove_all_visitor;

  elems.accept_remove_visitor(remove_all_visitor);
  queue.accept_remove_visitor(remove_all_visitor);
}

current_packet_first_element()

ACE_INLINE TransportQueueElement * OpenDDS::DCPS::TransportSendStrategy::current_packet_first_element ( ) const

protected

Definition at line 153 of file TransportSendStrategy.inl.

References elems_, OPENDDS_END_VERSIONED_NAMESPACE_DECL, and OpenDDS::DCPS::BasicQueue< T >::peek().

Referenced by OpenDDS::DCPS::RtpsUdpSendStrategy::send_bytes_i_helper().

{
  return this->elems_.peek();
}

current_space_available()

size_t OpenDDS::DCPS::TransportSendStrategy::current_space_available ( ) const

private

Like above, but use the current packet.

Definition at line 1961 of file TransportSendStrategy.cpp.

References header_, OpenDDS::DCPS::TransportHeader::length_, and space_available().

Referenced by get_packet_elems_from_queue(), and send().

{
  return space_available(header_.length_);
}

deliver_ack_request()

void OpenDDS::DCPS::TransportSendStrategy::deliver_ack_request

(

TransportQueueElement *

element

)

Definition at line 1940 of file TransportSendStrategy.cpp.

References OpenDDS::DCPS::TransportQueueElement::data_delivered(), do_remove_sample(), OpenDDS::DCPS::GUID_UNKNOWN, and lock_.

{
  const TransportQueueElement::MatchOnElement moe(element);
  {
    GuardType guard(lock_);
    do_remove_sample(GUID_UNKNOWN, moe);
  }

  element->data_delivered();
}

direct_send()

void OpenDDS::DCPS::TransportSendStrategy::direct_send ( bool  relink )

private

Called from send() when it is time to attempt to send our current packet to the socket while in MODE_DIRECT mode_. If backpressure occurs, our current packet will be adjusted to account for bytes that were sent, and the mode will be changed to MODE_QUEUE. If no backpressure occurs (ie, the entire packet is sent), then our current packet will be "reset" to be an empty packet following the send.

Definition at line 1465 of file TransportSendStrategy.cpp.

References DBG_ENTRY_LVL, OpenDDS::DCPS::TransportImpl::dump(), LM_DEBUG, LM_WARNING, mode_, mode_before_suspend_, MODE_QUEUE, MODE_SUSPEND, MODE_TERMINATED, OUTCOME_BACKPRESSURE, OUTCOME_PARTIAL_SEND, OUTCOME_PEER_LOST, OUTCOME_SEND_ERROR, prepare_packet(), relink(), send_packet(), transport_, OpenDDS::DCPS::Transport_debug_level, VDBG, and VDBG_LVL.

Referenced by send(), and send_stop().

{
  DBG_ENTRY_LVL("TransportSendStrategy", "direct_send", 6);

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Prepare the current packet for a direct send attempt.\n"));

  // Prepare the packet for sending.
  prepare_packet();

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Now attempt to send the packet.\n"));

  // We will try resend the packet if the send() fails and then connection
  // is re-established.  Only loops if the "continue" line is hit.
  while (true) {
    // Attempt to send the packet
    const SendPacketOutcome outcome = send_packet();

    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "The outcome of the send_packet() was %d.\n", outcome));

    if ((outcome == OUTCOME_BACKPRESSURE) ||
        (outcome == OUTCOME_PARTIAL_SEND)) {
      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "The outcome of the send_packet() was either "
                "OUTCOME_BACKPRESSURE or OUTCOME_PARTIAL_SEND.\n"), 5);

      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "Flip into the MODE_QUEUE mode_.\n"), 5);

      // We encountered backpressure, or only sent part of the packet.
      mode_ = MODE_QUEUE;

    } else if ((outcome == OUTCOME_PEER_LOST) ||
               (outcome == OUTCOME_SEND_ERROR)) {
      if (outcome == OUTCOME_SEND_ERROR) {
        VDBG_LVL((LM_WARNING,
                  "(%P|%t) WARNING: Problem detected in "
                  "send buffer management: %p.\n",
                  "send_bytes"), 1);

        if (Transport_debug_level > 0) {
          TransportImpl_rch transport = transport_.lock();
          if (transport) {
            transport->dump();
          }
        }
      } else {
        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "The outcome of the send_packet() was "
              "OUTCOME_PEER_LOST.\n"));
      }

      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "Now flip to MODE_SUSPEND before we try to reconnect.\n"), 5);

      if (mode_ != MODE_SUSPEND) {
        mode_before_suspend_ = mode_;
        mode_ = MODE_SUSPEND;
      }

      if (do_relink) {
        bool do_suspend = false;
        relink(do_suspend);

        if (mode_ == MODE_SUSPEND) {
          VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                    "The reconnect has not done yet and we are "
                    "still in MODE_SUSPEND.\n"), 5);

        } else if (mode_ == MODE_TERMINATED) {
          VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                    "Reconnect failed, we are in MODE_TERMINATED\n"), 5);

        } else {
          VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                    "Try send the packet again since the connection "
                    "is re-established.\n"), 5);

          // Try send the packet again since the connection is re-established.
          continue;
        }
      }

    } else {
      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "The outcome of the send_packet() must have been "
            "OUTCOME_COMPLETE_SEND.\n"));
      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "So, we will just stay in MODE_DIRECT.\n"));
    }

    break;
  }

  // We stay in MODE_DIRECT mode if we didn't encounter any backpressure.
}

do_remove_sample()

RemoveResult OpenDDS::DCPS::TransportSendStrategy::do_remove_sample ( const GUID_t &  pub_id, const TransportQueueElement::MatchCriteria &  criteria, bool  remove_all = false  )

protectedvirtual

Implement framework chain visitations to remove a sample.

Definition at line 1362 of file TransportSendStrategy.cpp.

References OpenDDS::DCPS::BasicQueue< T >::accept_remove_visitor(), OpenDDS::DCPS::BasicQueue< T >::accept_replace_visitor(), DBG_ENTRY_LVL, elems_, header_, header_block_, OpenDDS::DCPS::TransportHeader::length_, LM_DEBUG, mode_, MODE_DIRECT, pkt_chain_, queue_, OpenDDS::DCPS::REMOVE_ERROR, OpenDDS::DCPS::REMOVE_FOUND, OpenDDS::DCPS::REMOVE_NOT_FOUND, OpenDDS::DCPS::REMOVE_RELEASED, OpenDDS::DCPS::QueueRemoveVisitor::removed_bytes(), replaced_element_db_allocator_, replaced_element_mb_allocator_, OpenDDS::DCPS::BasicQueue< T >::size(), OpenDDS::DCPS::QueueRemoveVisitor::status(), OpenDDS::DCPS::PacketRemoveVisitor::status(), OpenDDS::DCPS::TransportQueueElement::MatchCriteria::unique(), and VDBG.

Referenced by deliver_ack_request(), remove_all_msgs(), and remove_sample().

{
  DBG_ENTRY_LVL("TransportSendStrategy", "do_remove_sample", 6);

  //ciju: Tim had the idea that we could do the following check
  // if ((mode_ == MODE_DIRECT) ||
  //     ((pkt_chain_ == 0) && (queue_ == empty)))
  // then we can assume that the sample can be safely removed (no need for
  // replacement) from the elems_ queue.
  if ((mode_ == MODE_DIRECT)
      || ((pkt_chain_ == 0) && (queue_.size() == 0))) {
    //ciju: I believe this is the only mode where a safe
    // assumption can be made that the samples
    // in the elems_ queue aren't part of a packet.
    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "The mode is MODE_DIRECT, or the queue is empty and no "
          "transport packet is in progress.\n"));

    QueueRemoveVisitor simple_rem_vis(criteria, remove_all);
    elems_.accept_remove_visitor(simple_rem_vis);

    const RemoveResult status = simple_rem_vis.status();

    if (status == REMOVE_RELEASED || status == REMOVE_FOUND) {
      header_.length_ -= simple_rem_vis.removed_bytes();

    } else if (status == REMOVE_NOT_FOUND) {
      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "Failed to find the sample to remove.\n"));
    }

    if (criteria.unique() || !remove_all) return status;
  }

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Visit the queue_ with the RemoveElementVisitor.\n"));

  QueueRemoveVisitor simple_rem_vis(criteria, remove_all);
  queue_.accept_remove_visitor(simple_rem_vis);

  RemoveResult status = simple_rem_vis.status();

  if (status == REMOVE_RELEASED || status == REMOVE_FOUND) {
    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "The sample was removed from the queue_.\n"));
    // This means that the visitor did not encounter any fatal error
    // along the way, *AND* the sample was found in the queue_,
    // and has now been removed.  We are done.
    if (criteria.unique() || !remove_all) return status;
  }

  if (status == REMOVE_ERROR) {
    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "The RemoveElementVisitor encountered a fatal error in queue_.\n"));
    // This means that the visitor encountered some fatal error along
    // the way (and it already reported something to the log).
    // Return our failure code.
    return status;
  }

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "The RemoveElementVisitor did not find the sample in queue_.\n"));

  // We get here if the visitor did not encounter any fatal error, but it
  // also didn't find the sample - and hence it didn't perform any
  // "remove sample" logic.

  // Now we need to turn our attention to the current transport packet,
  // since the packet is likely in a "partially sent" state, and the
  // sample may still be contributing unsent bytes in the pkt_chain_.

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Visit our elems_ with the PacketRemoveVisitor.\n"));

  PacketRemoveVisitor pac_rem_vis(criteria,
                                  pkt_chain_,
                                  header_block_,
                                  replaced_element_mb_allocator_,
                                  replaced_element_db_allocator_,
                                  remove_all);

  elems_.accept_replace_visitor(pac_rem_vis);

  status = pac_rem_vis.status();

  if (status == REMOVE_ERROR) {
    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "The PacketRemoveVisitor encountered a fatal error.\n"));

  } else if (status == REMOVE_NOT_FOUND) {
    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "The PacketRemoveVisitor didn't find the sample.\n"));

  } else {
    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "The PacketRemoveVisitor found the sample and removed it.\n"));
  }

  return status;
}

do_send_packet()

ssize_t OpenDDS::DCPS::TransportSendStrategy::do_send_packet ( const ACE_Message_Block *  packet, int &  bp  )

private

Form an IOV and call the send_bytes() template method.

Definition at line 1743 of file TransportSendStrategy.cpp.

References ACE_DEBUG, ACE_TEXT(), ACE_Message_Block::data_block(), DBG_ENTRY_LVL, LM_DEBUG, OpenDDS::DCPS::MAX_SEND_BLOCKS, mb_to_iov(), pre_send_packet(), OpenDDS::DCPS::unique_ptr< T, Deleter >::reset(), security_config(), send_bytes(), ACE_Message_Block::total_length(), OpenDDS::DCPS::Transport_debug_level, VDBG, and VDBG_LVL.

Referenced by OpenDDS::DCPS::TransportSendBuffer::resend_one(), and send_packet().

{
  if (Transport_debug_level > 9) {
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) TransportSendStrategy::do_send_packet() [%d] - ")
               ACE_TEXT("sending data at 0x%x.\n"),
               id(), packet));
  }
  DBG_ENTRY_LVL("TransportSendStrategy", "do_send_packet", 6);

#ifdef OPENDDS_SECURITY
  Message_Block_Ptr substitute;
  if (security_config()) {
    const DDS::Security::CryptoTransform_var crypto = security_config()->get_crypto_transform();
    // pre_send_packet may provide different data that takes the place of the
    // original "packet" (used for security encryption/authentication)
    if (crypto) {
      substitute.reset(pre_send_packet(packet));
      if (!substitute) {
        VDBG((LM_DEBUG, "(%P|%t) DBG:   pre_send_packet returned NULL, dropping.\n"));
        return packet->total_length();
      }
    }
  }
#endif

  VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
            "Populate the iovec array using the packet.\n"), 5);

  iovec iov[MAX_SEND_BLOCKS];

#ifdef OPENDDS_SECURITY
  const int num_blocks = mb_to_iov(substitute ? *substitute : *packet, iov);
#else
  const int num_blocks = mb_to_iov(*packet, iov);
#endif

  VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
            "There are [%d] number of entries in the iovec array.\n",
            num_blocks), 5);

  VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
            "Attempt to send_bytes() now.\n"), 5);

  const ssize_t num_bytes_sent = send_bytes(iov, num_blocks, bp);

  VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
            "The send_bytes() said that num_bytes_sent == [%d].\n",
            num_bytes_sent), 5);

#ifdef OPENDDS_SECURITY
  if (num_bytes_sent > 0 && substitute && packet->data_block() != substitute->data_block()) {
    // Although the "substitute" data took the place of "packet", the rest
    // of the framework needs to account for the bytes in "packet" being taken
    // care of, as if they were actually sent.
    // Since this is done with datagram sockets, partial sends aren't possible.
    return packet->total_length();
  }
#endif

  return num_bytes_sent;
}

fragmentation_helper()

bool OpenDDS::DCPS::TransportSendStrategy::fragmentation_helper	(	TransportQueueElement *	original_element,
		TqeVector &	elements_to_send
	)

Alternative to TransportSendStrategy::send for fragmentation

Parameters

original_element	data sample to send, may be larger than max msg size
elements_to_send	populated by this method with either original_element or fragments created from it. Elements need to be cleaned up by the caller using data_delivered or data_dropped.

Returns

operation succeeded

Definition at line 1988 of file TransportSendStrategy.cpp.

References ACE_ERROR, OpenDDS::DCPS::TransportQueueElement::increment_loan(), LM_DEBUG, LM_ERROR, OpenDDS::DCPS::null_tqe_pair, OPENDDS_END_VERSIONED_NAMESPACE_DECL, space_available(), and VDBG_LVL.

{
  original_element->increment_loan();
  const size_t space = space_available();
  for (TransportQueueElement* e = original_element; e;) {
    const size_t esize = e->msg()->total_length();
    if (esize > space) {
      VDBG_LVL((LM_DEBUG, "(%P|%t) TransportSendStrategy::fragmentation_helper: "
          "message size %B > space %B: Fragmenting\n", esize, space), 0);
      const TqePair pair = e->fragment(space);
      if (pair == null_tqe_pair) {
        ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: TransportSendStrategy::fragmentation_helper: "
          "Element Fragmentation Failed\n"));
        return false;
      }
      elements_to_send.push_back(pair.first);
      e = pair.second;
    } else {
      elements_to_send.push_back(e);
      e = 0;
    }
  }
  return true;
}

get_handle()

ACE_INLINE ACE_HANDLE OpenDDS::DCPS::TransportSendStrategy::get_handle ( void  )

virtual

Implements OpenDDS::DCPS::ThreadSynchWorker.

Reimplemented in OpenDDS::DCPS::TcpSendStrategy.

Definition at line 140 of file TransportSendStrategy.inl.

References ACE_INLINE.

Referenced by OpenDDS::DCPS::ScheduleOutputHandler::handle_exception(), and non_blocking_send().

{
  return ACE_INVALID_HANDLE;
}

get_packet_elems_from_queue()

void OpenDDS::DCPS::TransportSendStrategy::get_packet_elems_from_queue ( )

private

This method is used while in MODE_QUEUE mode, and a new packet needs to be formulated using elements from the queue_. This is the first step of formulating the new packet. It will extract elements from the queue_ and insert those elements into the pkt_elems_ collection.

After this step has been done, the prepare_packet() step can be performed, followed by the actual send_packet() call.

Definition at line 1565 of file TransportSendStrategy.cpp.

References ACE_ERROR, current_space_available(), DBG_ENTRY_LVL, elems_, OpenDDS::DCPS::TransportHeader::first_fragment_, OpenDDS::DCPS::BasicQueue< T >::get(), header_, OpenDDS::DCPS::TransportHeader::last_fragment_, OpenDDS::DCPS::TransportHeader::length_, LM_ERROR, LM_TRACE, max_message_size(), max_samples_, OpenDDS::DCPS::null_tqe_pair, optimum_size_, OpenDDS::DCPS::BasicQueue< T >::peek(), OpenDDS::DCPS::BasicQueue< T >::put(), queue_, OpenDDS::DCPS::BasicQueue< T >::replace_head(), OpenDDS::DCPS::BasicQueue< T >::size(), and VDBG_LVL.

Referenced by perform_work().

{
  DBG_ENTRY_LVL("TransportSendStrategy", "get_packet_elems_from_queue", 6);

  for (TransportQueueElement* element = queue_.peek(); element != 0;
       element = queue_.peek()) {

    // Total number of bytes in the current element's message block chain.
    size_t element_length = element->msg()->total_length();

    // Flag used to determine if the element requires a packet all to itself.
    const bool exclusive_packet = element->requires_exclusive_packet();

    const size_t avail = current_space_available();

    bool frag = false;
    if (element_length > avail) {
      // The current element won't fit into the current packet
      if (max_message_size()) { // fragmentation enabled
        header_.first_fragment_ = !element->is_fragment();
        VDBG_LVL((LM_TRACE, "(%P|%t) DBG:   Fragmenting from queue\n"), 0);
        const TqePair ep = element->fragment(avail);
        if (ep == null_tqe_pair) {
          ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: TransportSendStrategy::get_packet_elems_from_queue: "
            "Element Fragmentation Failed\n"));
          return;
        }
        element = ep.first;
        element_length = element->msg()->total_length();
        queue_.replace_head(ep.second);
        frag = true; // queue_ is already taken care of, don't get() later
      } else {
        break;
      }
    }

    // If exclusive and the current packet is empty, we won't violate the
    // exclusive_packet requirement by put()'ing the element
    // into the elems_ collection.
    if ((exclusive_packet && elems_.size() == 0)
        || !exclusive_packet) {
      // At this point, we have passed all of the pre-conditions and we can
      // now extract the current element from the queue_, put it into the
      // packet elems_, and adjust the packet header_.length_.
      elems_.put(frag ? element : queue_.get());
      if (header_.length_ == 0) {
        header_.last_fragment_ = !frag && element->is_fragment();
      }
      header_.length_ += static_cast<ACE_UINT32>(element_length);
      VDBG_LVL((LM_TRACE, "(%P|%t) DBG:   Packetizing from queue\n"), 0);
    }

    // With exclusive and (elems_.size() != 0), we don't use the current
    // element as part of the packet.  We know that there is already
    // at least one element in the packet, and the current element
    // is going to need its own (exclusive) packet.  We will just
    // use the packet elems_ as it is now.  Always break once
    // we've encountered and dealt with the exclusive_packet case.
    // Also break if fragmentation was required.
    if (exclusive_packet || frag
        // If the current number of packet elems_ has reached the maximum
        // number of samples per packet, then we are done.
        || elems_.size() == max_samples_
        // If the current value of the header_.length_ exceeds (or equals)
        // the optimum_size_ for a packet, then we are done.
        || header_.length_ >= optimum_size_) {
      break;
    }
  }
}

isDirectMode()

ACE_INLINE bool OpenDDS::DCPS::TransportSendStrategy::isDirectMode

(

)

Definition at line 128 of file TransportSendStrategy.inl.

References ACE_INLINE, mode_, and MODE_DIRECT.

Referenced by OpenDDS::DCPS::DataLink::resume_send().

{
  return this->mode_ == MODE_DIRECT;
}

link_released()

ACE_INLINE void OpenDDS::DCPS::TransportSendStrategy::link_released

(

bool

flag

)

Definition at line 49 of file TransportSendStrategy.inl.

References ACE_INLINE, DBG_ENTRY_LVL, link_released_, and lock_.

Referenced by OpenDDS::DCPS::TcpDataLink::do_association_actions(), OpenDDS::DCPS::DataLink::make_reservation(), OpenDDS::DCPS::MulticastSendStrategy::MulticastSendStrategy(), and OpenDDS::DCPS::ShmemDataLink::send_association_msg().

{
  DBG_ENTRY_LVL("TransportSendStrategy","link_released",6);

  GuardType guard(this->lock_);
  this->link_released_ = flag;
}

marshal_transport_header()

bool OpenDDS::DCPS::TransportSendStrategy::marshal_transport_header ( ACE_Message_Block *  mb )

privatevirtual

Reimplemented in OpenDDS::DCPS::RtpsUdpSendStrategy.

Definition at line 1723 of file TransportSendStrategy.cpp.

References header_.

Referenced by prepare_packet().

{
  return *mb << header_;
}

max_message_size()

ACE_INLINE size_t OpenDDS::DCPS::TransportSendStrategy::max_message_size ( void  ) const

protectedvirtual

The maximum size of a message allowed by the this TransportImpl, or 0 if there is no such limit. This is expected to be a constant, for example UDP/IPv4 can send messages of up to 65466 bytes. The transport framework will use the returned value (if > 0) to fragment larger messages. This fragmentation and reassembly will be transparent to the user.

Reimplemented in OpenDDS::DCPS::RtpsUdpSendStrategy, OpenDDS::DCPS::MulticastSendStrategy, and OpenDDS::DCPS::UdpSendStrategy.

Definition at line 147 of file TransportSendStrategy.inl.

References ACE_INLINE.

Referenced by get_packet_elems_from_queue(), send(), and space_available().

{
  return 0;
}

mb_to_iov()

int OpenDDS::DCPS::TransportSendStrategy::mb_to_iov ( const ACE_Message_Block &  msg, iovec *  iov  )

static

Convert ACE_Message_Block chain into iovec[] entries for send(), returns number of iovec[] entries used (up to MAX_SEND_BLOCKS). Precondition: iov must be an iovec[] of size MAX_SEND_BLOCKS or greater.

Definition at line 1967 of file TransportSendStrategy.cpp.

References OpenDDS::DCPS::MAX_SEND_BLOCKS.

Referenced by do_send_packet(), OpenDDS::DCPS::UdpDataLink::open(), OpenDDS::DCPS::RtpsUdpTransport::IceEndpoint::send(), and OpenDDS::DCPS::RtpsUdpSendStrategy::send_rtps_control().

{
  int num_blocks = 0;
#ifdef _MSC_VER
#pragma warning(push)
// iov_len is 32-bit on 64-bit VC++, but we don't want a cast here
// since on other platforms iov_len is 64-bit
#pragma warning(disable : 4267)
#endif
  for (const ACE_Message_Block* block = &msg;
       block && num_blocks < MAX_SEND_BLOCKS;
       block = block->cont()) {
    iov[num_blocks].iov_len = block->length();
    iov[num_blocks++].iov_base = block->rd_ptr();
  }
#ifdef _MSC_VER
#pragma warning(pop)
#endif
  return num_blocks;
}

mode()

ACE_INLINE TransportSendStrategy::SendMode OpenDDS::DCPS::TransportSendStrategy::mode

(

void

)

const

Access the current sending mode.

Definition at line 16 of file TransportSendStrategy.inl.

References ACE_INLINE, DBG_ENTRY_LVL, and mode_.

Referenced by OpenDDS::DCPS::ScheduleOutputHandler::handle_exception(), mode_as_str(), OpenDDS::DCPS::TcpSendStrategy::schedule_output(), OpenDDS::DCPS::ScheduleOutputHandler::schedule_output(), and send_delayed_notifications().

{
  DBG_ENTRY_LVL("TransportSendStrategy","mode",6);

  return mode_;
}

mode_as_str()

ACE_INLINE const char * OpenDDS::DCPS::TransportSendStrategy::mode_as_str ( SendMode  mode )

staticprivate

Helper function to debugging.

Definition at line 114 of file TransportSendStrategy.inl.

References ACE_INLINE, and mode().

Referenced by perform_work(), send(), and send_stop().

{
  static const char* SendModeStr[] = { "MODE_NOT_SET",
                                       "MODE_DIRECT",
                                       "MODE_QUEUE",
                                       "MODE_SUSPEND",
                                       "MODE_TERMINATED",
                                       "UNKNOWN"
                                     };

  return SendModeStr[mode];
}

non_blocking_send()

ssize_t OpenDDS::DCPS::TransportSendStrategy::non_blocking_send ( const iovec  iov[], int  n, int &  bp  )

protectedvirtual

Definition at line 1879 of file TransportSendStrategy.cpp.

References ACE_DEBUG, ACE_TEXT(), ENOBUFS, EWOULDBLOCK, get_handle(), LM_DEBUG, LM_ERROR, ACE::record_and_set_non_blocking_mode(), ACE::restore_non_blocking_mode(), send_bytes_i(), VDBG, and VDBG_LVL.

Referenced by OpenDDS::DCPS::TcpSendStrategy::send_bytes().

{
  int val = 0;
  ACE_HANDLE handle = get_handle();

  if (handle == ACE_INVALID_HANDLE)
    return -1;

  ACE::record_and_set_non_blocking_mode(handle, val);

  // Set the back-pressure flag to false.
  bp = 0;

  // Clear errno
  errno = 0;

  ssize_t result = send_bytes_i(iov, n);

  if (result == -1) {
    if ((errno == EWOULDBLOCK) || (errno == ENOBUFS)) {
      VDBG((LM_DEBUG,"(%P|%t) DBG:   "
            "Backpressure encountered.\n"));
      // Set the back-pressure flag to true
      bp = 1;

    } else {
      VDBG_LVL((LM_ERROR, "(%P|%t) TransportSendStrategy::send_bytes: ERROR: %p iovec count: %d\n",
                ACE_TEXT("sendv"), n),1);

      // try to get the application to core when "Bad Address" is returned
      // by looking at the iovec
      for (int ii = 0; ii < n; ii++) {
        ACE_DEBUG((LM_DEBUG, "(%P|%t) send_bytes: iov[%d].iov_len = %d .iov_base =%X\n",
                   ii, iov[ii].iov_len, iov[ii].iov_base));
      }
    }
  }

  VDBG_LVL((LM_DEBUG,"(%P|%t) DBG:   "
            "The sendv() returned [%d].\n", result), 5);

  ACE::restore_non_blocking_mode(handle, val);

  return result;
}

OPENDDS_VECTOR()

OpenDDS::DCPS::TransportSendStrategy::OPENDDS_VECTOR ( TQESendModePair  )

private

Referenced by OpenDDS::DCPS::RtpsUdpSendStrategy::encode_submessages(), and send_delayed_notifications().

perform_work()

ThreadSynchWorker::WorkOutcome OpenDDS::DCPS::TransportSendStrategy::perform_work ( void  )

virtual

Called by our ThreadSynch object when we should be able to start sending any partial packet bytes and/or compose a new packet using elements from the queue_.

Returns 0 to indicate that the ThreadSynch object doesn't need to call perform_work() again since the queue (and any unsent packet bytes) has been drained, and the mode_ has been switched to MODE_DIRECT.

Returns 1 to indicate that there is more work to do, and the ThreadSynch object should have this perform_work() method called again.

Implements OpenDDS::DCPS::ThreadSynchWorker.

Definition at line 132 of file TransportSendStrategy.cpp.

References DBG_ENTRY_LVL, get_packet_elems_from_queue(), header_, OpenDDS::DCPS::TransportHeader::length_, LM_DEBUG, lock_, mode_, mode_as_str(), MODE_DIRECT, MODE_QUEUE, MODE_SUSPEND, MODE_TERMINATED, OUTCOME_BACKPRESSURE, OUTCOME_COMPLETE_SEND, OUTCOME_PEER_LOST, OUTCOME_SEND_ERROR, prepare_packet(), queue_, relink(), send_delayed_notifications(), send_packet(), OpenDDS::DCPS::BasicQueue< T >::size(), synch_, VDBG_LVL, OpenDDS::DCPS::ThreadSynchWorker::WORK_OUTCOME_BROKEN_RESOURCE, OpenDDS::DCPS::ThreadSynchWorker::WORK_OUTCOME_CLOGGED_RESOURCE, OpenDDS::DCPS::ThreadSynchWorker::WORK_OUTCOME_MORE_TO_DO, and OpenDDS::DCPS::ThreadSynchWorker::WORK_OUTCOME_NO_MORE_TO_DO.

{
  DBG_ENTRY_LVL("TransportSendStrategy","perform_work",6);

  SendPacketOutcome outcome;
  bool no_more_work = false;

  { // scope for the guard(lock_);
    GuardType guard(lock_);

    VDBG_LVL((LM_DEBUG, "(%P|%t) DBG: perform_work mode: %C\n", mode_as_str(mode_)), 5);

    if (mode_ == MODE_TERMINATED) {
      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "Entered perform_work() and mode_ is MODE_TERMINATED - "
                "we lost connection and could not reconnect, just return "
                "WORK_OUTCOME_BROKEN_RESOURCE.\n"), 5);
      return WORK_OUTCOME_BROKEN_RESOURCE;
    }

    // The perform_work() is called by our synch_ object using
    // a thread designated to call this method when it thinks
    // we need to be called in order to "service" the queue_ and/or
    // deal with a partially sent current packet.
    //
    // We will return a 0 if we don't see a need to have our perform_work()
    // called again, and we will return a 1 if we do see the need to have our
    // perform_work() method called again.

    // First, make sure that the mode_ indicates that we are, indeed, in
    // the MODE_QUEUE mode.  If we are not in MODE_QUEUE mode (meaning we are
    // in MODE_DIRECT), then it means we didn't need to have this perform_work()
    // method called - in this case, do nothing other than return
    // WORK_OUTCOME_NO_MORE_TO_DO to tell our caller that we really don't
    // see a need for it to call our perform_work() again (at least not
    // right now).
    if (mode_ != MODE_QUEUE && mode_ != MODE_SUSPEND) {
      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "Entered perform_work() and mode_ is %C - just return "
                "WORK_OUTCOME_NO_MORE_TO_DO.\n", mode_as_str(mode_)), 5);
      return WORK_OUTCOME_NO_MORE_TO_DO;
    }

    // Check the "state" of the current packet.  We will either find that the
    // current packet is in a state of being "partially sent", or we will find
    // it in a state of being "empty".  When the current packet is "empty", it
    // means that it is time to build up the current packet using elements
    // extracted from the queue_, and then we will attempt to send the
    // packet.  When we find the current packet in the "partially sent" state,
    // we will not touch the queue_ - we will just try to send the unsent
    // bytes in the current (partially sent) packet.
    const size_t header_length = header_.length_;

    if (header_length == 0) {
      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "The current packet doesn't have any unsent bytes - we "
                "need to 'populate' the current packet with elems from "
                "the queue.\n"), 5);

      // The current packet is "empty".  Build up the current packet using
      // elements from the queue_, and prepare the current packet to be sent.

      // Before we build the packet from the queue_, let's make sure that
      // there is actually something on the queue_ to build from.
      if (queue_.size() == 0) {
        VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                  "But the queue is empty.  We have cleared the "
                  "backpressure situation.\n"),5);
        // We are here because the queue_ is empty, and there isn't
        // any "partial packet" bytes left to send.  We have overcome
        // the backpressure situation and don't have anything to do
        // right now.

        VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                  "Flip mode to MODE_DIRECT, and return "
                  "WORK_OUTCOME_NO_MORE_TO_DO.\n"), 5);

        // Flip the mode back to MODE_DIRECT.
        mode_ = MODE_DIRECT;

        // And return WORK_OUTCOME_NO_MORE_TO_DO to tell our caller that
        // perform_work() doesn't need to be called again (at this time).
        return WORK_OUTCOME_NO_MORE_TO_DO;
      }

      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "There is at least one elem in the queue - get the packet "
                "elems from the queue.\n"), 5);

      // There is stuff in the queue_ if we get to this point in the logic.
      // Build-up the current packet using element(s) from the queue_.
      get_packet_elems_from_queue();

      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "Prepare the packet from the packet elems_.\n"), 5);

      // Now we can prepare the new packet to be sent.
      prepare_packet();

      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "Packet has been prepared from packet elems_.\n"), 5);

    } else {
      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "We have a current packet that still has unsent bytes.\n"), 5);
    }

    VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
              "Attempt to send the current packet.\n"), 5);

    // Now we can attempt to send the current packet - whether it is
    // a "partially sent" packet or one that we just built-up using elements
    // from the queue_ (and subsequently prepared for sending) - it doesn't
    // matter.  Just attempt to send as many of the "unsent" bytes in the
    // packet as possible.
    outcome = send_packet();

    // If we sent the whole packet (eg, partial_send is false), and the queue_
    // is now empty, then we've cleared the backpressure situation.
    if ((outcome == OUTCOME_COMPLETE_SEND) && (queue_.size() == 0)) {
      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "Flip the mode to MODE_DIRECT, and then return "
                "WORK_OUTCOME_NO_MORE_TO_DO.\n"), 5);

      // Revert back to MODE_DIRECT mode.
      mode_ = MODE_DIRECT;
      no_more_work = true;
    }
  } // End of scope for guard(lock_);

  VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
            "The outcome of the send_packet() was %d.\n", outcome), 5);

  send_delayed_notifications();

  // If we sent the whole packet (eg, partial_send is false), and the queue_
  // is now empty, then we've cleared the backpressure situation.
  if (no_more_work) {
    VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
              "We sent the whole packet, and there is nothing left on "
              "the queue now.\n"), 5);

    // Return WORK_OUTCOME_NO_MORE_TO_DO to tell our caller that we
    // don't desire another call to this perform_work() method.
    return WORK_OUTCOME_NO_MORE_TO_DO;
  }

  VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
            "We still have unsent bytes in the current packet AND/OR there "
            "are still elements in the queue.\n"), 5);

  if ((outcome == OUTCOME_PEER_LOST) || (outcome == OUTCOME_SEND_ERROR)) {
    VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
              "We lost our connection, or had some fatal connection "
              "error.  Return WORK_OUTCOME_BROKEN_RESOURCE.\n"), 5);

    VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
              "Now flip to MODE_SUSPEND before we try to reconnect.\n"), 5);

    bool do_suspend = true;
    relink(do_suspend);

    if (mode_ == MODE_SUSPEND) {
      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "The reconnect has not done yet and we are still in MODE_SUSPEND. "
                "Return WORK_OUTCOME_CLOGGED_RESOURCE.\n"), 5);
      // Return WORK_OUTCOME_NO_MORE_TO_DO to tell our caller that we
      // don't desire another call to this perform_work() method.
      return WORK_OUTCOME_NO_MORE_TO_DO;

    } else if (mode_ == MODE_TERMINATED) {
      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "Reconnect failed, now we are in MODE_TERMINATED\n"), 5);
      return WORK_OUTCOME_BROKEN_RESOURCE;

    } else {
      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "Reconnect succeeded, Notify synch thread of work "
                "availability.\n"), 5);
      // If the datalink is re-established then notify the synch
      // thread to perform work.  We do not hold the object lock at
      // this point.
      synch_->work_available();
    }
  }

  VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
            "We still have an 'unbroken' connection.\n"), 5);

  if (outcome == OUTCOME_BACKPRESSURE) {
    VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
              "We experienced backpressure on our attempt to send the "
              "packet.  Return WORK_OUTCOME_CLOGGED_RESOURCE.\n"), 5);
    // We have a "clogged resource".
    return WORK_OUTCOME_CLOGGED_RESOURCE;
  }

  VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
            "We may have sent the whole current packet, but still have "
            "elements on the queue.\n"), 5);
  VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
            "Or, we may have only partially sent the current packet.\n"), 5);

  VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
            "Either way, we return WORK_OUTCOME_MORE_TO_DO now.\n"), 5);

  // We may have had an OUTCOME_COMPLETE_SEND, but there is still stuff
  // in the queue_ to be sent.  *OR* we have have had an OUTCOME_PARTIAL_SEND,
  // which equates to the same thing - we still have work to do.

  // We are still in MODE_QUEUE mode, thus there is still work to be
  // done to service the queue_ and/or a partially sent current packet.
  // Return WORK_OUTCOME_MORE_TO_DO so that our caller knows that we still
  // want it to call this perform_work() method.
  return WORK_OUTCOME_MORE_TO_DO;
}

pre_send_packet()

virtual ACE_Message_Block* OpenDDS::DCPS::TransportSendStrategy::pre_send_packet ( const ACE_Message_Block *  m )

inlineprivatevirtual

Derived classes can override to transform the data right before it's sent. If the returned value is non-NULL it will be sent instead of sending the parameter. If the returned value is NULL the original message will be dropped.

Reimplemented in OpenDDS::DCPS::RtpsUdpSendStrategy.

Definition at line 274 of file TransportSendStrategy.h.

References ACE_Message_Block::duplicate().

Referenced by do_send_packet().

  {
    return m->duplicate();
  }

prepare_header()

void OpenDDS::DCPS::TransportSendStrategy::prepare_header ( )

private

This method is responsible for updating the packet header. Called exclusively by prepare_packet.

Definition at line 1637 of file TransportSendStrategy.cpp.

References DBG_ENTRY_LVL, header_, header_sequence_, prepare_header_i(), and OpenDDS::DCPS::TransportHeader::sequence_.

Referenced by prepare_packet().

{
  DBG_ENTRY_LVL("TransportSendStrategy", "prepare_header", 6);

  // Increment header sequence for packet:
  header_.sequence_ = ++header_sequence_;

  // Allow the specific implementation the opportunity to set
  // values in the packet header.
  prepare_header_i();
}

prepare_header_i()

void OpenDDS::DCPS::TransportSendStrategy::prepare_header_i ( )

protectedvirtual

Specific implementation processing of prepared packet header.

Reimplemented in OpenDDS::DCPS::MulticastSendStrategy.

Definition at line 1650 of file TransportSendStrategy.cpp.

References DBG_ENTRY_LVL.

Referenced by prepare_header().

{
  DBG_ENTRY_LVL("TransportSendStrategy","prepare_header_i",6);

  // Default implementation does nothing.
}

prepare_packet()

void OpenDDS::DCPS::TransportSendStrategy::prepare_packet ( )

private

This method is responsible for actually "creating" the current send packet using the packet header and the collection of packet elements that are to make-up the packet's contents.

Definition at line 1658 of file TransportSendStrategy.cpp.

References OpenDDS::DCPS::BasicQueue< T >::accept_visitor(), ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY, ACE_NEW_MALLOC, OpenDDS::DCPS::BuildChainVisitor::chain(), ACE_Message_Block::cont(), DBG_ENTRY_LVL, ACE_Message_Block::duplicate(), elems_, DataBlockLockPool::get_lock(), header_block_, header_complete_, header_data_allocator_, header_db_allocator_, header_db_lock_pool_, header_mb_allocator_, LM_DEBUG, marshal_transport_header(), max_header_size_, ACE_Time_Value::max_time, ACE_Message_Block::MB_DATA, pkt_chain_, prepare_header(), prepare_packet_i(), ACE_Message_Block::release(), VDBG, and ACE_Time_Value::zero.

Referenced by direct_send(), and perform_work().

{
  DBG_ENTRY_LVL("TransportSendStrategy", "prepare_packet", 6);

  // Prepare the header for sending.
  prepare_header();

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Marshal the packet header.\n"));

  if (header_block_ != 0) {
    header_block_->release();
  }

  ACE_NEW_MALLOC(header_block_,
    static_cast<ACE_Message_Block*>(header_mb_allocator_->malloc()),
    ACE_Message_Block(max_header_size_,
                      ACE_Message_Block::MB_DATA,
                      0, // cont
                      0, // data
                      header_data_allocator_.get(),
                      header_db_lock_pool_->get_lock(),
                      ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
                      ACE_Time_Value::zero,
                      ACE_Time_Value::max_time,
                      header_db_allocator_.get(),
                      header_mb_allocator_.get()));

  marshal_transport_header(header_block_);

  pkt_chain_ = header_block_->duplicate();

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Use a BuildChainVisitor to visit the packet elems_.\n"));

  // Build up a chain of blocks by duplicating the message block chain
  // held by each element (in elems_), and then chaining the new duplicate
  // blocks together to form one long chain.
  BuildChainVisitor visitor;
  elems_.accept_visitor(visitor);

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Attach the visitor's chain of blocks to the lone (packet "
        "header) block currently in the pkt_chain_.\n"));

  // Attach the visitor's chain of blocks to the packet header block.
  pkt_chain_->cont(visitor.chain());

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Increment header sequence for next packet.\n"));

  // Allow the specific implementation the opportunity to process the
  // newly prepared packet.
  prepare_packet_i();

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "Set the header_complete_ flag to false.\n"));

  // Set the header_complete_ to false to indicate
  // that the first block in the pkt_chain_ is the packet header block
  // (actually a duplicate() of the packet header_block_).
  header_complete_ = false;
}

prepare_packet_i()

void OpenDDS::DCPS::TransportSendStrategy::prepare_packet_i ( )

protectedvirtual

Specific implementation processing of prepared packet.

Definition at line 1729 of file TransportSendStrategy.cpp.

References DBG_ENTRY_LVL.

Referenced by prepare_packet().

{
  DBG_ENTRY_LVL("TransportSendStrategy","prepare_packet_i",6);

  // Default implementation does nothing.
}

relink()

ACE_INLINE void OpenDDS::DCPS::TransportSendStrategy::relink ( bool  do_suspend = true )

virtual

The subclass needs to provide the implementation for re-establishing the datalink. This is called when send returns an error.

Reimplemented in OpenDDS::DCPS::TcpSendStrategy.

Definition at line 58 of file TransportSendStrategy.inl.

References ACE_INLINE, and DBG_ENTRY_LVL.

Referenced by direct_send(), and perform_work().

{
  DBG_ENTRY_LVL("TransportSendStrategy","relink",6);
  // The subsclass needs implement this function for re-establishing
  // the link upon send failure.
}

remove_all_msgs()

void OpenDDS::DCPS::TransportSendStrategy::remove_all_msgs

(

const GUID_t &

pub_id

)

Definition at line 1315 of file TransportSendStrategy.cpp.

References DBG_ENTRY_LVL, do_remove_sample(), lock_, OpenDDS::DCPS::TransportSendBuffer::retain_all(), send_buffer_, and send_delayed_notifications().

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::client_stop(), OpenDDS::DCPS::DataLink::remove_all_msgs(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_all_msgs().

{
  DBG_ENTRY_LVL("TransportSendStrategy","remove_all_msgs",6);

  const TransportQueueElement::MatchOnPubId match(pub_id);
  send_delayed_notifications(&match);

  GuardType guard(lock_);

  if (send_buffer_ != 0) {
    // If a secondary send buffer is bound, removed samples must
    // be retained in order to properly maintain the buffer:
    send_buffer_->retain_all(pub_id);
  }

  do_remove_sample(pub_id, match, true);
}

remove_sample()

RemoveResult OpenDDS::DCPS::TransportSendStrategy::remove_sample

(

const DataSampleElement *

sample

)

Our DataLink has been requested by some particular TransportClient to remove the supplied sample (basically, an "unsend" attempt) from this strategy object.

Definition at line 1334 of file TransportSendStrategy.cpp.

References ACE_Message_Block::cont(), DBG_ENTRY_LVL, do_remove_sample(), OpenDDS::DCPS::DataSampleElement::get_pub_id(), OpenDDS::DCPS::DataSampleElement::get_sample(), LM_DEBUG, lock_, ACE_Message_Block::rd_ptr(), OpenDDS::DCPS::REMOVE_RELEASED, send_delayed_notifications(), and VDBG_LVL.

Referenced by OpenDDS::DCPS::DataLink::remove_sample(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_sample().

{
  DBG_ENTRY_LVL("TransportSendStrategy", "remove_sample", 6);

  VDBG_LVL((LM_DEBUG, "(%P|%t)  Removing sample: %@\n", sample->get_sample()), 5);

  // The sample to remove is either in temporary delayed notification list or
  // internal list (elems_ or queue_). If it's going to be removed from temporary delayed
  // notification list by transport thread, it needs acquire WriterDataContainer lock for
  // data_dropped/data_delivered callback, then it needs wait for this remove_sample() call
  // complete as this call already hold the WriterContainer's lock. So this call is safe to
  // access the sample to remove. If it's going to be removed by this remove_sample() calling
  // thread, it will be removed either from delayed notification list or from internal list
  // in which case the element carry the info if the sample is released so the datalinkset
  // can stop calling rest datalinks to remove this sample if it's already released..

  const char* const payload = sample->get_sample()->cont()->rd_ptr();
  GUID_t pub_id = sample->get_pub_id();
  const TransportQueueElement::MatchOnDataPayload modp(payload);
  if (send_delayed_notifications(&modp)) {
    return REMOVE_RELEASED;
  }

  GuardType guard(lock_);
  return do_remove_sample(pub_id, modp);
}

resume_send()

ACE_INLINE void OpenDDS::DCPS::TransportSendStrategy::resume_send

(

)

This is called when connection is lost and reconnect succeeds. The send mode is set to the mode before suspend which is either MODE_QUEUE or MODE_DIRECT.

Definition at line 78 of file TransportSendStrategy.inl.

References ACE_ERROR, ACE_INLINE, DBG_ENTRY_LVL, elems_, header_, header_complete_, OpenDDS::DCPS::TransportHeader::length_, LM_ERROR, lock_, mode_, mode_before_suspend_, MODE_DIRECT, MODE_NOT_SET, MODE_QUEUE, MODE_SUSPEND, MODE_TERMINATED, pkt_chain_, queue_, OpenDDS::DCPS::BasicQueue< T >::size(), start_counter_, OpenDDS::DCPS::BasicQueue< T >::swap(), and synch_.

Referenced by OpenDDS::DCPS::DataLink::resume_send().

{
  DBG_ENTRY_LVL("TransportSendStrategy","resume_send",6);
  GuardType guard(this->lock_);

  // If this send strategy is reused when the connection is reestablished, then
  // we need re-initialize the mode_ and mode_before_suspend_.
  if (this->mode_ == MODE_TERMINATED) {
    this->header_.length_ = 0;
    this->pkt_chain_ = 0;
    this->header_complete_ = false;
    this->start_counter_ = 0;
    this->mode_ = MODE_DIRECT;
    this->mode_before_suspend_ = MODE_NOT_SET;
    this->delayed_delivered_notification_queue_.clear();

  } else if (this->mode_ == MODE_SUSPEND) {
    this->header_.length_ = 0;
    this->pkt_chain_ = 0;
    QueueType elems;
    elems.swap(this->elems_);
    this->mode_ = this->mode_before_suspend_;
    this->header_complete_ = false;
    this->mode_before_suspend_ = MODE_NOT_SET;
    if (this->queue_.size() > 0) {
      this->mode_ = MODE_QUEUE;
      this->synch_->work_available();
    }

  } else {
    ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: TransportSendStrategy::resume_send  The suspend or terminate"
               " is not called previously.\n"));
  }
}

security_config()

virtual Security::SecurityConfig_rch OpenDDS::DCPS::TransportSendStrategy::security_config ( ) const

inlineprotectedvirtual

Reimplemented in OpenDDS::DCPS::RtpsUdpSendStrategy.

Definition at line 219 of file TransportSendStrategy.h.

Referenced by do_send_packet().

{ return Security::SecurityConfig_rch(); }

send()

void OpenDDS::DCPS::TransportSendStrategy::send	(	TransportQueueElement *	element,
		bool	relink = true
	)

Our DataLink has been requested by some particular TransportClient to send the element.

Definition at line 935 of file TransportSendStrategy.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), add_delayed_notification(), current_space_available(), OpenDDS::DCPS::TransportQueueElement::data_dropped(), DBG_ENTRY_LVL, direct_send(), elems_, OpenDDS::DCPS::TransportHeader::first_fragment_, OpenDDS::DCPS::TransportQueueElement::fragment(), OpenDDS::DCPS::DataSampleHeader::get_max_serialized_size(), graceful_disconnecting_, header_, OpenDDS::DCPS::TransportHeader::last_fragment_, OpenDDS::DCPS::TransportHeader::length_, link_released_, LM_DEBUG, LM_ERROR, LM_TRACE, lock_, max_header_size_, max_message_size(), max_samples_, max_size_, mode_, mode_as_str(), MODE_DIRECT, MODE_QUEUE, MODE_SUSPEND, MODE_TERMINATED, OpenDDS::DCPS::TransportQueueElement::msg(), OpenDDS::DCPS::null_tqe_pair, optimum_size_, OpenDDS::DCPS::BasicQueue< T >::put(), queue_, ACE_Guard< ACE_LOCK >::release(), OpenDDS::DCPS::TransportQueueElement::requires_exclusive_packet(), send_delayed_notifications(), OpenDDS::DCPS::BasicQueue< T >::size(), synch_, ACE_Message_Block::total_length(), OpenDDS::DCPS::Transport_debug_level, VDBG, and VDBG_LVL.

Referenced by OpenDDS::DCPS::DataLink::send_i().

{
  if (Transport_debug_level > 9) {
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) TransportSendStrategy::send() [%d] - ")
               ACE_TEXT("sending data at 0x%x.\n"),
               id(), element));
  }

  DBG_ENTRY_LVL("TransportSendStrategy", "send", 6);

  {
    GuardType guard(lock_);

    if (link_released_) {
      add_delayed_notification(element);

    } else {
      if (mode_ == MODE_TERMINATED && !graceful_disconnecting_) {
        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "TransportSendStrategy::send: mode is MODE_TERMINATED and not in "
              "graceful disconnecting, so discard message.\n"));
        guard.release();
        element->data_dropped(true);
        return;
      }

      size_t element_length = element->msg()->total_length();

      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "Send element msg() has total_length() == [%d].\n",
            element_length));

      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "max_header_size_ == [%d].\n",
            max_header_size_));

      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "max_size_ == [%d].\n",
            max_size_));

      const size_t max_msg_size = max_message_size();

      // Really an assert.  We can't accept any element that wouldn't fit into
      // a transport packet by itself (ie, it would be the only element in the
      // packet).  This max_size_ is the user-configurable maximum, not based
      // on the transport's inherent maximum message size.  If max_msg_size
      // is non-zero, we will fragment so max_size_ doesn't apply per-element.
      if (max_msg_size == 0 &&
          max_header_size_ + element_length > max_size_) {
        ACE_ERROR((LM_ERROR,
                   "(%P|%t) ERROR: Element too large (%Q) "
                   "- won't fit into packet.\n", ACE_UINT64(element_length)));
        return;
      }

      // Check the mode_ to see if we simply put the element on the queue.
      if (mode_ == MODE_QUEUE || mode_ == MODE_SUSPEND) {
        VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                  "mode_ == %C, so queue elem and leave.\n",
                  mode_as_str(mode_)), 5);

        queue_.put(element);

        if (mode_ != MODE_SUSPEND) {
          synch_->work_available();
        }

        return;
      }

      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "mode_ == MODE_DIRECT.\n"));

      // We are in the MODE_DIRECT send mode.  When in this mode, the send()
      // calls will "build up" the transport packet to be sent directly when it
      // reaches the optimal size, contains the maximum number of samples, etc.

      // We need to check if the current element (the arg passed-in to this
      // send() method) should be appended to the transport packet, or if the
      // transport packet should be sent (directly) first, dealing with the
      // current element afterwards.

      // We will decide to send the packet as it is now, under two circumstances:
      //
      //    Either:
      //
      //    (1) The current element won't fit into the current packet since it
      //        would violate the max_packet_size_.
      //
      //    -OR-
      //
      //    (2) There is at least one element already in the current packet,
      //        and the current element says that it must be sent in an
      //        exclusive packet (ie, in a packet all by itself).
      //
      const bool exclusive = element->requires_exclusive_packet();

      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "The element %C require an exclusive packet.\n",
            (exclusive ? "DOES" : "does NOT")
          ));

      const size_t space_needed =
        (max_msg_size > 0)
        ? /* fragmenting */ DataSampleHeader::get_max_serialized_size() + MIN_FRAG
        : /* not fragmenting */ element_length;

      if ((exclusive && (elems_.size() != 0))
          || (current_space_available() < space_needed)) {

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Element won't fit in current packet or requires exclusive"
              " - send current packet (directly) now.\n"));

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "max_header_size_: %d, header_.length_: %d, element_length: %d\n"
              , max_header_size_, header_.length_, element_length));

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Tot possible length: %d, max_len: %d\n"
              , max_header_size_ + header_.length_ + element_length
              , max_size_));
        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "current elem size: %d\n"
              , elems_.size()));

        // Send the current packet, and deal with the current element
        // afterwards.
        // The invocation's relink status should dictate the direct_send's
        // do_relink. We don't want a (relink == false) invocation to end up
        // doing a relink. Think of (relink == false) as a non-blocking call.
        direct_send(relink);

        // Now check to see if we flipped into MODE_QUEUE, which would mean
        // that the direct_send() experienced backpressure, and the
        // packet was only partially sent.  If this has happened, we deal with
        // the current element by placing it on the queue (and then we are done).
        //
        // Otherwise, if the mode_ is still MODE_DIRECT, we can just
        // "drop" through to the next step in the logic where we append the
        // current element to the current packet.
        if (mode_ == MODE_QUEUE) {
          VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                    "We experienced backpressure on that direct send, as "
                    "the mode_ is now MODE_QUEUE or MODE_SUSPEND.  "
                    "Queue elem and leave.\n"), 5);
          queue_.put(element);
          synch_->work_available();

          return;
        }
      }

      // Loop for sending 'element', in fragments if needed
      bool first_pkt = true; // enter the loop 1st time through unconditionally
      for (TransportQueueElement* next_fragment = 0;
           (first_pkt || next_fragment)
           && (mode_ == MODE_DIRECT || mode_ == MODE_TERMINATED);) {
           // We do need to send in MODE_TERMINATED (GRACEFUL_DISCONNECT msg)

        if (next_fragment) {
          element = next_fragment;
          element_length = next_fragment->msg()->total_length();
          header_.first_fragment_ = false;
        }

        header_.last_fragment_ = false;
        if (max_msg_size) { // fragmentation enabled
          const size_t avail = current_space_available();
          if (element_length > avail) {
            VDBG_LVL((LM_TRACE, "(%P|%t) DBG:   Fragmenting %B > %B\n", element_length, avail), 0);
            const TqePair ep = element->fragment(avail);
            if (ep == null_tqe_pair) {
              ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: TransportSendStrategy::send: "
                "Element Fragmentation Failed\n"));
              return;
            }
            element = ep.first;
            element_length = element->msg()->total_length();
            next_fragment = ep.second;
            header_.first_fragment_ = first_pkt;
          } else if (next_fragment) {
            // We are sending the "tail" element of a previous fragment()
            // operation, and this element didn't itself require fragmentation
            header_.last_fragment_ = true;
            next_fragment = 0;
          }
        }
        first_pkt = false;

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Start the 'append elem' to current packet logic.\n"));

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Put element into current packet elems_.\n"));

        // Now that we know the current element should go into the current
        // packet, we can just go ahead and "append" the current element to
        // the current packet.

        // Add the current element to the collection of packet elements.
        elems_.put(element);

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Before, the header_.length_ == [%d].\n",
              header_.length_));

        // Adjust the header_.length_ to account for the length of the element.
        header_.length_ += static_cast<ACE_UINT32>(element_length);
        const size_t message_length = header_.length_;

        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "After adding element's length, the header_.length_ == [%d].\n",
              message_length));

        // The current packet now contains the current element.  We need to
        // check to see if the conditions are such that we should go ahead and
        // attempt to send the packet "directly" now, or if we can just leave
        // and send the current packet later (in another send() call or in a
        // send_stop() call).

        // There a few conditions that will cause us to attempt to send the
        // packet (directly) right now:
        // - Fragmentation was needed
        // - The current packet has the maximum number of samples per packet.
        // - The current packet's total length exceeds the optimum packet size.
        // - The current element (currently part of the packet elems_)
        //   requires an exclusive packet.
        //
        if (next_fragment || (elems_.size() >= max_samples_)
            || (max_header_size_ + message_length > optimum_size_)
            || exclusive) {
          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "Now the current packet looks full - send it (directly).\n"));

          direct_send(relink);

          if (next_fragment && mode_ != MODE_DIRECT) {
            if (mode_ == MODE_QUEUE) {
              queue_.put(next_fragment);
              synch_->work_available();

            } else {
              next_fragment->data_dropped(true /* dropped by transport */);
            }
          } else if (mode_ == MODE_QUEUE) {
            // Background thread handles packets in progress
            synch_->work_available();
          }

          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "Back from the direct_send() attempt.\n"));

          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "And we %C as a result of the direct_send() call.\n",
                ((mode_ == MODE_QUEUE) ? "flipped into MODE_QUEUE"
                                             : "stayed in MODE_DIRECT")));

        } else {
          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "Packet not sent. Send conditions weren't satisfied.\n"));
          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "elems_.size(): %d, max_samples_: %d\n",
                int(elems_.size()), int(max_samples_)));
          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "header_size_: %d, optimum_size_: %d\n",
                int(max_header_size_ + message_length),
                int(optimum_size_)));
          VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                "element_requires_exclusive_packet: %d\n", int(exclusive)));

          if (mode_ == MODE_QUEUE) {
            VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                  "We flipped into MODE_QUEUE.\n"));

          } else {
            VDBG((LM_DEBUG, "(%P|%t) DBG:   "
                  "We stayed in MODE_DIRECT.\n"));
          }
        }
      }
    }
  }

  send_delayed_notifications();
}

send_buffer()

void OpenDDS::DCPS::TransportSendStrategy::send_buffer

(

TransportSendBuffer *

send_buffer

)

Assigns an optional send buffer.

Definition at line 122 of file TransportSendStrategy.cpp.

References OpenDDS::DCPS::TransportSendBuffer::bind(), and send_buffer_.

Referenced by OpenDDS::DCPS::MulticastDataLink::MulticastDataLink(), and OpenDDS::DCPS::MulticastDataLink::~MulticastDataLink().

{
  send_buffer_ = send_buffer;

  if (send_buffer_ != 0) {
    send_buffer_->bind(this);
  }
}

send_bytes()

ACE_INLINE ssize_t OpenDDS::DCPS::TransportSendStrategy::send_bytes ( const iovec  iov[], int  n, int &  bp  )

protectedvirtual

Reimplemented in OpenDDS::DCPS::TcpSendStrategy.

Definition at line 134 of file TransportSendStrategy.inl.

References ACE_INLINE, and send_bytes_i().

Referenced by do_send_packet().

{
  return send_bytes_i(iov, n);
}

send_bytes_i()

virtual ssize_t OpenDDS::DCPS::TransportSendStrategy::send_bytes_i ( const iovec  iov[], int  n  )

protectedpure virtual

Implemented in OpenDDS::DCPS::RtpsUdpSendStrategy, OpenDDS::DCPS::TcpSendStrategy, OpenDDS::DCPS::MulticastSendStrategy, OpenDDS::DCPS::ShmemSendStrategy, and OpenDDS::DCPS::UdpSendStrategy.

Referenced by non_blocking_send(), and send_bytes().

send_delayed_notifications()

bool OpenDDS::DCPS::TransportSendStrategy::send_delayed_notifications ( const TransportQueueElement::MatchCriteria *  match = 0 )

protected

If delayed notifications were queued up, issue those callbacks here. The default match is "match all", otherwise match can be used to specify either a certain individual packet or a publication id. Returns true if anything in the delayed notification list matched.

Definition at line 657 of file TransportSendStrategy.cpp.

References OpenDDS::DCPS::TransportQueueElement::data_delivered(), OpenDDS::DCPS::TransportQueueElement::data_dropped(), DBG_ENTRY_LVL, OpenDDS::DCPS::TransportImpl::is_shut_down(), lock_, OpenDDS::DCPS::TransportQueueElement::MatchCriteria::matches(), mode(), MODE_NOT_SET, MODE_TERMINATED, OPENDDS_VECTOR(), OpenDDS::DCPS::TransportQueueElement::owned_by_transport(), and transport_.

Referenced by clear(), perform_work(), remove_all_msgs(), remove_sample(), send(), and send_stop().

{
  DBG_ENTRY_LVL("TransportSendStrategy","send_delayed_notifications",6);
  TransportQueueElement* sample = 0;
  SendMode mode = MODE_NOT_SET;

  OPENDDS_VECTOR(TQESendModePair) samples;

  size_t num_delayed_notifications = 0;
  bool found_element = false;

  {
    GuardType guard(lock_);

    num_delayed_notifications = delayed_delivered_notification_queue_.size();

    if (num_delayed_notifications == 0) {
      return false;

    } else if (num_delayed_notifications == 1) {
      // Optimization for the most common case (doesn't need vectors)

      if (!match || match->matches(*delayed_delivered_notification_queue_[0].first)) {
        found_element = true;
        sample = delayed_delivered_notification_queue_[0].first;
        mode = delayed_delivered_notification_queue_[0].second;

        delayed_delivered_notification_queue_.clear();
      }

    } else {
      OPENDDS_VECTOR(TQESendModePair)::iterator iter;
      for (iter = delayed_delivered_notification_queue_.begin(); iter != delayed_delivered_notification_queue_.end(); ) {
        sample = iter->first;
        mode = iter->second;
        if (!match || match->matches(*sample)) {
          found_element = true;
          samples.push_back(*iter);
          iter = delayed_delivered_notification_queue_.erase(iter);
        } else {
          ++iter;
        }
      }
    }
  }

  if (!found_element) {
    return false;
  }

  bool transport_shutdown = true;
  TransportImpl_rch transport = transport_.lock();
  if (transport) {
    transport_shutdown = transport->is_shut_down();
  }

  if (num_delayed_notifications == 1) {
    // optimization for the common case
    if (mode == MODE_TERMINATED) {
      if (!transport_shutdown || sample->owned_by_transport()) {
        sample->data_dropped(true);
      }
    } else {
      if (!transport_shutdown || sample->owned_by_transport()) {
        sample->data_delivered();
      }
    }

  } else {
    for (size_t i = 0; i < samples.size(); ++i) {
      if (samples[i].second == MODE_TERMINATED) {
        if (!transport_shutdown || samples[i].first->owned_by_transport()) {
          samples[i].first->data_dropped(true);
        }
      } else {
        if (!transport_shutdown || samples[i].first->owned_by_transport()) {
          samples[i].first->data_delivered();
        }
      }
    }
  }
  return true;
}

send_packet()

TransportSendStrategy::SendPacketOutcome OpenDDS::DCPS::TransportSendStrategy::send_packet ( )

private

This is called to send the current packet. The current packet will either be a "partially sent" packet, or a packet that has just been prepared via a call to prepare_packet().

Definition at line 1807 of file TransportSendStrategy.cpp.

References adjust_packet_after_send(), DBG_ENTRY_LVL, do_send_packet(), elems_, header_, OpenDDS::DCPS::TransportSendBuffer::insert(), LM_DEBUG, OUTCOME_BACKPRESSURE, OUTCOME_COMPLETE_SEND, OUTCOME_PARTIAL_SEND, OUTCOME_PEER_LOST, OUTCOME_SEND_ERROR, pkt_chain_, send_buffer_, OpenDDS::DCPS::TransportHeader::sequence_, VDBG, and VDBG_LVL.

Referenced by direct_send(), and perform_work().

{
  DBG_ENTRY_LVL("TransportSendStrategy", "send_packet", 6);

  int bp_flag = 0;
  const ssize_t num_bytes_sent =
    do_send_packet(pkt_chain_, bp_flag);

  if (num_bytes_sent == 0) {
    VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
              "Since num_bytes_sent == 0, return OUTCOME_PEER_LOST.\n"), 5);
    // This means that the peer has disconnected.
    return OUTCOME_PEER_LOST;
  }

  if (num_bytes_sent < 0) {
    VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
              "Since num_bytes_sent < 0, check the backpressure flag.\n"), 5);

    // Check for backpressure...
    if (bp_flag == 1) {
      VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
                "Since backpressure flag is true, return "
                "OUTCOME_BACKPRESSURE.\n"), 5);
      // Ok.  Not really an error - just backpressure.
      return OUTCOME_BACKPRESSURE;
    }

    VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
              "Since backpressure flag is false, return "
              "OUTCOME_SEND_ERROR.\n"), 5);

    // Not backpressure - it's a real error.
    // Note: moved this to send_bytes so the errno msg could be written.
    //ACE_ERROR((LM_ERROR,
    //           "(%P|%t) ERROR: Call to peer().send() failed with negative "
    //           "return code.\n"));

    return OUTCOME_SEND_ERROR;
  }

  if (send_buffer_ != 0) {
    // If a secondary send buffer is bound, sent samples must
    // be inserted in order to properly maintain the buffer:
    send_buffer_->insert(header_.sequence_,
      &elems_, pkt_chain_);
  }

  VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   "
            "Since num_bytes_sent > 0, adjust the packet to account for "
            "the bytes that did get sent.\n"),5);

  // We sent some bytes - adjust the current packet (elems_ and pkt_chain_)
  // to account for the bytes that have been sent.
  const int result =
    adjust_packet_after_send(num_bytes_sent);

  if (result == 0) {
    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "The adjustment logic says that the complete packet was "
          "sent.  Return OUTCOME_COMPLETE_SEND.\n"));
    return OUTCOME_COMPLETE_SEND;
  }

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "The adjustment logic says that only a part of the packet was "
        "sent. Return OUTCOME_PARTIAL_SEND.\n"));

  return OUTCOME_PARTIAL_SEND;
}

send_start()

ACE_INLINE void OpenDDS::DCPS::TransportSendStrategy::send_start

(

)

Invoked prior to one or more send() invocations from a particular TransportClient.

Definition at line 38 of file TransportSendStrategy.inl.

References ACE_INLINE, DBG_ENTRY_LVL, link_released_, lock_, and start_counter_.

Referenced by OpenDDS::DCPS::DataLink::send_start_i().

{
  DBG_ENTRY_LVL("TransportSendStrategy","send_start",6);

  GuardType guard(this->lock_);

  if (!this->link_released_)
    ++this->start_counter_;
}

send_stop()

void OpenDDS::DCPS::TransportSendStrategy::send_stop

(

GUID_t

repoId

)

Invoked after one or more send() invocations from a particular TransportClient.

Definition at line 1224 of file TransportSendStrategy.cpp.

References DBG_ENTRY_LVL, direct_send(), elems_, graceful_disconnecting_, header_, OpenDDS::DCPS::TransportHeader::length_, link_released_, LM_DEBUG, LM_ERROR, lock_, mode_, mode_as_str(), MODE_QUEUE, MODE_SUSPEND, MODE_TERMINATED, send_delayed_notifications(), OpenDDS::DCPS::BasicQueue< T >::size(), start_counter_, synch_, VDBG, and VDBG_LVL.

Referenced by OpenDDS::DCPS::DataLink::send_stop_i().

{
  DBG_ENTRY_LVL("TransportSendStrategy","send_stop",6);
  {
    GuardType guard(lock_);

    if (link_released_)
      return;

    if (start_counter_ == 0) {
      // This is an indication of a logic error.  This is more of an assert.
      VDBG_LVL((LM_ERROR,
                "(%P|%t) ERROR: Received unexpected send_stop() call.\n"), 5);
      return;
    }

    --start_counter_;

    if (start_counter_ != 0) {
      // This wasn't the last send_stop() that we are expecting.  We only
      // really honor the first send_start() and the last send_stop().
      // We can return without doing anything else in this case.
      return;
    }

    if (mode_ == MODE_TERMINATED && !graceful_disconnecting_) {
      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "TransportSendStrategy::send_stop: dont try to send current packet "
            "since mode is MODE_TERMINATED and not in graceful disconnecting.\n"));
      return;
    }

    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "This is an 'important' send_stop() event since our "
          "start_counter_ is 0.\n"));

    // We just caused the start_counter_ to become zero.  This
    // means that we aren't expecting another send() or send_stop() at any
    // time in the near future (ie, it isn't imminent).

    // If our mode_ is currently MODE_QUEUE or MODE_SUSPEND, then we don't have
    // anything to do here because samples have already been going to the
    // queue.

    // We only need to do something if the mode_ is
    // MODE_DIRECT.  It means that we may have some sample(s) in the
    // current packet that have never been sent.  This is our
    // opportunity to send the current packet directly if this is the case.
    if (mode_ == MODE_QUEUE || mode_ == MODE_SUSPEND) {
      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "But since we are in %C, we don't have to do "
            "anything more in this important send_stop().\n",
            mode_as_str(mode_)));
      // We don't do anything if we are in MODE_QUEUE.  Just leave.
      return;
    }

    size_t header_length = header_.length_;
    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "We are in MODE_DIRECT in an important send_stop() - "
          "header_.length_ == [%d].\n", header_length));

    // Only attempt to send the current packet (directly) if the current
    // packet actually contains something (it could be empty).
    if ((header_length > 0) &&
        //(elems_.size ()+not_yet_pac_q_->size() > 0))
        (elems_.size() > 0)) {
      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "There is something in the current packet - attempt to send "
            "it (directly) now.\n"));
      // If a relink needs to be done for this packet to be sent, do it.
      direct_send(true);
      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "Back from the attempt to send leftover packet directly.\n"));

      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "But we %C as a result.\n",
            ((mode_ == MODE_QUEUE)? "flipped into MODE_QUEUE":
                                          "stayed in MODE_DIRECT" )));
      if (mode_ == MODE_QUEUE  && mode_ != MODE_SUSPEND) {
        VDBG((LM_DEBUG, "(%P|%t) DBG:   "
              "Notify Synch thread of work availability\n"));
        synch_->work_available();
      }
    }
  }

  send_delayed_notifications();
}

set_graceful_disconnecting()

void OpenDDS::DCPS::TransportSendStrategy::set_graceful_disconnecting ( bool  flag )

protected

Set graceful disconnecting flag.

Definition at line 1737 of file TransportSendStrategy.cpp.

References graceful_disconnecting_.

Referenced by OpenDDS::DCPS::TcpSendStrategy::reset().

{
  graceful_disconnecting_ = flag;
}

set_header_source()

ACE_INLINE void OpenDDS::DCPS::TransportSendStrategy::set_header_source ( ACE_INT64  source )

protected

Definition at line 32 of file TransportSendStrategy.inl.

References ACE_INLINE, header_, and OpenDDS::DCPS::TransportHeader::source_.

Referenced by OpenDDS::DCPS::MulticastSendStrategy::prepare_header_i().

{
  header_.source_ = source;
}

space_available()

size_t OpenDDS::DCPS::TransportSendStrategy::space_available ( size_t  already_used = 0 ) const

private

How much space is available in packet with a given used space before we reach one of the limits: max_message_size() [transport's inherent limitation] or max_size_ [user's configured limit]

Definition at line 1951 of file TransportSendStrategy.cpp.

References max_header_size_, max_message_size(), and max_size_.

Referenced by current_space_available(), and fragmentation_helper().

{
  const size_t used = max_header_size_ + already_used;
  const size_t max_msg = max_message_size();
  if (max_msg) {
    return std::min(static_cast<size_t>(max_size_), max_msg) - used;
  }
  return max_size_ - used;
}

start()

int OpenDDS::DCPS::TransportSendStrategy::start

(

void

)

Start the TransportSendStrategy. This happens once, when the DataLink that "owns" this strategy object has established a connection.

Definition at line 831 of file TransportSendStrategy.cpp.

References ACE_ERROR_RETURN, OpenDDS::DCPS::TransportSendBuffer::capacity(), DBG_ENTRY_LVL, header_data_allocator_, header_db_allocator_, header_db_lock_pool_, header_mb_allocator_, LM_ERROR, lock_, max_header_size_, OpenDDS::DCPS::unique_ptr< T, Deleter >::reset(), send_buffer_, start_i(), synch_, and TheServiceParticipant.

Referenced by OpenDDS::DCPS::ShmemSendStrategy::send_bytes_i(), and OpenDDS::DCPS::DataLink::start().

{
  DBG_ENTRY_LVL("TransportSendStrategy","start",6);

  {
    GuardType guard(lock_);

    if (!start_i()) {
      return -1;
    }
  }

  size_t header_chunks(1);

  // If a secondary send buffer is bound, sent headers should
  // be cached to properly maintain the buffer:
  if (send_buffer_ != 0) {
    header_chunks += send_buffer_->capacity();

  } else {
    header_chunks += 1;
  }

  header_db_allocator_.reset( new TransportDataBlockAllocator(header_chunks));
  header_mb_allocator_.reset( new TransportMessageBlockAllocator(header_chunks));
  header_db_lock_pool_.reset(new DataBlockLockPool(static_cast<unsigned long>(TheServiceParticipant->n_chunks())));
  header_data_allocator_.reset(new DataAllocator(TheServiceParticipant->association_chunk_multiplier(), max_header_size_));

  // Since we (the TransportSendStrategy object) are a reference-counted
  // object, but the synch_ object doesn't necessarily know this, we need
  // to give a "copy" of a reference to ourselves to the synch_ object here.
  // We will do the reverse when we unregister ourselves (as a worker) from
  // the synch_ object.

  if (synch_->register_worker(*this) == -1) {

    ACE_ERROR_RETURN((LM_ERROR,
                      "(%P|%t) ERROR: TransportSendStrategy failed to register "
                      "as a worker with the ThreadSynch object.\n"),
                     -1);
  }

  return 0;
}

start_i()

virtual bool OpenDDS::DCPS::TransportSendStrategy::start_i ( )

inlinevirtual

Let the subclass start.

Reimplemented in OpenDDS::DCPS::ShmemSendStrategy.

Definition at line 136 of file TransportSendStrategy.h.

Referenced by start().

{ return true; }

stop()

void OpenDDS::DCPS::TransportSendStrategy::stop

(

void

)

Stop the TransportSendStrategy. This happens once, when the DataLink that "owns" this strategy object is going away.

Definition at line 877 of file TransportSendStrategy.cpp.

References OpenDDS::DCPS::BasicQueue< T >::accept_remove_visitor(), ACE_DEBUG, ACE_TEXT(), DBG_ENTRY_LVL, elems_, header_block_, LM_WARNING, lock_, pkt_chain_, queue_, ACE_Message_Block::release(), OpenDDS::DCPS::BasicQueue< T >::size(), stop_i(), OpenDDS::DCPS::BasicQueue< T >::swap(), synch_, and ACE_Message_Block::total_length().

Referenced by OpenDDS::DCPS::DataLink::start(), and OpenDDS::DCPS::DataLink::stop().

{
  DBG_ENTRY_LVL("TransportSendStrategy","stop",6);

  if (header_block_ != 0) {
    header_block_->release ();
    header_block_ = 0;
  }

  synch_->unregister_worker();

  QueueType elems;
  QueueType queue;
  {
    GuardType guard(lock_);

    if (pkt_chain_ != 0) {
      size_t size = pkt_chain_->total_length();
      if (size > 0) {
        pkt_chain_->release();
        ACE_DEBUG((LM_WARNING,
                   ACE_TEXT("(%P|%t) WARNING: TransportSendStrategy::stop() - ")
                   ACE_TEXT("terminating with %d unsent bytes.\n"),
                   size));
      }
      pkt_chain_ = 0;
    }

    if (elems_.size()) {
      elems_.swap(elems);
      ACE_DEBUG((LM_WARNING,
                 ACE_TEXT("(%P|%t) WARNING: TransportSendStrategy::stop() - ")
                 ACE_TEXT("terminating with %d unsent elements.\n"),
                 elems_.size()));
    }

    if (queue_.size()) {
      queue_.swap(queue);
      ACE_DEBUG((LM_WARNING,
                 ACE_TEXT("(%P|%t) WARNING: TransportSendStrategy::stop() - ")
                 ACE_TEXT("terminating with %d queued elements.\n"),
                 queue_.size()));
    }
  }

  RemoveAllVisitor remove_all_visitor;

  elems.accept_remove_visitor(remove_all_visitor);
  queue.accept_remove_visitor(remove_all_visitor);

  {
    GuardType guard(lock_);

    stop_i();
  }
}

stop_i()

virtual void OpenDDS::DCPS::TransportSendStrategy::stop_i ( )

pure virtual

Let the subclass stop.

Implemented in OpenDDS::DCPS::TcpSendStrategy, OpenDDS::DCPS::RtpsUdpSendStrategy, OpenDDS::DCPS::ShmemSendStrategy, OpenDDS::DCPS::MulticastSendStrategy, and OpenDDS::DCPS::UdpSendStrategy.

Referenced by stop().

suspend_send()

ACE_INLINE void OpenDDS::DCPS::TransportSendStrategy::suspend_send

(

)

This is called when first time reconnect is attempted. The send mode is set to MODE_SUSPEND. Messages are queued at this state.

Definition at line 66 of file TransportSendStrategy.inl.

References ACE_INLINE, DBG_ENTRY_LVL, lock_, mode_, mode_before_suspend_, MODE_SUSPEND, and MODE_TERMINATED.

{
  DBG_ENTRY_LVL("TransportSendStrategy","suspend_send",6);
  GuardType guard(this->lock_);

  if (this->mode_ != MODE_TERMINATED && this->mode_ != MODE_SUSPEND) {
    this->mode_before_suspend_ = this->mode_;
    this->mode_ = MODE_SUSPEND;
  }
}

synch()

ACE_INLINE ThreadSynch * OpenDDS::DCPS::TransportSendStrategy::synch ( ) const

protected

Definition at line 24 of file TransportSendStrategy.inl.

References ACE_INLINE, DBG_ENTRY_LVL, and synch_.

Referenced by OpenDDS::DCPS::TcpSendStrategy::schedule_output().

{
  DBG_ENTRY_LVL("TransportSendStrategy","synch",6);

  return synch_.get();
}

terminate_send()

void OpenDDS::DCPS::TransportSendStrategy::terminate_send

(

bool

graceful_disconnecting = false

)

Remove all samples in the backpressure queue and packet queue.

This is called whenver the connection is lost and reconnect fails. It removes all samples in the backpressure queue and packet queue.

Definition at line 743 of file TransportSendStrategy.cpp.

References clear(), DBG_ENTRY_LVL, graceful_disconnecting_, LM_DEBUG, lock_, mode_, MODE_SUSPEND, MODE_TERMINATED, and VDBG.

Referenced by OpenDDS::DCPS::TcpDataLink::send_graceful_disconnect_message(), and OpenDDS::DCPS::DataLink::terminate_send().

{
  DBG_ENTRY_LVL("TransportSendStrategy","terminate_send",6);

  bool reset_flag = true;

  {
    GuardType guard(lock_);

    // If the terminate_send call due to a non-graceful disconnection before
    // a datalink shutdown then we will not try to send the graceful disconnect
    // message.
    if ((mode_ == MODE_TERMINATED || mode_ == MODE_SUSPEND)
        && !graceful_disconnecting_) {
      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "It was already terminated non gracefully, will not set to graceful disconnecting\n"));
      reset_flag = false;
    }
  }

  VDBG((LM_DEBUG, "(%P|%t) DBG:  Now flip to MODE_TERMINATED\n"));

  clear(MODE_TERMINATED);

  if (reset_flag) {
    GuardType guard(lock_);
    graceful_disconnecting_ = graceful_disconnecting;
  }
}

terminate_send_if_suspended()

void OpenDDS::DCPS::TransportSendStrategy::terminate_send_if_suspended ( )

virtual

Reimplemented in OpenDDS::DCPS::TcpSendStrategy.

Definition at line 774 of file TransportSendStrategy.cpp.

Referenced by OpenDDS::DCPS::DataLink::terminate_send_if_suspended().

{
}

Friends And Related Function Documentation

TransportSendBuffer

friend class TransportSendBuffer

friend

Definition at line 436 of file TransportSendStrategy.h.

Member Data Documentation

elems_

QueueType OpenDDS::DCPS::TransportSendStrategy::elems_

private

Current elements that have contributed blocks to the current transport packet.

Definition at line 368 of file TransportSendStrategy.h.

Referenced by adjust_packet_after_send(), clear(), current_packet_first_element(), do_remove_sample(), get_packet_elems_from_queue(), prepare_packet(), resume_send(), send(), send_packet(), send_stop(), and stop().

graceful_disconnecting_

bool OpenDDS::DCPS::TransportSendStrategy::graceful_disconnecting_

private

Definition at line 427 of file TransportSendStrategy.h.

Referenced by send(), send_stop(), set_graceful_disconnecting(), and terminate_send().

header_

TransportHeader OpenDDS::DCPS::TransportSendStrategy::header_

private

Current transport packet header.

Definition at line 439 of file TransportSendStrategy.h.

Referenced by adjust_packet_after_send(), clear(), current_space_available(), do_remove_sample(), get_packet_elems_from_queue(), marshal_transport_header(), perform_work(), prepare_header(), resume_send(), send(), send_packet(), send_stop(), and set_header_source().

header_block_

ACE_Message_Block* OpenDDS::DCPS::TransportSendStrategy::header_block_

private

Current transport packet header, marshalled.

Definition at line 361 of file TransportSendStrategy.h.

Referenced by do_remove_sample(), prepare_packet(), and stop().

header_complete_

bool OpenDDS::DCPS::TransportSendStrategy::header_complete_

private

Set to false when the packet header hasn't been fully sent. Set to true once the packet header has been fully sent.

Definition at line 376 of file TransportSendStrategy.h.

Referenced by adjust_packet_after_send(), clear(), prepare_packet(), and resume_send().

header_data_allocator_

unique_ptr<DataAllocator> OpenDDS::DCPS::TransportSendStrategy::header_data_allocator_

private

Definition at line 412 of file TransportSendStrategy.h.

Referenced by prepare_packet(), and start().

header_db_allocator_

unique_ptr<TransportDataBlockAllocator> OpenDDS::DCPS::TransportSendStrategy::header_db_allocator_

private

Allocator for header message block.

Definition at line 405 of file TransportSendStrategy.h.

Referenced by prepare_packet(), and start().

header_db_lock_pool_

unique_ptr<DataBlockLockPool> OpenDDS::DCPS::TransportSendStrategy::header_db_lock_pool_

private

DataBlockLockPool.

Definition at line 408 of file TransportSendStrategy.h.

Referenced by prepare_packet(), and start().

header_mb_allocator_

unique_ptr<TransportMessageBlockAllocator> OpenDDS::DCPS::TransportSendStrategy::header_mb_allocator_

private

Allocator for header data block.

Definition at line 402 of file TransportSendStrategy.h.

Referenced by prepare_packet(), and start().

header_sequence_

SequenceNumber OpenDDS::DCPS::TransportSendStrategy::header_sequence_

private

Current transport header sequence number.

Definition at line 364 of file TransportSendStrategy.h.

Referenced by prepare_header().

link_released_

bool OpenDDS::DCPS::TransportSendStrategy::link_released_

private

Definition at line 429 of file TransportSendStrategy.h.

Referenced by link_released(), send(), send_start(), and send_stop().

lock_

LockType OpenDDS::DCPS::TransportSendStrategy::lock_

private

This lock will protect critical sections of code that play a role in the sending of data.

Definition at line 419 of file TransportSendStrategy.h.

Referenced by clear(), deliver_ack_request(), link_released(), OpenDDS::DCPS::RtpsUdpSendStrategy::marshal_transport_header(), perform_work(), remove_all_msgs(), remove_sample(), resume_send(), send(), send_delayed_notifications(), send_start(), send_stop(), start(), stop(), suspend_send(), and terminate_send().

max_header_size_

size_t OpenDDS::DCPS::TransportSendStrategy::max_header_size_

private

Maximum marshalled size of the transport packet header.

Definition at line 358 of file TransportSendStrategy.h.

Referenced by prepare_packet(), send(), space_available(), start(), and TransportSendStrategy().

max_samples_

size_t OpenDDS::DCPS::TransportSendStrategy::max_samples_

private

Configuration - max number of samples per transport packet.

Definition at line 342 of file TransportSendStrategy.h.

Referenced by add_delayed_notification(), get_packet_elems_from_queue(), send(), and TransportSendStrategy().

max_size_

ACE_UINT32 OpenDDS::DCPS::TransportSendStrategy::max_size_

private

Configuration - max transport packet size (bytes)

Definition at line 348 of file TransportSendStrategy.h.

Referenced by send(), space_available(), and TransportSendStrategy().

mode_

Atomic<SendMode> OpenDDS::DCPS::TransportSendStrategy::mode_

private

This mode determines how send() calls will be handled.

Definition at line 390 of file TransportSendStrategy.h.

Referenced by add_delayed_notification(), clear(), direct_send(), do_remove_sample(), isDirectMode(), mode(), perform_work(), resume_send(), send(), send_stop(), suspend_send(), and terminate_send().

mode_before_suspend_

SendMode OpenDDS::DCPS::TransportSendStrategy::mode_before_suspend_

private

This mode remembers the mode before send is suspended and is used after the send is resumed because the connection is re-established.

Definition at line 395 of file TransportSendStrategy.h.

Referenced by clear(), direct_send(), resume_send(), and suspend_send().

optimum_size_

ACE_UINT32 OpenDDS::DCPS::TransportSendStrategy::optimum_size_

private

Configuration - optimum transport packet size (bytes)

Definition at line 345 of file TransportSendStrategy.h.

Referenced by get_packet_elems_from_queue(), send(), and TransportSendStrategy().

pkt_chain_

ACE_Message_Block* OpenDDS::DCPS::TransportSendStrategy::pkt_chain_

private

Current (head of chain) block containing unsent bytes for the current transport packet.

Definition at line 372 of file TransportSendStrategy.h.

Referenced by adjust_packet_after_send(), clear(), do_remove_sample(), prepare_packet(), resume_send(), send_packet(), and stop().

queue_

QueueType OpenDDS::DCPS::TransportSendStrategy::queue_

private

Used during backpressure situations to hold samples that have not yet been made to be part of a transport packet, and are completely unsent. Also used as a bucket for packets which still have to become part of a packet.

Definition at line 355 of file TransportSendStrategy.h.

Referenced by clear(), do_remove_sample(), get_packet_elems_from_queue(), perform_work(), resume_send(), send(), and stop().

replaced_element_db_allocator_

DataBlockAllocator OpenDDS::DCPS::TransportSendStrategy::replaced_element_db_allocator_

private

Definition at line 423 of file TransportSendStrategy.h.

Referenced by do_remove_sample().

replaced_element_mb_allocator_

MessageBlockAllocator OpenDDS::DCPS::TransportSendStrategy::replaced_element_mb_allocator_

private

Cached allocator for TransportReplaceElement.

Definition at line 422 of file TransportSendStrategy.h.

Referenced by do_remove_sample().

send_buffer_

TransportSendBuffer* OpenDDS::DCPS::TransportSendStrategy::send_buffer_

private

Definition at line 431 of file TransportSendStrategy.h.

Referenced by remove_all_msgs(), send_buffer(), send_packet(), and start().

start_counter_

unsigned OpenDDS::DCPS::TransportSendStrategy::start_counter_

private

Counter that, when greater than zero, indicates that we still expect to receive a send_stop() event. Incremented once for each call to our send_start() method, and decremented once for each call to our send_stop() method. We only care about the transitions of the start_counter_ value from 0 to 1, and from 1 to 0. This accommodates the case where more than one TransportClient is sending to us at the same time. We use this counter to enable a "composite" send_start() and send_stop().

Definition at line 387 of file TransportSendStrategy.h.

Referenced by clear(), resume_send(), send_start(), and send_stop().

synch_

unique_ptr<ThreadSynch> OpenDDS::DCPS::TransportSendStrategy::synch_

private

The thread synch object.

Definition at line 415 of file TransportSendStrategy.h.

Referenced by perform_work(), resume_send(), send(), send_stop(), start(), stop(), synch(), and TransportSendStrategy().

transport_

WeakRcHandle<TransportImpl> OpenDDS::DCPS::TransportSendStrategy::transport_

private

Definition at line 425 of file TransportSendStrategy.h.

Referenced by direct_send(), and send_delayed_notifications().

UDP_MAX_MESSAGE_SIZE

const size_t OpenDDS::DCPS::TransportSendStrategy::UDP_MAX_MESSAGE_SIZE = 65466

static

Put the maximum UDP payload size here so that it can be shared by all UDP-based transports. This is the worst-case (conservative) value for UDP/IPv4. If there are no IP options, or if IPv6 is used, it could actually be a little larger.

Definition at line 159 of file TransportSendStrategy.h.

Referenced by OpenDDS::DCPS::RtpsUdpSendStrategy::send_single_i().

---

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

• TransportSendStrategy.h
• TransportSendStrategy.cpp
• TransportSendStrategy.inl