manage the states of a received data instance. More...

#include <InstanceState.h>

Inheritance diagram for OpenDDS::DCPS::InstanceState:

[legend]

Collaboration diagram for OpenDDS::DCPS::InstanceState:

[legend]

Classes
struct	CancelCommand

struct	CommandBase

struct	ScheduleCommand

Public Member Functions
	InstanceState (const DataReaderImpl_rch &reader, ACE_Recursive_Thread_Mutex &lock, DDS::InstanceHandle_t handle)

virtual	~InstanceState ()

void	sample_info (DDS::SampleInfo &si, const ReceivedDataElement *de)
	Populate the SampleInfo structure. More...

DDS::InstanceStateKind	instance_state () const
	Access instance state. More...

DDS::ViewStateKind	view_state () const
	Access view state. More...

bool	match (DDS::ViewStateMask view, DDS::InstanceStateMask inst) const

size_t	disposed_generation_count () const
	Access disposed generation count. More...

size_t	no_writers_generation_count () const
	Access no writers generation count. More...

bool	dispose_was_received (const GUID_t &writer_id)

bool	unregister_was_received (const GUID_t &writer_id)

void	data_was_received (const GUID_t &writer_id)
	Data sample received for this instance. More...

void	lively (const GUID_t &writer_id)
	LIVELINESS message received for this DataWriter. More...

void	accessed ()
	A read or take operation has been performed on this instance. More...

bool	most_recent_generation (ReceivedDataElement *item) const

bool	empty (bool value)
	DataReader has become empty. Returns true if the instance was released. More...

void	schedule_pending ()
	Schedule a pending release of resources. More...

void	schedule_release ()
	Schedule an immediate release of resources. More...

void	cancel_release ()
	Cancel a scheduled or pending release of resources. More...

bool	release_if_empty ()

void	release ()
	Remove the instance immediately. More...

bool	writes_instance (const GUID_t &writer_id) const
	Returns true if the writer is a writer of this instance. More...

WeakRcHandle< DataReaderImpl >	data_reader () const

void	state_updated () const

virtual int	handle_timeout (const ACE_Time_Value &current_time, const void *arg)

void	set_owner (const GUID_t &owner)

GUID_t	get_owner ()

bool	is_exclusive () const

bool	registered ()

void	registered (bool flag)

bool	is_last (const GUID_t &pub)

bool	no_writer () const

void	reset_ownership (DDS::InstanceHandle_t instance)

DDS::InstanceHandle_t	instance_handle () const

const char *	instance_state_string () const
	Return string of the name of the current instance state. More...

Public Member Functions inherited from OpenDDS::DCPS::ReactorInterceptor
CommandPtr	execute_or_enqueue (CommandPtr command)

virtual void	reactor (ACE_Reactor *reactor)

virtual ACE_Reactor *	reactor () const

Public Member Functions inherited from OpenDDS::DCPS::RcEventHandler
	RcEventHandler ()

ACE_Event_Handler::Reference_Count	add_reference ()

ACE_Event_Handler::Reference_Count	remove_reference ()

Public Member Functions inherited from ACE_Event_Handler
virtual	~ACE_Event_Handler (void)

virtual ACE_HANDLE	get_handle (void) const

virtual void	set_handle (ACE_HANDLE)

virtual int	priority (void) const

virtual void	priority (int priority)

