OpenDDS::DCPS::DataReaderImpl::LivelinessTimer Class Reference

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

[legend]

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

[legend]

List of all members.

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
Private Member Functions
	~LivelinessTimer ()
void	check_liveliness_i (bool cancel, const ACE_Time_Value &current_time)
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

---

Detailed Description

Definition at line 746 of file DataReaderImpl.h.

---

Constructor & Destructor Documentation

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

Definition at line 748 of file DataReaderImpl.h.

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

OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::~LivelinessTimer

(

)

[inline, private]

Definition at line 774 of file DataReaderImpl.h.

{ }

---

Member Function Documentation

void OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::cancel_timer

(

void

)

[inline]

Definition at line 762 of file DataReaderImpl.h.

References c.

Referenced by check_liveliness_i().

    {
      CancelCommand c(this);
      execute_or_enqueue(c);
    }

Here is the caller graph for this function:

void OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::check_liveliness

(

)

[inline]

Definition at line 756 of file DataReaderImpl.h.

References c.

    {
      CheckLivelinessCommand c(this);
      execute_or_enqueue(c);
    }

void OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::check_liveliness_i	(	bool	cancel,
		const ACE_Time_Value &	current_time
	)			[private]

Definition at line 1851 of file DataReaderImpl.cpp.

References ACE_TEXT(), cancel_timer(), data_reader_, OpenDDS::DCPS::DCPS_debug_level, liveliness_timer_id_, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::WeakRcHandle< T >::lock(), ACE_Time_Value::max_time, OPENDDS_STRING, ACE_Reactor::purge_pending_notifications(), ACE_Event_Handler::reactor(), and ACE_Reactor::schedule_timer().

Referenced by handle_timeout().

{
  // 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) {
      GuidConverter converter(data_reader->get_subscription_id());
      ACE_DEBUG((LM_DEBUG,
                 ACE_TEXT("(%P|%t) DataReaderImpl::handle_timeout: ")
                 ACE_TEXT(" canceling timer for reader %C.\n"),
                 OPENDDS_STRING(converter).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::handle_timeout: ")
                 ACE_TEXT(" %p. \n"), ACE_TEXT("cancel_timer")));
    }

    timer_was_reset = true;
  }

  // Used after the lock scope ends.
  ACE_Time_Value smallest(ACE_Time_Value::max_time);
  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;
  }

  ACE_Time_Value next_absolute;

  // Iterate over each writer to this reader
  {
    ACE_READ_GUARD(ACE_RW_Thread_Mutex,
        read_guard,
        data_reader->writers_lock_);

    for (WriterMapType::iterator iter = data_reader->writers_.begin();
        iter != data_reader->writers_.end();
        ++iter) {
      // deal with possibly not being alive or
      // tell when it will not be alive next (if no activity)
      next_absolute = iter->second->check_activity(now);

      if (next_absolute != ACE_Time_Value::max_time) {
        alive_writers++;

        if (next_absolute < smallest) {
          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) {
    GuidConverter converter(data_reader->get_subscription_id());
    ACE_DEBUG((LM_DEBUG,
        ACE_TEXT("(%P|%t) DataReaderImpl::handle_timeout: ")
        ACE_TEXT("reader %C has %d live writers; from_reactor=%d\n"),
        OPENDDS_STRING(converter).c_str(),
        alive_writers,
        !cancel));
  }

  // Call into the reactor after releasing the sample lock.
  if (alive_writers) {
    ACE_Time_Value relative;

    // compare the time now with the earliest(smallest) deadline we found
    if (now < smallest) {
      relative = smallest - now;
    } else {
      relative = ACE_Time_Value(0,1); // ASAP
    }

    liveliness_timer_id_ = this->reactor()->schedule_timer(this, 0, relative);

    if (liveliness_timer_id_ == -1) {
      ACE_ERROR((LM_ERROR,
          ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::handle_timeout: ")
          ACE_TEXT(" %p. \n"), ACE_TEXT("schedule_timer")));
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

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

Reimplemented from ACE_Event_Handler.

Definition at line 1843 of file DataReaderImpl.cpp.

References check_liveliness_i().

{
  check_liveliness_i(false, tv);
  return 0;
}

Here is the call graph for this function:

virtual bool OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::reactor_is_shut_down

(

)

const [inline, virtual]

Implements OpenDDS::DCPS::ReactorInterceptor.

Definition at line 768 of file DataReaderImpl.h.

References TheServiceParticipant.

    {
      return TheServiceParticipant->is_shut_down();
    }

---

Member Data Documentation

WeakRcHandle<DataReaderImpl> OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::data_reader_ [private]

Definition at line 776 of file DataReaderImpl.h.

Referenced by check_liveliness_i().

long OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::liveliness_timer_id_ [private]

liveliness timer id; -1 if no timer is set

Definition at line 779 of file DataReaderImpl.h.

Referenced by check_liveliness_i().

---

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

• DataReaderImpl.h
• DataReaderImpl.cpp