#include <stdio.h>
#include <string.h>
#include <arpa/inet.h>
+#include <fecpp.h>
#include <map>
#include "rsdecoder.h"
#include "rs_parm.h"
-extern "C" {
-#include <fec.h>
-}
-
#define RS_GROUP_HISTORY 3
using namespace std;
RSDecoder::RSDecoder(Sender *sender)
- : sender(sender)
+ : sender(sender),
+ rs(RS_PAYLOAD_SIZE, RS_GROUP_SIZE)
{
- rs = init_rs_char(RS_SYM_SIZE, RS_GF_POLY, 1, 1, RS_PARITY_SIZE, RS_PAD);
}
void RSDecoder::send_packet(uint16_t proto, const std::string &data, int incoming_seq)
if (num_regular < RS_PAYLOAD_SIZE) {
// Piece the data back into the different RS groups.
vector<string> padded_packets;
- vector<int> missing_packets;
- for (int i = 0; i < RS_GROUP_SIZE; ++i) {
- int packet_num = (i < RS_PAYLOAD_SIZE) ? rs_group * RS_PAYLOAD_SIZE + i :
- rs_group * RS_PAYLOAD_SIZE - 1 - (i - RS_PAYLOAD_SIZE);
+ std::map<size_t, const fecpp::byte *> shares;
+ for (const auto &packet_pair : group.packets) {
+ int packet_seq = packet_pair.first;
+ int share_num;
string p;
p.resize(max_length);
- const auto it = group.packets.find(packet_num);
- if (it == group.packets.end()) {
- missing_packets.push_back(i);
- } else if (i < RS_PAYLOAD_SIZE) {
+ const GREPacket &packet = packet_pair.second;
+ if (packet_seq >= rs_group * RS_PAYLOAD_SIZE) {
// Regular packet.
- const GREPacket &packet = it->second;
+ share_num = packet_seq - rs_group * RS_PAYLOAD_SIZE;
uint16_t proto_be = htons(packet.proto);
memcpy(&p[0], &proto_be, sizeof(uint16_t));
uint16_t len_be = htons(packet.data.size());
memcpy(&p[4], packet.data.data(), packet.data.size());
} else {
// RS packet.
- const GREPacket &packet = it->second;
+ share_num = RS_PAYLOAD_SIZE + (rs_group * RS_PAYLOAD_SIZE - packet_seq - 1);
memcpy(&p[0], packet.data.data(), packet.data.size());
}
padded_packets.push_back(p);
- }
-
- // Now reconstruct the missing pieces.
- unsigned char ch[RS_GROUP_SIZE];
- for (int i = 0; i < max_length; ++i) {
- for (int j = 0; j < RS_GROUP_SIZE; ++j) {
- ch[j] = padded_packets[j][i];
- }
- int ret = decode_rs_char(rs, ch, &missing_packets[0], missing_packets.size());
- if (ret == -1) {
- printf("Failed reconstruction!\n");
- // We might get more data later, so don't remove it.
- return;
- }
- for (int j = 0; j < RS_GROUP_SIZE; ++j) {
- padded_packets[j][i] = ch[j];
- }
+ shares[share_num] = reinterpret_cast<const fecpp::byte *>(padded_packets.back().data());
}
// Output all packets we didn't have before. They will come
// out-of-order, which will be the job of the Reorderer to fix.
- for (int i = 0; i < RS_PAYLOAD_SIZE; ++i) {
- int packet_num = rs_group * RS_PAYLOAD_SIZE + i;
- if (group.packets.count(packet_num) != 0) {
+ rs.decode(shares, max_length, [&](size_t share_num, size_t num_shares, const fecpp::byte data[], size_t len){
+ if (shares.count(share_num)) {
// Already had this packet.
- continue;
+ return;
}
- const string &p = padded_packets[i];
+ int packet_num = rs_group * RS_PAYLOAD_SIZE + share_num;
uint16_t proto_be, len_be;
- memcpy(&proto_be, &p[0], sizeof(uint16_t));
- memcpy(&len_be, &p[2], sizeof(uint16_t));
- string s(&p[4], &p[4 + ntohs(len_be)]); // TODO: security
+ memcpy(&proto_be, &data[0], sizeof(uint16_t));
+ memcpy(&len_be, &data[2], sizeof(uint16_t));
+ string s(&data[4], &data[4 + ntohs(len_be)]); // TODO: security
sender->send_packet(ntohs(proto_be), s, packet_num);
- printf("Reconstructed packet %d\n", packet_num);
- }
+ const unsigned char *ptr = &data[4];
+ printf("Reconstructed packet %d (proto=0x%04x len=%d, data=%02x %02x %02x %02x %02x %02x %02x %02x)\n", packet_num,
+ ntohs(proto_be), ntohs(len_be),
+ ptr[0], ptr[1], ptr[2], ptr[3], ptr[4], ptr[5], ptr[6], ptr[7]);
+
+ });
}
group.done = true;
#include <stdlib.h>
#include <sys/select.h>
#include <sys/socket.h>
-extern "C" {
-#include <fec.h>
-}
+#include <fecpp.h>
#include "reorderer.h"
#include "rsencoder.h"
using namespace std;
RSEncoder::RSEncoder(Sender *sender)
- : sender(sender)
+ : sender(sender),
+ rs(RS_PAYLOAD_SIZE, RS_GROUP_SIZE)
{
- rs = init_rs_char(RS_SYM_SIZE, RS_GF_POLY, 1, 1, RS_PARITY_SIZE, RS_PAD);
}
void RSEncoder::send_packet(uint16_t proto, const std::string &data, int incoming_seq)
max_length = max<int>(max_length, packet_history[i].data.size());
}
- vector<string> padded_packets;
+ string padded_packets;
+ padded_packets.reserve((max_length + 4) * packet_history.size());
+ // TODO: RS_PAD
for (int i = 0; i < packet_history.size(); ++i) {
string p;
p.resize(max_length + 4);
uint16_t len_be = htons(packet_history[i].data.size());
memcpy(&p[2], &len_be, sizeof(uint16_t));
memcpy(&p[4], packet_history[i].data.data(), packet_history[i].data.size());
- padded_packets.push_back(p);
+ padded_packets += p;
}
- // Now construct RS packets.
- vector<string> rs_packets;
- for (int i = 0; i < RS_PARITY_SIZE; ++i) {
- string p;
- p.resize(max_length + 4);
- memset(&p[0], 0, max_length + 4);
- rs_packets.push_back(p);
- }
- string data, parity;
- data.resize(RS_PAYLOAD_SIZE);
- parity.resize(RS_PARITY_SIZE);
- for (int i = 0; i < max_length + 4; ++i) {
- for (int j = 0; j < packet_history.size(); ++j) {
- data[j] = packet_history[j].data[i];
- }
- encode_rs_char(rs,
- reinterpret_cast<unsigned char *>(&data[0]),
- reinterpret_cast<unsigned char *>(&parity[0]));
- for (int j = 0; j < RS_PARITY_SIZE; ++j) {
- rs_packets[j][i] = parity[j];
- }
- }
+ // Now construct and send RS packets.
+ rs.encode(reinterpret_cast<const fecpp::byte*>(padded_packets.data()),
+ padded_packets.size(),
+ [&](size_t packet_num, size_t num_packets, const fecpp::byte data[], size_t size) {
+ // The first N packets are just the original ones; ignore them.
+ if (packet_num < RS_PAYLOAD_SIZE) {
+ return;
+ }
- // Actually send the RS packets.
- int start_seq = packet_history[0].seq - 1;
- for (int i = 0; i < RS_PARITY_SIZE; ++i) {
- sender->send_packet(0xffff, rs_packets[i], start_seq - i);
- }
+ const char *sdata = reinterpret_cast<const char *>(data);
+ int start_seq = packet_history[0].seq - 1;
+ for (int i = 0; i < RS_PARITY_SIZE; ++i) {
+ sender->send_packet(0xffff, string(sdata, size), start_seq - (packet_num - RS_PAYLOAD_SIZE));
+ }
+ });
packet_history.clear();
}
projects / greproxy / commitdiff