#include <ShmemReceiveStrategy.h>

Inheritance diagram for OpenDDS::DCPS::ShmemReceiveStrategy:

[legend]

Collaboration diagram for OpenDDS::DCPS::ShmemReceiveStrategy:

[legend]

Public Member Functions
	ShmemReceiveStrategy (ShmemDataLink *link)

void	read ()

Public Member Functions inherited from OpenDDS::DCPS::TransportReceiveStrategy<>
virtual	~TransportReceiveStrategy ()

int	start ()

void	stop ()

int	handle_dds_input (ACE_HANDLE fd)

virtual void	relink (bool do_suspend=true)

const TransportHeader &	received_header () const

TransportHeader &	received_header ()

const DataSampleHeader &	received_sample_header () const

DataSampleHeader &	received_sample_header ()

ACE_Message_Block *	to_msgblock (const ReceivedDataSample &sample)

Public Member Functions inherited from OpenDDS::DCPS::TransportStrategy
virtual	~TransportStrategy ()

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	receive_bytes (iovec iov[], int n, ACE_INET_Addr &remote_address, ACE_HANDLE fd, bool &stop)
	Only our subclass knows how to do this. More...

virtual void	deliver_sample (ReceivedDataSample &sample, const ACE_INET_Addr &remote_address)
	Called when there is a ReceivedDataSample to be delivered. More...

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

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

Protected Member Functions inherited from OpenDDS::DCPS::TransportReceiveStrategy<>
	TransportReceiveStrategy (const TransportInst_rch &config, size_t receive_buffers_count=RECEIVE_BUFFERS)

virtual bool	check_header (const TransportHeader &header)
	Check the transport header for suitability. More...

virtual bool	check_header (const DataSampleHeader &header)
	Check the data sample header for suitability. More...

virtual void	begin_transport_header_processing ()
	Begin Current Transport Header Processing. More...

virtual void	end_transport_header_processing ()
	End Current Transport Header Processing. More...

virtual void	finish_message ()

int	skip_bad_pdus ()
	Ignore bad PDUs by skipping over them. More...

void	reset ()

size_t	pdu_remaining () const

size_t	successor_index (size_t index) const
	Manage an index into the receive buffer array. More...

void	update_buffer_index (bool &done)

virtual bool	reassemble (ReceivedDataSample &data)

	OPENDDS_VECTOR (ACE_Message_Block *) receive_buffers_
	Set of receive buffers in use. More...

Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()

Private Attributes
ShmemDataLink *	link_

std::string	bound_name_

ShmemData *	current_data_

size_t	partial_recv_remaining_

const char *	partial_recv_ptr_

ACE_Thread_Mutex	mutex_

Additional Inherited Members
Static Public Attributes inherited from OpenDDS::DCPS::TransportReceiveConstants
static const size_t	RECEIVE_BUFFERS = DEFAULT_TRANSPORT_RECEIVE_BUFFERS

static const size_t	BUFFER_LOW_WATER = 4096

static const size_t	MESSAGE_BLOCKS = 1000

static const size_t	DATA_BLOCKS = 100

Protected Attributes inherited from OpenDDS::DCPS::TransportReceiveStrategy<>
bool	gracefully_disconnected_
	Flag indicates if the GRACEFUL_DISCONNECT message is received. More...

size_t	receive_sample_remaining_
	Bytes remaining in the current DataSample. More...

TransportHeader	receive_transport_header_
	Current receive TransportHeader. More...

TransportMessageBlockAllocator	mb_allocator_

TransportDataBlockAllocator	db_allocator_

TransportDataAllocator	data_allocator_

ACE_Lock_Adapter< ACE_SYNCH_MUTEX >	receive_lock_
	Locking strategy for the allocators. More...

size_t	buffer_index_
	Current receive buffer index in use. More...

DataSampleHeader	data_sample_header_
	Current data sample header. More...

ACE_Message_Block *	payload_

bool	good_pdu_

size_t	pdu_remaining_
	Amount of the current PDU that has not been processed yet. More...

Detailed Description

