Make Reorderer handle multi-timeouts.

[greproxy] / reorderer.cpp

diff --git a/reorderer.cpp b/reorderer.cpp
index fe3b7b64cd98027b4be2b49eab8399ec8618d89c..2533aadc042064a7577e55c45a56718ad04faa6f 100644 (file)
--- a/reorderer.cpp
+++ b/reorderer.cpp
@@ -2,30 +2,66 @@
 
 #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(Protocol* sender)
+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::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)
+{
+       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;
        }
 
+       // 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 +79,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 +91,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 +112,5 @@ void Reorderer::send_packet(uint16_t proto, const string &data, bool silence)
                        }
                }
        }
-       sender->send_packet(proto, data);
 }