#include "reorderer.h"
#include "protocol.h"
+#include "timeutil.h"
-#define PACKET_BUFFER_SIZE 100
+#define PACKET_BUFFER_SIZE 1000
+#define TIMEOUT_SEC 1.000
using namespace std;
Reorderer::Reorderer(Sender* sender)
: sender(sender), last_seq(-1)
{
+ gettimeofday(&last_sent_packet, NULL);
}
-void Reorderer::send_packet(uint16_t proto, const string& data, int seq)
+void Reorderer::possibly_adjust_tv(timeval *tv)
{
+ if (packet_buffer.empty()) {
+ return;
+ }
+
+ timeval now;
+ gettimeofday(&now, NULL);
+ timeval tdiff = subtract_timeval_saturate(
+ offset_timeval_seconds(packet_buffer.top().ts, TIMEOUT_SEC), now);
+ if (less_than(tdiff, *tv)) {
+ *tv = tdiff;
+ }
+}
+
+void Reorderer::send_packet(uint16_t proto, const string& data, uint32_t seq)
+{
+ timeval now;
+ gettimeofday(&now, NULL);
+
bool silence = false;
if (packet_buffer.size() >= PACKET_BUFFER_SIZE) {
- printf("Gave up waiting for packets [%d,%d>\n",
+ printf("Gave up waiting for packets [%d,%d> (buffer full)\n",
+ last_seq + 1, packet_buffer.top().seq);
+ silence = true;
+ num_lost_packets += packet_buffer.top().seq - (last_seq + 1);
+ last_seq = packet_buffer.top().seq - 1;
+ } else if (!packet_buffer.empty() &&
+ tdiff(packet_buffer.top().ts, now) > TIMEOUT_SEC) {
+ printf("Gave up waiting for packets [%d,%d> (timeout)\n",
last_seq + 1, packet_buffer.top().seq);
silence = true;
+ num_lost_packets += packet_buffer.top().seq - (last_seq + 1);
last_seq = packet_buffer.top().seq - 1;
}
+ // In case of restarts.
+ if (packet_buffer.empty() &&
+ seq < last_seq &&
+ tdiff(last_sent_packet, now) > 5.0) {
+ printf("No good data for five seconds, resetting sequence to %u\n", seq);
+ last_seq = seq - 1;
+ }
+
GREPacket packet;
packet.seq = seq;
packet.proto = proto;
packet.data = data;
+ packet.ts = now;
packet_buffer.push(packet);
+ bool did_reorder = false;
+
while (!packet_buffer.empty() &&
(last_seq == -1 || packet_buffer.top().seq <= last_seq + 1)) {
- int front_seq = packet_buffer.top().seq;
+ uint32_t front_seq = packet_buffer.top().seq;
if (front_seq < last_seq + 1) {
- printf("Duplicate packet or way out-of-order: seq=%d front_seq=%d\n",
+ printf("Duplicate packet or way out-of-order: seq=%u front_seq=%u\n",
front_seq, last_seq + 1);
packet_buffer.pop();
continue;
//}
const string &data = packet_buffer.top().data;
check_ts_discontinuity(packet_buffer.top().proto, data, silence);
- sender->send_packet(proto, data, seq);
+ sender->send_packet(packet_buffer.top().proto, data, packet_buffer.top().seq);
packet_buffer.pop();
last_seq = front_seq;
- if (!silence && !packet_buffer.empty()) {
- printf("Reordering with packet buffer size %d: seq=%d new_front_seq=%d\n", int(packet_buffer.size()), front_seq, packet_buffer.top().seq);
- silence = true;
+ last_sent_packet = now;
+ if (!packet_buffer.empty()) {
+ did_reorder = true;
+ if (!silence) {
+ printf("Reordering with packet buffer size %d: seq=%d new_front_seq=%d\n", int(packet_buffer.size()), front_seq, packet_buffer.top().seq);
+ silence = true;
+ }
}
}
+
+ if (did_reorder) {
+ ++num_reorders;
+ }
}
void Reorderer::check_ts_discontinuity(uint16_t proto, const string &data, bool silence)
int cc = pkt[3] & 0xf;
if (has_payload) {
int last_cc = ccs[pid];
- if (!silence && cc != ((last_cc + 1) & 0xf)) {
- printf("Pid %d discontinuity (expected %d, got %d)\n", pid, (last_cc + 1) & 0xf, cc);
+ if (cc != ((last_cc + 1) & 0xf)) {
+ if (!silence) {
+ printf("Pid %d discontinuity (expected %d, got %d)\n", pid, (last_cc + 1) & 0xf, cc);
+ }
+ ++num_ts_discontinuities;
}
ccs[pid] = cc;
}