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.

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
DataReaderImpl *	data_reader_
long	liveliness_timer_id_
	liveliness timer id; -1 if no timer is set
Classes
class	CancelCommand
class	CheckLivelinessCommand
class	CommandBase

---

Detailed Description

Definition at line 733 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 735 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 761 of file DataReaderImpl.h.

{ }

---

Member Function Documentation

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

(

)

[inline]

Definition at line 749 of file DataReaderImpl.h.

Referenced by OpenDDS::DCPS::DataReaderImpl::cleanup(), and OpenDDS::DCPS::DataReaderImpl::~DataReaderImpl().

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

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

(

)

[inline]

Definition at line 743 of file DataReaderImpl.h.

Referenced by OpenDDS::DCPS::DataReaderImpl::transport_assoc_done(), and OpenDDS::DCPS::DataReaderImpl::writer_became_alive().

    {
      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 1992 of file DataReaderImpl.cpp.

References data_reader_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::DataReaderImpl::get_subscription_id(), liveliness_timer_id_, OPENDDS_STRING, OpenDDS::DCPS::DataReaderImpl::writers_, and OpenDDS::DCPS::DataReaderImpl::writers_lock_.

Referenced by handle_timeout().

{
  // Working copy of the active timer Id.
  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")));
    }
  }
}

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

Definition at line 1984 of file DataReaderImpl.cpp.

References check_liveliness_i().

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

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

(

)

const [inline, virtual]

Implements OpenDDS::DCPS::ReactorInterceptor.

Definition at line 755 of file DataReaderImpl.h.

References TheServiceParticipant.

    {
      return TheServiceParticipant->is_shut_down();
    }

---

Member Data Documentation

DataReaderImpl* OpenDDS::DCPS::DataReaderImpl::LivelinessTimer::data_reader_ [private]

Definition at line 763 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 766 of file DataReaderImpl.h.

Referenced by check_liveliness_i().

---

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

• DataReaderImpl.h
• DataReaderImpl.cpp

---