PacketScheduler.cpp

Go to the documentation of this file.

#include "PacketScheduler.h"
#include "madara/utility/Utility.h"
#include "madara/logger/GlobalLogger.h"
 
const uint64_t max_stride(150000000);
 
class StrideTask
{
public:
  StrideTask(double rate, bool op) : operation(op)
  {
    set_rate(rate);
  }
  bool operator<(const StrideTask& rhs) const
  {
    return pass > rhs.pass;
  }
 
  bool operator==(const StrideTask& rhs) const
  {
    return pass == rhs.pass;
  }
 
  bool operator>(const StrideTask& rhs) const
  {
    return pass < rhs.pass;
  }
 
  void operator++(void)
  {
    pass += stride;
  }
 
  void set_rate(double rate)
  {
    tickets = (uint64_t)(1000000 * rate);
 
    if (tickets > 0)
      stride = max_stride / tickets;
    else
      stride = 1;
 
    pass = stride;
  }
 
  uint64_t stride;
 
  uint64_t pass;
 
  uint64_t tickets;
 
  bool operation;
};
 
madara::transport::PacketScheduler::PacketScheduler(
    const QoSTransportSettings* settings)
  : settings_(settings),
    sent_messages_(0),
    dropped_messages_(0),
    consecutive_drops_(0)
{
}
 
madara::transport::PacketScheduler::PacketScheduler(const PacketScheduler& rhs)
  : settings_(rhs.settings_),
    sent_messages_(rhs.sent_messages_),
    dropped_messages_(rhs.dropped_messages_),
    consecutive_drops_(rhs.consecutive_drops_)
{
}
 
madara::transport::PacketScheduler::~PacketScheduler() {}
 
void madara::transport::PacketScheduler::operator=(const PacketScheduler& rhs)
{
  MADARA_GUARD_TYPE guard(mutex_);
  if (this != &rhs)
  {
    settings_ = rhs.settings_;
    sent_messages_ = rhs.sent_messages_;
    dropped_messages_ = rhs.dropped_messages_;
    consecutive_drops_ = rhs.consecutive_drops_;
  }
}
 
bool madara::transport::PacketScheduler::add(void)
{
  MADARA_GUARD_TYPE guard(mutex_);
 
  if (settings_)
  {
    double drop_rate = settings_->get_drop_rate();
    int drop_type = settings_->get_drop_type();
    uint64_t drop_burst = settings_->get_drop_burst();
 
    bool result = true;
 
    // if the user has specified a positive drop rate
    if (drop_rate > 0)
    {
      if (drop_rate >= 100 || (consecutive_drops_ > 0 && drop_burst > 1 &&
                                  consecutive_drops_ < drop_burst))
      {
        result = false;
      }
      else
      {
        if (drop_type == PACKET_DROP_PROBABLISTIC)
        {
          // if burst is greater than 1, then divide the rate by burst
          if (drop_burst > 1)
            drop_rate = drop_rate / (drop_burst - 1);
 
          // easiest drop rate policy. Just drop messages with a probability.
          if (utility::rand_double() <= drop_rate)
          {
            result = false;
          }
        }
        else
        {
          // Reset queue if this is the first time we have used this policy
          if (queue_.empty())
          {
            reset();
          }
 
          // remove current operation from the priority queue
          StrideTask current = queue_.top();
          queue_.pop();
 
          // set the result to the current operation
          result = current.operation;
 
          // reinsert the operation into the queue
          ++current;
          queue_.push(current);
        }
      }
    }
 
    if (result)
    {
      ++sent_messages_;
      consecutive_drops_ = 0;
    }
    else
    {
      ++dropped_messages_;
      ++consecutive_drops_;
    }
 
    return result;
  }  // end if settings_
  else
  {
    madara_logger_ptr_log(logger::global_logger.get(), logger::LOG_EMERGENCY,
        "PacketScheduler::add: ERROR: Invalid settings class\n");
 
    return true;
  }
}
 
void madara::transport::PacketScheduler::attach(
    const QoSTransportSettings* settings)
{
  settings_ = settings;
}
 
void madara::transport::PacketScheduler::clear(void)
{
  MADARA_GUARD_TYPE guard(mutex_);
 
  sent_messages_ = 0;
  dropped_messages_ = 0;
  consecutive_drops_ = 0;
  while (!queue_.empty())
    queue_.pop();
}
 
void madara::transport::PacketScheduler::reset(void)
{
  MADARA_GUARD_TYPE guard(mutex_);
 
  if (settings_)
  {
    double drop_rate = settings_->get_drop_rate();
    uint64_t burst = settings_->get_drop_burst();
 
    // if burst is greater than 1, then divide the rate by burst
    if (burst > 1)
      drop_rate = drop_rate / (burst - 1);
 
    StrideTask drop_message(drop_rate, false);
    StrideTask send_message(1 - drop_rate, true);
 
    while (!queue_.empty())
      queue_.pop();
 
    queue_.push(drop_message);
    queue_.push(send_message);
  }
  else
  {
    madara_logger_ptr_log(logger::global_logger.get(), logger::LOG_EMERGENCY,
        "PacketScheduler::reset: ERROR: Invalid settings class\n");
  }
}
 
void madara::transport::PacketScheduler::print_status(
    unsigned int log_level, const char* prefix)
{
  MADARA_GUARD_TYPE guard(mutex_);
 
  madara_logger_ptr_log(logger::global_logger.get(), (int)log_level,
      "%s: %d sent, %d dropped, %d consec dropped\n", prefix, sent_messages_,
      dropped_messages_, consecutive_drops_);
}
 
uint64_t madara::transport::PacketScheduler::get_dropped(void)
{
  MADARA_GUARD_TYPE guard(mutex_);
  return dropped_messages_;
}
 
uint64_t madara::transport::PacketScheduler::get_sent(void)
{
  MADARA_GUARD_TYPE guard(mutex_);
  return sent_messages_;
}