Inheritance diagram for OpenDDS::DCPS::DataReaderImpl::LivelinessTimer:

[legend]

Collaboration diagram for OpenDDS::DCPS::DataReaderImpl::LivelinessTimer:

[legend]

Classes
class	CancelCommand

class	CheckLivelinessCommand

class	CommandBase

Public Member Functions
	LivelinessTimer (ACE_Reactor reactor, ACE_thread_t owner, DataReaderImpl data_reader)

void	check_liveliness ()

void	cancel_timer ()

virtual bool	reactor_is_shut_down () const

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

Private Member Functions
	~LivelinessTimer ()

void	check_liveliness_i (bool cancel, const MonotonicTimePoint &now)

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

Private Attributes
WeakRcHandle< DataReaderImpl >	data_reader_

long	liveliness_timer_id_
	liveliness timer id; -1 if no timer is set More...

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

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)

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

Definition at line 918 of file DataReaderImpl.h.

Constructor & Destructor Documentation

◆ LivelinessTimer()

OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::LivelinessTimer	(	ACE_Reactor *	reactor,
		ACE_thread_t	owner,
		DataReaderImpl *	data_reader
	)

inline

Definition at line 920 of file DataReaderImpl.h.

       : ReactorInterceptor(reactor, owner)
       , data_reader_(*data_reader)
       , liveliness_timer_id_(-1)
     { }

◆ ~LivelinessTimer()

OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::~LivelinessTimer ( )

inlineprivate

Definition at line 941 of file DataReaderImpl.h.

941 { }

Member Function Documentation

◆ cancel_timer()

void OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::cancel_timer ( void )

inline

Definition at line 930 of file DataReaderImpl.h.

     {
       execute_or_enqueue(make_rch<CancelCommand>(this));
     }

◆ check_liveliness()

void OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::check_liveliness ( )

Definition at line 1819 of file DataReaderImpl.cpp.

 {
   execute_or_enqueue(make_rch<CheckLivelinessCommand>(this));
 }

◆ check_liveliness_i()

void OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::check_liveliness_i	(	bool	cancel,
		const MonotonicTimePoint &	now
	)

private

Definition at line 1835 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_READ_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::max_value, and OpenDDS::DCPS::TimeDuration::value().

 {
   // Working copy of the active timer Id.
 
   RcHandle<DataReaderImpl> data_reader = data_reader_.lock();
   if (! data_reader) {
     this->reactor()->purge_pending_notifications(this);
     return;
   }
 
   long local_timer_id = liveliness_timer_id_;
   bool timer_was_reset = false;
 
   if (local_timer_id != -1 && cancel) {
     if (DCPS_debug_level >= 5) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) DataReaderImpl::LivelinessTimer::check_liveliness_i: ")
                  ACE_TEXT(" canceling timer for reader %C.\n"),
                  LogGuid(data_reader->get_guid()).c_str()));
     }
 
     // called from add_associations and there is already a timer
     // so cancel the existing timer.
     if (this->reactor()->cancel_timer(local_timer_id) == -1) {
       // this could fail because the reactor's call and
       // the add_associations' call to this could overlap
       // so it is not a failure.
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::LivelinessTimer::check_liveliness_i: ")
                  ACE_TEXT(" %p.\n"), ACE_TEXT("cancel_timer")));
     }
 
     timer_was_reset = true;
   }
 
   // Used after the lock scope ends.
   MonotonicTimePoint smallest(MonotonicTimePoint::max_value);
   int alive_writers = 0;
 
   // This is a bit convoluted.  The reasoning goes as follows:
   // 1) We grab the current timer Id value when we enter the method.
   // 2) We *might* cancel the timer if it is active.
   // 3) The timer *might* be rescheduled while we do not hold the sample lock.
   // 4) If we (or another thread) canceled the timer that we can tell, then
   // 5) we should clear the Id value,
   // 6) unless it has been rescheduled.
   // We are using a changed timer Id value as a proxy for having been
   // rescheduled.
   if( timer_was_reset && (liveliness_timer_id_ == local_timer_id)) {
     liveliness_timer_id_ = -1;
   }
 
   // Iterate over each writer to this reader
   {
     ACE_READ_GUARD(ACE_RW_Thread_Mutex,
         read_guard,
         data_reader->writers_lock_);
     WriterMapType writers = data_reader->writers_;
     read_guard.release();
 
     for (WriterMapType::iterator iter = writers.begin();
         iter != writers.end();
         ++iter) {
       // deal with possibly not being alive or
       // tell when it will not be alive next (if no activity)
       const MonotonicTimePoint next_absolute(iter->second->check_activity(now));
       if (!next_absolute.is_max()) {
         alive_writers++;
         smallest = std::min(smallest, next_absolute);
       }
     }
   }
 
   if (!alive_writers) {
     // no live writers so no need to schedule a timer
     // but be sure we don't try to cancel the timer later.
     liveliness_timer_id_ = -1;
   }
 
   if (DCPS_debug_level >= 5) {
     ACE_DEBUG((LM_DEBUG,
         ACE_TEXT("(%P|%t) DataReaderImpl::LivelinessTimer::check_liveliness_i: ")
         ACE_TEXT("reader %C has %d live writers; from_reactor=%d\n"),
         LogGuid(data_reader->get_guid()).c_str(),
         alive_writers,
         !cancel));
   }
 
   // Call into the reactor after releasing the sample lock.
   if (alive_writers) {
     // compare the time now with the earliest(smallest) deadline we found
     TimeDuration relative;
     if (now < smallest) {
       relative = smallest - now;
     } else {
       relative = TimeDuration(0, 1); // ASAP
     }
     liveliness_timer_id_ = this->reactor()->schedule_timer(this, 0, relative.value());
 
     if (liveliness_timer_id_ == -1) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::LivelinessTimer::check_liveliness_i: ")
           ACE_TEXT(" %p.\n"), ACE_TEXT("schedule_timer")));
     }
   }
 }

◆ handle_timeout()

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

privatevirtual

Reimplemented from ACE_Event_Handler.

Definition at line 1825 of file DataReaderImpl.cpp.

References TheServiceParticipant.

 {
   ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());
 
   check_liveliness_i(false, MonotonicTimePoint(tv));
   return 0;
 }

◆ reactor_is_shut_down()

virtual bool OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::reactor_is_shut_down ( ) const

inlinevirtual

Implements OpenDDS::DCPS::ReactorInterceptor.

Definition at line 935 of file DataReaderImpl.h.

References TheServiceParticipant.

     {
       return TheServiceParticipant->is_shut_down();
     }

Member Data Documentation

◆ data_reader_

WeakRcHandle<DataReaderImpl> OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::data_reader_

private

Definition at line 943 of file DataReaderImpl.h.

◆ liveliness_timer_id_

long OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::liveliness_timer_id_

private

liveliness timer id; -1 if no timer is set

Definition at line 946 of file DataReaderImpl.h.

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

Classes

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ LivelinessTimer()

◆ ~LivelinessTimer()

Member Function Documentation

◆ cancel_timer()

◆ check_liveliness()

◆ check_liveliness_i()

◆ handle_timeout()

◆ reactor_is_shut_down()

Member Data Documentation

◆ data_reader_

◆ liveliness_timer_id_