virtual int	handle_input (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_output (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_exit (ACE_Process *)

virtual int	handle_close (ACE_HANDLE handle, ACE_Reactor_Mask close_mask)

virtual int	handle_signal (int signum, siginfo_t =0, ucontext_t =0)

virtual int	resume_handler (void)

virtual int	handle_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual int	handle_group_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual ACE_Reactor_Timer_Interface *	reactor_timer_interface (void) const

Reference_Counting_Policy &	reference_counting_policy (void)

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

virtual void	_add_ref ()

virtual void	_remove_ref ()

long	ref_count () const

WeakObject *	_get_weak_object () const

Static Public Member Functions
static const char *	instance_state_string (DDS::InstanceStateKind value)
	Return string of the name of the instance state kind passed. More...

static OPENDDS_STRING	instance_state_mask_string (DDS::InstanceStateMask mask)
	Return string representation of the instance state mask passed. More...

Static Public Member Functions inherited from ACE_Event_Handler
static ACE_THR_FUNC_RETURN	read_adapter (void *event_handler)

static int	register_stdin_handler (ACE_Event_Handler eh, ACE_Reactor reactor, ACE_Thread_Manager *thr_mgr, int flags=THR_DETACHED)

static int	remove_stdin_handler (ACE_Reactor reactor, ACE_Thread_Manager thr_mgr)

Private Member Functions
bool	reactor_is_shut_down () const

Private Attributes
ACE_Recursive_Thread_Mutex &	lock_

ACE_Thread_Mutex	owner_lock_

DDS::InstanceStateKind	instance_state_

DDS::ViewStateKind	view_state_

size_t	disposed_generation_count_

size_t	no_writers_generation_count_

bool	empty_

bool	release_pending_

long	release_timer_id_

WeakRcHandle< DataReaderImpl >	reader_

DDS::InstanceHandle_t	handle_

RepoIdSet	writers_

GUID_t	owner_

bool	exclusive_

bool	registered_

Additional Inherited Members
Public Types inherited from OpenDDS::DCPS::ReactorInterceptor
typedef RcHandle< Command >	CommandPtr

Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

Public Attributes inherited from ACE_Event_Handler
	LO_PRIORITY

	HI_PRIORITY

	NULL_MASK

	READ_MASK

	WRITE_MASK

	EXCEPT_MASK

	ACCEPT_MASK

	CONNECT_MASK

	TIMER_MASK

	QOS_MASK

	GROUP_QOS_MASK

	SIGNAL_MASK

	ALL_EVENTS_MASK

	RWE_MASK

	DONT_CALL

	ACE_EVENT_HANDLER_NOT_RESUMED

	ACE_REACTOR_RESUMES_HANDLER

	ACE_APPLICATION_RESUMES_HANDLER

Protected Types inherited from OpenDDS::DCPS::ReactorInterceptor
enum	ReactorState { RS_NONE, RS_NOTIFIED, RS_PROCESSING }

Protected Types inherited from ACE_Event_Handler
typedef ACE_Atomic_Op< ACE_SYNCH_MUTEX, Reference_Count >	Atomic_Reference_Count

Protected Member Functions inherited from OpenDDS::DCPS::ReactorInterceptor
	ReactorInterceptor (ACE_Reactor *reactor, ACE_thread_t owner)

virtual	~ReactorInterceptor ()

int	handle_exception (ACE_HANDLE)

void	process_command_queue_i (ACE_Guard< ACE_Thread_Mutex > &guard)

typedef	OPENDDS_VECTOR (CommandPtr) Queue

Protected Member Functions inherited from ACE_Event_Handler
	ACE_Event_Handler (ACE_Reactor *=0, int priority=ACE_Event_Handler::LO_PRIORITY)

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

Protected Attributes inherited from OpenDDS::DCPS::ReactorInterceptor
ACE_thread_t	owner_

ACE_Thread_Mutex	mutex_

Queue	command_queue_

ReactorState	state_

Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

manage the states of a received data instance.

Provide a mechanism to manage the view state and instance state values for an instance contained within a DataReader. The instance_state and view_state are managed by this class. Accessors are provided to query the current value of each of these states.

Definition at line 49 of file InstanceState.h.

Constructor & Destructor Documentation

◆ InstanceState()

OpenDDS::DCPS::InstanceState::InstanceState	(	const DataReaderImpl_rch &	reader,
		ACE_Recursive_Thread_Mutex &	lock,
		DDS::InstanceHandle_t	handle
	)

Definition at line 28 of file InstanceState.cpp.

   : ReactorInterceptor(TheServiceParticipant->reactor(),
                        TheServiceParticipant->reactor_owner()),
     lock_(lock),
     instance_state_(0),
     view_state_(0),
     disposed_generation_count_(0),
     no_writers_generation_count_(0),
     empty_(true),
     release_pending_(false),
     release_timer_id_(-1),
     reader_(reader),
     handle_(handle),
     owner_(GUID_UNKNOWN),
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
     exclusive_(reader->qos_.ownership.kind == DDS::EXCLUSIVE_OWNERSHIP_QOS),
 #endif
     registered_(false)
 {}

◆ ~InstanceState()

OpenDDS::DCPS::InstanceState::~InstanceState ( )

virtual

Definition at line 50 of file InstanceState.cpp.

References OpenDDS::DCPS::WeakRcHandle< T >::lock(), reader_, registered_, and OpenDDS::DCPS::OwnershipManager::remove_instance().

 {
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   if (registered_) {
     RcHandle<DataReaderImpl> reader = reader_.lock();
     if (reader) {
       DataReaderImpl::OwnershipManagerPtr om = reader->ownership_manager();
       if (om) om->remove_instance(this);
     }
   }
 #endif
 }

Member Function Documentation

◆ accessed()

ACE_INLINE void OpenDDS::DCPS::InstanceState::accessed ( )

A read or take operation has been performed on this instance.

Definition at line 22 of file InstanceState.inl.

References ACE_GUARD, ACE_INLINE, DDS::ANY_VIEW_STATE, lock_, DDS::NOT_NEW_VIEW_STATE, state_updated(), and view_state_.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::copy_into(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_next_sample(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::store_synthetic_data(), and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::take_next_sample().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, lock_);
   //
   // Manage the view state due to data access here.
   //
   if (view_state_ & DDS::ANY_VIEW_STATE) {
     const bool updated = (view_state_ != DDS::NOT_NEW_VIEW_STATE);
     view_state_ = DDS::NOT_NEW_VIEW_STATE;
     if (updated) {
       state_updated();
     }
   }
 }

◆ cancel_release()

void OpenDDS::DCPS::InstanceState::cancel_release ( )

Cancel a scheduled or pending release of resources.

Definition at line 240 of file InstanceState.cpp.

References OpenDDS::DCPS::ReactorInterceptor::execute_or_enqueue(), and release_pending_.

Referenced by data_was_received(), lively(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::purge_data(), and schedule_release().

 {
   release_pending_ = false;
   execute_or_enqueue(make_rch<CancelCommand>(this));
 }

◆ data_reader()

OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL ACE_INLINE OpenDDS::DCPS::WeakRcHandle< OpenDDS::DCPS::DataReaderImpl > OpenDDS::DCPS::InstanceState::data_reader ( ) const

Definition at line 15 of file InstanceState.inl.

References ACE_INLINE, and reader_.

Referenced by OpenDDS::DCPS::SubscriptionInstance::~SubscriptionInstance().

 {
   return reader_;
 }

◆ data_was_received()

ACE_INLINE void OpenDDS::DCPS::InstanceState::data_was_received ( const GUID_t & writer_id )

Data sample received for this instance.

Definition at line 71 of file InstanceState.inl.

References ACE_GUARD, ACE_INLINE, DDS::ALIVE_INSTANCE_STATE, cancel_release(), disposed_generation_count_, instance_state_, lock_, DDS::NEW_VIEW_STATE, no_writers_generation_count_, DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE, DDS::NOT_ALIVE_INSTANCE_STATE, DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE, DDS::NOT_NEW_VIEW_STATE, state_updated(), view_state_, and writers_.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::finish_store_instance_data().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, lock_);
   cancel_release();
 
   //
   // Update the view state here, since only sample data received affects
   // this state value.  Then manage the data sample only transitions
   // here.  Let the lively() method manage the other transitions.
   //
   writers_.insert(writer_id);
 
   const CORBA::ULong old_view_state = view_state_;
   const CORBA::ULong old_instance_state = instance_state_;
 
   switch (view_state_) {
   case DDS::NEW_VIEW_STATE:
     break;
 
   case DDS::NOT_NEW_VIEW_STATE:
     if (instance_state_ & DDS::NOT_ALIVE_INSTANCE_STATE) {
       view_state_ = DDS::NEW_VIEW_STATE;
     }
     break;
 
   default:
     view_state_ = DDS::NEW_VIEW_STATE;
     break;
   }
 
   switch (instance_state_) {
   case DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE:
     ++disposed_generation_count_;
     break;
 
   case DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE:
     ++no_writers_generation_count_;
     break;
 
   default:
     break;
   }
 
   instance_state_ = DDS::ALIVE_INSTANCE_STATE;
 
   if (view_state_ != old_view_state || instance_state_ != old_instance_state) {
     state_updated();
   }
 }

◆ dispose_was_received()

bool OpenDDS::DCPS::InstanceState::dispose_was_received ( const GUID_t & writer_id )

DISPOSE message received for this instance. Return flag indicates whether the instance state was changed. This flag is used by concrete DataReader to determine whether it should notify listener. If state is not changed, the dispose message is ignored.

Definition at line 119 of file InstanceState.cpp.

References ACE_GUARD_RETURN, DDS::ALIVE_INSTANCE_STATE, exclusive_, handle_, instance_state_, OpenDDS::DCPS::OwnershipManager::is_owner(), OpenDDS::DCPS::WeakRcHandle< T >::lock(), lock_, DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE, reader_, schedule_release(), state_updated(), and writers_.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::finish_store_instance_data().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, lock_, false);
   writers_.erase(writer_id);
 
   //
   // Manage the instance state on disposal here.
   //
   // If disposed by owner then the owner is not re-elected, it can
   // resume if the writer sends message again.
   if (instance_state_ & DDS::ALIVE_INSTANCE_STATE) {
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
     RcHandle<DataReaderImpl> reader = reader_.lock();
     if (reader) {
       DataReaderImpl::OwnershipManagerPtr owner_manager = reader->ownership_manager();
       if (! exclusive_
         || (owner_manager && owner_manager->is_owner (handle_, writer_id))) {
 #endif
         instance_state_ = DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE;
         state_updated();
         schedule_release();
         return true;
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
       }
     }
 #endif
   }
 
   return false;
 }