Definition at line 25 of file ShmemReceiveStrategy.h.

Constructor & Destructor Documentation

◆ ShmemReceiveStrategy()

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

explicit

Definition at line 23 of file ShmemReceiveStrategy.cpp.

   : TransportReceiveStrategy<>(link->config())
   , link_(link)
   , current_data_(0)
   , partial_recv_remaining_(0)
   , partial_recv_ptr_(0)
 {
 }

Member Function Documentation

◆ deliver_sample()

void OpenDDS::DCPS::ShmemReceiveStrategy::deliver_sample	(	ReceivedDataSample &	sample,
		const ACE_INET_Addr &	remote_address
	)

protectedvirtual

Called when there is a ReceivedDataSample to be delivered.

Implements OpenDDS::DCPS::TransportReceiveStrategy<>.

Definition at line 181 of file ShmemReceiveStrategy.cpp.

References OpenDDS::DCPS::ShmemDataLink::control_received(), OpenDDS::DCPS::DataLink::data_received(), OpenDDS::DCPS::ReceivedDataSample::header_, link_, OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::REQUEST_ACK, OpenDDS::DCPS::ShmemDataLink::request_ack_received(), and OpenDDS::DCPS::TRANSPORT_CONTROL.

 {
   switch (sample.header_.message_id_) {
 
   case REQUEST_ACK:
     link_->request_ack_received(sample);
     break;
 
   case TRANSPORT_CONTROL:
     link_->control_received(sample);
     break;
 
   default:
     link_->data_received(sample);
   }
 }

◆ read()

void OpenDDS::DCPS::ShmemReceiveStrategy::read ( void )

Definition at line 33 of file ShmemReceiveStrategy.cpp.

References bound_name_, current_data_, OpenDDS::DCPS::ShmemData::EndOfAlloc, ACE_Malloc_T< class, ACE_LOCK, ACE_CB >::find(), OpenDDS::DCPS::ShmemData::Free, OpenDDS::DCPS::TransportReceiveStrategy<>::handle_dds_input(), link_, LM_DEBUG, OpenDDS::DCPS::ShmemDataLink::local_address(), partial_recv_remaining_, OpenDDS::DCPS::ShmemDataLink::peer_allocator(), OpenDDS::DCPS::ShmemData::RecvDone, OpenDDS::DCPS::TransportReceiveStrategy<>::start(), OpenDDS::DCPS::ShmemData::status_, VDBG, and VDBG_LVL.

 {
   if (partial_recv_remaining_) {
     VDBG((LM_DEBUG, "(%P|%t) ShmemReceiveStrategy::read link %@ "
           "resuming partial recv\n", link_));
     handle_dds_input(ACE_INVALID_HANDLE);
     return;
   }
 
   if (bound_name_.empty()) {
     bound_name_ = "Write-" + link_->local_address();
   }
 
   ShmemAllocator* alloc = link_->peer_allocator();
   void* mem = 0;
   if (alloc == 0 || -1 == alloc->find(bound_name_.c_str(), mem)) {
     VDBG_LVL((LM_DEBUG, "(%P|%t) ShmemReceiveStrategy::read link %@ "
               "peer allocator not found, receive_bytes will close link\n",
               link_), 1);
     handle_dds_input(ACE_INVALID_HANDLE); // will return 0 to the TRecvStrateg.
     return;
   }
 
   if (!current_data_) {
     current_data_ = reinterpret_cast<ShmemData*>(mem);
   }
 
   for (ShmemData* start = 0; current_data_->status_ == ShmemData::Free ||
          current_data_->status_ == ShmemData::RecvDone; ++current_data_) {
     if (!start) {
       start = current_data_;
     } else if (start == current_data_) {
       return; // none found => don't call handle_dds_input()
     }
     if (current_data_[1].status_ == ShmemData::EndOfAlloc) {
       current_data_ = reinterpret_cast<ShmemData*>(mem) - 1; // incremented by the for loop
     }
   }
 
   VDBG((LM_DEBUG, "(%P|%t) ShmemReceiveStrategy::read link %@ "
         "reading at control block #%d\n",
         link_, current_data_ - reinterpret_cast<ShmemData*>(mem)));
   // If we get this far, current_data_ points to the first ShmemData::DataInUse.
   // handle_dds_input() will call our receive_bytes() to get the data.
   handle_dds_input(ACE_INVALID_HANDLE);
 }

