A container for instances sample data. More...

#include <WriteDataContainer.h>

Inheritance diagram for OpenDDS::DCPS::WriteDataContainer:

Collaboration diagram for OpenDDS::DCPS::WriteDataContainer:

[legend]

Public Member Functions
	WriteDataContainer (DataWriterImpl writer, CORBA::Long max_samples_per_instance, CORBA::Long history_depth, CORBA::Long max_durable_per_instance, DDS::Duration_t max_blocking_time, size_t n_chunks, DDS::DomainId_t domain_id, char const topic_name, char const type_name, DataDurabilityCache durability_cache, DDS::DurabilityServiceQosPolicy const &durability_service, CORBA::Long max_instances, CORBA::Long max_total_samples, ACE_Recursive_Thread_Mutex &deadline_status_lock, DDS::OfferedDeadlineMissedStatus &deadline_status, CORBA::Long &deadline_last_total_count)

	~WriteDataContainer ()

DDS::ReturnCode_t	enqueue_control (DataSampleElement *control_sample)

DDS::ReturnCode_t	enqueue (DataSampleElement *sample, DDS::InstanceHandle_t instance)

DDS::ReturnCode_t	reenqueue_all (const GUID_t &reader_id, const DDS::LifespanQosPolicy &lifespan, const OPENDDS_STRING &filterClassName, const FilterEvaluator *eval, const DDS::StringSeq &params)

DDS::ReturnCode_t	register_instance (DDS::InstanceHandle_t &instance_handle, Message_Block_Ptr &registered_sample)

DDS::ReturnCode_t	unregister (DDS::InstanceHandle_t handle, Message_Block_Ptr &registered_sample, bool dup_registered_sample=true)

DDS::ReturnCode_t	dispose (DDS::InstanceHandle_t handle, Message_Block_Ptr &registered_sample, bool dup_registered_sample=true)

DDS::ReturnCode_t	num_samples (DDS::InstanceHandle_t handle, size_t &size)

size_t	num_all_samples ()

ACE_UINT64	get_unsent_data (SendStateDataSampleList &list)

SendStateDataSampleList	get_resend_data ()

void	data_delivered (const DataSampleElement *sample)

void	data_dropped (const DataSampleElement *element, bool dropped_by_transport)

DDS::ReturnCode_t	obtain_buffer_for_control (DataSampleElement *&element)

DDS::ReturnCode_t	obtain_buffer (DataSampleElement *&element, DDS::InstanceHandle_t handle)

void	release_buffer (DataSampleElement *element)

void	unregister_all ()

PublicationInstance_rch	get_handle_instance (DDS::InstanceHandle_t handle)

bool	persist_data ()

void	wait_pending (const MonotonicTimePoint &deadline)

typedef	OPENDDS_VECTOR (DDS::InstanceHandle_t) InstanceHandleVec

void	get_instance_handles (InstanceHandleVec &instance_handles)

DDS::ReturnCode_t	wait_ack_of_seq (const MonotonicTimePoint &abs_deadline, bool deadline_is_infinite, const SequenceNumber &sequence)

bool	sequence_acknowledged (const SequenceNumber &sequence)

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

Private Types
typedef ConditionVariable< ACE_Recursive_Thread_Mutex >	ConditionVariableType

typedef ConditionVariable< ACE_Thread_Mutex >	WfaConditionVariableType

Private Member Functions
DDS::ReturnCode_t	remove_instance (PublicationInstance_rch instance, Message_Block_Ptr &registered_sample, bool dup_registered_sample)

	WriteDataContainer (WriteDataContainer const &)

WriteDataContainer &	operator= (WriteDataContainer const &)

bool	pending_data ()

void	copy_and_prepend (SendStateDataSampleList &list, const SendStateDataSampleList &appended, const GUID_t &reader_id, const DDS::LifespanQosPolicy &lifespan, const OPENDDS_STRING &filterClassName, const FilterEvaluator *eval, const DDS::StringSeq &params, ssize_t &max_resend_samples)

void	remove_excess_durable ()

DDS::ReturnCode_t	remove_oldest_sample (InstanceDataSampleList &instance_list, bool &released)

void	wakeup_blocking_writers (DataSampleElement *stale)

void	add_reader_acks (const GUID_t &reader, const SequenceNumber &base)

void	remove_reader_acks (const GUID_t &reader)

void	log_send_state_lists (OPENDDS_STRING description)

typedef	OPENDDS_MAP_CMP (GUID_t, DisjointSequence, GUID_tKeyLessThan) AckedSequenceMap

SequenceNumber	get_cumulative_ack ()

SequenceNumber	get_last_ack ()

void	update_acked (const SequenceNumber &seq, const GUID_t &id=GUID_UNKNOWN)

bool	sequence_acknowledged_i (const SequenceNumber &sequence)

typedef	OPENDDS_MULTIMAP (MonotonicTimePoint, PublicationInstance_rch) DeadlineMapType

void	set_deadline_period (const TimeDuration &deadline_period)

void	process_deadlines (const MonotonicTimePoint &now)

void	extend_deadline (const PublicationInstance_rch &instance)

void	cancel_deadline (const PublicationInstance_rch &instance)

Private Attributes
AckedSequenceMap	acked_sequences_

SequenceNumber	cached_cumulative_ack_

bool	cached_cumulative_ack_valid_

SendStateDataSampleList	unsent_data_
	List of data that has not been sent yet. More...

ACE_UINT64	transaction_id_

SendStateDataSampleList	sending_data_
	List of data that is currently being sent. More...

SendStateDataSampleList	sent_data_
	List of data that has already been sent. More...

SendStateDataSampleList	orphaned_to_transport_

WriterDataSampleList	data_holder_

SendStateDataSampleList	resend_data_

PublicationInstanceMapType	instances_
	The individual instance queue threads in the data. More...

GUID_t	publication_id_
	The publication Id from repo. More...

DataWriterImpl *	writer_
	The writer that owns this container. More...

CORBA::Long	max_samples_per_instance_

CORBA::Long	history_depth_

CORBA::Long	max_durable_per_instance_

CORBA::Long	max_num_instances_

CORBA::Long	max_num_samples_

DDS::Duration_t	max_blocking_time_

bool	waiting_on_release_
	The block waiting flag. More...

ACE_Recursive_Thread_Mutex	lock_

ConditionVariableType	condition_

ConditionVariableType	empty_condition_

ACE_Thread_Mutex	wfa_lock_
	Lock used for wait_for_acks() processing. More...

WfaConditionVariableType	wfa_condition_
	Used to block in wait_for_acks(). More...

size_t	n_chunks_

DataSampleElementAllocator	sample_list_element_allocator_

bool	shutdown_
	The flag indicates the datawriter will be destroyed. More...

DDS::DomainId_t const	domain_id_
	Domain ID. More...

char const *const	topic_name_
	Topic name. More...

char const *const	type_name_
	Type name. More...

DataDurabilityCache *const	durability_cache_
	Pointer to the data durability cache. More...

DDS::DurabilityServiceQosPolicy const &	durability_service_
	DURABILITY_SERVICE QoS specific to the DataWriter. More...

RcHandle< DCPS::PmfSporadicTask< WriteDataContainer > >	deadline_task_
	Timer responsible for reporting missed offered deadlines. More...

TimeDuration	deadline_period_

DeadlineMapType	deadline_map_

ACE_Recursive_Thread_Mutex &	deadline_status_lock_
	Lock for synchronization of `status_` member. More...

DDS::OfferedDeadlineMissedStatus &	deadline_status_
	Reference to the missed requested deadline status structure. More...

CORBA::Long &	deadline_last_total_count_
	Last total_count when status was last checked. More...

Friends
class	DataWriterImpl

class	::DDS_TEST

Additional Inherited Members
Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()

Detailed Description

A container for instances sample data.

This container is instantiated per DataWriter. It maintains list of PublicationInstance objects which is internally referenced by the instance handle.

This container contains threaded lists of all data written to a given DataWriter. The real data sample is represented by the DataSampleElement. The data_holder_ holds all DataSampleElement in the writing order via the next_writer_sample_/previous_writer_sample_ thread. The instance list in PublicationInstance links samples via the next_instance_sample_ thread.

There are three state transition lists used during write operations: unsent, sending, and sent. These lists are linked via the next_send_sample_/previous_send_sample_ thread. Any DataSampleElement should be in one of these three lists and SHOULD NOT be shared between these three lists. A normal transition of a DataSampleElement would be unsent->sending->sent. A DataSampleElement transitions from unsent to sent naturally during a write operation when get_unsent_data is called. A DataSampleElement transitions from sending to sent when data_delivered is called notifying the container that the transport is finished with the sample and it can be marked as a historical sample. A DataSampleElement may transition back to unsent from sending if the transport notifies that the sample was dropped and should be resent. A DataSampleElement is removed from sent list and freed when the instance queue or container (for another instance) needs more space. A DataSampleElement may be removed and freed directly from sending when and unreliable writer needs space. In this case the transport will be notified that the sample should be removed. A DataSampleElement may also be removed directly from unsent if an unreliable writer needs space for new samples. Note: The real data sample will be freed when the reference counting goes 0. The resend list is only used when the datawriter uses TRANSIENT_LOCAL_DURABILITY_QOS. It holds the DataSampleElements for the data sample duplicates of the sending and sent list and hands this list off to the transport.

Note: : 1) The PublicationInstance object is not removed from this container until the instance is unregistered. The same instance handle is reused for re-registration. The instance data is deleted when this container is deleted. This would simplify instance data memory management. An alternative way is to remove the handle from the instance list when unregister occurs and delete the instance data after the transport is done with the instance, but we do not have a way to know when the transport is done since we have the sending list for all instances in the same datawriter. 2) This container has the ownership of the instance data samples when the data is written. The type-specific datawriter that allocates the memory for the sample data gives the ownership to its base class. 3) It is the responsibility of the owner of objects of this class to ensure that access to the lists are properly locked. This means using the same lock/condition to access the lists via the enqueue(), get*(), and data_delivered() methods. For the case where these are all called from the same (client) thread, this should be a recursive lock so that: 1) we do not deadlock; and, 2) we incur the cost of obtaining the lock only once.

Definition at line 122 of file WriteDataContainer.h.

Member Typedef Documentation

◆ ConditionVariableType

typedef ConditionVariable<ACE_Recursive_Thread_Mutex> OpenDDS::DCPS::WriteDataContainer::ConditionVariableType

private

Definition at line 512 of file WriteDataContainer.h.

◆ WfaConditionVariableType

typedef ConditionVariable<ACE_Thread_Mutex> OpenDDS::DCPS::WriteDataContainer::WfaConditionVariableType

private

Definition at line 519 of file WriteDataContainer.h.

Constructor & Destructor Documentation

◆ WriteDataContainer() [1/2]

OpenDDS::DCPS::WriteDataContainer::WriteDataContainer	(	DataWriterImpl *	writer,
		CORBA::Long	max_samples_per_instance,
		CORBA::Long	history_depth,
		CORBA::Long	max_durable_per_instance,
		DDS::Duration_t	max_blocking_time,
		size_t	n_chunks,
		DDS::DomainId_t	domain_id,
		char const *	topic_name,
		char const *	type_name,
		DataDurabilityCache *	durability_cache,
		DDS::DurabilityServiceQosPolicy const &	durability_service,
		CORBA::Long	max_instances,
		CORBA::Long	max_total_samples,
		ACE_Recursive_Thread_Mutex &	deadline_status_lock,
		DDS::OfferedDeadlineMissedStatus &	deadline_status,
		CORBA::Long &	deadline_last_total_count
	)

No default constructor, must be initialized.

Parameters

writer	The writer which owns this container.
max_samples_per_instance	Max samples kept within each instance
max_durable_per_instance	Max durable samples sent for each instance
max_blocking_time	The timeout for write.
n_chunks	The number of chunks that the DataSampleElementAllocator needs allocate.
domain_id	Domain ID.
topic_name	Topic name.
type_name	Type name.
durability_cache	The data durability cache for unsent data.
durability_service	DURABILITY_SERVICE QoS specific to the DataWriter.
max_instances	maximum number of instances, 0 for unlimited
max_total_samples	maximum total number of samples, 0 for unlimited

Definition at line 68 of file WriteDataContainer.cpp.