◆ disposed_generation_count()

ACE_INLINE size_t OpenDDS::DCPS::InstanceState::disposed_generation_count ( ) const

Access disposed generation count.

Definition at line 58 of file InstanceState.inl.

References ACE_INLINE, and disposed_generation_count_.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::finish_store_instance_data().

 {
   return disposed_generation_count_;
 }

◆ empty()

ACE_INLINE bool OpenDDS::DCPS::InstanceState::empty ( bool value )

DataReader has become empty. Returns true if the instance was released.

Definition at line 147 of file InstanceState.inl.

References ACE_INLINE, empty_, release_if_empty(), and release_pending_.

Referenced by OpenDDS::DCPS::ReceivedDataElementList::add(), and OpenDDS::DCPS::ReceivedDataElementList::remove().

 {
   //
   // Manage the instance state due to the DataReader becoming empty
   // here.
   //
   if ((empty_ = value) && release_pending_) {
     return release_if_empty();
   } else {
     return false;
   }
 }

◆ get_owner()

GUID_t OpenDDS::DCPS::InstanceState::get_owner ( )

Definition at line 272 of file InstanceState.cpp.

References owner_, and owner_lock_.

Referenced by OpenDDS::DCPS::DataReaderImpl::ownership_filter_instance().

 {
   ACE_Guard<ACE_Thread_Mutex> guard(owner_lock_);
   return owner_;
 }

◆ handle_timeout()

