7 #define PACKET_BUFFER_SIZE 1000
8 #define TIMEOUT_SEC 1.000
12 Reorderer::Reorderer(Sender* sender)
13 : sender(sender), last_seq(-1)
15 gettimeofday(&last_sent_packet, NULL);
18 void Reorderer::possibly_adjust_tv(timeval *tv)
20 if (packet_buffer.empty()) {
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)) {
33 void Reorderer::send_packet(uint16_t proto, const string& data, int seq)
36 gettimeofday(&now, NULL);
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);
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);
50 num_lost_packets += packet_buffer.top().seq - (last_seq + 1);
51 last_seq = packet_buffer.top().seq - 1;
54 // In case of restarts.
55 if (packet_buffer.empty() &&
57 tdiff(last_sent_packet, now) > 5.0) {
58 printf("No good data for five seconds, resetting sequence to %d\n", seq);
67 packet_buffer.push(packet);
69 while (!packet_buffer.empty() &&
70 (last_seq == -1 || packet_buffer.top().seq <= last_seq + 1)) {
71 int front_seq = packet_buffer.top().seq;
72 if (front_seq < last_seq + 1) {
73 printf("Duplicate packet or way out-of-order: seq=%d front_seq=%d\n",
74 front_seq, last_seq + 1);
78 //if (packet_buffer.size() > 1) {
79 // printf("seq=%d (REORDER %d)\n", front_seq, int(packet_buffer.size()));
81 // printf("seq=%d\n", front_seq);
83 const string &data = packet_buffer.top().data;
84 check_ts_discontinuity(packet_buffer.top().proto, data, silence);
85 sender->send_packet(packet_buffer.top().proto, data, packet_buffer.top().seq);
88 last_sent_packet = now;
89 if (!packet_buffer.empty()) {
92 printf("Reordering with packet buffer size %d: seq=%d new_front_seq=%d\n", int(packet_buffer.size()), front_seq, packet_buffer.top().seq);
99 void Reorderer::check_ts_discontinuity(uint16_t proto, const string &data, bool silence)
101 if (data.size() == 1344) {
102 for (int i = 0; i < 7; ++i) {
103 const char *pkt = &data[i * 188 + 28];
104 int pid = (ntohl(*(uint32_t *)(pkt)) & 0x1fff00) >> 8;
109 int has_payload = pkt[3] & 0x10;
110 int cc = pkt[3] & 0xf;
112 int last_cc = ccs[pid];
113 if (cc != ((last_cc + 1) & 0xf)) {
115 printf("Pid %d discontinuity (expected %d, got %d)\n", pid, (last_cc + 1) & 0xf, cc);
117 ++num_ts_discontinuities;