References ACE_DEBUG, acked_sequences_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::GUID_UNKNOWN, LM_DEBUG, n_chunks_, sample_list_element_allocator_, and OpenDDS::DCPS::SequenceNumber::ZERO().

   : cached_cumulative_ack_valid_(false)
   , transaction_id_(0)
   , publication_id_(GUID_UNKNOWN)
   , writer_(writer)
   , max_samples_per_instance_(max_samples_per_instance)
   , history_depth_(history_depth)
   , max_durable_per_instance_(max_durable_per_instance)
   , max_num_instances_(max_instances)
   , max_num_samples_(max_total_samples)
   , max_blocking_time_(max_blocking_time)
   , waiting_on_release_(false)
   , condition_(lock_)
   , empty_condition_(lock_)
   , wfa_condition_(wfa_lock_)
   , n_chunks_(n_chunks)
   , sample_list_element_allocator_(2 * n_chunks_)
   , shutdown_(false)
   , domain_id_(domain_id)
   , topic_name_(topic_name)
   , type_name_(type_name)
 #ifndef OPENDDS_NO_PERSISTENCE_PROFILE
   , durability_cache_(durability_cache)
   , durability_service_(durability_service)
 #endif
   , deadline_task_(DCPS::make_rch<DCPS::PmfSporadicTask<WriteDataContainer> >(TheServiceParticipant->time_source(), TheServiceParticipant->interceptor(), rchandle_from(this), &WriteDataContainer::process_deadlines))
   , deadline_period_(TimeDuration::max_value)
   , deadline_status_lock_(deadline_status_lock)
   , deadline_status_(deadline_status)
   , deadline_last_total_count_(deadline_last_total_count)
 {
   if (DCPS_debug_level >= 2) {
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) WriteDataContainer "
                "sample_list_element_allocator %x with %d chunks\n",
                &sample_list_element_allocator_, n_chunks_));
   }
   acked_sequences_[GUID_UNKNOWN].insert(SequenceNumber::ZERO());
 }

◆ ~WriteDataContainer()

OpenDDS::DCPS::WriteDataContainer::~WriteDataContainer ( )

Definition at line 126 of file WriteDataContainer.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, deadline_task_, OpenDDS::DCPS::SendStateDataSampleList::dequeue_head(), OpenDDS::DCPS::TransportRegistry::instance(), LM_DEBUG, LM_ERROR, LM_WARNING, orphaned_to_transport_, release_buffer(), OpenDDS::DCPS::TransportRegistry::released(), sending_data_, sent_data_, shutdown_, OpenDDS::DCPS::SendStateDataSampleList::size(), and unsent_data_.

 {
   deadline_task_->cancel();
 
   if (this->unsent_data_.size() > 0) {
     ACE_DEBUG((LM_WARNING,
                ACE_TEXT("(%P|%t) WARNING: WriteDataContainer::~WriteDataContainer() - ")
                ACE_TEXT("destroyed with %d samples unsent.\n"),
                this->unsent_data_.size()));
   }
 
   if (this->sending_data_.size() > 0) {
     if (TransportRegistry::instance()->released()) {
       for (DataSampleElement* e; sending_data_.dequeue_head(e);) {
         release_buffer(e);
       }
     }
     if (sending_data_.size() && DCPS_debug_level) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: WriteDataContainer::~WriteDataContainer() - ")
                  ACE_TEXT("destroyed with %d samples sending.\n"),
                  this->sending_data_.size()));
     }
   }
 
   if (this->sent_data_.size() > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) WriteDataContainer::~WriteDataContainer() - ")
                ACE_TEXT("destroyed with %d samples sent.\n"),
                this->sent_data_.size()));
   }
 
   if (this->orphaned_to_transport_.size() > 0) {
     if (DCPS_debug_level > 0) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) WriteDataContainer::~WriteDataContainer() - ")
                  ACE_TEXT("destroyed with %d samples orphaned_to_transport.\n"),
                  this->orphaned_to_transport_.size()));
     }
   }
 
   if (!shutdown_) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: ")
                ACE_TEXT("WriteDataContainer::~WriteDataContainer, ")
                ACE_TEXT("The container has not been cleaned.\n")));
   }
 }

◆ WriteDataContainer() [2/2]

OpenDDS::DCPS::WriteDataContainer::WriteDataContainer ( WriteDataContainer const & )

private

Member Function Documentation

◆ add_reader_acks()

void OpenDDS::DCPS::WriteDataContainer::add_reader_acks	(	const GUID_t &	reader,
		const SequenceNumber &	base
	)

private

Definition at line 176 of file WriteDataContainer.cpp.

References acked_sequences_, cached_cumulative_ack_valid_, OpenDDS::DCPS::DisjointSequence::insert(), OpenDDS::DCPS::DisjointSequence::reset(), OpenDDS::DCPS::SequenceNumber::SEQUENCENUMBER_UNKNOWN(), wfa_lock_, and OpenDDS::DCPS::SequenceNumber::ZERO().

 {
   ACE_Guard<ACE_Thread_Mutex> guard(wfa_lock_);
 
   DisjointSequence& ds = acked_sequences_[reader];
   ds.reset();
   if (base == SequenceNumber::SEQUENCENUMBER_UNKNOWN()) {
     ds.insert(SequenceNumber::ZERO());
   } else {
     ds.insert(SequenceRange(SequenceNumber(), base));
   }
   cached_cumulative_ack_valid_ = false;
 }

◆ cancel_deadline()

void OpenDDS::DCPS::WriteDataContainer::cancel_deadline ( const PublicationInstance_rch & instance )

private

Definition at line 1737 of file WriteDataContainer.cpp.

References OpenDDS::DCPS::PublicationInstance::deadline_, deadline_map_, deadline_period_, deadline_task_, OpenDDS::DCPS::TimeDuration::max_value, and OPENDDS_END_VERSIONED_NAMESPACE_DECL.

Referenced by remove_instance().

 {
   // Call comes from DataWriterImpl_t which should arleady have the lock_.
 
   if (deadline_period_ == TimeDuration::max_value) {
     return;
   }
 
   std::pair<DeadlineMapType::iterator, DeadlineMapType::iterator> r = deadline_map_.equal_range(instance->deadline_);
   while (r.first != r.second && r.first->second != instance) {
     ++r.first;
   }
   if (r.first != r.second) {
     deadline_map_.erase(r.first);
     if (deadline_map_.empty()) {
       deadline_task_->cancel();
     }
   }
 }

◆ copy_and_prepend()

void OpenDDS::DCPS::WriteDataContainer::copy_and_prepend	(	SendStateDataSampleList &	list,
		const SendStateDataSampleList &	appended,
		const GUID_t &	reader_id,
		const DDS::LifespanQosPolicy &	lifespan,
		const OPENDDS_STRING &	filterClassName,
		const FilterEvaluator *	eval,
		const DDS::StringSeq &	params,
		ssize_t &	max_resend_samples
	)

private

Definition at line 1370 of file WriteDataContainer.cpp.

References ACE_DEBUG, ACE_NEW_MALLOC, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::PublicationInstance::durable_samples_remaining_, OpenDDS::DCPS::SendStateDataSampleList::enqueue_head(), OpenDDS::DCPS::DataWriterImpl::filter_out(), LM_DEBUG, OpenDDS::DCPS::Cached_Allocator_With_Overflow< T, ACE_LOCK >::malloc(), OpenDDS::DCPS::SendStateDataSampleList::rbegin(), OpenDDS::DCPS::SendStateDataSampleList::rend(), OpenDDS::DCPS::resend_data_expired(), sample_list_element_allocator_, OpenDDS::DCPS::DataSampleElement::set_num_subs(), OpenDDS::DCPS::DataSampleElement::set_sub_id(), and writer_.

Referenced by reenqueue_all().

 {
   for (SendStateDataSampleList::const_reverse_iterator cur = appended.rbegin();
        cur != appended.rend() && max_resend_samples; ++cur) {
 
     if (resend_data_expired(*cur, lifespan))
       continue;
 
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
     if (eval && writer_->filter_out(*cur, filterClassName, *eval, params))
       continue;
 #endif
 
     PublicationInstance_rch inst = cur->get_handle();
 
     if (!inst) {
       // *cur is a control message, just skip it
       continue;
     }
 
     if (inst->durable_samples_remaining_ == 0)
       continue;
     --inst->durable_samples_remaining_;
 
     DataSampleElement* element = 0;
     ACE_NEW_MALLOC(element,
                    static_cast<DataSampleElement*>(
                      sample_list_element_allocator_.malloc(
                        sizeof(DataSampleElement))),
                    DataSampleElement(*cur));
 
     element->set_num_subs(1);
     element->set_sub_id(0, reader_id);
 
     if (DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) WriteDataContainer::copy_and_prepend added seq# %q\n",
                  cur->get_header().sequence_.getValue()));
     }
 
     list.enqueue_head(element);
     --max_resend_samples;
   }
 }

◆ data_delivered()

void OpenDDS::DCPS::WriteDataContainer::data_delivered ( const DataSampleElement * sample )

Acknowledge the delivery of data. The sample that resides in this container will be moved from sending_data_ list to the internal sent_data_ list. If there are any threads waiting for available space, it wakes up these threads.

Definition at line 651 of file WriteDataContainer.cpp.

Referenced by data_dropped().

 {
   DBG_ENTRY_LVL("WriteDataContainer","data_delivered",6);
 
   if (DCPS_debug_level >= 2) {
     ACE_DEBUG ((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::data_delivered")
                           ACE_TEXT(" %@\n"), sample));
   }
 
   ACE_GUARD(ACE_Recursive_Thread_Mutex,
             guard,
             lock_);
 
   // Delivered samples _must_ be on sending_data_ list
 
   // If it is not found in one of the lists, an invariant
   // exception is declared.
 
   // The element now needs to be removed from the sending_data_
   // list, and appended to the end of the sent_data_ list here
 
   DataSampleElement* stale = const_cast<DataSampleElement*>(sample);
 
   // If sample is on a SendStateDataSampleList it should be on the
   // sending_data_ list signifying it was given to the transport to
   // deliver and now the transport is signaling it has been delivered
   if (!sending_data_.dequeue(sample)) {
     //
     // Should be on sending_data_.  If it is in sent_data_
     // or unsent_data there was a problem.
     //
     SendStateDataSampleList* send_lists[] = {
       &sent_data_,
       &unsent_data_,
       &orphaned_to_transport_};
     const SendStateDataSampleList* containing_list =
       SendStateDataSampleList::send_list_containing_element(stale, send_lists);
 
     if (containing_list == &sent_data_) {
       ACE_ERROR((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: ")
                  ACE_TEXT("WriteDataContainer::data_delivered, ")
                  ACE_TEXT("The delivered sample is not in sending_data_ and ")
                  ACE_TEXT("WAS IN sent_data_.\n")));
     } else if (containing_list == &unsent_data_) {
       ACE_ERROR((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: ")
                  ACE_TEXT("WriteDataContainer::data_delivered, ")
                  ACE_TEXT("The delivered sample is not in sending_data_ and ")
                  ACE_TEXT("WAS IN unsent_data_ list.\n")));
     } else {
 
       //No-op: elements may be removed from all WriteDataContainer lists during shutdown
       //and inform transport of their release.  Transport will call data-delivered on the
       //elements as it processes the removal but they will already be gone from the send lists.
       if (stale->get_header().message_id_ != SAMPLE_DATA) {
         //this message was a control message so release it
         if (DCPS_debug_level > 9) {
           ACE_DEBUG((LM_DEBUG,
                      ACE_TEXT("(%P|%t) WriteDataContainer::data_delivered: ")
                      ACE_TEXT("domain %d topic %C publication %C control message delivered.\n"),
                      domain_id_,
                      topic_name_,
                      LogGuid(publication_id_).c_str()));
         }
         writer_->controlTracker.message_delivered();
       }
 
       if (containing_list == &orphaned_to_transport_) {
         orphaned_to_transport_.dequeue(sample);
         release_buffer(stale);
 
       } else if (!containing_list) {
         // samples that were retrieved from get_resend_data()
         ACE_Guard<ACE_SYNCH_MUTEX> wfa_guard(wfa_lock_);
         const CORBA::ULong num_subs = stale->get_num_subs();
         for (CORBA::ULong i = 0; i < num_subs; ++i) {
           update_acked(stale->get_header().sequence_, stale->get_sub_id(i));
         }
         wfa_guard.release();
         SendStateDataSampleList::remove(stale);
         release_buffer(stale);
       }
 
       if (!pending_data()) {
         empty_condition_.notify_all();
       }
     }
 
     return;
   }
   ACE_GUARD(ACE_SYNCH_MUTEX, wfa_guard, wfa_lock_);
   SequenceNumber acked_seq = stale->get_header().sequence_;
   SequenceNumber prev_max = get_cumulative_ack();
 
   if (stale->get_header().message_id_ != SAMPLE_DATA) {
     //this message was a control message so release it
     if (DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) WriteDataContainer::data_delivered: ")
                  ACE_TEXT("domain %d topic %C publication %C control message delivered.\n"),
                  domain_id_,
                  topic_name_,
                  LogGuid(publication_id_).c_str()));
     }
     release_buffer(stale);
     stale = 0;
     writer_->controlTracker.message_delivered();
   } else {
 
     if (max_durable_per_instance_ && !shutdown_ && InstanceDataSampleList::on_some_list(sample)) {
       const_cast<DataSampleElement*>(sample)->get_header().historic_sample_ = true;
       DataSampleHeader::set_flag(HISTORIC_SAMPLE_FLAG, sample->get_sample());
       sent_data_.enqueue_tail(sample);
 
     } else {
       if (InstanceDataSampleList::on_some_list(sample)) {
         PublicationInstance_rch inst = sample->get_handle();
         inst->samples_.dequeue(sample);
       }
       release_buffer(stale);
       stale = 0;
     }
 
     if (DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) WriteDataContainer::data_delivered: ")
                  ACE_TEXT("domain %d topic %C publication %C seq# %q %s.\n"),
                  domain_id_,
                  topic_name_,
                  LogGuid(publication_id_).c_str(),
                  acked_seq.getValue(),
                  max_durable_per_instance_
                  ? ACE_TEXT("stored for durability")
                  : ACE_TEXT("released")));
     }
 
     wakeup_blocking_writers(stale);
   }
   if (DCPS_debug_level > 9) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::data_delivered: ")
                          ACE_TEXT("Inserting acked_sequence: %q\n"),
                          acked_seq.getValue()));
   }
 
   update_acked(acked_seq);
 
   if (prev_max == SequenceNumber::SEQUENCENUMBER_UNKNOWN() ||
       prev_max < get_cumulative_ack()) {
 
     if (DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) WriteDataContainer::data_delivered - ")
                  ACE_TEXT("broadcasting wait_for_acknowledgments update.\n")));
     }
 
     wfa_condition_.notify_all();
   }
 
   // Signal if there is no pending data.
   if (!pending_data()) {
     empty_condition_.notify_all();
   }
 }