int OpenDDS::DCPS::InstanceState::handle_timeout	(	const ACE_Time_Value &	current_time,
		const void *	arg
	)

virtual

Reimplemented from ACE_Event_Handler.

Definition at line 103 of file InstanceState.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, handle_, LM_NOTICE, release(), and TheServiceParticipant.

 {
   ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());
 
   if (DCPS_debug_level) {
     ACE_DEBUG((LM_NOTICE,
                ACE_TEXT("(%P|%t) NOTICE:")
                ACE_TEXT(" InstanceState::handle_timeout:")
                ACE_TEXT(" autopurging samples with instance handle 0x%x!\n"),
                handle_));
   }
   release();
 
   return 0;
 }

◆ instance_handle()

DDS::InstanceHandle_t OpenDDS::DCPS::InstanceState::instance_handle ( ) const

inline

Definition at line 144 of file InstanceState.h.

References OPENDDS_STRING.

Referenced by OpenDDS::DCPS::ReceivedDataElementList::decrement_not_read_count(), OpenDDS::DCPS::ReceivedDataElementList::decrement_read_count(), OpenDDS::DCPS::ReceivedDataElementList::increment_not_read_count(), OpenDDS::DCPS::ReceivedDataElementList::increment_read_count(), and OpenDDS::DCPS::OwnershipManager::remove_instance().

144 { return handle_; }

OpenDDS::DCPS::InstanceState::handle_

DDS::InstanceHandle_t handle_

Definition: InstanceState.h:221

◆ instance_state()

ACE_INLINE DDS::InstanceStateKind OpenDDS::DCPS::InstanceState::instance_state ( ) const

Access instance state.

Definition at line 39 of file InstanceState.inl.

References ACE_INLINE, and instance_state_.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_i(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_instance_i(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::take_i(), and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::take_instance_i().

 {
   return instance_state_;
 }

◆ instance_state_mask_string()

OPENDDS_STRING OpenDDS::DCPS::InstanceState::instance_state_mask_string ( DDS::InstanceStateMask mask )

static

Return string representation of the instance state mask passed.

Definition at line 366 of file InstanceState.cpp.

References DDS::ALIVE_INSTANCE_STATE, DDS::ANY_INSTANCE_STATE, instance_state_string(), DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE, DDS::NOT_ALIVE_INSTANCE_STATE, DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE, OPENDDS_END_VERSIONED_NAMESPACE_DECL, and OPENDDS_STRING.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_instance_i().

 {
   if (mask == DDS::ANY_INSTANCE_STATE) {
     return instance_state_string(DDS::ANY_INSTANCE_STATE);
   }
   if (mask == DDS::NOT_ALIVE_INSTANCE_STATE) {
     return instance_state_string(DDS::NOT_ALIVE_INSTANCE_STATE);
   }
   OPENDDS_STRING str;
   if (mask & DDS::ALIVE_INSTANCE_STATE) {
     str = instance_state_string(DDS::ALIVE_INSTANCE_STATE);
   }
   if (mask & DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE) {
     if (!str.empty()) str += " | ";
     str += instance_state_string(DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE);
   }
   if (mask & DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE) {
     if (!str.empty()) str += " | ";
     str += instance_state_string(DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE);
   }
   return str;
 }

◆ instance_state_string() [1/2]

ACE_INLINE const char * OpenDDS::DCPS::InstanceState::instance_state_string ( ) const

Return string of the name of the current instance state.

Definition at line 180 of file InstanceState.inl.

References instance_state_, and OPENDDS_END_VERSIONED_NAMESPACE_DECL.

Referenced by instance_state_mask_string(), and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_instance_i().

 {
   return instance_state_string(instance_state_);
 }

◆ instance_state_string() [2/2]

const char * OpenDDS::DCPS::InstanceState::instance_state_string ( DDS::InstanceStateKind value )

static

Return string of the name of the instance state kind passed.

Definition at line 344 of file InstanceState.cpp.

References ACE_ERROR, ACE_TEXT(), DDS::ALIVE_INSTANCE_STATE, DDS::ANY_INSTANCE_STATE, LM_ERROR, DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE, DDS::NOT_ALIVE_INSTANCE_STATE, and DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE.

 {
   switch (value) {
   case DDS::ALIVE_INSTANCE_STATE:
     return "ALIVE_INSTANCE_STATE";
   case DDS::NOT_ALIVE_INSTANCE_STATE:
     return "NOT_ALIVE_INSTANCE_STATE";
   case DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE:
     return "NOT_ALIVE_DISPOSED_INSTANCE_STATE";
   case DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE:
     return "NOT_ALIVE_NO_WRITERS_INSTANCE_STATE";
   case DDS::ANY_INSTANCE_STATE:
     return "ANY_INSTANCE_STATE";
   default:
     ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: InstanceState::instance_state_string: ")
       ACE_TEXT("%d is either invalid or not recognized.\n"),
       value));
 
     return "Invalid instance state";
   }
 }

◆ is_exclusive()

bool OpenDDS::DCPS::InstanceState::is_exclusive ( ) const

Definition at line 278 of file InstanceState.cpp.

References exclusive_.

Referenced by OpenDDS::DCPS::DataReaderImpl::process_deadline().

 {
   return exclusive_;
 }

