X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;f=reorderer.cpp;h=71b90c54d7433308ea0684574768b6bd990ecfa5;hb=HEAD;hp=7b19d1d3435ab9e5b1ba702b38c6e70d295e3408;hpb=1bacf4bb3d9dbd9f07e33b5676f5a1be43cd83a5;p=greproxy

diff --git a/reorderer.cpp b/reorderer.cpp
index 7b19d1d..71b90c5 100644
--- a/reorderer.cpp
+++ b/reorderer.cpp
@@ -2,24 +2,35 @@
 
 #include "reorderer.h"
 #include "protocol.h"
+#include "timeutil.h"
 
-#define PACKET_BUFFER_SIZE 100
-#define TIMEOUT_SEC 0.100
+#define PACKET_BUFFER_SIZE 1000
+#define TIMEOUT_SEC 1.000
 
 using namespace std;
 
-double tdiff(const timeval& a, const timeval& b)
+Reorderer::Reorderer(Sender* sender)
+	: sender(sender), last_seq(-1)
 {
-	return b.tv_sec - a.tv_sec +
-		1e-6 * (b.tv_usec - a.tv_usec);
+	gettimeofday(&last_sent_packet, NULL);
 }
 
-Reorderer::Reorderer(Sender* sender)
-	: sender(sender), last_seq(-1)
+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, int seq)
+void Reorderer::send_packet(uint16_t proto, const string& data, uint32_t seq)
 {
 	timeval now;
 	gettimeofday(&now, NULL);
@@ -29,15 +40,23 @@ void Reorderer::send_packet(uint16_t proto, const string& data, int seq)
 		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;
-		// TODO: Rerun immediately after we've cleared out,
-		// in case there are more timeouts.
+	}
+
+	// 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;
@@ -47,11 +66,13 @@ void Reorderer::send_packet(uint16_t proto, const string& data, int seq)
 	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;
@@ -63,14 +84,22 @@ void Reorderer::send_packet(uint16_t proto, const string& data, int seq)
 		//}
 		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)
@@ -87,8 +116,11 @@ void Reorderer::check_ts_discontinuity(uint16_t proto, const string &data, bool
 			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;
 			}