◆ data_dropped()

void OpenDDS::DCPS::WriteDataContainer::data_dropped	(	const DataSampleElement *	element,
		bool	dropped_by_transport
	)

This method is called by the transport to notify the sample is dropped. Which the transport was told to do by the publication code by calling TransportClient::remove_sample(). If the sample was "sending" then it is moved to the "unsent" list. If there are any threads waiting for available space then it needs wake up these threads. The dropped_by_transport flag true indicates the dropping initiated by transport when the transport send strategy is in a MODE_TERMINATED. The dropped_by_transport flag false indicates the dropping is initiated by the remove_sample and data_dropped() is a result of remove_sample().

Definition at line 817 of file WriteDataContainer.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DataWriterImpl::controlTracker, data_delivered(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::SendStateDataSampleList::dequeue(), domain_id_, empty_condition_, OpenDDS::DCPS::SendStateDataSampleList::enqueue_tail(), OpenDDS::DCPS::DataSampleElement::get_header(), LM_DEBUG, LM_WARNING, lock_, OpenDDS::DCPS::MessageTracker::message_dropped(), OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), OpenDDS::DCPS::InstanceDataSampleList::on_some_list(), orphaned_to_transport_, pending_data(), publication_id_, release_buffer(), OpenDDS::DCPS::SendStateDataSampleList::remove(), OpenDDS::DCPS::SAMPLE_DATA, OpenDDS::DCPS::SendStateDataSampleList::send_list_containing_element(), sending_data_, sent_data_, shutdown_, topic_name_, unsent_data_, wakeup_blocking_writers(), and writer_.

 {
   DBG_ENTRY_LVL("WriteDataContainer","data_dropped",6);
 
   if (DCPS_debug_level >= 2) {
     ACE_DEBUG ((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::data_dropped")
                           ACE_TEXT(" sample %X dropped_by_transport %d\n"),
                           sample, dropped_by_transport));
   }
 
   // If the transport initiates the data dropping, we need do same thing
   // as data_delivered. e.g. remove the sample from the internal list
   // and the instance list. We do not need acquire the lock here since
   // the data_delivered acquires the lock.
   if (dropped_by_transport) {
     data_delivered(sample);
     return;
   }
 
   //The data_dropped could be called from the thread initiating sample remove
   //which already hold the lock. In this case, it's not necessary to acquire
   //lock here. It also could be called from the transport thread in a delayed
   //notification, it's necessary to acquire lock here to protect the internal
   //structures in this class.
 
   ACE_GUARD (ACE_Recursive_Thread_Mutex,
     guard,
     lock_);
 
   // The dropped sample should be in the sending_data_ list.
   // Otherwise an exception will be raised.
   //
   // We are now been notified by transport, so we can
   // keep the sample from the sending_data_ list still in
   // sample list since we will send it.
 
   DataSampleElement* stale = const_cast<DataSampleElement*>(sample);
 
   // If sample is on a SendStateDataSampleList it should be on the
   // sending_data_ list signifying it was given to the transport to
   // deliver and now the transport is signaling it has been dropped
 
   if (sending_data_.dequeue(sample)) {
     // else: The data_dropped is called as a result of remove_sample()
     // called from reenqueue_all() which supports the TRANSIENT_LOCAL
     // qos. The samples that are sending by transport are dropped from
     // transport and will be moved to the unsent list for resend.
     if (!shutdown_ && InstanceDataSampleList::on_some_list(sample)) {
       unsent_data_.enqueue_tail(sample);
     } else {
       SendStateDataSampleList::remove(stale);
       release_buffer(stale);
       stale = 0;
     }
 
   } else {
     //
     // If it is in sent_data_ or unsent_data there was a problem.
     //
     SendStateDataSampleList* send_lists[] = {
       &sent_data_,
       &unsent_data_,
       &orphaned_to_transport_};
     const SendStateDataSampleList* containing_list =
       SendStateDataSampleList::send_list_containing_element(stale, send_lists);
 
     if (containing_list == &sent_data_) {
       ACE_ERROR((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: ")
                  ACE_TEXT("WriteDataContainer::data_dropped, ")
                  ACE_TEXT("The dropped sample is not in sending_data_ and ")
                  ACE_TEXT("WAS IN sent_data_.\n")));
     } else if (containing_list == &unsent_data_) {
       ACE_ERROR((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: ")
                  ACE_TEXT("WriteDataContainer::data_dropped, ")
                  ACE_TEXT("The dropped sample is not in sending_data_ and ")
                  ACE_TEXT("WAS IN unsent_data_ list.\n")));
     } else {
 
       //No-op: elements may be removed from all WriteDataContainer lists during shutdown
       //and inform transport of their release.  Transport will call data-dropped on the
       //elements as it processes the removal but they will already be gone from the send lists.
       if (stale->get_header().message_id_ != SAMPLE_DATA) {
         //this message was a control message so release it
         if (DCPS_debug_level > 9) {
           ACE_DEBUG((LM_DEBUG,
                      ACE_TEXT("(%P|%t) WriteDataContainer::data_dropped: ")
                      ACE_TEXT("domain %d topic %C publication %C control message dropped.\n"),
                      domain_id_,
                      topic_name_,
                      LogGuid(publication_id_).c_str()));
         }
         writer_->controlTracker.message_dropped();
       }
 
       if (containing_list == &orphaned_to_transport_) {
         orphaned_to_transport_.dequeue(sample);
         release_buffer(stale);
         stale = 0;
         if (!pending_data()) {
           empty_condition_.notify_all();
         }
 
       } else if (!containing_list) {
         // samples that were retrieved from get_resend_data()
         SendStateDataSampleList::remove(stale);
         release_buffer(stale);
         stale = 0;
       }
     }
 
     return;
   }
 
   wakeup_blocking_writers(stale);
 
   if (!pending_data()) {
     empty_condition_.notify_all();
   }
 }

◆ dispose()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::dispose	(	DDS::InstanceHandle_t	handle,
		Message_Block_Ptr &	registered_sample,
		bool	dup_registered_sample = `true`
	)

Delete the samples for the provided instance. A shallow copy of the sample data will be given to datawriter as part of the control message if the dup_registered_sample is true.

This method returns error if the instance is not registered.

Definition at line 482 of file WriteDataContainer.cpp.

References ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::find(), instances_, LM_ERROR, lock_, remove_instance(), DDS::RETCODE_ERROR, and DDS::RETCODE_PRECONDITION_NOT_MET.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    lock_,
                    DDS::RETCODE_ERROR);
 
   PublicationInstance_rch instance;
 
   int const find_attempt = find(instances_, instance_handle, instance);
 
   if (0 != find_attempt) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("WriteDataContainer::dispose, ")
                       ACE_TEXT("The instance(handle=%X) ")
                       ACE_TEXT("is not registered yet.\n"),
                       instance_handle),
                      DDS::RETCODE_PRECONDITION_NOT_MET);
   }
 
   return remove_instance(instance, registered_sample, dup_registered_sample);
 }

◆ enqueue()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::enqueue	(	DataSampleElement *	sample,
		DDS::InstanceHandle_t	instance
	)

Enqueue the data sample in its instance thread. This method assumes there is an available space for the sample in the instance list.

Definition at line 294 of file WriteDataContainer.cpp.

References OpenDDS::DCPS::InstanceDataSampleList::enqueue_tail(), OpenDDS::DCPS::SendStateDataSampleList::enqueue_tail(), extend_deadline(), get_handle_instance(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::PublicationInstance::samples_, shutdown_, and unsent_data_.

 {
   if (shutdown_) {
     return DDS::RETCODE_ERROR;
   }
 
   // Get the PublicationInstance pointer from InstanceHandle_t.
   PublicationInstance_rch instance =
     get_handle_instance(instance_handle);
   // Extract the instance queue.
   InstanceDataSampleList& instance_list = instance->samples_;
 
   extend_deadline(instance);
 
   //
   // Enqueue to the next_send_sample_ thread of unsent_data_
   // will link samples with the next_sample/previous_sample and
   // also next_send_sample_.
   // This would save time when we actually send the data.
 
   unsent_data_.enqueue_tail(sample);
 
   //
   // Add this sample to the INSTANCE scope list.
   instance_list.enqueue_tail(sample);
 
   return DDS::RETCODE_OK;
 }

◆ enqueue_control()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::enqueue_control ( DataSampleElement * control_sample )

Definition at line 276 of file WriteDataContainer.cpp.

References OpenDDS::DCPS::SendStateDataSampleList::enqueue_tail(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, shutdown_, and unsent_data_.

 {
   // Enqueue to the next_send_sample_ thread of unsent_data_
   // will link samples with the next_sample/previous_sample and
   // also next_send_sample_.
   // This would save time when we actually send the data.
 
   if (shutdown_) {
     return DDS::RETCODE_ERROR;
   }
 
   unsent_data_.enqueue_tail(control_sample);
 
   return DDS::RETCODE_OK;
 }

◆ extend_deadline()

void OpenDDS::DCPS::WriteDataContainer::extend_deadline ( const PublicationInstance_rch & instance )

private

Definition at line 1712 of file WriteDataContainer.cpp.

References OpenDDS::DCPS::PublicationInstance::deadline_, deadline_map_, deadline_period_, deadline_task_, OpenDDS::DCPS::TimeDuration::max_value, and OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now().

Referenced by enqueue(), and register_instance().

 {
   // Call comes from DataWriterImpl_t which should arleady have the lock_.
 
   if (deadline_period_ == TimeDuration::max_value) {
     return;
   }
 
   std::pair<DeadlineMapType::iterator, DeadlineMapType::iterator> r = deadline_map_.equal_range(instance->deadline_);
   while (r.first != r.second && r.first->second != instance) {
     ++r.first;
   }
   if (r.first != r.second) {
     // The instance was in the map.
     deadline_map_.erase(r.first);
   }
   instance->deadline_ = MonotonicTimePoint::now() + deadline_period_;
   bool schedule = deadline_map_.empty();
   deadline_map_.insert(std::make_pair(instance->deadline_, instance));
   if (schedule) {
     deadline_task_->schedule(deadline_period_);
   }
 }

◆ get_cumulative_ack()

SequenceNumber OpenDDS::DCPS::WriteDataContainer::get_cumulative_ack ( )

private

Definition at line 207 of file WriteDataContainer.cpp.

References acked_sequences_, cached_cumulative_ack_, cached_cumulative_ack_valid_, and OpenDDS::DCPS::SequenceNumber::SEQUENCENUMBER_UNKNOWN().

Referenced by data_delivered(), remove_reader_acks(), and sequence_acknowledged_i().

 {
   if (acked_sequences_.empty()) {
     return SequenceNumber::SEQUENCENUMBER_UNKNOWN();
   }
 
   if (cached_cumulative_ack_valid_) {
     return cached_cumulative_ack_;
   }
 
   SequenceNumber result = SequenceNumber::SEQUENCENUMBER_UNKNOWN();
   for (AckedSequenceMap::const_iterator it = acked_sequences_.begin(); it != acked_sequences_.end(); ++it) {
     if (!it->second.empty()) {
       result = result == SequenceNumber::SEQUENCENUMBER_UNKNOWN() ? it->second.cumulative_ack() : std::min(result, it->second.cumulative_ack());
     }
   }
   cached_cumulative_ack_ = result;
   cached_cumulative_ack_valid_ = true;
   return result;
 }

◆ get_handle_instance()

PublicationInstance_rch OpenDDS::DCPS::WriteDataContainer::get_handle_instance ( DDS::InstanceHandle_t handle )

Todo:: remove/document this!

Definition at line 1355 of file WriteDataContainer.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::find(), instances_, and LM_DEBUG.

Referenced by enqueue(), and obtain_buffer().

 {
   PublicationInstance_rch instance;
 
   if (0 != find(instances_, handle, instance)) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) ")
                ACE_TEXT("WriteDataContainer::get_handle_instance, ")
                ACE_TEXT("lookup for %d failed\n"), handle));
   }
 
   return instance;
 }