◆ is_last()

ACE_INLINE bool OpenDDS::DCPS::InstanceState::is_last ( const GUID_t & pub )

Definition at line 163 of file InstanceState.inl.

References ACE_GUARD_RETURN, ACE_INLINE, lock_, and writers_.

Referenced by OpenDDS::DCPS::DataReaderImpl::data_received().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, lock_, false);
 
   return writers_.size() == 1 && *writers_.begin() == pub;
 }

◆ lively()

ACE_INLINE void OpenDDS::DCPS::InstanceState::lively ( const GUID_t & writer_id )

LIVELINESS message received for this DataWriter.

Definition at line 123 of file InstanceState.inl.

References ACE_GUARD, ACE_INLINE, DDS::ALIVE_INSTANCE_STATE, cancel_release(), instance_state_, lock_, no_writers_generation_count_, DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE, state_updated(), and writers_.

Referenced by OpenDDS::DCPS::DataReaderImpl::signal_liveliness(), and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::store_instance_data().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, lock_);
 
   //
   // Manage transisitions in the instance state that do not require a
   // data sample, but merely the notion of liveliness.
   //
   writers_.insert(writer_id);
 
   if (instance_state_ == DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE) {
     cancel_release(); // cancel unregister
 
     ++no_writers_generation_count_;
     const bool updated = (instance_state_ != DDS::ALIVE_INSTANCE_STATE);
     instance_state_ = DDS::ALIVE_INSTANCE_STATE;
     if (updated) {
       state_updated();
     }
   }
 }

◆ match()

ACE_INLINE bool OpenDDS::DCPS::InstanceState::match	(	DDS::ViewStateMask	view,
		DDS::InstanceStateMask	inst
	)		const

Definition at line 52 of file InstanceState.inl.

References ACE_INLINE, instance_state_, and view_state_.

Referenced by OpenDDS::DCPS::DataReaderImpl::get_ordered_data(), OpenDDS::DCPS::SubscriptionInstance::matches(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_instance_i(), and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::take_instance_i().

 {
   return (view_state_ & view) && (instance_state_ & inst);
 }

◆ most_recent_generation()

bool OpenDDS::DCPS::InstanceState::most_recent_generation ( ReceivedDataElement * item ) const

Definition at line 309 of file InstanceState.cpp.

References OpenDDS::DCPS::ReceivedDataElement::disposed_generation_count_, disposed_generation_count_, OpenDDS::DCPS::ReceivedDataElement::no_writers_generation_count_, and no_writers_generation_count_.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::copy_into(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_next_sample(), and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::take_next_sample().

 {
   return item->disposed_generation_count_ == disposed_generation_count_
     && item->no_writers_generation_count_ == no_writers_generation_count_;
 }

◆ no_writer()

ACE_INLINE bool OpenDDS::DCPS::InstanceState::no_writer ( ) const

Definition at line 172 of file InstanceState.inl.

References ACE_GUARD_RETURN, ACE_INLINE, lock_, and writers_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, lock_, true);
 
   return writers_.empty();
 }

◆ no_writers_generation_count()

ACE_INLINE size_t OpenDDS::DCPS::InstanceState::no_writers_generation_count ( ) const

Access no writers generation count.

Definition at line 64 of file InstanceState.inl.

References ACE_INLINE, and no_writers_generation_count_.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::finish_store_instance_data().

 {
   return no_writers_generation_count_;
 }

◆ reactor_is_shut_down()

bool OpenDDS::DCPS::InstanceState::reactor_is_shut_down ( ) const

privatevirtual

Implements OpenDDS::DCPS::ReactorInterceptor.

Definition at line 315 of file InstanceState.cpp.

References TheServiceParticipant.

 {
   return TheServiceParticipant->is_shut_down();
 }

◆ registered() [1/2]

bool OpenDDS::DCPS::InstanceState::registered ( )

Definition at line 283 of file InstanceState.cpp.

References lock_, and registered_.

Referenced by OpenDDS::DCPS::OwnershipManager::select_owner().

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(lock_);
   const bool ret = registered_;
   registered_ = true;
   return ret;
 }

◆ registered() [2/2]

void OpenDDS::DCPS::InstanceState::registered ( bool flag )

Definition at line 291 of file InstanceState.cpp.

References lock_, and registered_.

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(lock_);
   registered_ = flag;
 }

◆ release()

void OpenDDS::DCPS::InstanceState::release ( void )

Remove the instance immediately.

Definition at line 258 of file InstanceState.cpp.

References handle_, OpenDDS::DCPS::WeakRcHandle< T >::lock(), and reader_.

Referenced by handle_timeout(), and release_if_empty().

 {
   RcHandle<DataReaderImpl> reader = reader_.lock();
   if (reader) {
     reader->release_instance(handle_);
   }
 }

◆ release_if_empty()

bool OpenDDS::DCPS::InstanceState::release_if_empty ( )

Remove the instance if it's instance has no samples and no writers. Returns true if the instance was released.

Definition at line 246 of file InstanceState.cpp.

References empty_, release(), schedule_pending(), and writers_.

