#include <ShmemSendStrategy.h>

Inheritance diagram for OpenDDS::DCPS::ShmemSendStrategy:

[legend]

Collaboration diagram for OpenDDS::DCPS::ShmemSendStrategy:

[legend]

Public Member Functions
	ShmemSendStrategy (ShmemDataLink *link)

virtual bool	start_i ()
	Let the subclass start. More...

virtual void	stop_i ()
	Let the subclass stop. More...

Public Member Functions inherited from OpenDDS::DCPS::TransportSendStrategy
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 ()

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

Protected Member Functions
virtual ssize_t	send_bytes_i (const iovec iov[], int n)

Protected Member Functions inherited from OpenDDS::DCPS::TransportSendStrategy
	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 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 Attributes
ShmemDataLink *	link_

std::string	bound_name_

ACE_sema_t	peer_semaphore_

ShmemData *	current_data_

const size_t	datalink_control_size_

Additional Inherited Members
Public Types inherited from OpenDDS::DCPS::TransportSendStrategy
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 }

Static Public Member Functions inherited from OpenDDS::DCPS::TransportSendStrategy
static int	mb_to_iov (const ACE_Message_Block &msg, iovec *iov)

Static Public Attributes inherited from OpenDDS::DCPS::TransportSendStrategy
static const size_t	UDP_MAX_MESSAGE_SIZE = 65466

Detailed Description

Definition at line 29 of file ShmemSendStrategy.h.

Constructor & Destructor Documentation

◆ ShmemSendStrategy()

OpenDDS::DCPS::ShmemSendStrategy::ShmemSendStrategy ( ShmemDataLink * link )

Definition at line 21 of file ShmemSendStrategy.cpp.

References memset(), and peer_semaphore_.

   : TransportSendStrategy(0, link->impl(),
                           0,  // synch_resource
                           link->transport_priority(),
                           make_rch<NullSynchStrategy>())
   , link_(link)
   , current_data_(0)
   , datalink_control_size_(link->config()->datalink_control_size_)
 {
 #ifdef OPENDDS_SHMEM_UNIX
   memset(&peer_semaphore_, 0, sizeof(peer_semaphore_));
 #endif
 }

Member Function Documentation

◆ send_bytes_i()

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

protectedvirtual

Implements OpenDDS::DCPS::TransportSendStrategy.