◆ get_instance_handles()

void OpenDDS::DCPS::WriteDataContainer::get_instance_handles ( InstanceHandleVec & instance_handles )

Definition at line 1513 of file WriteDataContainer.cpp.

References ACE_GUARD, instances_, and lock_.

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex,
             guard,
             lock_);
   PublicationInstanceMapType::iterator it = instances_.begin();
 
   while (it != instances_.end()) {
     instance_handles.push_back(it->second->instance_handle_);
     ++it;
   }
 }

◆ get_last_ack()

SequenceNumber OpenDDS::DCPS::WriteDataContainer::get_last_ack ( )

private

Definition at line 229 of file WriteDataContainer.cpp.

References acked_sequences_, and OpenDDS::DCPS::SequenceNumber::SEQUENCENUMBER_UNKNOWN().

 {
   if (acked_sequences_.empty()) {
     return SequenceNumber::SEQUENCENUMBER_UNKNOWN();
   }
 
   SequenceNumber result = SequenceNumber::SEQUENCENUMBER_UNKNOWN();
   for (AckedSequenceMap::const_iterator it = acked_sequences_.begin(); it != acked_sequences_.end(); ++it) {
     if (!it->second.empty()) {
       result = result == SequenceNumber::SEQUENCENUMBER_UNKNOWN() ? it->second.last_ack() : std::max(result, it->second.last_ack());
     }
   }
   return result;
 }

◆ get_resend_data()

SendStateDataSampleList OpenDDS::DCPS::WriteDataContainer::get_resend_data ( )

Obtain a list of data for resending. This is only used when TRANSIENT_LOCAL_DURABILITY_QOS is used. The data on the list returned is not put on any SendStateDataSampleList.

Definition at line 622 of file WriteDataContainer.cpp.

References DBG_ENTRY_LVL, resend_data_, and OpenDDS::DCPS::SendStateDataSampleList::reset().

 {
   DBG_ENTRY_LVL("WriteDataContainer","get_resend_data",6);
 
   //
   // The samples in unsent_data are added to the sending_data
   // during enqueue.
   //
   SendStateDataSampleList list = this->resend_data_;
 
   //
   // Clear the unsent data list.
   //
   this->resend_data_.reset();
   //
   // Return the moved list.
   //
   return list;
 }

◆ get_unsent_data()

ACE_UINT64 OpenDDS::DCPS::WriteDataContainer::get_unsent_data ( SendStateDataSampleList & list )

Obtain a list of data that has not yet been sent. The data on the list returned is moved from the internal unsent_data_ list to the internal sending_data_ list as part of this call. The entire list is linked via the DataSampleElement.next_send_sample_ link as well.

Definition at line 583 of file WriteDataContainer.cpp.

References OpenDDS::DCPS::SendStateDataSampleList::begin(), DBG_ENTRY_LVL, OpenDDS::DCPS::SendStateDataSampleList::end(), OpenDDS::DCPS::SendStateDataSampleList::enqueue_tail(), OpenDDS::DCPS::SendStateDataSampleList::reset(), sending_data_, OpenDDS::DCPS::DataSampleElement::set_transaction_id(), transaction_id_, and unsent_data_.

 {
   DBG_ENTRY_LVL("WriteDataContainer","get_unsent_data",6);
   //
   // The samples in unsent_data are added to the local datawriter
   // list and enqueued to the sending_data_ signifying they have
   // been passed to the transport to send in a transaction
   //
   list = this->unsent_data_;
 
   // Increment send counter for this send operation
   ++transaction_id_;
 
   // Mark all samples with current send counter
   SendStateDataSampleList::iterator iter = list.begin();
   while (iter != list.end()) {
     iter->set_transaction_id(this->transaction_id_);
     ++iter;
   }
 
   //
   // The unsent_data_ already linked with the
   // next_send_sample during enqueue.
   // Append the unsent_data_ to current sending_data_
   // list.
   sending_data_.enqueue_tail(list);
 
   //
   // Clear the unsent data list.
   //
   this->unsent_data_.reset();
 
   //
   // Return the moved list.
   //
   return transaction_id_;
 }

◆ log_send_state_lists()

void OpenDDS::DCPS::WriteDataContainer::log_send_state_lists ( OPENDDS_STRING description )

private

Definition at line 1593 of file WriteDataContainer.cpp.

References ACE_DEBUG, instances_, LM_DEBUG, num_all_samples(), orphaned_to_transport_, sending_data_, sent_data_, OpenDDS::DCPS::SendStateDataSampleList::size(), and unsent_data_.

Referenced by wait_pending().

 {
   ACE_DEBUG((LM_DEBUG, "(%P|%t) WriteDataContainer::log_send_state_lists: %C -- unsent(%d), sending(%d), sent(%d), orphaned_to_transport(%d), num_all_samples(%d), num_instances(%d)\n",
              description.c_str(),
              unsent_data_.size(),
              sending_data_.size(),
              sent_data_.size(),
              orphaned_to_transport_.size(),
              num_all_samples(),
              instances_.size()));
 }

◆ num_all_samples()

size_t OpenDDS::DCPS::WriteDataContainer::num_all_samples ( )

Return the number of samples for all instances.

Definition at line 563 of file WriteDataContainer.cpp.

References ACE_GUARD_RETURN, instances_, and lock_.

Referenced by log_send_state_lists(), and obtain_buffer().

 {
   size_t size = 0;
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    lock_,
                    0);
 
   for (PublicationInstanceMapType::iterator iter = instances_.begin();
        iter != instances_.end();
        ++iter)
   {
     size += iter->second->samples_.size();
   }
 
   return size;
 }

◆ num_samples()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::num_samples	(	DDS::InstanceHandle_t	handle,
		size_t &	size
	)

Return the number of samples for the given instance.

Definition at line 542 of file WriteDataContainer.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::find(), instances_, lock_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::PublicationInstance::samples_, and OpenDDS::DCPS::InstanceDataSampleList::size().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    lock_,
                    DDS::RETCODE_ERROR);
   PublicationInstance_rch instance;
 
   int const find_attempt = find(instances_, handle, instance);
 
   if (0 != find_attempt) {
     return DDS::RETCODE_ERROR;
 
   } else {
     size = instance->samples_.size();
     return DDS::RETCODE_OK;
   }
 }

◆ obtain_buffer()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::obtain_buffer	(	DataSampleElement *&	element,
		DDS::InstanceHandle_t	handle
	)

Allocate a DataSampleElement object and check the space availability for newly allocated element according to qos settings. For the blocking write case, if resource limits or history qos limits are reached, then it blocks for max blocking time for a previous sample to be delivered or dropped by the transport. In non-blocking write case, if resource limits or history qos limits are reached, will attempt to remove oldest samples (forcing the transport to drop samples if necessary) to make space. If there are several threads waiting then the first one in the waiting list can enqueue, others continue waiting. Note: the lock should be held before calling this method

Definition at line 1138 of file WriteDataContainer.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_NEW_MALLOC_RETURN, ACE_TEXT(), condition_, OpenDDS::DCPS::CvStatus_Error, OpenDDS::DCPS::CvStatus_NoTimeout, OpenDDS::DCPS::CvStatus_Timeout, data_holder_, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::WriterDataSampleList::enqueue_tail(), get_handle_instance(), history_depth_, instances_, DDS::ReliabilityQosPolicy::kind, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::Cached_Allocator_With_Overflow< T, ACE_LOCK >::malloc(), max_blocking_time_, max_num_samples_, max_samples_per_instance_, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), num_all_samples(), publication_id_, OpenDDS::DCPS::DataWriterImpl::qos_, release_buffer(), DDS::DataWriterQos::reliability, DDS::RELIABLE_RELIABILITY_QOS, remove_excess_durable(), remove_oldest_sample(), DDS::RETCODE_BAD_PARAMETER, DDS::RETCODE_ERROR, DDS::RETCODE_OK, DDS::RETCODE_TIMEOUT, sample_list_element_allocator_, OpenDDS::DCPS::PublicationInstance::samples_, shutdown_, TheServiceParticipant, OpenDDS::DCPS::ConditionVariable< Mutex >::wait_until(), waiting_on_release_, and writer_.

 {
   DBG_ENTRY_LVL("WriteDataContainer","obtain_buffer", 6);
 
   remove_excess_durable();
 
   PublicationInstance_rch instance = get_handle_instance(handle);
 
   if (!instance) {
     return DDS::RETCODE_BAD_PARAMETER;
   }
 
   ACE_NEW_MALLOC_RETURN(
     element,
     static_cast<DataSampleElement*>(
       sample_list_element_allocator_.malloc(
         sizeof(DataSampleElement))),
     DataSampleElement(publication_id_,
                           this->writer_,
                           instance),
     DDS::RETCODE_ERROR);
 
   // Extract the current instance queue.
   InstanceDataSampleList& instance_list = instance->samples_;
   DDS::ReturnCode_t ret = DDS::RETCODE_OK;
 
   bool set_timeout = true;
   MonotonicTimePoint timeout;
 
   //max_num_samples_ covers ResourceLimitsQosPolicy max_samples and
   //max_instances and max_instances * depth
   ThreadStatusManager& thread_status_manager = TheServiceParticipant->get_thread_status_manager();
   while ((instance_list.size() >= max_samples_per_instance_) ||
          ((this->max_num_samples_ > 0) &&
          ((CORBA::Long) this->num_all_samples () >= this->max_num_samples_))) {
 
     if (this->writer_->qos_.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS) {
       if (instance_list.size() >= history_depth_) {
         if (DCPS_debug_level >= 2) {
           ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::obtain_buffer")
                      ACE_TEXT(" instance %d attempting to remove")
                      ACE_TEXT(" its oldest sample (reliable)\n"),
                      handle));
         }
         bool oldest_released = false;
         ret = remove_oldest_sample(instance_list, oldest_released);
         if (oldest_released) {
           break;
         }
       }
       // Reliable writers can wait
       if (set_timeout) {
         timeout = MonotonicTimePoint::now() + TimeDuration(max_blocking_time_);
         set_timeout = false;
       }
       if (!shutdown_ && MonotonicTimePoint::now() < timeout) {
         if (DCPS_debug_level >= 2) {
           ACE_DEBUG ((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::obtain_buffer")
                                 ACE_TEXT(" instance %d waiting for samples to be released by transport\n"),
                       handle));
         }
 
         waiting_on_release_ = true;
         switch (condition_.wait_until(timeout, thread_status_manager)) {
         case CvStatus_NoTimeout:
           remove_excess_durable();
           break;
 
         case CvStatus_Timeout:
           if (DCPS_debug_level >= 2) {
             ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::obtain_buffer")
               ACE_TEXT(" instance %d timed out waiting for samples to be released by transport\n"),
               handle));
           }
           ret = DDS::RETCODE_TIMEOUT;
           break;
 
         case CvStatus_Error:
           if (DCPS_debug_level) {
             ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: WriteDataContainer::obtain_buffer: "
               "error in wait_until\n"));
           }
           ret = DDS::RETCODE_ERROR;
           break;
         }
 
       } else {
         //either shutdown has been signaled or max_blocking_time
         //has surpassed so treat as timeout
         ret = DDS::RETCODE_TIMEOUT;
       }
 
     } else {
       //BEST EFFORT
       bool oldest_released = false;
 
       //try to remove stale samples from this instance
       // The remove_oldest_sample() method removes the oldest sample
       // from instance list and removes it from the internal lists.
       if (instance_list.size() > 0) {
         if (DCPS_debug_level >= 2) {
           ACE_DEBUG ((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::obtain_buffer")
                                 ACE_TEXT(" instance %d attempting to remove")
                                 ACE_TEXT(" its oldest sample\n"),
                                 handle));
         }
         ret = remove_oldest_sample(instance_list, oldest_released);
       }
       //else try to remove stale samples from other instances which are full
       if (ret == DDS::RETCODE_OK && !oldest_released) {
         if (DCPS_debug_level >= 2) {
           ACE_DEBUG ((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::obtain_buffer")
                                 ACE_TEXT(" instance %d attempting to remove")
                                 ACE_TEXT(" oldest sample from any full instances\n"),
                                 handle));
         }
         PublicationInstanceMapType::iterator it = instances_.begin();
 
         while (!oldest_released && it != instances_.end() && ret == DDS::RETCODE_OK) {
           if (it->second->samples_.size() >= max_samples_per_instance_) {
             ret = remove_oldest_sample(it->second->samples_, oldest_released);
           }
           ++it;
         }
       }
       //else try to remove stale samples from other non-full instances
       if (ret == DDS::RETCODE_OK && !oldest_released) {
         if (DCPS_debug_level >= 2) {
           ACE_DEBUG ((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::obtain_buffer")
                                 ACE_TEXT(" instance %d attempting to remove")
                                 ACE_TEXT(" oldest sample from any instance with samples currently\n"),
                                 handle));
         }
         PublicationInstanceMapType::iterator it = instances_.begin();
 
         while (!oldest_released && it != instances_.end() && ret == DDS::RETCODE_OK) {
           if (it->second->samples_.size() > 0) {
             ret = remove_oldest_sample(it->second->samples_, oldest_released);
           }
           ++it;
         }
       }
       if (!oldest_released) {
         //This means that no instances have samples to remove and yet
         //still hitting resource limits.
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: ")
                    ACE_TEXT("WriteDataContainer::obtain_buffer, ")
                    ACE_TEXT("hitting resource limits with no samples to remove\n")));
         ret = DDS::RETCODE_ERROR;
       }
     }  //END BEST EFFORT
 
     if (ret != DDS::RETCODE_OK) {
       if (DCPS_debug_level >= 2) {
         ACE_DEBUG ((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::obtain_buffer")
                               ACE_TEXT(" instance %d could not obtain buffer for sample")
                               ACE_TEXT(" releasing allotted sample and returning\n"),
                               handle));
       }
       this->release_buffer(element);
       return ret;
     }
   }  //END WHILE
 
   data_holder_.enqueue_tail(element);
 
   return ret;
 }