Referenced by empty().

 {
   bool released = false;
   if (empty_ && writers_.empty()) {
     release();
     released = true;
   } else {
     schedule_pending();
   }
   return released;
 }

◆ reset_ownership()

void OpenDDS::DCPS::InstanceState::reset_ownership ( DDS::InstanceHandle_t instance )

Definition at line 297 of file InstanceState.cpp.

References OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::WeakRcHandle< T >::lock(), lock_, reader_, registered_, and set_owner().

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(lock_);
   set_owner(GUID_UNKNOWN);
   registered_ = false;
 
   RcHandle<DataReaderImpl> reader = reader_.lock();
   if (reader) {
     reader->reset_ownership(instance);
   }
 }

◆ sample_info()

void OpenDDS::DCPS::InstanceState::sample_info	(	DDS::SampleInfo &	si,
		const ReceivedDataElement *	de
	)

Populate the SampleInfo structure.

Definition at line 63 of file InstanceState.cpp.

References DDS::SampleInfo::absolute_generation_rank, DDS::SampleInfo::disposed_generation_count, OpenDDS::DCPS::ReceivedDataElement::disposed_generation_count_, disposed_generation_count_, DDS::SampleInfo::generation_rank, OpenDDS::DCPS::SequenceNumber::getValue(), handle_, DDS::HANDLE_NIL, DDS::SampleInfo::instance_handle, DDS::SampleInfo::instance_state, instance_state_, OpenDDS::DCPS::WeakRcHandle< T >::lock(), DDS::SampleInfo::no_writers_generation_count, OpenDDS::DCPS::ReceivedDataElement::no_writers_generation_count_, no_writers_generation_count_, DDS::SampleInfo::opendds_reserved_publication_seq, OpenDDS::DCPS::ReceivedDataElement::pub_, DDS::SampleInfo::publication_handle, reader_, DDS::SampleInfo::sample_rank, DDS::SampleInfo::sample_state, OpenDDS::DCPS::ReceivedDataElement::sample_state_, OpenDDS::DCPS::ReceivedDataElement::sequence_, DDS::SampleInfo::source_timestamp, OpenDDS::DCPS::ReceivedDataElement::source_timestamp_, DDS::SampleInfo::valid_data, OpenDDS::DCPS::ReceivedDataElement::valid_data_, DDS::SampleInfo::view_state, and view_state_.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::copy_into(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_next_sample(), and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::take_next_sample().

 {
   si.sample_state = de->sample_state_;
   si.view_state = view_state_;
   si.instance_state = instance_state_;
   si.disposed_generation_count =
     static_cast<CORBA::Long>(disposed_generation_count_);
   si.no_writers_generation_count =
     static_cast<CORBA::Long>(no_writers_generation_count_);
   si.source_timestamp = de->source_timestamp_;
   si.instance_handle = handle_;
   RcHandle<DataReaderImpl> reader = reader_.lock();
   if (reader) {
     RcHandle<DomainParticipantImpl> participant = reader->participant_servant_.lock();
     si.publication_handle = participant ? participant->lookup_handle(de->pub_) : DDS::HANDLE_NIL;
   } else {
     si.publication_handle = DDS::HANDLE_NIL;
   }
   si.valid_data = de->valid_data_;
   /*
    * These are actually calculated later...
    */
   si.sample_rank = 0;
 
   // these aren't the real value, they're being saved
   // for a later calculation. the actual value is
   // calculated in DataReaderImpl::sample_info using
   // these values.
   si.generation_rank =
     static_cast<CORBA::Long>(de->disposed_generation_count_ +
                              de->no_writers_generation_count_);
   si.absolute_generation_rank =
     static_cast<CORBA::Long>(de->disposed_generation_count_ +
                              de->no_writers_generation_count_);
 
   si.opendds_reserved_publication_seq = de->sequence_.getValue();
 }

◆ schedule_pending()

void OpenDDS::DCPS::InstanceState::schedule_pending ( )

Schedule a pending release of resources.

Definition at line 183 of file InstanceState.cpp.

References release_pending_.

Referenced by release_if_empty(), and schedule_release().

 {
   release_pending_ = true;
 }

◆ schedule_release()

void OpenDDS::DCPS::InstanceState::schedule_release ( )

Schedule an immediate release of resources.

Definition at line 196 of file InstanceState.cpp.

References ACE_ERROR, ACE_TEXT(), DDS::ReaderDataLifecycleQosPolicy::autopurge_disposed_samples_delay, DDS::ReaderDataLifecycleQosPolicy::autopurge_nowriter_samples_delay, cancel_release(), DDS::DURATION_INFINITE_NSEC, DDS::DURATION_INFINITE_SEC, OpenDDS::DCPS::ReactorInterceptor::execute_or_enqueue(), instance_state_, LM_ERROR, OpenDDS::DCPS::WeakRcHandle< T >::lock(), DDS::Duration_t::nanosec, DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE, DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE, reader_, DDS::DataReaderQos::reader_data_lifecycle, schedule_pending(), and DDS::Duration_t::sec.

Referenced by dispose_was_received(), and unregister_was_received().

 {
   DDS::DataReaderQos qos;
   RcHandle<DataReaderImpl> reader = reader_.lock();
   if (reader) {
     qos = reader->qos_;
   } else {
     cancel_release();
     return;
   }
 
   DDS::Duration_t delay;
 
   switch (instance_state_) {
   case DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE:
     delay = qos.reader_data_lifecycle.autopurge_nowriter_samples_delay;
     break;
 
   case DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE:
     delay = qos.reader_data_lifecycle.autopurge_disposed_samples_delay;
     break;
 
   default:
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: InstanceState::schedule_release:")
                ACE_TEXT(" Unsupported instance state: %d!\n"),
                instance_state_));
     return;
   }
 
   if (delay.sec != DDS::DURATION_INFINITE_SEC &&
       delay.nanosec != DDS::DURATION_INFINITE_NSEC) {
 
     execute_or_enqueue(make_rch<ScheduleCommand>(this, TimeDuration(delay)));
 
   } else {
     // N.B. instance transitions are always followed by a non-valid
     // sample being queued to the ReceivedDataElementList; marking
     // the release as pending prevents this sample from being lost
     // if all samples have been already removed from the instance.
     schedule_pending();
   }
 }

