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

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