◆ obtain_buffer_for_control()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::obtain_buffer_for_control ( DataSampleElement *& element )

Definition at line 1120 of file WriteDataContainer.cpp.

References ACE_NEW_MALLOC_RETURN, DBG_ENTRY_LVL, OpenDDS::DCPS::Cached_Allocator_With_Overflow< T, ACE_LOCK >::malloc(), publication_id_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, sample_list_element_allocator_, and writer_.

 {
   DBG_ENTRY_LVL("WriteDataContainer","obtain_buffer_for_control", 6);
 
   ACE_NEW_MALLOC_RETURN(
     element,
     static_cast<DataSampleElement*>(
       sample_list_element_allocator_.malloc(
         sizeof(DataSampleElement))),
     DataSampleElement(publication_id_,
                       this->writer_,
                       PublicationInstance_rch()),
     DDS::RETCODE_ERROR);
 
   return DDS::RETCODE_OK;
 }

◆ OPENDDS_MAP_CMP()

typedef OpenDDS::DCPS::WriteDataContainer::OPENDDS_MAP_CMP	(	GUID_t	,
		DisjointSequence	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_MULTIMAP()

typedef OpenDDS::DCPS::WriteDataContainer::OPENDDS_MULTIMAP	(	MonotonicTimePoint	,
		PublicationInstance_rch
	)

private

◆ OPENDDS_VECTOR()

typedef OpenDDS::DCPS::WriteDataContainer::OPENDDS_VECTOR ( DDS::InstanceHandle_t )

Returns a vector of handles for the instances registered for this data writer.

◆ operator=()

WriteDataContainer& OpenDDS::DCPS::WriteDataContainer::operator= ( WriteDataContainer const & )

private

◆ pending_data()

bool OpenDDS::DCPS::WriteDataContainer::pending_data ( )

private

Returns if pending data exists. This includes sending, and unsent data.

Definition at line 643 of file WriteDataContainer.cpp.

References orphaned_to_transport_, sending_data_, OpenDDS::DCPS::SendStateDataSampleList::size(), and unsent_data_.

Referenced by data_delivered(), data_dropped(), remove_oldest_sample(), and wait_pending().

 {
   return this->sending_data_.size() != 0
          || this->orphaned_to_transport_.size() != 0
          || this->unsent_data_.size() != 0;
 }

◆ persist_data()

bool OpenDDS::DCPS::WriteDataContainer::persist_data ( )

Copy sent data to data DURABILITY cache.

Definition at line 1425 of file WriteDataContainer.cpp.

References ACE_ERROR, ACE_TEXT(), domain_id_, durability_cache_, durability_service_, OpenDDS::DCPS::DataDurabilityCache::insert(), LM_ERROR, sent_data_, topic_name_, and type_name_.

 {
   bool result = true;
 
   // ------------------------------------------------------------
   // Transfer sent data to data DURABILITY cache.
   // ------------------------------------------------------------
   if (this->durability_cache_) {
     // A data durability cache is available for TRANSIENT or
     // PERSISTENT data durability.  Cache the data samples.
 
     //
     //  We only cache data that is not still in use outside of
     //  this instance of WriteDataContainer
     //  (only cache samples in sent_data_ meaning transport has delivered).
     bool const inserted =
       this->durability_cache_->insert(this->domain_id_,
                                       this->topic_name_,
                                       this->type_name_,
                                       this->sent_data_,
                                       this->durability_service_
                                      );
 
     result = inserted;
 
     if (!inserted)
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: ")
                  ACE_TEXT("WriteDataContainer::persist_data, ")
                  ACE_TEXT("failed to make data durable for ")
                  ACE_TEXT("(domain, topic, type) = (%d, %C, %C)\n"),
                  this->domain_id_,
                  this->topic_name_,
                  this->type_name_));
   }
 
   return result;
 }

◆ process_deadlines()

void OpenDDS::DCPS::WriteDataContainer::process_deadlines ( const MonotonicTimePoint & now )

private

Definition at line 1655 of file WriteDataContainer.cpp.

References ACE_GUARD, OpenDDS::DCPS::PublicationInstance::deadline_, deadline_last_total_count_, deadline_map_, deadline_period_, deadline_status_, deadline_status_lock_, deadline_task_, OpenDDS::DCPS::PublicationInstance::instance_handle_, DDS::OfferedDeadlineMissedStatus::last_instance_handle, OpenDDS::DCPS::DataWriterImpl::listener_for(), lock_, OpenDDS::DCPS::EntityImpl::notify_status_condition(), DDS::OFFERED_DEADLINE_MISSED_STATUS, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), DDS::OfferedDeadlineMissedStatus::total_count, DDS::OfferedDeadlineMissedStatus::total_count_change, and writer_.

 {
   // Lock the DataWriterImpl.
   ACE_GUARD (ACE_Recursive_Thread_Mutex, dwi_guard, deadline_status_lock_);
   // Lock ourselves.
   ACE_GUARD (ACE_Recursive_Thread_Mutex, wdc_guard, lock_);
 
   if (deadline_map_.empty()) {
     return;
   }
 
   bool notify = false;
 
   for (DeadlineMapType::iterator pos = deadline_map_.begin(), limit = deadline_map_.end();
        pos != limit && pos->first < now; pos = deadline_map_.begin()) {
 
     PublicationInstance_rch instance = pos->second;
     deadline_map_.erase(pos);
 
     ++deadline_status_.total_count;
     deadline_status_.total_count_change = deadline_status_.total_count - deadline_last_total_count_;
     deadline_status_.last_instance_handle = instance->instance_handle_;
 
     writer_->set_status_changed_flag(DDS::OFFERED_DEADLINE_MISSED_STATUS, true);
     notify = true;
 
     DDS::DataWriterListener_var listener = writer_->listener_for(DDS::OFFERED_DEADLINE_MISSED_STATUS);
 
     if (listener) {
       // Copy before releasing the lock.
       const DDS::OfferedDeadlineMissedStatus status = deadline_status_;
 
       // Release the lock during the upcall.
       ACE_Reverse_Lock<ACE_Recursive_Thread_Mutex> deadline_reverse_status_lock(deadline_status_lock_);
       ACE_GUARD(ACE_Reverse_Lock<ACE_Recursive_Thread_Mutex>, rev_dwi_guard, deadline_reverse_status_lock);
 
       // @todo Will this operation ever throw?  If so we may want to
       //       catch all exceptions, and act accordingly.
       listener->on_offered_deadline_missed(writer_, status);
 
       // We need to update the last total count value to our current total
       // so that the next time we will calculate the correct total_count_change;
       deadline_last_total_count_ = deadline_status_.total_count;
     }
 
     instance->deadline_ += deadline_period_;
     deadline_map_.insert(std::make_pair(instance->deadline_, instance));
   }
 
   if (notify) {
     writer_->notify_status_condition();
   }
 
   deadline_task_->schedule(deadline_map_.begin()->first - now);
 }

◆ reenqueue_all()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::reenqueue_all	(	const GUID_t &	reader_id,
		const DDS::LifespanQosPolicy &	lifespan,
		const OPENDDS_STRING &	filterClassName,
		const FilterEvaluator *	eval,
		const DDS::StringSeq &	params
	)

Create a resend list with the copies of all current "sending" and "sent" samples. The samples will be sent to the subscriber specified.

Definition at line 326 of file WriteDataContainer.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), acked_sequences_, OpenDDS::DCPS::SendStateDataSampleList::begin(), cached_cumulative_ack_valid_, copy_and_prepend(), OpenDDS::DCPS::DCPS_debug_level, domain_id_, OpenDDS::DCPS::SendStateDataSampleList::end(), OpenDDS::DCPS::DisjointSequence::erase(), OpenDDS::DCPS::DataSampleElement::get_header(), OpenDDS::DCPS::GUID_UNKNOWN, instances_, LM_DEBUG, lock_, max_durable_per_instance_, publication_id_, resend_data_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, sending_data_, sent_data_, OpenDDS::DCPS::DataSampleHeader::sequence_, OpenDDS::DCPS::SendStateDataSampleList::size(), topic_name_, and wfa_lock_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    lock_,
                    DDS::RETCODE_ERROR);
 
   ssize_t total_size = 0;
   for (PublicationInstanceMapType::iterator it = instances_.begin();
        it != instances_.end(); ++it) {
     const ssize_t durable = std::min(it->second->samples_.size(),
                                      ssize_t(max_durable_per_instance_));
     total_size += durable;
     it->second->durable_samples_remaining_ = durable;
   }
 
   copy_and_prepend(resend_data_,
                    sending_data_,
                    reader_id,
                    lifespan,
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
                    filterClassName, eval, expression_params,
 #endif
                    total_size);
 
   copy_and_prepend(resend_data_,
                    sent_data_,
                    reader_id,
                    lifespan,
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
                    filterClassName, eval, expression_params,
 #endif
                    total_size);
 
   {
     ACE_Guard<ACE_SYNCH_MUTEX> guard(wfa_lock_);
     cached_cumulative_ack_valid_ = false;
     DisjointSequence& ds = acked_sequences_[reader_id];
     ds = acked_sequences_[GUID_UNKNOWN];
 
     // Remove exactly what will be sent
     SendStateDataSampleList::iterator iter = resend_data_.begin();
     while (iter != resend_data_.end()) {
       ds.erase(iter->get_header().sequence_);
       ++iter;
     }
   }
 
   if (DCPS_debug_level > 9 && resend_data_.size()) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) WriteDataContainer::reenqueue_all: ")
                ACE_TEXT("domain %d topic %C publication %C copying ")
                ACE_TEXT("sending/sent to resend to %C.\n"),
                domain_id_,
                topic_name_,
                LogGuid(publication_id_).c_str(),
                LogGuid(reader_id).c_str()));
   }
 
   return DDS::RETCODE_OK;
 }

◆ register_instance()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::register_instance	(	DDS::InstanceHandle_t &	instance_handle,
		Message_Block_Ptr &	registered_sample
	)

Dynamically allocate a PublicationInstance object and add to the instances_ list.

Note: : The registered_sample is an input and output parameter. A shallow copy of the sample data will be given to datawriter as part of the control message.

Definition at line 397 of file WriteDataContainer.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::bind(), ACE_Message_Block::duplicate(), extend_deadline(), OpenDDS::DCPS::find(), OpenDDS::DCPS::DataWriterImpl::get_next_handle(), DDS::HANDLE_NIL, OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::PublicationInstance::instance_handle_, instances_, LM_ERROR, max_num_instances_, OpenDDS::DCPS::move(), OpenDDS::DCPS::PublicationInstance::registered_sample_, OpenDDS::DCPS::RcHandle< T >::reset(), OpenDDS::DCPS::unique_ptr< T, Deleter >::reset(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, DDS::RETCODE_OUT_OF_RESOURCES, and writer_.

 {
   PublicationInstance_rch instance;
 
   if (instance_handle == DDS::HANDLE_NIL) {
     if (max_num_instances_ > 0
         && max_num_instances_ <= (CORBA::Long) instances_.size()) {
       return DDS::RETCODE_OUT_OF_RESOURCES;
     }
 
     // registered the instance for the first time.
     instance.reset(new PublicationInstance(move(registered_sample)), keep_count());
 
     instance_handle = this->writer_->get_next_handle();
 
     int const insert_attempt = OpenDDS::DCPS::bind(instances_, instance_handle, instance);
 
     if (0 != insert_attempt) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: ")
                  ACE_TEXT("WriteDataContainer::register_instance, ")
                  ACE_TEXT("failed to insert instance handle=%X\n"),
                  instance.in()));
       return DDS::RETCODE_ERROR;
     } // if (0 != insert_attempt)
 
     instance->instance_handle_ = instance_handle;
 
     extend_deadline(instance);
 
   } else {
 
     int const find_attempt = find(instances_, instance_handle, instance);
 
     if (0 != find_attempt) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: ")
                  ACE_TEXT("WriteDataContainer::register_instance, ")
                  ACE_TEXT("The provided instance handle=%X is not a valid")
                  ACE_TEXT("handle.\n"),
                  instance_handle));
 
       return DDS::RETCODE_ERROR;
     } // if (0 != find_attempt)
   }
 
   // The registered_sample is shallow copied.
   registered_sample.reset(instance->registered_sample_->duplicate());
 
   return DDS::RETCODE_OK;
 }