◆ receive_bytes()

ssize_t OpenDDS::DCPS::ShmemReceiveStrategy::receive_bytes	(	iovec	iov[],
		int	n,
		ACE_INET_Addr &	remote_address,
		ACE_HANDLE	fd,
		bool &	stop
	)

protectedvirtual

Only our subclass knows how to do this.

Implements OpenDDS::DCPS::TransportReceiveStrategy<>.

Definition at line 81 of file ShmemReceiveStrategy.cpp.

References bound_name_, current_data_, ENOBUFS, ACE_Malloc_T< class, ACE_LOCK, ACE_CB >::find(), OpenDDS::DCPS::TransportHeader::get_length(), OpenDDS::DCPS::TransportReceiveStrategy<>::gracefully_disconnected_, OpenDDS::DCPS::ShmemData::InUse, link_, LM_DEBUG, LM_ERROR, ACE_Malloc_T< class, ACE_LOCK, ACE_CB >::memory_pool(), partial_recv_ptr_, partial_recv_remaining_, OpenDDS::DCPS::ShmemData::payload_, OpenDDS::DCPS::ShmemDataLink::peer_allocator(), OpenDDS::DCPS::ShmemData::RecvDone, OpenDDS::DCPS::ShmemDataLink::signal_semaphore(), OpenDDS::DCPS::ShmemData::status_, OpenDDS::DCPS::ShmemData::transport_header_, VDBG, and VDBG_LVL.

 {
   VDBG((LM_DEBUG,
         "(%P|%t) ShmemReceiveStrategy::receive_bytes link %@\n", link_));
 
   // check that the writer's shared memory is still available
   ShmemAllocator* alloc = link_->peer_allocator();
   void* mem;
   if (!alloc || -1 == alloc->find(bound_name_.c_str(), mem) || !current_data_
       || current_data_->status_ != ShmemData::InUse) {
     VDBG_LVL((LM_DEBUG, "(%P|%t) ShmemReceiveStrategy::receive_bytes closing\n"),
              1);
     gracefully_disconnected_ = true; // do not attempt reconnect via relink()
     return 0; // close "connection"
   }
 
   ssize_t total = 0;
 
   const char* src_iter;
   char* dst_iter = 0;
   int i = 0; // current iovec index in iov[]
   size_t remaining;
 
   if (partial_recv_remaining_) {
     remaining = partial_recv_remaining_;
     src_iter = partial_recv_ptr_;
     // iov_base is void* on POSIX but char* on Win32, we'll have to cast:
     dst_iter = (char*)iov[0].iov_base;
 
   } else {
     remaining = TransportHeader::get_length(current_data_->transport_header_);
     const size_t hdr_sz = sizeof(current_data_->transport_header_);
     // BUFFER_LOW_WATER in the framework ensures a large enough buffer
     if (static_cast<size_t>(iov[0].iov_len) <= hdr_sz) {
       VDBG_LVL((LM_ERROR, "(%P|%t) ERROR: ShmemReceiveStrategy::receive_bytes "
                 "receive buffer of length %d is too small\n",
                 iov[0].iov_len), 0);
       errno = ENOBUFS;
       return -1;
     }
 
     VDBG((LM_DEBUG, "(%P|%t) ShmemReceiveStrategy::receive_bytes "
           "header %@ payload %@ len %B\n", current_data_->transport_header_,
           (char*)current_data_->payload_, remaining));
     std::memcpy(iov[0].iov_base, current_data_->transport_header_, hdr_sz);
     total += hdr_sz;
     src_iter = current_data_->payload_;
     if (static_cast<size_t>(iov[0].iov_len) > hdr_sz) {
       dst_iter = (char*)iov[0].iov_base + hdr_sz;
     } else if (n > 1) {
       dst_iter = (char*)iov[1].iov_base;
       i = 1;
     }
   }
 
   for (; i < n && remaining; ++i) {
     const size_t space = (i == 0) ? iov[i].iov_len - total : iov[i].iov_len,
       chunk = std::min(space, remaining);
 
 #ifdef OPENDDS_SHMEM_WINDOWS
     if (alloc->memory_pool().remap((void*)(src_iter + chunk - 1)) == -1) {
       VDBG_LVL((LM_ERROR, "(%P|%t) ERROR: ShmemReceiveStrategy::receive_bytes "
                 "shared memory pool couldn't be extended\n"), 0);
       errno = ENOMEM;
       return -1;
     }
 #endif
 
     std::memcpy(dst_iter, src_iter, chunk);
     if (i < n - 1) {
       dst_iter = (char*)iov[i + 1].iov_base;
     }
     remaining -= chunk;
     total += chunk;
     src_iter += chunk;
   }
 
   if (remaining) {
     partial_recv_remaining_ = remaining;
     partial_recv_ptr_ = src_iter;
     VDBG((LM_DEBUG, "(%P|%t) ShmemReceiveStrategy::receive_bytes "
           "receive was partial\n"));
     link_->signal_semaphore();
 
   } else {
     partial_recv_remaining_ = 0;
     partial_recv_ptr_ = 0;
     VDBG((LM_DEBUG, "(%P|%t) ShmemReceiveStrategy::receive_bytes "
           "receive done\n"));
     current_data_->status_ = ShmemData::RecvDone;
   }
 
   return total;
 }