◆ set_owner()

void OpenDDS::DCPS::InstanceState::set_owner ( const GUID_t & owner )

Definition at line 266 of file InstanceState.cpp.

References owner_, and owner_lock_.

Referenced by reset_ownership().

 {
   ACE_Guard<ACE_Thread_Mutex> guard(owner_lock_);
   owner_ = owner;
 }

◆ state_updated()

void OpenDDS::DCPS::InstanceState::state_updated ( ) const

Definition at line 188 of file InstanceState.cpp.

References handle_, OpenDDS::DCPS::WeakRcHandle< T >::lock(), and reader_.

Referenced by accessed(), data_was_received(), dispose_was_received(), lively(), and unregister_was_received().

 {
   RcHandle<DataReaderImpl> reader = reader_.lock();
   if (reader) {
     reader->state_updated(handle_);
   }
 }

◆ unregister_was_received()

bool OpenDDS::DCPS::InstanceState::unregister_was_received ( const GUID_t & writer_id )

UNREGISTER message received for this instance. Return flag indicates whether the instance state was changed. This flag is used by concrete DataReader to determine whether it should notify listener. If state is not changed, the unregister message is ignored.

Definition at line 150 of file InstanceState.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), DDS::ALIVE_INSTANCE_STATE, OpenDDS::DCPS::DCPS_debug_level, exclusive_, handle_, instance_state_, LM_DEBUG, OpenDDS::DCPS::WeakRcHandle< T >::lock(), lock_, DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE, reader_, OpenDDS::DCPS::OwnershipManager::remove_writer(), schedule_release(), state_updated(), and writers_.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::finish_store_instance_data().

 {
   if (DCPS_debug_level > 1) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) InstanceState::unregister_was_received on %C\n"),
       LogGuid(writer_id).c_str()
     ));
   }
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, lock_, false);
   writers_.erase(writer_id);
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   if (exclusive_) {
     // If unregistered by owner then the ownership should be transferred to another
     // writer.
     RcHandle<DataReaderImpl> reader = reader_.lock();
     if (reader) {
       DataReaderImpl::OwnershipManagerPtr owner_manager = reader->ownership_manager();
       if (owner_manager)
         owner_manager->remove_writer (handle_, writer_id);
     }
   }
 #endif
 
   if (writers_.empty() && (instance_state_ & DDS::ALIVE_INSTANCE_STATE)) {
     instance_state_ = DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE;
     state_updated();
     schedule_release();
     return true;
   }
 
   return false;
 }

◆ view_state()

ACE_INLINE DDS::ViewStateKind OpenDDS::DCPS::InstanceState::view_state ( ) const

Access view state.

Definition at line 46 of file InstanceState.inl.

References ACE_INLINE, and view_state_.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_instance_i().

 {
   return view_state_;
 }

◆ writes_instance()

bool OpenDDS::DCPS::InstanceState::writes_instance ( const GUID_t & writer_id ) const

inline

Returns true if the writer is a writer of this instance.

Definition at line 121 of file InstanceState.h.

References ACE_GUARD_RETURN, and lock_.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::store_instance_data_check().

   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, lock_, false);
     return writers_.count(writer_id);
   }

Member Data Documentation

◆ disposed_generation_count_

size_t OpenDDS::DCPS::InstanceState::disposed_generation_count_

private

Number of times the instance state changes from NOT_ALIVE_DISPOSED to ALIVE.

Definition at line 193 of file InstanceState.h.

Referenced by data_was_received(), disposed_generation_count(), most_recent_generation(), and sample_info().

◆ empty_

bool OpenDDS::DCPS::InstanceState::empty_

private

Keep track of whether the DataReader is empty or not.

Definition at line 202 of file InstanceState.h.

Referenced by empty(), and release_if_empty().

◆ exclusive_

bool OpenDDS::DCPS::InstanceState::exclusive_

private

Definition at line 225 of file InstanceState.h.

Referenced by dispose_was_received(), is_exclusive(), and unregister_was_received().

◆ handle_

DDS::InstanceHandle_t OpenDDS::DCPS::InstanceState::handle_

private

Definition at line 221 of file InstanceState.h.

