X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;f=reorderer.cpp;h=051280287c7754d066c71906438923de610131a6;hb=a8e9b75339d48294145a9b102e43046a10b0bc36;hp=b6ac6a9079906b8aaac5944ef8398f88fdc341e9;hpb=8e81c01b2de70c5e7e1a73d25ebbece0a95bd2c8;p=greproxy diff --git a/reorderer.cpp b/reorderer.cpp index b6ac6a9..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::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() && @@ -47,6 +75,7 @@ void Reorderer::send_packet(uint16_t proto, const string& data, int seq) 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;