git.sesse.net Git - greproxy/blob - rsencoder.cpp

   1 #include <string.h>
   2 #include <arpa/inet.h>
   3 #include <netinet/in.h>
   4 #include <stdio.h>
   5 #include <stdlib.h>
   6 #include <sys/select.h>
   7 #include <sys/socket.h>
   8 #include <fecpp.h>
   9
  10 #include "reorderer.h"
  11 #include "rsencoder.h"
  12 #include "rs_parm.h"
  13
  14 #include <algorithm>
  15
  16 using namespace std;
  17
  18 RSEncoder::RSEncoder(Sender *sender)
  19         : sender(sender),
  20           rs(RS_PAYLOAD_SIZE, RS_GROUP_SIZE)
  21 {
  22 }
  23
  24 void RSEncoder::send_packet(uint16_t proto, const std::string &data, int incoming_seq)
  25 {
  26         if (!packet_history.empty() &&
  27             incoming_seq <= packet_history.back().seq) {
  28                 // Reorderer should have done this for us.
  29                 return;
  30         }
  31         if (!packet_history.empty() &&
  32             incoming_seq / RS_PAYLOAD_SIZE !=
  33                 packet_history.back().seq / RS_PAYLOAD_SIZE) {
  34                 // Received an unfinished group.
  35                 packet_history.clear();
  36         }
  37         sender->send_packet(proto, data, incoming_seq);
  38         packet_history.emplace_back(GREPacket{incoming_seq, proto, data});
  39         if (packet_history.size() == RS_PAYLOAD_SIZE) {
  40                 finish_group();
  41         }
  42 }
  43
  44 void RSEncoder::finish_group()
  45 {
  46         // Our RS packets need to have the same max length as the longest one.
  47         int max_length = 0;
  48         for (int i = 0; i < packet_history.size(); ++i) {
  49                 max_length = max<int>(max_length, packet_history[i].data.size());
  50         }
  51
  52         string padded_packets;
  53         padded_packets.reserve((max_length + 4) * packet_history.size());
  54         // TODO: RS_PAD
  55         for (int i = 0; i < packet_history.size(); ++i) {
  56                 string p;
  57                 p.resize(max_length + 4);
  58                 memset(&p[0], 0, max_length + 4);
  59                 uint16_t proto_be = htons(packet_history[i].proto);
  60                 memcpy(&p[0], &proto_be, sizeof(uint16_t));
  61                 uint16_t len_be = htons(packet_history[i].data.size());
  62                 memcpy(&p[2], &len_be, sizeof(uint16_t));
  63                 memcpy(&p[4], packet_history[i].data.data(), packet_history[i].data.size());
  64                 padded_packets += p;
  65         }
  66
  67         // Now construct and send RS packets.
  68         rs.encode(reinterpret_cast<const fecpp::byte*>(padded_packets.data()),
  69                   padded_packets.size(),
  70                   [&](size_t packet_num, size_t num_packets, const fecpp::byte data[], size_t size) {
  71                         // The first N packets are just the original ones; ignore them.
  72                         if (packet_num < RS_PAYLOAD_SIZE) {
  73                                 return;
  74                         }
  75
  76                         const char *sdata = reinterpret_cast<const char *>(data);
  77                         int start_seq = packet_history[0].seq - 1;
  78                         for (int i = 0; i < RS_PARITY_SIZE; ++i) {
  79                                 sender->send_packet(0xffff, string(sdata, size), start_seq - (packet_num - RS_PAYLOAD_SIZE));
  80                         }
  81                 });
  82
  83         packet_history.clear();
  84 }
  85