◆ start_i()

int OpenDDS::DCPS::ShmemReceiveStrategy::start_i ( )

protectedvirtual

Let the subclass start.

Implements OpenDDS::DCPS::TransportReceiveStrategy<>.

Definition at line 200 of file ShmemReceiveStrategy.cpp.

 {
   return 0;
 }

◆ stop_i()

void OpenDDS::DCPS::ShmemReceiveStrategy::stop_i ( )

protectedvirtual

Let the subclass stop.

Implements OpenDDS::DCPS::TransportReceiveStrategy<>.

Definition at line 206 of file ShmemReceiveStrategy.cpp.

References OPENDDS_END_VERSIONED_NAMESPACE_DECL.

207 {

208 }

Member Data Documentation

◆ bound_name_

std::string OpenDDS::DCPS::ShmemReceiveStrategy::bound_name_

private

Definition at line 47 of file ShmemReceiveStrategy.h.

Referenced by read(), and receive_bytes().

◆ current_data_

ShmemData* OpenDDS::DCPS::ShmemReceiveStrategy::current_data_

private

Definition at line 48 of file ShmemReceiveStrategy.h.

Referenced by read(), and receive_bytes().

◆ link_

ShmemDataLink* OpenDDS::DCPS::ShmemReceiveStrategy::link_

private

Definition at line 46 of file ShmemReceiveStrategy.h.

Referenced by deliver_sample(), read(), and receive_bytes().

◆ mutex_

ACE_Thread_Mutex OpenDDS::DCPS::ShmemReceiveStrategy::mutex_

private

Definition at line 51 of file ShmemReceiveStrategy.h.

◆ partial_recv_ptr_

const char* OpenDDS::DCPS::ShmemReceiveStrategy::partial_recv_ptr_

private

Definition at line 50 of file ShmemReceiveStrategy.h.

Referenced by receive_bytes().

◆ partial_recv_remaining_

size_t OpenDDS::DCPS::ShmemReceiveStrategy::partial_recv_remaining_

private

Definition at line 49 of file ShmemReceiveStrategy.h.

Referenced by read(), and receive_bytes().

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

◆ ShmemReceiveStrategy()

Member Function Documentation

◆ deliver_sample()

◆ read()

◆ receive_bytes()

◆ start_i()

◆ stop_i()

Member Data Documentation

◆ bound_name_

◆ current_data_

◆ link_

◆ mutex_

◆ partial_recv_ptr_

◆ partial_recv_remaining_