◆ release_buffer()

void OpenDDS::DCPS::WriteDataContainer::release_buffer ( DataSampleElement * element )

Release the memory previously allocated. This method is corresponding to the obtain_buffer method. If the memory is allocated by some allocator then the memory needs to be released to the allocator.

Definition at line 1310 of file WriteDataContainer.cpp.

References ACE_DES_FREE, data_holder_, OpenDDS::DCPS::WriterDataSampleList::dequeue(), OpenDDS::DCPS::Cached_Allocator_With_Overflow< T, ACE_LOCK >::free(), OpenDDS::DCPS::DataSampleElement::get_header(), OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::SAMPLE_DATA, and sample_list_element_allocator_.

Referenced by data_delivered(), data_dropped(), obtain_buffer(), remove_excess_durable(), remove_oldest_sample(), and ~WriteDataContainer().

 {
   if (element->get_header().message_id_ == SAMPLE_DATA)
     data_holder_.dequeue(element);
   // Release the memory to the allocator.
   ACE_DES_FREE(element,
                sample_list_element_allocator_.free,
                DataSampleElement);
 }

◆ remove_excess_durable()

void OpenDDS::DCPS::WriteDataContainer::remove_excess_durable ( )

private

Remove the oldest "n" samples from each instance list that are in a state such that they could only be used for durability purposes (see reenqueue_all). "n" is determined by max_durable_per_instance_, so these samples are truly unneeded – there are max_durable_per_instance_ newer samples available in the instance.

Definition at line 941 of file WriteDataContainer.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::InstanceDataSampleList::dequeue(), OpenDDS::DCPS::SendStateDataSampleList::dequeue(), domain_id_, OpenDDS::DCPS::HISTORIC_SAMPLE_FLAG, instances_, LM_DEBUG, max_durable_per_instance_, OpenDDS::DCPS::InstanceDataSampleList::prev(), publication_id_, release_buffer(), sent_data_, OpenDDS::DCPS::InstanceDataSampleList::tail(), OpenDDS::DCPS::DataSampleHeader::test_flag(), and topic_name_.

Referenced by obtain_buffer().

 {
   if (!max_durable_per_instance_)
     return;
 
   size_t n_released = 0;
 
   for (PublicationInstanceMapType::iterator iter = instances_.begin();
        iter != instances_.end();
        ++iter) {
 
     CORBA::Long durable_allowed = max_durable_per_instance_;
     InstanceDataSampleList& instance_list = iter->second->samples_;
 
     for (DataSampleElement* it = instance_list.tail(), *prev; it; it = prev) {
       prev = InstanceDataSampleList::prev(it);
 
       if (DataSampleHeader::test_flag(HISTORIC_SAMPLE_FLAG, it->get_sample())) {
 
         if (durable_allowed) {
           --durable_allowed;
         } else {
           instance_list.dequeue(it);
           sent_data_.dequeue(it);
           release_buffer(it);
           ++n_released;
         }
       }
     }
   }
 
   if (n_released && DCPS_debug_level > 9) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) WriteDataContainer::remove_excess_durable: ")
                ACE_TEXT("domain %d topic %C publication %C %B samples removed ")
                ACE_TEXT("from durable data.\n"), domain_id_, topic_name_,
                LogGuid(publication_id_).c_str(), n_released));
   }
 }

◆ remove_instance()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::remove_instance	(	PublicationInstance_rch	instance,
		Message_Block_Ptr &	registered_sample,
		bool	dup_registered_sample
	)

private

Definition at line 509 of file WriteDataContainer.cpp.

References cancel_deadline(), ACE_Message_Block::duplicate(), OpenDDS::DCPS::PublicationInstance::registered_sample_, remove_oldest_sample(), OpenDDS::DCPS::unique_ptr< T, Deleter >::reset(), DDS::RETCODE_OK, OpenDDS::DCPS::PublicationInstance::samples_, and OpenDDS::DCPS::InstanceDataSampleList::size().

Referenced by dispose(), unregister(), and unregister_all().

 {
   if (dup_registered_sample) {
     // The registered_sample is shallow copied.
     registered_sample.reset(instance->registered_sample_->duplicate());
   }
 
   // Note: The DDS specification is unclear as to if samples in the process
   // of being sent should be removed or not.
   // The advantage of calling remove_sample() on them is that the
   // cached allocator memory for them is freed.  The disadvantage
   // is that the slow reader may see multiple disposes without
   // any write sample between them and hence not temporarily move into the
   // Alive state.
   // We have chosen to NOT remove the sending samples.
   InstanceDataSampleList& instance_list = instance->samples_;
 
   while (instance_list.size() > 0) {
     bool released = false;
     const DDS::ReturnCode_t ret = remove_oldest_sample(instance_list, released);
     if (ret != DDS::RETCODE_OK) {
       return ret;
     }
   }
 
   cancel_deadline(instance);
 
   return DDS::RETCODE_OK;
 }

◆ remove_oldest_sample()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::remove_oldest_sample	(	InstanceDataSampleList &	instance_list,
		bool &	released
	)

private

Remove the oldest sample (head) from the instance history list. This method also updates the internal lists to reflect the change. If the sample is in the unsent_data_ or sent_data_ list then it will be released. If the sample is in the sending_data_ list then the transport will be notified to release the sample, then the sample will be released. Otherwise an error is returned. The "released" boolean value indicates whether the sample is released.

Definition at line 983 of file WriteDataContainer.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_ERROR_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::SendStateDataSampleList::dequeue(), OpenDDS::DCPS::InstanceDataSampleList::dequeue_head(), domain_id_, empty_condition_, LM_DEBUG, LM_ERROR, LM_WARNING, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), orphaned_to_transport_, pending_data(), publication_id_, release_buffer(), OpenDDS::DCPS::TransportClient::remove_sample(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::SendStateDataSampleList::send_list_containing_element(), sending_data_, sent_data_, topic_name_, unsent_data_, and writer_.

Referenced by obtain_buffer(), and remove_instance().

 {
   DataSampleElement* stale = 0;
 
   //
   // Remove the oldest sample from the instance list.
   //
   if (!instance_list.dequeue_head(stale)) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("WriteDataContainer::remove_oldest_sample, ")
                       ACE_TEXT("dequeue_head_next_sample failed\n")),
                      DDS::RETCODE_ERROR);
   }
 
   //
   // Remove the stale data from the next_writer_sample_ list.  The
   // sending_data_/next_send_sample_ list is not managed within the
   // container, it is only used external to the container and does
   // not need to be managed internally.
   //
   // The next_writer_sample_ link is being used in one of the sent_data_,
   // sending_data_, or unsent_data lists.  Removal from the doubly
   // linked list needs to repair the list only when the stale sample
   // is either the head or tail of the list.
   //
 
   //
   // Locate the head of the list that the stale data is in.
   //
   SendStateDataSampleList* send_lists[] = {
     &sending_data_,
     &sent_data_,
     &unsent_data_,
     &orphaned_to_transport_};
   const SendStateDataSampleList* containing_list =
     SendStateDataSampleList::send_list_containing_element(stale, send_lists);
 
   //
   // Identify the list that the stale data is in.
   // The stale data should be in one of the sent_data_, sending_data_
   // or unsent_data_. It should not be in released_data_ list since
   // this function is the only place a sample is moved from
   // sending_data_ to released_data_ list.
 
   // Remove the element from the internal list.
   bool result = false;
 
   if (containing_list == &this->sending_data_) {
     if (DCPS_debug_level > 2) {
       ACE_ERROR((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: ")
                  ACE_TEXT("WriteDataContainer::remove_oldest_sample, ")
                  ACE_TEXT("removing from sending_data_ so must notify transport to remove sample\n")));
     }
 
     // This means transport is still using the sample that needs to
     // be released currently so notify transport that sample is being removed.
 
     if (this->writer_->remove_sample(stale)) {
       if (this->sent_data_.dequeue(stale)) {
         release_buffer(stale);
       }
       result = true;
 
     } else {
       if (this->sending_data_.dequeue(stale)) {
         this->orphaned_to_transport_.enqueue_tail(stale);
       } else if (this->sent_data_.dequeue(stale)) {
         release_buffer(stale);
         result = true;
       }
       result = true;
     }
     released = true;
 
   } else if (containing_list == &this->sent_data_) {
     // No one is using the data sample, so we can release it back to
     // its allocator.
     //
     result = this->sent_data_.dequeue(stale) != 0;
     release_buffer(stale);
     released = true;
 
     if (DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) WriteDataContainer::remove_oldest_sample: ")
                  ACE_TEXT("domain %d topic %C publication %C sample removed from HISTORY.\n"),
                  this->domain_id_,
                  this->topic_name_,
                  LogGuid(publication_id_).c_str()));
     }
 
   } else if (containing_list == &this->unsent_data_) {
     //
     // No one is using the data sample, so we can release it back to
     // its allocator.
     //
     result = this->unsent_data_.dequeue(stale) != 0;
     release_buffer(stale);
     released = true;
 
     if (DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) WriteDataContainer::remove_oldest_sample: ")
                  ACE_TEXT("domain %d topic %C publication %C sample removed from unsent.\n"),
                  this->domain_id_,
                  this->topic_name_,
                  LogGuid(publication_id_).c_str()));
     }
   } else {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("WriteDataContainer::remove_oldest_sample, ")
                       ACE_TEXT("The oldest sample is not in any internal list.\n")),
                      DDS::RETCODE_ERROR);
   }
 
   if (!pending_data()) {
     empty_condition_.notify_all();
   }
 
   if (!result) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("WriteDataContainer::remove_oldest_sample, ")
                       ACE_TEXT("dequeue_next_send_sample from internal list failed.\n")),
                      DDS::RETCODE_ERROR);
 
   }
 
   return DDS::RETCODE_OK;
 }

◆ remove_reader_acks()

void OpenDDS::DCPS::WriteDataContainer::remove_reader_acks ( const GUID_t & reader )

private

Definition at line 191 of file WriteDataContainer.cpp.

References acked_sequences_, cached_cumulative_ack_valid_, get_cumulative_ack(), OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), wfa_condition_, and wfa_lock_.

 {
   ACE_Guard<ACE_Thread_Mutex> guard(wfa_lock_);
 
   const SequenceNumber prev_cum_ack = get_cumulative_ack();
   const AckedSequenceMap::iterator it = acked_sequences_.find(reader);
   if (it != acked_sequences_.end()) {
     acked_sequences_.erase(it);
     cached_cumulative_ack_valid_ = false;
     if (prev_cum_ack != get_cumulative_ack()) {
       wfa_condition_.notify_all();
     }
   }
 }

◆ sequence_acknowledged()

bool OpenDDS::DCPS::WriteDataContainer::sequence_acknowledged ( const SequenceNumber & sequence )

Definition at line 1557 of file WriteDataContainer.cpp.

References sequence_acknowledged_i(), and wfa_lock_.

 {
   ACE_Guard<ACE_SYNCH_MUTEX> guard(wfa_lock_);
   return sequence_acknowledged_i(sequence);
 }

◆ sequence_acknowledged_i()

bool OpenDDS::DCPS::WriteDataContainer::sequence_acknowledged_i ( const SequenceNumber & sequence )

private

Definition at line 1564 of file WriteDataContainer.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, get_cumulative_ack(), OpenDDS::DCPS::SequenceNumber::getValue(), LM_DEBUG, publication_id_, and OpenDDS::DCPS::SequenceNumber::SEQUENCENUMBER_UNKNOWN().

Referenced by sequence_acknowledged(), and wait_ack_of_seq().

 {
   if (sequence == SequenceNumber::SEQUENCENUMBER_UNKNOWN()) {
     //return true here so that wait_for_acknowledgments doesn't block
     return true;
   }
 
   SequenceNumber acked = get_cumulative_ack();
   if (DCPS_debug_level >= 10) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::sequence_acknowledged_i ")
                           ACE_TEXT("- %C cumulative ack is currently: %q\n"), DCPS::LogGuid(publication_id_).c_str(), acked.getValue()));
   }
   if (acked == SequenceNumber::SEQUENCENUMBER_UNKNOWN() || acked < sequence){
     return false;
   }
   return true;
 }

◆ set_deadline_period()

void OpenDDS::DCPS::WriteDataContainer::set_deadline_period ( const TimeDuration & deadline_period )

private

Definition at line 1606 of file WriteDataContainer.cpp.

