rs.h

 
/*
 * Copyright (c) 2021 Cisco and/or its affiliates.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#pragma once
 
#include <arpa/inet.h>
#include <hicn/transport/portability/c_portability.h>
#include <hicn/transport/utils/membuf.h>
#include <protocols/fec/fec_info.h>
#include <protocols/fec_base.h>
 
#include <array>
#include <cstdint>
#include <map>
#include <unordered_set>
#include <vector>
 
namespace transport {
namespace protocol {
 
namespace fec {
 
#define foreach_rs_fec_type \
  _(RS, 1, 3)               \
  _(RS, 4, 5)               \
  _(RS, 4, 6)               \
  _(RS, 4, 7)               \
  _(RS, 6, 10)              \
  _(RS, 8, 10)              \
  _(RS, 8, 11)              \
  _(RS, 8, 12)              \
  _(RS, 8, 14)              \
  _(RS, 8, 16)              \
  _(RS, 8, 32)              \
  _(RS, 10, 30)             \
  _(RS, 10, 40)             \
  _(RS, 10, 60)             \
  _(RS, 10, 90)             \
  _(RS, 16, 18)             \
  _(RS, 16, 21)             \
  _(RS, 16, 23)             \
  _(RS, 16, 24)             \
  _(RS, 16, 27)             \
  _(RS, 17, 21)             \
  _(RS, 17, 34)             \
  _(RS, 32, 36)             \
  _(RS, 32, 41)             \
  _(RS, 32, 46)             \
  _(RS, 32, 54)             \
  _(RS, 34, 42)             \
  _(RS, 35, 70)             \
  _(RS, 52, 62)
 
static const constexpr uint16_t MAX_SOURCE_BLOCK_SIZE = 128;
 
using Packets = std::array<std::tuple</* index */ uint32_t, /* buffer */ buffer,
                                      uint32_t /* offset */>,
                           MAX_SOURCE_BLOCK_SIZE>;
 
struct fec_header {
  uint32_t seq_number;
 
  uint8_t encoded_symbol_id;
 
  uint8_t source_block_len;
 
  uint8_t n_fec_symbols;
 
  uint8_t padding;
 
  void setSeqNumberBase(uint32_t suffix) { seq_number = htonl(suffix); }
  uint32_t getSeqNumberBase() { return ntohl(seq_number); }
  void setEncodedSymbolId(uint8_t esi) { encoded_symbol_id = esi; }
  uint8_t getEncodedSymbolId() { return encoded_symbol_id; }
  void setSourceBlockLen(uint8_t k) { source_block_len = k; }
  uint8_t getSourceBlockLen() { return source_block_len; }
  void setNFecSymbols(uint8_t n_r) { n_fec_symbols = n_r; }
  uint8_t getNFecSymbols() { return n_fec_symbols; }
};
 
class rs;
 
class BlockCode : public Packets {
  static constexpr std::size_t LEN_SIZE_BYTES = 2;
 
 public:
  BlockCode(uint32_t k, uint32_t n, uint32_t seq_offset, struct fec_parms *code,
            rs &params);
 
  bool addRepairSymbol(const fec::buffer &packet, uint32_t i,
                       uint32_t offset = 0);
 
  bool addSourceSymbol(const fec::buffer &packet, uint32_t i,
                       uint32_t offset = 0);
 
  std::size_t length() { return current_block_size_; }
 
  uint32_t getN() { return n_; }
 
  uint32_t getK() { return k_; }
 
  void clear();
 
 private:
  bool addSymbol(const fec::buffer &packet, uint32_t i, uint32_t offset,
                 std::size_t size);
 
  void encode();
 
  void decode();
 
 private:
  uint32_t k_;
  uint32_t n_;
  uint32_t seq_offset_;
  struct fec_parms *code_;
  std::size_t max_buffer_size_;
  std::size_t current_block_size_;
  std::vector<uint32_t> sorted_index_;
  bool to_decode_;
  rs &params_;
};
 
class rs : public virtual FECBase {
  friend class BlockCode;
 
  struct MatrixDeleter {
    void operator()(struct fec_parms *params);
  };
 
  using Matrix = std::unique_ptr<struct fec_parms, MatrixDeleter>;
 
  using Code = std::pair<std::uint32_t /* k */, std::uint32_t /* n */>;
 
  struct CodeHasher {
    std::size_t operator()(const Code &code) const {
      uint64_t ret = uint64_t(code.first) << 32 | uint64_t(code.second);
      return std::hash<uint64_t>{}(ret);
    }
  };
 
 protected:
  using PacketsReady = std::function<void(std::vector<buffer> &)>;
 
  using SNBase = std::uint32_t;
 
  using SourceBlocks = std::unordered_map<SNBase, BlockCode>;
 
  using Codes = std::unordered_map<Code, Matrix, CodeHasher>;
 
 public:
  rs(uint32_t k, uint32_t n, uint32_t seq_offset = 0);
  ~rs() = default;
 
  virtual void clear() { processed_source_blocks_.clear(); }
 
  bool isSymbol(uint32_t index) { return ((index - seq_offset_) % n_) >= k_; }
 
 private:
  static Codes createCodes();
 
 protected:
  bool processed(SNBase seq_base) {
    return processed_source_blocks_.find(seq_base) !=
           processed_source_blocks_.end();
  }
 
  void setProcessed(SNBase seq_base) {
    processed_source_blocks_.emplace(seq_base);
  }
 
  std::uint32_t k_;
  std::uint32_t n_;
  std::uint32_t seq_offset_;
 
  std::unordered_set<SNBase> processed_source_blocks_;
 
  static Codes codes_;
};
 
class RSEncoder : public rs, public ProducerFEC {
 public:
  RSEncoder(uint32_t k, uint32_t n, uint32_t seq_offset = 0);
  void consume(const fec::buffer &packet, uint32_t index, uint32_t offset = 0);
 
  void onPacketProduced(core::ContentObject &content_object,
                        uint32_t offset) override;
 
  std::size_t getFecHeaderSize() override { return 0; }
 
  void clear() override {
    rs::clear();
    source_block_.clear();
  }
 
  void reset() override { clear(); }
 
 private:
  struct fec_parms *current_code_;
  BlockCode source_block_;
};
 
class RSDecoder : public rs, public ConsumerFEC {
 public:
  RSDecoder(uint32_t k, uint32_t n, uint32_t seq_offset = 0);
 
  void consumeSource(const fec::buffer &packet, uint32_t i,
                     uint32_t offset = 0);
 
  void consumeRepair(const fec::buffer &packet, uint32_t offset = 0);
 
  void onDataPacket(core::ContentObject &content_object,
                    uint32_t offset) override;
 
  std::size_t getFecHeaderSize() override { return 0; }
 
  void clear() override {
    rs::clear();
    src_blocks_.clear();
    parked_packets_.clear();
  }
 
  void reset() override { clear(); }
 
 private:
  void recoverPackets(SourceBlocks::iterator &src_block_it);
 
 private:
  SourceBlocks src_blocks_;
 
  std::unordered_map<uint32_t, std::vector<std::pair<buffer, uint32_t>>>
      parked_packets_;
};
 
}  // namespace fec
 
}  // namespace protocol
 
}  // namespace transport