Referenced by dispose_was_received(), handle_timeout(), release(), sample_info(), state_updated(), and unregister_was_received().

◆ instance_state_

DDS::InstanceStateKind OpenDDS::DCPS::InstanceState::instance_state_

private

Current instance state.

Can have values defined as:

DDS::ALIVE_INSTANCE_STATE DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE

and can be checked with the masks:

DDS::ANY_INSTANCE_STATE DDS::NOT_ALIVE_INSTANCE_STATE

Definition at line 175 of file InstanceState.h.

Referenced by data_was_received(), dispose_was_received(), OpenDDS::DCPS::InstanceState::CancelCommand::execute(), OpenDDS::DCPS::InstanceState::ScheduleCommand::execute(), instance_state(), instance_state_string(), lively(), match(), sample_info(), schedule_release(), and unregister_was_received().

◆ lock_

ACE_Recursive_Thread_Mutex& OpenDDS::DCPS::InstanceState::lock_

private

Definition at line 158 of file InstanceState.h.

Referenced by accessed(), data_was_received(), dispose_was_received(), is_last(), lively(), no_writer(), registered(), reset_ownership(), and unregister_was_received().

◆ no_writers_generation_count_

size_t OpenDDS::DCPS::InstanceState::no_writers_generation_count_

private

Number of times the instance state changes from NOT_ALIVE_NO_WRITERS to ALIVE.

Definition at line 197 of file InstanceState.h.

Referenced by data_was_received(), lively(), most_recent_generation(), no_writers_generation_count(), and sample_info().

◆ owner_

GUID_t OpenDDS::DCPS::InstanceState::owner_

private

Definition at line 224 of file InstanceState.h.

Referenced by get_owner(), and set_owner().

◆ owner_lock_

ACE_Thread_Mutex OpenDDS::DCPS::InstanceState::owner_lock_

private

Definition at line 159 of file InstanceState.h.

Referenced by get_owner(), and set_owner().

◆ reader_

WeakRcHandle<DataReaderImpl> OpenDDS::DCPS::InstanceState::reader_

private

Reference to our containing reader. This is used to call back and notify the reader that liveliness has been lost on this instance. It is also queried to determine if the DataReader is empty – that it contains no more sample data.

Definition at line 220 of file InstanceState.h.

Referenced by data_reader(), dispose_was_received(), release(), reset_ownership(), sample_info(), schedule_release(), state_updated(), unregister_was_received(), and ~InstanceState().

◆ registered_

bool OpenDDS::DCPS::InstanceState::registered_

private

registered with participant so it can be called back as the owner is updated.

Definition at line 228 of file InstanceState.h.

Referenced by registered(), reset_ownership(), and ~InstanceState().

◆ release_pending_

bool OpenDDS::DCPS::InstanceState::release_pending_

private

Keep track of whether the instance is waiting to be released.

Definition at line 207 of file InstanceState.h.

Referenced by cancel_release(), empty(), and schedule_pending().

◆ release_timer_id_

long OpenDDS::DCPS::InstanceState::release_timer_id_

private

Keep track of a scheduled release timer.

Definition at line 212 of file InstanceState.h.

◆ view_state_

DDS::ViewStateKind OpenDDS::DCPS::InstanceState::view_state_

private

Current instance view state.

Can have values defined as:

DDS::NEW_VIEW_STATE DDS::NOT_NEW_VIEW_STATE

and can be checked with the mask:

DDS::ANY_VIEW_STATE

Definition at line 189 of file InstanceState.h.

Referenced by accessed(), data_was_received(), match(), sample_info(), and view_state().

◆ writers_

RepoIdSet OpenDDS::DCPS::InstanceState::writers_

private

Definition at line 223 of file InstanceState.h.

Referenced by data_was_received(), dispose_was_received(), is_last(), lively(), no_writer(), release_if_empty(), and unregister_was_received().

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

Classes

Public Member Functions

Static Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ InstanceState()

◆ ~InstanceState()

Member Function Documentation

◆ accessed()

◆ cancel_release()

◆ data_reader()

◆ data_was_received()

◆ dispose_was_received()

◆ disposed_generation_count()

◆ empty()

◆ get_owner()

◆ handle_timeout()

◆ instance_handle()

◆ instance_state()

◆ instance_state_mask_string()

◆ instance_state_string() [1/2]

◆ instance_state_string() [2/2]

◆ is_exclusive()

◆ is_last()

◆ lively()

◆ match()

◆ most_recent_generation()

◆ no_writer()

◆ no_writers_generation_count()

◆ reactor_is_shut_down()

◆ registered() [1/2]

◆ registered() [2/2]

◆ release()

◆ release_if_empty()

◆ reset_ownership()

◆ sample_info()

◆ schedule_pending()

◆ schedule_release()

◆ set_owner()

◆ state_updated()

◆ unregister_was_received()

◆ view_state()

◆ writes_instance()

Member Data Documentation

◆ disposed_generation_count_

◆ empty_

◆ exclusive_

◆ handle_

◆ instance_state_

◆ lock_

◆ no_writers_generation_count_

◆ owner_

◆ owner_lock_

◆ reader_

◆ registered_

◆ release_pending_

◆ release_timer_id_

◆ view_state_

◆ writers_