git.sesse.net Git - greproxy/blob - pacer.cpp

   1 #include "pacer.h"
   2
   3 #include <algorithm>
   4
   5 using namespace std;
   6
   7 Pacer::Pacer(Sender *sender, int max_rate_kbit_per_sec, int burst_num_packets)
   8         : sender(sender), burst_num_packets(burst_num_packets), next_send_packet{0, 0}
   9 {
  10         const int max_rate_byte_per_sec = 1000 * max_rate_kbit_per_sec / 8;
  11         seconds_per_byte = 1.0 / max_rate_byte_per_sec;
  12 }
  13
  14 bool less_than(const timeval &a, const timeval &b)
  15 {
  16         return make_pair(a.tv_sec, a.tv_usec) < make_pair(b.tv_sec, b.tv_usec);
  17 }
  18
  19 void Pacer::send_packet(uint16_t proto, const string &data, int incoming_seq)
  20 {
  21         waiting_packets.push_back(GREPacket{incoming_seq, proto, data, {0, 0}});
  22         possibly_flush_packets();
  23 }
  24
  25 void Pacer::possibly_adjust_tv(timeval *tv)
  26 {
  27         if (waiting_packets.empty()) {
  28                 return;
  29         }
  30
  31         timeval now;
  32         gettimeofday(&now, NULL);
  33         if (less_than(next_send_packet, now)) {
  34                 // Should send packets immediately.
  35                 tv->tv_sec = tv->tv_usec = 0;
  36                 return;
  37         }
  38
  39         timeval tdiff;
  40         tdiff.tv_sec = next_send_packet.tv_sec - now.tv_sec;
  41         tdiff.tv_usec = next_send_packet.tv_usec - now.tv_usec;
  42         if (tdiff.tv_usec < 0) {
  43                 tdiff.tv_usec += 1000000;
  44                 --tdiff.tv_sec;
  45         }
  46
  47         if (less_than(tdiff, *tv)) {
  48                 *tv = tdiff;
  49         }
  50 }
  51
  52 void Pacer::possibly_flush_packets()
  53 {
  54         if (waiting_packets.empty()) {
  55                 return;
  56         }
  57
  58         timeval now;
  59         gettimeofday(&now, NULL);
  60         if (less_than(now, next_send_packet)) {
  61                 return;
  62         }
  63
  64         int bytes_sent = 0;
  65         for (int i = 0; i < burst_num_packets && !waiting_packets.empty(); ++i) {
  66                 const GREPacket &packet = waiting_packets.front();
  67                 sender->send_packet(packet.proto, packet.data, packet.seq);
  68                 bytes_sent += packet.data.size();
  69                 waiting_packets.pop_front();
  70         }
  71
  72         int usec_to_add = lrint(1e6 * seconds_per_byte * bytes_sent);
  73         int sec_to_add = usec_to_add / 1000000;
  74         usec_to_add %= 1000000;
  75
  76         next_send_packet = now;
  77         next_send_packet.tv_usec += usec_to_add;
  78         next_send_packet.tv_sec += sec_to_add;
  79         next_send_packet.tv_sec += next_send_packet.tv_usec / 1000000;
  80         next_send_packet.tv_usec %= 1000000;
  81 }
  82