X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;f=reorderer.cpp;h=051280287c7754d066c71906438923de610131a6;hb=ed84bfc7a5e30f49b35d9a864c6618cb58beb789;hp=fe3b7b64cd98027b4be2b49eab8399ec8618d89c;hpb=b93c68671f62724af6e0b21e4d09af7ca91e9749;p=greproxy

diff --git a/reorderer.cpp b/reorderer.cpp
index fe3b7b6..0512802 100644
--- a/reorderer.cpp
+++ b/reorderer.cpp
@@ -3,29 +3,57 @@
 #include "reorderer.h"
 #include "protocol.h"
 
-#define PACKET_BUFFER_SIZE 100
+#define PACKET_BUFFER_SIZE 1000
+#define TIMEOUT_SEC 1.000
 
 using namespace std;
 
-Reorderer::Reorderer(Protocol* sender)
+double tdiff(const timeval& a, const timeval& b)
+{
+	return b.tv_sec - a.tv_sec +
+		1e-6 * (b.tv_usec - a.tv_usec);
+}
+
+Reorderer::Reorderer(Sender* sender)
 	: sender(sender), last_seq(-1)
 {
+	gettimeofday(&last_sent_packet, NULL);
 }
 
-void Reorderer::handle_packet(uint16_t proto, const string& data, int seq)
+void Reorderer::send_packet(uint16_t proto, const string& data, int 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;
 		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;
+		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 %d\n", seq);
+		last_seq = seq - 1;
 	}
 
 	GREPacket packet;
 	packet.seq = seq;
 	packet.proto = proto;
 	packet.data = data;
+	packet.ts = now;
 	packet_buffer.push(packet);
 
 	while (!packet_buffer.empty() &&
@@ -43,9 +71,11 @@ void Reorderer::handle_packet(uint16_t proto, const string& data, int seq)
 		//	printf("seq=%d\n", front_seq);
 		//}
 		const string &data = packet_buffer.top().data;
-		send_packet(packet_buffer.top().proto, data, silence);
+		check_ts_discontinuity(packet_buffer.top().proto, data, silence);
+		sender->send_packet(proto, data, seq);
 		packet_buffer.pop();
 		last_seq = front_seq;
+		last_sent_packet = now;
 		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;
@@ -53,7 +83,7 @@ void Reorderer::handle_packet(uint16_t proto, const string& data, int seq)
 	}
 }
 
-void Reorderer::send_packet(uint16_t proto, const string &data, bool silence)
+void Reorderer::check_ts_discontinuity(uint16_t proto, const string &data, bool silence)
 {
 	if (data.size() == 1344) {
 		for (int i = 0; i < 7; ++i) {
@@ -74,6 +104,5 @@ void Reorderer::send_packet(uint16_t proto, const string &data, bool silence)
 			}
 		}
 	}
-	sender->send_packet(proto, data);
 }