ACE/protocols/ace/RMCast/Flow.cpp

   1 // author    : Boris Kolpackov <boris@kolpackov.net>
   2 #include "Flow.h"
   3
   4 #include "ace/ACE.h"
   5 #include "ace/OS_NS_unistd.h"   // sleep
   6 #include "ace/OS_NS_sys_time.h" // gettimeofday
   7
   8 #include "ace/os_include/os_math.h" // exp
   9
  10 namespace ACE_RMCast
  11 {
  12   Flow::
  13   Flow (Parameters const& )
  14     : //params_ (params),
  15         nak_time_ (0, 0),
  16         sample_start_time_ (0, 0),
  17         sample_bytes_ (0),
  18         current_tput_ (0.0),
  19         cap_tput_ (0.0)
  20   {
  21   }
  22
  23   void Flow::send (Message_ptr m)
  24   {
  25     if (Data const* data = static_cast<Data const*> (m->find (Data::id)))
  26     {
  27       ACE_Time_Value now_time (ACE_OS::gettimeofday ());
  28
  29       Lock l (mutex_);
  30       sample_bytes_ += data->size ();
  31
  32       if (sample_start_time_ == ACE_Time_Value (0, 0))
  33       {
  34         sample_start_time_ = now_time;
  35       }
  36       else
  37       {
  38         ACE_Time_Value delta (now_time - sample_start_time_);
  39
  40         if (delta > ACE_Time_Value (0, 2000))
  41         {
  42           current_tput_ =
  43             double (sample_bytes_) / (delta.sec () * 1000000 + delta.usec ());
  44
  45           // cerr << "tput: " << current_tput_ << " bytes/usec" << endl;
  46
  47           sample_bytes_ = 0;
  48           sample_start_time_ = ACE_Time_Value (0, 0);
  49         }
  50       }
  51
  52       if (!ACE::is_equal (cap_tput_, 0.0) &&
  53           !ACE::is_equal (current_tput_, 0.0) &&
  54           current_tput_ > cap_tput_)
  55       {
  56         double dev = (current_tput_ - cap_tput_) / current_tput_;
  57
  58         // cerr << "deviation: " << dev << endl;
  59
  60         // Cap decay algorithm.
  61         //
  62         {
  63           ACE_Time_Value delta (now_time - nak_time_);
  64
  65           unsigned long msec = delta.msec ();
  66
  67           double x = msec / -16000.0;
  68           double y = 1.0 * exp (x);
  69           cap_tput_ = cap_tput_ / y;
  70
  71           // cerr << "cap decay: " << cap_tput_ << " bytes/usec" << endl;
  72         }
  73
  74         l.release ();
  75
  76
  77         timespec time;
  78         time.tv_sec = 0;
  79         time.tv_nsec = static_cast<unsigned long> (dev * 500000.0);
  80
  81         // Don't bother to sleep if the time is less than 10 usec.
  82         //
  83         if (time.tv_nsec > 10000)
  84           ACE_OS::sleep (ACE_Time_Value (time));
  85       }
  86     }
  87
  88     out_->send (m);
  89   }
  90
  91   void Flow::recv (Message_ptr m)
  92   {
  93     if (NAK const* nak = static_cast<NAK const*> (m->find (NAK::id)))
  94     {
  95       Address to (static_cast<To const*> (m->find (To::id))->address ());
  96
  97       if (nak->address () == to)
  98       {
  99         // This one is for us.
 100         //
 101
 102         //cerr << "NAK from "
 103         //     << static_cast<From const*> (m->find (From::id))->address ()
 104         //     << " for " << nak->count () << " sns." << endl;
 105
 106
 107         ACE_Time_Value nak_time (ACE_OS::gettimeofday ());
 108
 109         Lock l (mutex_);
 110
 111         nak_time_ = nak_time;
 112
 113         if (ACE::is_equal (cap_tput_, 0.0))
 114           cap_tput_ = current_tput_;
 115
 116         if (!ACE::is_equal (cap_tput_, 0.0))
 117         {
 118           cap_tput_ = cap_tput_ - cap_tput_ / 6.0;
 119
 120           // cerr << "cap: " << cap_tput_ << " bytes/usec" << endl;
 121         }
 122       }
 123     }
 124
 125     in_->recv (m);
 126   }
 127 }