Main Page | Namespace List | Class Hierarchy | Alphabetical List | Class List | File List | Namespace Members | Class Members | File Members | Search for 

ACE_RMCast::Retransmit Class Reference

#include <Retransmit.h>

Inheritance diagram for ACE_RMCast::Retransmit:

Inheritance graph
[legend]
Collaboration diagram for ACE_RMCast::Retransmit:

Collaboration graph
[legend]
List of all members.

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

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_

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 20 of file Retransmit.cpp. 

00021       : params_ (params),
00022         cond_ (mutex_),
00023         stop_ (false)
00024   {
00025   }

Member Function Documentation

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

Reimplemented from ACE_RMCast::Out_Element.

Definition at line 28 of file Retransmit.cpp.

References ACE_Thread_Manager::spawn(). 

00029   {
00030     Element::out_start (out);
00031 
00032     tracker_mgr_.spawn (track_thunk, this);
00033   }

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

Reimplemented from ACE_RMCast::Out_Element.

Definition at line 36 of file Retransmit.cpp.

References ACE_RMCast::Lock, ACE_Condition< MUTEX >::signal(), and ACE_Thread_Manager::wait(). 

00037   {
00038     {
00039       Lock l (mutex_);
00040       stop_ = true;
00041       cond_.signal ();
00042     }
00043 
00044     tracker_mgr_.wait ();
00045 
00046     Element::out_stop ();
00047   }

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

Reimplemented from ACE_RMCast::In_Element.

Definition at line 64 of file Retransmit.cpp.

References ACE_RMCast::Address, ACE_Hash_Map_Manager_Ex< u64, Descr, ACE_Hash< u64 >, ACE_Equal_To< u64 >, ACE_Null_Mutex >::find(), ACE_RMCast::Lock, ACE_RMCast::Message_ptr, ACE_RMCast::Profile_ptr, ACE_RMCast::In_Element::recv(), and ACE_RMCast::u64. 

00065   {
00066     if (NAK const* nak = static_cast<NAK const*> (m->find (NAK::id)))
00067     {
00068       Address to (static_cast<To const*> (m->find (To::id))->address ());
00069 
00070       if (nak->address () == to)
00071       {
00072         Lock l (mutex_);
00073 
00074         for (NAK::iterator j (const_cast<NAK*> (nak)->begin ());
00075              !j.done ();
00076              j.advance ())
00077         {
00078           u64* psn;
00079           j.next (psn);
00080 
00081           Message_ptr m;
00082 
00083           Queue::ENTRY* pair;
00084 
00085           if (queue_.find (*psn, pair) == 0)
00086           {
00087             //cerr << 5 << "PRTM " << to << " " << pair->ext_id_ << endl;
00088 
00089             m = pair->int_id_.message ();
00090 
00091             pair->int_id_.reset ();
00092           }
00093           else
00094           {
00095             //cerr << 4 << "message " << *psn << " not available" << endl;
00096 
00097             m = Message_ptr (new Message);
00098             m->add (Profile_ptr (new SN (*psn)));
00099             m->add (Profile_ptr (new NoData));
00100           }
00101 
00102           out_->send (m);
00103         }
00104       }
00105     }
00106 
00107     in_->recv (m);
00108   }

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

Reimplemented from ACE_RMCast::Out_Element.

Definition at line 50 of file Retransmit.cpp.

References ACE_Hash_Map_Manager_Ex< u64, Descr, ACE_Hash< u64 >, ACE_Equal_To< u64 >, ACE_Null_Mutex >::bind(), ACE_RMCast::Lock, ACE_RMCast::Message_ptr, and ACE_RMCast::SN::num(). 

00051   {
00052     if (m->find (Data::id) != 0)
00053     {
00054       SN const* sn = static_cast<SN const*> (m->find (SN::id));
00055 
00056       Lock l (mutex_);
00057       queue_.bind (sn->num (), Descr (m->clone ()));
00058     }
00059 
00060     out_->send (m);
00061   }

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

Definition at line 118 of file Retransmit.cpp.

References ACE_OS::abort(), ETIME, ACE_RMCast::Lock, ACE_RMCast::Parameters::retention_timeout(), ACE_RMCast::Parameters::tick(), ACE_RMCast::u64, ACE_Hash_Map_Manager_Ex< u64, Descr, ACE_Hash< u64 >, ACE_Equal_To< u64 >, ACE_Null_Mutex >::unbind(), and ACE_Condition< MUTEX >::wait(). 

00119   {
00120     while (true)
00121     {
00122       Lock l (mutex_);
00123 
00124       for (Queue::iterator i (queue_); !i.done ();)
00125       {
00126         if ((*i).int_id_.inc () >= params_.retention_timeout ())
00127         {
00128           u64 sn ((*i).ext_id_);
00129           i.advance ();
00130           queue_.unbind (sn);
00131         }
00132         else
00133         {
00134           i.advance ();
00135         }
00136       }
00137 
00138       // Go to sleep but watch for "manual cancellation" request.
00139       //
00140       ACE_Time_Value time (ACE_OS::gettimeofday ());
00141       time += params_.tick ();
00142 
00143       while (!stop_)
00144       {
00145         if (cond_.wait (&time) == -1)
00146         {
00147           if (errno != ETIME)
00148             ACE_OS::abort ();
00149           else
00150             break;
00151         }
00152       }
00153 
00154       if (stop_)
00155         break;
00156     }
00157   }

ACE_THR_FUNC_RETURN ACE_RMCast::Retransmit::track_thunk (  void *  obj  )  [static, private]
 

Definition at line 111 of file Retransmit.cpp. 

00112   {
00113     reinterpret_cast<Retransmit*> (obj)->track ();
00114     return 0;
00115   }

Member Data Documentation

Condition ACE_RMCast::Retransmit::cond_ [private]
 

Definition at line 90 of file Retransmit.h.

Mutex ACE_RMCast::Retransmit::mutex_ [private]
 

Definition at line 89 of file Retransmit.h.

Parameters const& ACE_RMCast::Retransmit::params_ [private]
 

Definition at line 86 of file Retransmit.h.

Queue ACE_RMCast::Retransmit::queue_ [private]
 

Definition at line 88 of file Retransmit.h.

bool ACE_RMCast::Retransmit::stop_ [private]
 

Definition at line 92 of file Retransmit.h.

ACE_Thread_Manager ACE_RMCast::Retransmit::tracker_mgr_ [private]
 

Definition at line 93 of file Retransmit.h.

The documentation for this class was generated from the following files:
Generated on Thu Nov 9 11:41:21 2006 for ACE_RMCast by doxygen 1.3.6