#include <Retransmit.h>

Inheritance diagram for ACE_RMCast::Retransmit:

Collaboration diagram for ACE_RMCast::Retransmit:

Classes
struct	Descr
Public Member Functions
	Retransmit (Parameters const &params)
virtual void	out_start (Out_Element *out)
virtual void	out_stop ()
virtual void	send (Message_ptr m)
virtual void	recv (Message_ptr m)
Private Types
typedef ACE_Hash_Map_Manager < u64, Descr, ACE_Null_Mutex >	Queue
Private Member Functions
void	track ()
Static Private Member Functions
static ACE_THR_FUNC_RETURN	track_thunk (void *obj)
Private Attributes
Parameters const &	params_
Queue	queue_
Mutex	mutex_
Condition	cond_
bool	stop_
ACE_Thread_Manager	tracker_mgr_

Detailed Description

Definition at line 18 of file Retransmit.h.

Member Typedef Documentation

typedef ACE_Hash_Map_Manager<u64, Descr, ACE_Null_Mutex> ACE_RMCast::Retransmit::Queue [private]

Definition at line 76 of file Retransmit.h.

Constructor & Destructor Documentation

ACE_RMCast::Retransmit::Retransmit ( Parameters const & params )

Definition at line 12 of file Retransmit.cpp.

Member Function Documentation

void ACE_RMCast::Retransmit::out_start ( Out_Element * out ) [virtual]

Reimplemented from ACE_RMCast::Out_Element.

Definition at line 20 of file Retransmit.cpp.

      : params_ (params),
        cond_ (mutex_),
        stop_ (false)
  {
  }

void ACE_RMCast::Retransmit::out_stop ( ) [virtual]

Reimplemented from ACE_RMCast::Out_Element.

Definition at line 28 of file Retransmit.cpp.

  {
    Element::out_start (out);

    tracker_mgr_.spawn (track_thunk, this);
  }

  void Retransmit::
  out_stop ()
  {
    {
      Lock l (mutex_);

void ACE_RMCast::Retransmit::recv ( Message_ptr m ) [virtual]

Definition at line 54 of file Retransmit.cpp.

  {
    if (NAK const* nak = static_cast<NAK const*> (m->find (NAK::id)))
    {
      Address to (static_cast<To const*> (m->find (To::id))->address ());

      if (nak->address () == to)
      {
        Lock l (mutex_);

        for (NAK::iterator j (const_cast<NAK*> (nak)->begin ());
             !j.done ();
             j.advance ())
        {
          u64* psn;
          j.next (psn);

          Message_ptr m;

          Queue::ENTRY* pair;

          if (queue_.find (*psn, pair) == 0)
          {
            //cerr << 5 << "PRTM " << to << " " << pair->ext_id_ << endl;

            m = pair->int_id_.message ();

            pair->int_id_.reset ();
          }
          else
          {
            //cerr << 4 << "message " << *psn << " not available" << endl;

            m = Message_ptr (new Message);
            m->add (Profile_ptr (new SN (*psn)));
            m->add (Profile_ptr (new NoData));
          }

void ACE_RMCast::Retransmit::send ( Message_ptr m ) [virtual]

Reimplemented from ACE_RMCast::Out_Element.

Definition at line 41 of file Retransmit.cpp.

  {
    if (m->find (Data::id) != 0)
    {

void ACE_RMCast::Retransmit::track ( ) [private]

Definition at line 108 of file Retransmit.cpp.

  {
    reinterpret_cast<Retransmit*> (obj)->track ();
    return 0;
  }

  void Retransmit::
  track ()
  {
    while (true)
    {
      Lock l (mutex_);

      for (Queue::iterator i (queue_); !i.done ();)
      {
        if ((*i).int_id_.inc () >= params_.retention_timeout ())
        {
          u64 sn ((*i).ext_id_);
          i.advance ();
          queue_.unbind (sn);
        }
        else
        {
          i.advance ();
        }
      }

      //FUZZ: disable check_for_lack_ACE_OS
      // Go to sleep but watch for "manual cancellation" request.
      //
      ACE_Time_Value time (ACE_OS::gettimeofday ());
      //FUZZ: enable check_for_lack_ACE_OS

      time += params_.tick ();

      while (!stop_)
      {
        if (cond_.wait (&time) == -1)
        {
          if (errno != ETIME)
            ACE_OS::abort ();
          else