Definition at line 77 of file ShmemSendStrategy.cpp.

 {
   const size_t hdr_sz = sizeof(current_data_->transport_header_);
   if (static_cast<size_t>(iov[0].iov_len) != hdr_sz) {
     VDBG_LVL((LM_ERROR, "(%P|%t) ERROR: ShmemSendStrategy for link %@ "
               "expecting iov[0] of size %B, got %B\n",
               link_, hdr_sz, iov[0].iov_len), 0);
     return -1;
   }
   // see TransportHeader::valid(), but this is for the marshaled form
   if (std::memcmp(&TransportHeader::DCPS_PROTOCOL[0], iov[0].iov_base,
                   sizeof(TransportHeader::DCPS_PROTOCOL)) != 0) {
     VDBG_LVL((LM_ERROR, "(%P|%t) ERROR: ShmemSendStrategy for link %@ "
               "expecting iov[0] to contain the transport header\n", link_), 0);
     return -1;
   }
 
   //FUTURE: use the ShmemTransport object to see if we already have the
   //        same payload data available in the pool (from other DataLinks),
   //        and if so, add a refcount to the start of the "from_pool" allocation
 
   size_t pool_alloc_size = 0;
   for (int i = 1 /* skip TransportHeader in [0] */; i < n; ++i) {
     pool_alloc_size += iov[i].iov_len;
   }
 
   ShmemAllocator* alloc = link_->local_allocator();
   void* from_pool = 0;
   if (alloc == 0 || (from_pool = alloc->malloc(pool_alloc_size)) == 0) {
     VDBG_LVL((LM_ERROR, "(%P|%t) ERROR: ShmemSendStrategy for link %@ failed "
               "to allocate %B bytes for data\n", link_, pool_alloc_size), 0);
     errno = ENOMEM;
     return -1;
   }
 
   char* payload = reinterpret_cast<char*>(from_pool);
   char* iter = payload;
   for (int i = 1 /* skip TransportHeader in [0] */; i < n; ++i) {
     std::memcpy(iter, iov[i].iov_base, iov[i].iov_len);
     iter += iov[i].iov_len;
   }
 
   void* mem = 0;
   if (-1 == alloc->find(bound_name_.c_str(), mem) || mem == 0) {
     VDBG_LVL((LM_ERROR, "(%P|%t) ERROR: ShmemSendStrategy for link %@ failed "
               "to find control segment with bound name %C\n", link_, bound_name_.c_str()), 0);
     errno = ENOENT;
     return -1;
   }
 
   for (ShmemData* iter = reinterpret_cast<ShmemData*>(mem);
        iter->status_ != ShmemData::EndOfAlloc; ++iter) {
     if (iter->status_ == ShmemData::RecvDone) {
       alloc->free(iter->payload_);
       // This will eventually be refcounted so instead of a free(), the previous
       // statement would decrement the refcount and check for 0 before free().
       // See the 'FUTURE' comment above.
       iter->status_ = ShmemData::Free;
       VDBG_LVL((LM_DEBUG, "(%P|%t) ShmemSendStrategy for link %@ "
                 "releasing control block #%d\n", link_,
                 iter - reinterpret_cast<ShmemData*>(mem)), 5);
     }
   }
 
   if (!current_data_) {
     current_data_ = reinterpret_cast<ShmemData*>(mem);
   }
 
   for (ShmemData* start = 0; current_data_->status_ == ShmemData::InUse ||
          current_data_->status_ == ShmemData::RecvDone; ++current_data_) {
     if (!start) {
       start = current_data_;
     } else if (start == current_data_) {
       VDBG_LVL((LM_ERROR, "(%P|%t) ERROR: ShmemSendStrategy for link %@ out of "
                 "space for control\n", link_), 0);
       return -1;
     }
     if (current_data_[1].status_ == ShmemData::EndOfAlloc) {
       current_data_ = reinterpret_cast<ShmemData*>(mem) - 1; // incremented by the for loop
     }
   }
 
   if (current_data_->status_ == ShmemData::Free) {
     VDBG((LM_DEBUG, "(%P|%t) ShmemSendStrategy for link %@ "
           "writing at control block #%d header %@ payload %@ len %B\n",
           link_, current_data_ - reinterpret_cast<ShmemData*>(mem),
           current_data_->transport_header_, payload, pool_alloc_size));
     std::memcpy(current_data_->transport_header_, iov[0].iov_base,
                 sizeof(current_data_->transport_header_));
     current_data_->payload_ = payload;
     current_data_->status_ = ShmemData::InUse;
   } else {
     VDBG_LVL((LM_ERROR, "(%P|%t) ERROR: ShmemSendStrategy for link %@ "
               "failed to find space for control\n", link_), 0);
     return -1;
   }
 
   ACE_OS::sema_post(&peer_semaphore_);
 
   return pool_alloc_size + iov[0].iov_len;
 }

◆ start_i()

bool OpenDDS::DCPS::ShmemSendStrategy::start_i ( )

virtual

Let the subclass start.

Reimplemented from OpenDDS::DCPS::TransportSendStrategy.

Definition at line 36 of file ShmemSendStrategy.cpp.

