ACE/protocols/ace/RMCast/Retransmit.cpp

   1 // author    : Boris Kolpackov <boris@kolpackov.net>
   2 #include "ace/Time_Value.h"     // ACE_Time_Value
   3 #include "ace/OS_NS_stdlib.h"   // abort
   4 #include "ace/OS_NS_sys_time.h" // gettimeofday
   5
   6 #include "Retransmit.h"
   7
   8 namespace ACE_RMCast
   9 {
  10   Retransmit::
  11   Retransmit (Parameters const& params)
  12       : params_ (params),
  13         cond_ (mutex_),
  14         stop_ (false)
  15   {
  16   }
  17
  18   void Retransmit::
  19   out_start (Out_Element* out)
  20   {
  21     Element::out_start (out);
  22
  23     tracker_mgr_.spawn (track_thunk, this);
  24   }
  25
  26   void Retransmit::
  27   out_stop ()
  28   {
  29     {
  30       Lock l (mutex_);
  31       stop_ = true;
  32       cond_.signal ();
  33     }
  34
  35     tracker_mgr_.wait ();
  36
  37     Element::out_stop ();
  38   }
  39
  40   void Retransmit::send (Message_ptr m)
  41   {
  42     if (m->find (Data::id) != 0)
  43     {
  44       SN const* sn = static_cast<SN const*> (m->find (SN::id));
  45
  46       Lock l (mutex_);
  47       queue_.bind (sn->num (), Descr (m->clone ()));
  48     }
  49
  50     out_->send (m);
  51   }
  52
  53   void Retransmit::recv (Message_ptr m)
  54   {
  55     if (NAK const* nak = static_cast<NAK const*> (m->find (NAK::id)))
  56     {
  57       Address to (static_cast<To const*> (m->find (To::id))->address ());
  58
  59       if (nak->address () == to)
  60       {
  61         Lock l (mutex_);
  62
  63         for (NAK::iterator j (const_cast<NAK*> (nak)->begin ());
  64              !j.done ();
  65              j.advance ())
  66         {
  67           u64* psn;
  68           j.next (psn);
  69
  70           Message_ptr m;
  71
  72           Queue::ENTRY* pair;
  73
  74           if (queue_.find (*psn, pair) == 0)
  75           {
  76             //cerr << 5 << "PRTM " << to << " " << pair->ext_id_ << endl;
  77
  78             m = pair->int_id_.message ();
  79
  80             pair->int_id_.reset ();
  81           }
  82           else
  83           {
  84             //cerr << 4 << "message " << *psn << " not available" << endl;
  85
  86             m = Message_ptr (new Message);
  87             m->add (Profile_ptr (new SN (*psn)));
  88             m->add (Profile_ptr (new NoData));
  89           }
  90
  91           out_->send (m);
  92         }
  93       }
  94     }
  95
  96     in_->recv (m);
  97   }
  98
  99   ACE_THR_FUNC_RETURN Retransmit::
 100   track_thunk (void* obj)
 101   {
 102     reinterpret_cast<Retransmit*> (obj)->track ();
 103     return 0;
 104   }
 105
 106   void Retransmit::
 107   track ()
 108   {
 109     while (true)
 110     {
 111       Lock l (mutex_);
 112
 113       for (Queue::iterator i (queue_); !i.done ();)
 114       {
 115         if ((*i).int_id_.inc () >= params_.retention_timeout ())
 116         {
 117           u64 sn ((*i).ext_id_);
 118           i.advance ();
 119           queue_.unbind (sn);
 120         }
 121         else
 122         {
 123           i.advance ();
 124         }
 125       }
 126
 127       //FUZZ: disable check_for_lack_ACE_OS
 128       // Go to sleep but watch for "manual cancellation" request.
 129       //
 130       ACE_Time_Value time (ACE_OS::gettimeofday ());
 131       //FUZZ: enable check_for_lack_ACE_OS
 132
 133       time += params_.tick ();
 134
 135       while (!stop_)
 136       {
 137         if (cond_.wait (&time) == -1)
 138         {
 139           if (errno != ETIME)
 140             ACE_OS::abort ();
 141           else
 142             break;
 143         }
 144       }
 145
 146       if (stop_)
 147         break;
 148     }
 149   }
 150 }