-struct GREPacket {
- int seq;
- string data;
-
- bool operator> (const GREPacket &other) const {
- return seq > other.seq;
- }
-};
-
-class Reorderer {
-public:
- Reorderer(int sock, const in6_addr &dst);
- void handle_packet(const char* buf, size_t size, int seq);
-
-private:
- void send_packet(const string &data);
-
- int sock;
- sockaddr_in6 dstaddr;
- int last_seq;
-
- priority_queue<GREPacket, vector<GREPacket>, greater<GREPacket>> packet_buffer;
- map<int, int> ccs;
-};
-
-Reorderer::Reorderer(int sock, const in6_addr &dst)
- : sock(sock), last_seq(-1)
-{
- memset(&dstaddr, 0, sizeof(dstaddr));
- dstaddr.sin6_family = AF_INET6;
- dstaddr.sin6_addr = dst;
-}
-
-#define PACKET_BUFFER_SIZE 100
-
-void Reorderer::handle_packet(const char* buf, size_t size, int seq)
-{
- if (packet_buffer.size() >= PACKET_BUFFER_SIZE) {
- printf("Gave up waiting for packets [%d,%d>\n",
- last_seq + 1, packet_buffer.top().seq);
- last_seq = packet_buffer.top().seq - 1;
- }
-
- GREPacket packet;
- packet.seq = seq;
- packet.data = string(buf, buf + size);
- packet_buffer.push(packet);
-
- bool silence = false;
- while (!packet_buffer.empty() &&
- (last_seq == -1 || packet_buffer.top().seq <= last_seq + 1)) {
- int 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",
- seq, front_seq);
- packet_buffer.pop();
- continue;
- }
- //if (packet_buffer.size() > 1) {
- // printf("seq=%d (REORDER %d)\n", front_seq, int(packet_buffer.size()));
- //} else {
- // printf("seq=%d\n", front_seq);
- //}
- const string &data = packet_buffer.top().data;
- send_packet(data);
- 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;
- }
- }
-}
-
-void Reorderer::send_packet(const string &data)
-{
- if (data.size() == 1352) {
- for (int i = 0; i < 7; ++i) {
- const char *pkt = &data[i * 188 + 36];
- int pid = (ntohl(*(uint32_t *)(pkt)) & 0x1fff00) >> 8;
- int has_payload = pkt[3] & 0x10;
- int cc = pkt[3] & 0xf;
- if (has_payload) {
- int last_cc = ccs[pid];
- if (cc != ((last_cc + 1) & 0xf)) {
- printf("Pid %d discontinuity (expected %d, got %d)\n", pid, (last_cc + 1) & 0xf, cc);
- }
- ccs[pid] = cc;
- }
- }
- }
- if (sendto(sock, data.data(), data.size(), 0, (sockaddr *)&dstaddr, sizeof(dstaddr)) == -1) {
- perror("sendto");
- return;
- }
-}
-