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

   1 #include <arpa/inet.h>
   2
   3 #include "reorderer.h"
   4 #include "protocol.h"
   5 #include "timeutil.h"
   6
   7 #define PACKET_BUFFER_SIZE 1000
   8 #define TIMEOUT_SEC 1.000
   9
  10 using namespace std;
  11
  12 Reorderer::Reorderer(Sender* sender)
  13         : sender(sender), last_seq(-1)
  14 {
  15         gettimeofday(&last_sent_packet, NULL);
  16 }
  17
  18 void Reorderer::possibly_adjust_tv(timeval *tv)
  19 {
  20         if (packet_buffer.empty()) {
  21                 return;
  22         }
  23
  24         timeval now;
  25         gettimeofday(&now, NULL);
  26         timeval tdiff = subtract_timeval_saturate(
  27                 offset_timeval_seconds(packet_buffer.top().ts, TIMEOUT_SEC), now);
  28         if (less_than(tdiff, *tv)) {
  29                 *tv = tdiff;
  30         }
  31 }
  32
  33 void Reorderer::send_packet(uint16_t proto, const string& data, uint32_t seq)
  34 {
  35         timeval now;
  36         gettimeofday(&now, NULL);
  37
  38         bool silence = false;
  39         if (packet_buffer.size() >= PACKET_BUFFER_SIZE) {
  40                 printf("Gave up waiting for packets [%d,%d> (buffer full)\n",
  41                         last_seq + 1, packet_buffer.top().seq);
  42                 silence = true;
  43                 num_lost_packets += packet_buffer.top().seq - (last_seq + 1);
  44                 last_seq = packet_buffer.top().seq - 1;
  45         } else if (!packet_buffer.empty() &&
  46                    tdiff(packet_buffer.top().ts, now) > TIMEOUT_SEC) {
  47                 printf("Gave up waiting for packets [%d,%d> (timeout)\n",
  48                         last_seq + 1, packet_buffer.top().seq);
  49                 silence = true;
  50                 num_lost_packets += packet_buffer.top().seq - (last_seq + 1);
  51                 last_seq = packet_buffer.top().seq - 1;
  52         }
  53
  54         // In case of restarts.
  55         if (packet_buffer.empty() &&
  56             seq < last_seq &&
  57             tdiff(last_sent_packet, now) > 5.0) {
  58                 printf("No good data for five seconds, resetting sequence to %u\n", seq);
  59                 last_seq = seq - 1;
  60         }
  61
  62         GREPacket packet;
  63         packet.seq = seq;
  64         packet.proto = proto;
  65         packet.data = data;
  66         packet.ts = now;
  67         packet_buffer.push(packet);
  68
  69         bool did_reorder = false;
  70
  71         while (!packet_buffer.empty() &&
  72                (last_seq == -1 || packet_buffer.top().seq <= last_seq + 1)) {
  73                 uint32_t front_seq = packet_buffer.top().seq;
  74                 if (front_seq < last_seq + 1) {
  75                         printf("Duplicate packet or way out-of-order: seq=%u front_seq=%u\n",
  76                                 front_seq, last_seq + 1);
  77                         packet_buffer.pop();
  78                         continue;
  79                 }
  80                 //if (packet_buffer.size() > 1) {
  81                 //      printf("seq=%d (REORDER %d)\n", front_seq, int(packet_buffer.size()));
  82                 //} else {
  83                 //      printf("seq=%d\n", front_seq);
  84                 //}
  85                 const string &data = packet_buffer.top().data;
  86                 check_ts_discontinuity(packet_buffer.top().proto, data, silence);
  87                 sender->send_packet(packet_buffer.top().proto, data, packet_buffer.top().seq);
  88                 packet_buffer.pop();
  89                 last_seq = front_seq;
  90                 last_sent_packet = now;
  91                 if (!packet_buffer.empty()) {
  92                         did_reorder = true;
  93                         if (!silence) {
  94                                 printf("Reordering with packet buffer size %d: seq=%d new_front_seq=%d\n", int(packet_buffer.size()), front_seq, packet_buffer.top().seq);
  95                                 silence = true;
  96                         }
  97                 }
  98         }
  99
 100         if (did_reorder) {
 101                 ++num_reorders;
 102         }
 103 }
 104
 105 void Reorderer::check_ts_discontinuity(uint16_t proto, const string &data, bool silence)
 106 {
 107         if (data.size() == 1344) {
 108                 for (int i = 0; i < 7; ++i) {
 109                         const char *pkt = &data[i * 188 + 28];
 110                         int pid = (ntohl(*(uint32_t *)(pkt)) & 0x1fff00) >> 8;
 111                         if (pid == 8191) {
 112                                 // stuffing, ignore
 113                                 continue;
 114                         }
 115                         int has_payload = pkt[3] & 0x10;
 116                         int cc = pkt[3] & 0xf;
 117                         if (has_payload) {
 118                                 int last_cc = ccs[pid];
 119                                 if (cc != ((last_cc + 1) & 0xf)) {
 120                                         if (!silence) {
 121                                                 printf("Pid %d discontinuity (expected %d, got %d)\n", pid, (last_cc + 1) & 0xf, cc);
 122                                         }
 123                                         ++num_ts_discontinuities;
 124                                 }
 125                                 ccs[pid] = cc;
 126                         }
 127                 }
 128         }
 129 }
 130