References deadline_map_, deadline_period_, deadline_task_, instances_, OpenDDS::DCPS::TimeDuration::max_value, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OPENDDS_ASSERT, and OpenDDS::DCPS::swap().

 {
   // Call comes from DataWriterImpl_t which should arleady have the lock_.
 
   // Deadline for all instances starting from now.
   const MonotonicTimePoint deadline = MonotonicTimePoint::now() + deadline_period;
 
   // Reset the deadline timer if the period has changed.
   if (deadline_period_ != deadline_period) {
     if (deadline_period_ == TimeDuration::max_value) {
       OPENDDS_ASSERT(deadline_map_.empty());
 
       for (PublicationInstanceMapType::iterator iter = instances_.begin();
            iter != instances_.end();
            ++iter) {
         iter->second->deadline_ = deadline;
         deadline_map_.insert(std::make_pair(deadline, iter->second));
       }
 
       if (!deadline_map_.empty()) {
         deadline_task_->schedule(deadline_period);
       }
     } else if (deadline_period == TimeDuration::max_value) {
       if (!deadline_map_.empty()) {
         deadline_task_->cancel();
       }
 
       deadline_map_.clear();
     } else {
       DeadlineMapType new_map;
       for (PublicationInstanceMapType::iterator iter = instances_.begin();
            iter != instances_.end();
            ++iter) {
         iter->second->deadline_ = deadline;
         new_map.insert(std::make_pair(iter->second->deadline_, iter->second));
       }
       std::swap(new_map, deadline_map_);
 
       if (!deadline_map_.empty()) {
         deadline_task_->cancel();
         deadline_task_->schedule(deadline_map_.begin()->first - MonotonicTimePoint::now());
       }
     }
 
     deadline_period_ = deadline_period;
   }
 }

◆ unregister()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::unregister	(	DDS::InstanceHandle_t	handle,
		Message_Block_Ptr &	registered_sample,
		bool	dup_registered_sample = `true`
	)

Remove the provided instance from the instances_ list. The registered sample data will be released upon the deletion of the PublicationInstance. A shallow copy of the sample data will be given to datawriter as part of the control message if the dup_registered_sample is true.

This method returns error if the instance is not registered.

Definition at line 452 of file WriteDataContainer.cpp.

References ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_TEXT(), instances_, LM_ERROR, lock_, remove_instance(), DDS::RETCODE_ERROR, and DDS::RETCODE_PRECONDITION_NOT_MET.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    lock_,
                    DDS::RETCODE_ERROR);
 
   PublicationInstance_rch instance;
   {
     PublicationInstanceMapType::iterator pos = instances_.find(instance_handle);
     if (pos == instances_.end()) {
       ACE_ERROR_RETURN((LM_ERROR,
                         ACE_TEXT("(%P|%t) ERROR: ")
                         ACE_TEXT("WriteDataContainer::unregister, ")
                         ACE_TEXT("The instance(handle=%X) ")
                         ACE_TEXT("is not registered yet.\n"),
                         instance_handle),
                        DDS::RETCODE_PRECONDITION_NOT_MET);
     }
     instance = pos->second;
     instances_.erase(pos);
   }
 
   return remove_instance(instance, registered_sample, dup_registered_sample);
 }

◆ unregister_all()

void OpenDDS::DCPS::WriteDataContainer::unregister_all ( )

Unregister all instances managed by this data containers.

Definition at line 1321 of file WriteDataContainer.cpp.

References ACE_ERROR, ACE_GUARD, condition_, DBG_ENTRY_LVL, instances_, LM_ERROR, lock_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), OpenDDS::DCPS::TransportClient::remove_all_msgs(), remove_instance(), DDS::RETCODE_OK, OpenDDS::DCPS::DataWriterImpl::return_handle(), shutdown_, waiting_on_release_, and writer_.

 {
   DBG_ENTRY_LVL("WriteDataContainer","unregister_all",6);
   shutdown_ = true;
 
   //The internal list needs protection since this call may result from the
   //the delete_datawriter call which does not acquire the lock in advance.
   ACE_GUARD(ACE_Recursive_Thread_Mutex,
             guard,
             lock_);
   // Tell transport remove all control messages currently
   // transport is processing.
   (void) this->writer_->remove_all_msgs();
 
   // Broadcast to wake up all waiting threads.
   if (waiting_on_release_) {
     condition_.notify_all();
   }
 
   Message_Block_Ptr registered_sample;
 
   for (PublicationInstanceMapType::iterator pos = instances_.begin(), limit = instances_.end(); pos != limit;) {
     // Release the instance data.
     if (remove_instance(pos->second, registered_sample, false) != DDS::RETCODE_OK) {
       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: WriteDataContainer::unregister_all, "
                  "remove_instance %X failed\n", pos->first));
     }
 
     writer_->return_handle(pos->first);
     instances_.erase(pos++);
   }
 }

◆ update_acked()

void OpenDDS::DCPS::WriteDataContainer::update_acked	(	const SequenceNumber &	seq,
		const GUID_t &	id = `GUID_UNKNOWN`
	)

private

Definition at line 245 of file WriteDataContainer.cpp.

References acked_sequences_, cached_cumulative_ack_valid_, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), and wfa_condition_.

Referenced by data_delivered().

 {
   bool do_notify = false;
   if (id == GUID_UNKNOWN) {
     for (AckedSequenceMap::iterator it = acked_sequences_.begin(); it != acked_sequences_.end(); ++it) {
       SequenceNumber prev_cum_ack = it->second.cumulative_ack();
       it->second.insert(seq);
       cached_cumulative_ack_valid_ = false;
       if (prev_cum_ack != it->second.cumulative_ack()) {
         do_notify = true;
       }
     }
   } else {
     const AckedSequenceMap::iterator it = acked_sequences_.find(id);
     if (it != acked_sequences_.end()) {
       SequenceNumber prev_cum_ack = it->second.cumulative_ack();
       if (prev_cum_ack < seq) {
         it->second.insert(SequenceRange(prev_cum_ack, seq));
         cached_cumulative_ack_valid_ = false;
         if (prev_cum_ack != it->second.cumulative_ack()) {
           do_notify = true;
         }
       }
     }
   }
   if (do_notify) {
     wfa_condition_.notify_all();
   }
 }

◆ wait_ack_of_seq()

DDS::ReturnCode_t OpenDDS::DCPS::WriteDataContainer::wait_ack_of_seq	(	const MonotonicTimePoint &	abs_deadline,
		bool	deadline_is_infinite,
		const SequenceNumber &	sequence
	)

Definition at line 1527 of file WriteDataContainer.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::CvStatus_Error, OpenDDS::DCPS::CvStatus_NoTimeout, OpenDDS::DCPS::CvStatus_Timeout, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::SequenceNumber::getValue(), LM_DEBUG, LM_ERROR, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, DDS::RETCODE_TIMEOUT, sequence_acknowledged_i(), TheServiceParticipant, OpenDDS::DCPS::ConditionVariable< Mutex >::wait(), OpenDDS::DCPS::ConditionVariable< Mutex >::wait_until(), wfa_condition_, and wfa_lock_.

 {
   ACE_Guard<ACE_SYNCH_MUTEX> guard(wfa_lock_);
   ThreadStatusManager& thread_status_manager = TheServiceParticipant->get_thread_status_manager();
   while ((deadline_is_infinite || MonotonicTimePoint::now() < deadline) && !sequence_acknowledged_i(sequence)) {
     switch (deadline_is_infinite ? wfa_condition_.wait(thread_status_manager) : wfa_condition_.wait_until(deadline, thread_status_manager)) {
     case CvStatus_NoTimeout:
       break;
     case CvStatus_Timeout:
       if (DCPS_debug_level >= 2) {
         ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::wait_ack_of_seq")
                    ACE_TEXT(" timed out waiting for sequence %q to be acked\n"),
                    sequence.getValue()));
         }
       return DDS::RETCODE_TIMEOUT;
     case CvStatus_Error:
       if (DCPS_debug_level) {
         ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: WriteDataContainer::wait_ack_of_seq: "
                    "error in wait/wait_until\n"));
       }
       return DDS::RETCODE_ERROR;
     }
   }
 
   return sequence_acknowledged_i(sequence) ? DDS::RETCODE_OK : DDS::RETCODE_TIMEOUT;
 }

◆ wait_pending()

void OpenDDS::DCPS::WriteDataContainer::wait_pending ( const MonotonicTimePoint & deadline )

Block until pending samples have either been delivered, dropped, or the deadline has passed. Blocks indefinitely if deadline is zero.

Definition at line 1465 of file WriteDataContainer.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::CvStatus_Error, OpenDDS::DCPS::CvStatus_NoTimeout, OpenDDS::DCPS::CvStatus_Timeout, OpenDDS::DCPS::DCPS_debug_level, empty_condition_, OpenDDS::DCPS::TimePoint_T< AceClock >::is_zero(), LM_DEBUG, LM_ERROR, LM_INFO, lock_, log_send_state_lists(), pending_data(), TheServiceParticipant, OpenDDS::DCPS::TimePoint_T< AceClock >::value(), and OpenDDS::DCPS::ConditionVariable< Mutex >::wait_until().

 {
   const bool no_deadline = deadline.is_zero();
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, lock_);
   const bool report = DCPS_debug_level > 0 && pending_data();
   if (report) {
     if (no_deadline) {
       ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::wait_pending no timeout\n")));
     } else {
       ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::wait_pending ")
         ACE_TEXT("timeout at %#T\n"),
         &deadline.value()));
     }
   }
 
   bool loop = true;
   ThreadStatusManager& thread_status_manager = TheServiceParticipant->get_thread_status_manager();
   while (loop && pending_data()) {
     switch (empty_condition_.wait_until(deadline, thread_status_manager)) {
     case CvStatus_NoTimeout:
       break;
 
     case CvStatus_Timeout:
       if (pending_data()) {
         if (DCPS_debug_level >= 2) {
           ACE_DEBUG((LM_INFO, "(%P|%t) WriteDataContainer::wait_pending: "
             "Timed out waiting for messages to be transported\n"));
           log_send_state_lists("WriteDataContainer::wait_pending - wait timedout: ");
         }
       }
       loop = false;
       break;
 
     case CvStatus_Error:
       if (DCPS_debug_level) {
         ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: WriteDataContainer::wait_pending: "
           "error in wait_until\n"));
       }
       loop = false;
       break;
     }
   }
   if (report) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) WriteDataContainer::wait_pending done\n")));
   }
 }

◆ wakeup_blocking_writers()

void OpenDDS::DCPS::WriteDataContainer::wakeup_blocking_writers ( DataSampleElement * stale )

private

Called when data has been dropped or delivered and any blocked writers should be notified

Definition at line 1583 of file WriteDataContainer.cpp.

References condition_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), and waiting_on_release_.

Referenced by data_delivered(), and data_dropped().

 {
   if (!stale && waiting_on_release_) {
     waiting_on_release_ = false;
 
     condition_.notify_all();
   }
 }

Friends And Related Function Documentation

◆ ::DDS_TEST

friend class ::DDS_TEST

friend

Definition at line 356 of file WriteDataContainer.h.

◆ DataWriterImpl

friend class DataWriterImpl

friend

Definition at line 125 of file WriteDataContainer.h.

Member Data Documentation

◆ acked_sequences_

AckedSequenceMap OpenDDS::DCPS::WriteDataContainer::acked_sequences_

private

Definition at line 426 of file WriteDataContainer.h.

Referenced by add_reader_acks(), get_cumulative_ack(), get_last_ack(), reenqueue_all(), remove_reader_acks(), update_acked(), and WriteDataContainer().

◆ cached_cumulative_ack_

SequenceNumber OpenDDS::DCPS::WriteDataContainer::cached_cumulative_ack_

private

Definition at line 427 of file WriteDataContainer.h.

Referenced by get_cumulative_ack().

◆ cached_cumulative_ack_valid_

bool OpenDDS::DCPS::WriteDataContainer::cached_cumulative_ack_valid_

private

Definition at line 428 of file WriteDataContainer.h.

Referenced by add_reader_acks(), get_cumulative_ack(), reenqueue_all(), remove_reader_acks(), and update_acked().

◆ condition_

ConditionVariableType OpenDDS::DCPS::WriteDataContainer::condition_

private

Definition at line 513 of file WriteDataContainer.h.

Referenced by obtain_buffer(), unregister_all(), and wakeup_blocking_writers().

◆ data_holder_

WriterDataSampleList OpenDDS::DCPS::WriteDataContainer::data_holder_

private

The list of all samples written to this datawriter in writing order.

Definition at line 454 of file WriteDataContainer.h.

Referenced by obtain_buffer(), and release_buffer().

◆ deadline_last_total_count_

CORBA::Long& OpenDDS::DCPS::WriteDataContainer::deadline_last_total_count_

private

Last total_count when status was last checked.

Definition at line 571 of file WriteDataContainer.h.

Referenced by process_deadlines().

◆ deadline_map_

DeadlineMapType OpenDDS::DCPS::WriteDataContainer::deadline_map_

private

Definition at line 562 of file WriteDataContainer.h.

Referenced by cancel_deadline(), extend_deadline(), process_deadlines(), and set_deadline_period().

◆ deadline_period_

TimeDuration OpenDDS::DCPS::WriteDataContainer::deadline_period_

private

Definition at line 560 of file WriteDataContainer.h.

Referenced by cancel_deadline(), extend_deadline(), process_deadlines(), and set_deadline_period().

◆ deadline_status_