References ACE_Malloc_T< class, ACE_LOCK, ACE_CB >::bind(), bound_name_, ACE_Malloc_T< class, ACE_LOCK, ACE_CB >::calloc(), datalink_control_size_, OpenDDS::DCPS::ShmemData::EndOfAlloc, ACE_Malloc_T< class, ACE_LOCK, ACE_CB >::find(), link_, LM_ERROR, OpenDDS::DCPS::ShmemDataLink::local_allocator(), OpenDDS::DCPS::ShmemDataLink::peer_address(), OpenDDS::DCPS::ShmemDataLink::peer_allocator(), OpenDDS::DCPS::ShmemDataLink::peer_pid(), peer_semaphore_, OpenDDS::DCPS::ShmemData::status_, and VDBG_LVL.

 {
   bound_name_ = "Write-" + link_->peer_address();
   ShmemAllocator* alloc = link_->local_allocator();
 
   const size_t n_elems = datalink_control_size_ / sizeof(ShmemData),
     extra = datalink_control_size_ % sizeof(ShmemData);
 
   void* mem = 0;
   if (alloc == 0 || (mem = alloc->calloc(datalink_control_size_)) == 0) {
     VDBG_LVL((LM_ERROR, "(%P|%t) ERROR: ShmemSendStrategy for link %@ failed "
               "to allocate %B bytes for control\n", link_, datalink_control_size_), 0);
     return false;
   }
 
   ShmemData* data = reinterpret_cast<ShmemData*>(mem);
   const size_t limit = (extra >= sizeof(int)) ? n_elems : (n_elems - 1);
   data[limit].status_ = ShmemData::EndOfAlloc;
   alloc->bind(bound_name_.c_str(), mem);
 
   ShmemAllocator* peer = link_->peer_allocator();
   peer->find("Semaphore", mem);
   ShmemSharedSemaphore* sem = reinterpret_cast<ShmemSharedSemaphore*>(mem);
 #if defined OPENDDS_SHMEM_WINDOWS
   HANDLE srcProc = ::OpenProcess(PROCESS_DUP_HANDLE, false /*bInheritHandle*/,
                                  link_->peer_pid());
   ::DuplicateHandle(srcProc, *sem, GetCurrentProcess(), &peer_semaphore_,
                     0 /*dwDesiredAccess -- ignored*/,
                     false /*bInheritHandle*/,
                     DUPLICATE_SAME_ACCESS /*dwOptions*/);
   ::CloseHandle(srcProc);
 #elif defined OPENDDS_SHMEM_UNIX
   peer_semaphore_.sema_ = sem;
   peer_semaphore_.name_ = 0;
 #else
   ACE_UNUSED_ARG(sem);
 #endif
   return true;
 }

◆ stop_i()

void OpenDDS::DCPS::ShmemSendStrategy::stop_i ( )

virtual

Let the subclass stop.

Implements OpenDDS::DCPS::TransportSendStrategy.

Definition at line 180 of file ShmemSendStrategy.cpp.

References OPENDDS_END_VERSIONED_NAMESPACE_DECL, and peer_semaphore_.

 {
 #ifdef OPENDDS_SHMEM_WINDOWS
   ::CloseHandle(peer_semaphore_);
 #endif
 }

Member Data Documentation

◆ bound_name_

std::string OpenDDS::DCPS::ShmemSendStrategy::bound_name_

private

Definition at line 42 of file ShmemSendStrategy.h.

Referenced by send_bytes_i(), and start_i().

◆ current_data_

ShmemData* OpenDDS::DCPS::ShmemSendStrategy::current_data_

private

Definition at line 44 of file ShmemSendStrategy.h.

Referenced by send_bytes_i().

◆ datalink_control_size_

const size_t OpenDDS::DCPS::ShmemSendStrategy::datalink_control_size_

private

Definition at line 45 of file ShmemSendStrategy.h.

Referenced by start_i().

◆ link_

ShmemDataLink* OpenDDS::DCPS::ShmemSendStrategy::link_

private

Definition at line 41 of file ShmemSendStrategy.h.

Referenced by send_bytes_i(), and start_i().

◆ peer_semaphore_

ACE_sema_t OpenDDS::DCPS::ShmemSendStrategy::peer_semaphore_

private

Definition at line 43 of file ShmemSendStrategy.h.

Referenced by send_bytes_i(), ShmemSendStrategy(), start_i(), and stop_i().

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

Public Member Functions

Protected Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ ShmemSendStrategy()

Member Function Documentation

◆ send_bytes_i()

◆ start_i()

◆ stop_i()

Member Data Documentation

◆ bound_name_

◆ current_data_

◆ datalink_control_size_

◆ link_

◆ peer_semaphore_