ACE/protocols/ace/RMCast/Acknowledge.cpp

   1 // author    : Boris Kolpackov <boris@kolpackov.net>
   2 #include "ace/Time_Value.h"     // ACE_Time_Value
   3 #include "ace/OS_NS_unistd.h"
   4 #include "ace/OS_NS_stdlib.h"   // abort
   5 #include "ace/OS_NS_sys_time.h" // gettimeofday
   6 #include "Acknowledge.h"
   7 #include <memory>
   8
   9 namespace ACE_RMCast
  10 {
  11   Acknowledge::
  12   Acknowledge (Parameters const& params)
  13       : params_ (params),
  14         hold_ (params.addr_map_size ()),
  15         cond_ (mutex_),
  16         nrtm_timer_ (params_.nrtm_timeout ()),
  17         stop_ (false)
  18   {
  19   }
  20
  21   void Acknowledge::
  22   in_start (In_Element* in)
  23   {
  24     Element::in_start (in);
  25   }
  26
  27   void Acknowledge::
  28   out_start (Out_Element* out)
  29   {
  30     Element::out_start (out);
  31
  32     tracker_mgr_.spawn (track_thunk, this);
  33   }
  34
  35   void Acknowledge::
  36   out_stop ()
  37   {
  38     {
  39       Lock l (mutex_);
  40       stop_ = true;
  41       cond_.signal ();
  42     }
  43
  44     tracker_mgr_.wait ();
  45
  46     Element::out_stop ();
  47   }
  48
  49   void Acknowledge::
  50   collapse (Queue& q)
  51   {
  52     // I would normally use iterators in the logic below but ACE_Map_Manager
  53     // iterates over entries in no particular order so it is pretty much
  54     // unusable here. Instead we will do slow and cumbersome find's.
  55     //
  56
  57     u64 sn (q.sn () + 1);
  58
  59     for (;; ++sn)
  60     {
  61       Queue::ENTRY* e = 0;
  62
  63       if (q.find (sn, e) == -1 || e->int_id_.lost ()) break;
  64
  65       Message_ptr m (e->int_id_.msg ());
  66       q.unbind (sn);
  67
  68       in_->recv (m);
  69     }
  70
  71     q.sn (sn - 1);
  72   }
  73
  74   void Acknowledge::
  75   track ()
  76   {
  77     while (true)
  78     {
  79       Messages msgs;
  80
  81       {
  82         Lock l (mutex_);
  83
  84         if (stop_)
  85           break;
  86
  87         if (hold_.current_size () != 0)
  88         {
  89           for (Map::iterator i (hold_.begin ()), e (hold_.end ());
  90                i != e;
  91                ++i)
  92           {
  93             Queue& q = (*i).int_id_;
  94
  95             if (q.current_size () == 0) continue;
  96
  97             track_queue ((*i).ext_id_, q, msgs);
  98           }
  99         }
 100
 101         if (--nrtm_timer_ == 0)
 102         {
 103           nrtm_timer_ = params_.nrtm_timeout ();
 104
 105           // Send NRTM.
 106           //
 107           unsigned short max_payload_size (
 108             params_.max_packet_size () - max_service_size);
 109
 110           u32 max_elem (NRTM::max_count (max_payload_size));
 111
 112           Profile_ptr nrtm (create_nrtm (max_elem));
 113
 114           if (!nrtm.null ())
 115           {
 116             Message_ptr m (new Message);
 117             m->add (nrtm);
 118             msgs.push_back (m);
 119           }
 120         }
 121       }
 122
 123       // Send stuff off.
 124       //
 125       for (Messages::Iterator i (msgs); !i.done (); i.advance ())
 126       {
 127         Message_ptr* ppm;
 128         i.next (ppm);
 129
 130         //FUZZ: disable check_for_lack_ACE_OS
 131         send (*ppm);
 132         //FUZZ: enable check_for_lack_ACE_OS
 133       }
 134
 135       // Go to sleep but watch for "manual cancellation" request.
 136       //
 137       {
 138         //FUZZ: disable check_for_lack_ACE_OS
 139         ACE_Time_Value time (ACE_OS::gettimeofday ());
 140         //FUZZ: enable check_for_lack_ACE_OS
 141
 142         time += params_.tick ();
 143
 144         Lock l (mutex_);
 145
 146         while (!stop_)
 147         {
 148           if (cond_.wait (&time) == -1)
 149           {
 150             if (errno != ETIME)
 151               ACE_OS::abort ();
 152             else
 153               break;
 154           }
 155         }
 156
 157         if (stop_)
 158           break;
 159       }
 160     }
 161   }
 162
 163   void Acknowledge::
 164   track_queue (Address const& addr, Queue& q, Messages& msgs)
 165   {
 166     unsigned short max_payload_size (
 167       params_.max_packet_size () - max_service_size);
 168
 169     u32 max_elem (NAK::max_count (max_payload_size));
 170     u32 count (0);
 171
 172     Queue::iterator i (q.begin ()), e (q.end ());
 173
 174     // Track existing losses.
 175     //
 176     while (i != e)
 177     {
 178       std::unique_ptr<NAK> nak (new NAK (addr));
 179
 180       // Inner loop that fills NAK profile with up to max_elem elements.
 181       //
 182       for (; i != e && nak->count () < max_elem; ++i)
 183       {
 184         u64 sn ((*i).ext_id_);
 185         Descr& d = (*i).int_id_;
 186
 187         if (d.lost ())
 188         {
 189           d.timer (d.timer () - 1);
 190
 191           if (d.timer () == 0)
 192           {
 193             //@@ Need exp fallback.
 194             //
 195             d.nak_count (d.nak_count () + 1);
 196             d.timer ((d.nak_count () + 1) * params_.nak_timeout ());
 197
 198             nak->add (sn);
 199
 200             ++count;
 201
 202             // cerr << 6 << "NAK # " << d.nak_count () << ": "
 203             // << addr << " " << sn << endl;
 204           }
 205         }
 206       }
 207
 208       // Send this NAK.
 209       //
 210       if (nak->count ())
 211       {
 212         // cerr << 5 << "NAK: " << addr << " " << nak->count () << " sns"
 213         //     << endl;
 214
 215         Message_ptr m (new Message);
 216
 217         m->add (Profile_ptr (nak.release ()));
 218
 219         msgs.push_back (m);
 220       }
 221     }
 222
 223     // Detect and record new losses.
 224     //
 225     for (u64 sn (q.sn () + 1), end (q.max_sn ()); sn < end; ++sn)
 226     {
 227       if (q.find (sn) == -1)
 228       {
 229         q.bind (sn, Descr (1));
 230       }
 231     }
 232   }
 233
 234   void Acknowledge::recv (Message_ptr m)
 235   {
 236     // Handle NRTM. There could be some nasty interaction with code
 237     // that handles data below (like missing message and NAK). This
 238     // is why I hold the lock at the beginning (which may be not very
 239     // efficient).
 240     //
 241     Lock l (mutex_);
 242
 243     if (NRTM const* nrtm = static_cast<NRTM const*> (m->find (NRTM::id)))
 244     {
 245       for (Map::iterator i (hold_.begin ()), e (hold_.end ()); i != e; ++i)
 246       {
 247         u64 sn (nrtm->find ((*i).ext_id_));
 248
 249         if (sn != 0)
 250         {
 251           Queue& q = (*i).int_id_;
 252
 253           u64 old (q.max_sn ());
 254
 255           if (old < sn)
 256           {
 257             // Mark as lost.
 258             //
 259             q.bind (sn, Descr (1));
 260           }
 261         }
 262       }
 263     }
 264
 265     if (m->find (Data::id) || m->find (NoData::id))
 266     {
 267       Address from (
 268         static_cast<From const*> (m->find (From::id))->address ());
 269
 270       u64 sn (static_cast<SN const*> (m->find (SN::id))->num ());
 271
 272       Map::ENTRY* e = 0;
 273
 274       if (hold_.find (from, e) == -1)
 275       {
 276         // First message from this source.
 277         //
 278         hold_.bind (from, Queue (sn));
 279         in_->recv (m);
 280       }
 281       else
 282       {
 283         Queue& q = e->int_id_;
 284
 285         if (sn <= q.sn ())
 286         {
 287           // Duplicate.
 288           //
 289           //cerr << 6 << "DUP " << from << " " << q.sn () << " >= " << sn
 290           //     << endl;
 291         }
 292         else if (sn == q.sn () + 1)
 293         {
 294           // Next message.
 295           //
 296
 297           q.rebind (sn, Descr (m));
 298           collapse (q);
 299         }
 300         else
 301         {
 302           // Some messages are missing. Insert this one into the queue.
 303           //
 304           q.rebind (sn, Descr (m));
 305         }
 306       }
 307     }
 308     else
 309     {
 310       l.release ();
 311
 312       // Just forward it up.
 313       //
 314       in_->recv (m);
 315     }
 316   }
 317
 318   void Acknowledge::send (Message_ptr m)
 319   {
 320     if (Data const* data = static_cast<Data const*> (m->find (Data::id)))
 321     {
 322       size_t max_payload_size (
 323         params_.max_packet_size () - max_service_size);
 324
 325       if (max_payload_size > data->size ())
 326       {
 327         u32 max_size (max_payload_size - data->size ());
 328         u32 max_elem (NRTM::max_count (max_size));
 329
 330         if (max_elem > 0)
 331         {
 332           Lock l (mutex_);
 333
 334           Profile_ptr nrtm (create_nrtm (max_elem));
 335
 336           if (nrtm.get ())
 337             m->add (nrtm);
 338         }
 339       }
 340
 341       nrtm_timer_ = params_.nrtm_timeout (); // Reset timer.
 342     }
 343
 344     out_->send (m);
 345   }
 346
 347   Profile_ptr Acknowledge::
 348   create_nrtm (u32 max_elem)
 349   {
 350     // Prepare NRTM.
 351     //
 352     std::unique_ptr<NRTM> nrtm (new NRTM ());
 353
 354     // Gather the information.
 355     //
 356     {
 357       for (Map::iterator i (hold_.begin ()), e (hold_.end ()); i != e; ++i)
 358       {
 359         Address addr ((*i).ext_id_);
 360         Queue& q = (*i).int_id_;
 361
 362         //@@ Should look for the highest known number.
 363         //
 364         nrtm->insert (addr, q.sn ());
 365
 366         if (--max_elem == 0)
 367           break;
 368       }
 369     }
 370
 371     if (nrtm->empty ())
 372       return Profile_ptr (0);
 373     else
 374       return Profile_ptr (nrtm.release ());
 375   }
 376
 377   ACE_THR_FUNC_RETURN Acknowledge::
 378   track_thunk (void* obj)
 379   {
 380     reinterpret_cast<Acknowledge*> (obj)->track ();
 381     return 0;
 382   }
 383 }