DDS::OfferedDeadlineMissedStatus& OpenDDS::DCPS::WriteDataContainer::deadline_status_

private

Reference to the missed requested deadline status structure.

Definition at line 568 of file WriteDataContainer.h.

Referenced by process_deadlines().

◆ deadline_status_lock_

ACE_Recursive_Thread_Mutex& OpenDDS::DCPS::WriteDataContainer::deadline_status_lock_

private

Lock for synchronization of status_ member.

Definition at line 565 of file WriteDataContainer.h.

Referenced by process_deadlines().

◆ deadline_task_

RcHandle<DCPS::PmfSporadicTask<WriteDataContainer> > OpenDDS::DCPS::WriteDataContainer::deadline_task_

private

Timer responsible for reporting missed offered deadlines.

Definition at line 559 of file WriteDataContainer.h.

Referenced by cancel_deadline(), extend_deadline(), process_deadlines(), set_deadline_period(), and ~WriteDataContainer().

◆ domain_id_

DDS::DomainId_t const OpenDDS::DCPS::WriteDataContainer::domain_id_

private

Domain ID.

Definition at line 535 of file WriteDataContainer.h.

Referenced by data_delivered(), data_dropped(), persist_data(), reenqueue_all(), remove_excess_durable(), and remove_oldest_sample().

◆ durability_cache_

DataDurabilityCache* const OpenDDS::DCPS::WriteDataContainer::durability_cache_

private

Pointer to the data durability cache.

This a pointer to the data durability cache owned by the Service Participant singleton, which means this cache is also a singleton.

Definition at line 551 of file WriteDataContainer.h.

Referenced by persist_data().

◆ durability_service_

DDS::DurabilityServiceQosPolicy const& OpenDDS::DCPS::WriteDataContainer::durability_service_

private

DURABILITY_SERVICE QoS specific to the DataWriter.

Definition at line 554 of file WriteDataContainer.h.

Referenced by persist_data().

◆ empty_condition_

ConditionVariableType OpenDDS::DCPS::WriteDataContainer::empty_condition_

private

Definition at line 514 of file WriteDataContainer.h.

Referenced by data_delivered(), data_dropped(), remove_oldest_sample(), and wait_pending().

◆ history_depth_

CORBA::Long OpenDDS::DCPS::WriteDataContainer::history_depth_

private

Definition at line 475 of file WriteDataContainer.h.

Referenced by obtain_buffer().

◆ instances_

PublicationInstanceMapType OpenDDS::DCPS::WriteDataContainer::instances_

private

The individual instance queue threads in the data.

Definition at line 463 of file WriteDataContainer.h.

Referenced by dispose(), get_handle_instance(), get_instance_handles(), log_send_state_lists(), num_all_samples(), num_samples(), obtain_buffer(), reenqueue_all(), register_instance(), remove_excess_durable(), set_deadline_period(), unregister(), and unregister_all().

◆ lock_

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::WriteDataContainer::lock_

mutableprivate

This lock is used to protect the container and the map in the type-specific DataWriter. This lock can be accessible via the datawriter. This lock is made to be globally accessible for performance concern. The lock is acquired as the external call (e.g. FooDataWriterImpl::write) started and the same lock will be used by the transport thread to notify the datawriter the data is delivered. Other internal operations will not lock.

Definition at line 511 of file WriteDataContainer.h.

Referenced by data_delivered(), data_dropped(), dispose(), get_instance_handles(), num_all_samples(), num_samples(), process_deadlines(), reenqueue_all(), unregister(), unregister_all(), and wait_pending().

◆ max_blocking_time_

DDS::Duration_t OpenDDS::DCPS::WriteDataContainer::max_blocking_time_

private

The maximum time to block on write operation. This comes from DataWriter's QoS HISTORY.max_blocking_time

Definition at line 497 of file WriteDataContainer.h.

Referenced by obtain_buffer().

◆ max_durable_per_instance_

CORBA::Long OpenDDS::DCPS::WriteDataContainer::max_durable_per_instance_

private

The maximum number of samples from each instance that can be added to the resend_data_ for durability.

Definition at line 479 of file WriteDataContainer.h.

Referenced by data_delivered(), reenqueue_all(), and remove_excess_durable().

◆ max_num_instances_

CORBA::Long OpenDDS::DCPS::WriteDataContainer::max_num_instances_

private

The maximum number of instances allowed or zero to indicate unlimited. It corresponds to the QoS.RESOURCE_LIMITS.max_instances when QoS.RELIABILITY.kind == DDS::RELIABLE_RELIABILITY_QOS

Definition at line 485 of file WriteDataContainer.h.

Referenced by register_instance().

◆ max_num_samples_

CORBA::Long OpenDDS::DCPS::WriteDataContainer::max_num_samples_

private

The maximum number of samples allowed or zero to indicate unlimited. It corresponds to the QoS.RESOURCE_LIMITS.max_instances when QoS.RELIABILITY.kind == DDS::RELIABLE_RELIABILITY_QOS It also covers QoS.RESOURCE_LIMITS.max_samples and max_instances * max_samples_per_instance

Definition at line 493 of file WriteDataContainer.h.

Referenced by obtain_buffer().

◆ max_samples_per_instance_

CORBA::Long OpenDDS::DCPS::WriteDataContainer::max_samples_per_instance_

private

The maximum size a container should allow for an instance sample list

Definition at line 473 of file WriteDataContainer.h.

Referenced by obtain_buffer().

◆ n_chunks_

size_t OpenDDS::DCPS::WriteDataContainer::n_chunks_

private

The number of chunks that sample_list_element_allocator_ needs initialize.

Definition at line 525 of file WriteDataContainer.h.

Referenced by WriteDataContainer().

◆ orphaned_to_transport_

SendStateDataSampleList OpenDDS::DCPS::WriteDataContainer::orphaned_to_transport_

private

List of data that has been released by WriteDataContainer but is still in process of delivery (or dropping) by transport

Definition at line 450 of file WriteDataContainer.h.

Referenced by data_delivered(), data_dropped(), log_send_state_lists(), pending_data(), remove_oldest_sample(), and ~WriteDataContainer().

◆ publication_id_

GUID_t OpenDDS::DCPS::WriteDataContainer::publication_id_

private

The publication Id from repo.

Definition at line 466 of file WriteDataContainer.h.

Referenced by data_delivered(), data_dropped(), obtain_buffer(), obtain_buffer_for_control(), reenqueue_all(), remove_excess_durable(), remove_oldest_sample(), and sequence_acknowledged_i().

◆ resend_data_

SendStateDataSampleList OpenDDS::DCPS::WriteDataContainer::resend_data_

private

List of the data reenqueued to support the TRANSIENT_LOCAL_DURABILITY_QOS policy. It duplicates the samples in sent and sending list. This list will be passed to the transport for re-sending.

Definition at line 460 of file WriteDataContainer.h.

Referenced by get_resend_data(), and reenqueue_all().

◆ sample_list_element_allocator_

DataSampleElementAllocator OpenDDS::DCPS::WriteDataContainer::sample_list_element_allocator_

private

The cached allocator to allocate DataSampleElement objects.

Definition at line 529 of file WriteDataContainer.h.

Referenced by copy_and_prepend(), obtain_buffer(), obtain_buffer_for_control(), release_buffer(), and WriteDataContainer().

◆ sending_data_

SendStateDataSampleList OpenDDS::DCPS::WriteDataContainer::sending_data_

private

List of data that is currently being sent.

Definition at line 443 of file WriteDataContainer.h.

Referenced by data_delivered(), data_dropped(), get_unsent_data(), log_send_state_lists(), pending_data(), reenqueue_all(), remove_oldest_sample(), and ~WriteDataContainer().

◆ sent_data_

SendStateDataSampleList OpenDDS::DCPS::WriteDataContainer::sent_data_

private

List of data that has already been sent.

Definition at line 446 of file WriteDataContainer.h.

Referenced by data_delivered(), data_dropped(), log_send_state_lists(), persist_data(), reenqueue_all(), remove_excess_durable(), remove_oldest_sample(), and ~WriteDataContainer().

◆ shutdown_

bool OpenDDS::DCPS::WriteDataContainer::shutdown_

private

The flag indicates the datawriter will be destroyed.

Definition at line 532 of file WriteDataContainer.h.

Referenced by data_delivered(), data_dropped(), enqueue(), enqueue_control(), obtain_buffer(), unregister_all(), and ~WriteDataContainer().

◆ topic_name_

char const* const OpenDDS::DCPS::WriteDataContainer::topic_name_

private

Topic name.

Definition at line 538 of file WriteDataContainer.h.

Referenced by data_delivered(), data_dropped(), persist_data(), reenqueue_all(), remove_excess_durable(), and remove_oldest_sample().

◆ transaction_id_

ACE_UINT64 OpenDDS::DCPS::WriteDataContainer::transaction_id_

private

Id used to keep track of which send transaction DataWriter is currently creating

Definition at line 440 of file WriteDataContainer.h.

Referenced by get_unsent_data().

◆ type_name_

char const* const OpenDDS::DCPS::WriteDataContainer::type_name_

private

Type name.

Definition at line 541 of file WriteDataContainer.h.

Referenced by persist_data().

◆ unsent_data_

SendStateDataSampleList OpenDDS::DCPS::WriteDataContainer::unsent_data_

private

List of data that has not been sent yet.

Definition at line 436 of file WriteDataContainer.h.

Referenced by data_delivered(), data_dropped(), enqueue(), enqueue_control(), get_unsent_data(), log_send_state_lists(), pending_data(), remove_oldest_sample(), and ~WriteDataContainer().

◆ waiting_on_release_

bool OpenDDS::DCPS::WriteDataContainer::waiting_on_release_

private

The block waiting flag.

Definition at line 500 of file WriteDataContainer.h.

Referenced by obtain_buffer(), unregister_all(), and wakeup_blocking_writers().

◆ wfa_condition_

WfaConditionVariableType OpenDDS::DCPS::WriteDataContainer::wfa_condition_

private

Used to block in wait_for_acks().

Definition at line 521 of file WriteDataContainer.h.

Referenced by data_delivered(), remove_reader_acks(), update_acked(), and wait_ack_of_seq().

◆ wfa_lock_

ACE_Thread_Mutex OpenDDS::DCPS::WriteDataContainer::wfa_lock_

private

Lock used for wait_for_acks() processing.

Definition at line 517 of file WriteDataContainer.h.

Referenced by add_reader_acks(), data_delivered(), reenqueue_all(), remove_reader_acks(), sequence_acknowledged(), and wait_ack_of_seq().

◆ writer_

DataWriterImpl* OpenDDS::DCPS::WriteDataContainer::writer_

private

The writer that owns this container.

Definition at line 469 of file WriteDataContainer.h.

Referenced by copy_and_prepend(), data_delivered(), data_dropped(), obtain_buffer(), obtain_buffer_for_control(), process_deadlines(), register_instance(), remove_oldest_sample(), and unregister_all().

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

Public Member Functions

Private Types

Private Member Functions

Private Attributes

Friends

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ ConditionVariableType

◆ WfaConditionVariableType

Constructor & Destructor Documentation

◆ WriteDataContainer() [1/2]

◆ ~WriteDataContainer()

◆ WriteDataContainer() [2/2]

Member Function Documentation

◆ add_reader_acks()

◆ cancel_deadline()

◆ copy_and_prepend()

◆ data_delivered()

◆ data_dropped()

◆ dispose()

◆ enqueue()

◆ enqueue_control()

◆ extend_deadline()

◆ get_cumulative_ack()

◆ get_handle_instance()

◆ get_instance_handles()

◆ get_last_ack()

◆ get_resend_data()

◆ get_unsent_data()

◆ log_send_state_lists()

◆ num_all_samples()

◆ num_samples()

◆ obtain_buffer()

◆ obtain_buffer_for_control()

◆ OPENDDS_MAP_CMP()

◆ OPENDDS_MULTIMAP()

◆ OPENDDS_VECTOR()

◆ operator=()

◆ pending_data()

◆ persist_data()

◆ process_deadlines()

◆ reenqueue_all()

◆ register_instance()

◆ release_buffer()

◆ remove_excess_durable()

◆ remove_instance()

◆ remove_oldest_sample()

◆ remove_reader_acks()

◆ sequence_acknowledged()

◆ sequence_acknowledged_i()

◆ set_deadline_period()

◆ unregister()

◆ unregister_all()

◆ update_acked()

◆ wait_ack_of_seq()

◆ wait_pending()

◆ wakeup_blocking_writers()

Friends And Related Function Documentation

◆ ::DDS_TEST

◆ DataWriterImpl

Member Data Documentation

◆ acked_sequences_

◆ cached_cumulative_ack_

◆ cached_cumulative_ack_valid_

◆ condition_

◆ data_holder_

◆ deadline_last_total_count_

◆ deadline_map_

◆ deadline_period_

◆ deadline_status_

◆ deadline_status_lock_

◆ deadline_task_

◆ domain_id_

◆ durability_cache_

◆ durability_service_

◆ empty_condition_

◆ history_depth_

◆ instances_

◆ lock_