Newer
Older
/*
* Copyright (c) 2015-2016, Luca Fulchir<luca@fulchir.it>, All rights reserved.
*
* This file is part of "libRaptorQ".
*
* libRaptorQ is free software: you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as
* published by the Free Software Foundation, either version 3
* of the License, or (at your option) any later version.
*
* libRaptorQ is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* and a copy of the GNU Lesser General Public License
* along with libRaptorQ. If not, see <http://www.gnu.org/licenses/>.
*/
#pragma once
#include "RaptorQ/v1/common.hpp"
#include "RaptorQ/v1/block_sizes.hpp"
#include "RaptorQ/v1/Encoder.hpp"
#include "RaptorQ/v1/Decoder.hpp"
namespace RaptorQ__v1 {
namespace Impl {
template <typename Rnd_It, typename Fwd_It>
class RAPTORQ_LOCAL Encoder
{
public:
~Encoder();
// used for precomputation
Encoder (const Block_Size symbols, const size_t symbol_size);
size_t symbol_size() const; //FIXME: max smbol size is same as signed size_t
uint32_t max_repair() const;
RaptorQ__v1::It::Encoder::Symbol_Iterator<Rnd_It, Fwd_It> begin_source();
RaptorQ__v1::It::Encoder::Symbol_Iterator<Rnd_It, Fwd_It> end_source();
RaptorQ__v1::It::Encoder::Symbol_Iterator<Rnd_It, Fwd_It> begin_repair();
RaptorQ__v1::It::Encoder::Symbol_Iterator<Rnd_It, Fwd_It> end_repair
bool has_data() const;
size_t set_data (const Rnd_It &from, const Rnd_It &to);
bool precompute_sync();
bool compute_sync();
std::future<Error> precompute();
size_t encode (Fwd_It &output, const Fwd_It end, const uint32_t id);
bool have_data;
Raw_Encoder<Rnd_It, Fwd_It, without_interleaver> encoder;
DenseMtx precomputed;
std::thread waiting;
Rnd_It _from, _to;
static uint16_t calc_symbols (const Rnd_It data_from, const Rnd_It data_to,
bool forced_precomputation,
std::promise<Error> p);
};
template <typename In_It, typename Fwd_It>
class RAPTORQ_LOCAL Decoder
{
public:
enum class RAPTORQ_LOCAL Report : uint8_t {
PARTIAL_FROM_BEGINNING = RQ_COMPUTE_PARTIAL_FROM_BEGINNING,
PARTIAL_ANY = RQ_COMPUTE_PARTIAL_ANY,
COMPLETE = RQ_COMPUTE_COMPLETE
Decoder (const Block_Size symbols, const size_t symbol_size,
RaptorQ__v1::It::Decoder::Symbol_Iterator<In_It, Fwd_It> begin();
RaptorQ__v1::It::Decoder::Symbol_Iterator<In_It, Fwd_It> end();
Error add_symbol (In_It &from, const In_It to, const uint32_t esi);
void set_max_concurrency (const uint16_t max_threads);
using Decoder_Result = typename Raw_Decoder<In_It>::Decoder_Result;
Decoder_Result decode_once();
std::pair<Error, uint16_t> wait_sync();
std::future<std::pair<Error, uint16_t>> wait();
Error decode_symbol (Fwd_It &start, const Fwd_It end,const uint16_t esi);
std::pair<size_t, size_t> decode_bytes (Fwd_It &start, const Fwd_It end,
const size_t from_byte, const size_t skip);
uint16_t _max_threads;
const uint16_t _symbols;
const size_t _symbol_size;
// 2* symbols. Actually tracks available and reported symbols.
// each symbol gets 2 bool: 1= available, 2=reported
std::mutex _mtx;
std::condition_variable _cond;
std::vector<std::thread> waiting;
static void waiting_thread (Decoder<In_It, Fwd_It> *obj,
std::promise<std::pair<Error, uint16_t>> p);
};
///////////////////
//// Encoder
///////////////////
template <typename Rnd_It, typename Fwd_It>
uint16_t Encoder<Rnd_It, Fwd_It>::calc_symbols (const Rnd_It data_from,
const Rnd_It data_to,
{
using T = typename std::iterator_traits<Rnd_It>::value_type;
uint64_t size = static_cast<uint64_t> (data_to - data_from) * sizeof(T);
uint16_t symbols = static_cast<uint16_t> (size / symbol_size);
if ((size % symbol_size) != 0)
++symbols;
return symbols;
}
template <typename Rnd_It, typename Fwd_It>
Encoder<Rnd_It, Fwd_It>::~Encoder()
{
encoder.stop();
if (waiting.joinable())
waiting.join();
}
template <typename Rnd_It, typename Fwd_It>
Encoder<Rnd_It, Fwd_It>::Encoder (const Block_Size symbols,
const size_t symbol_size)
: _symbol_size (symbol_size), _symbols (static_cast<uint16_t> (symbols)),
encoder (symbols, _symbol_size)
{
IS_RANDOM(Rnd_It, "RaptorQ__v1::Encoder");
IS_FORWARD(Fwd_It, "RaptorQ__v1::Encoder");
}
template <typename Rnd_It, typename Fwd_It>
uint16_t Encoder<Rnd_It, Fwd_It>::symbols() const
{
return _symbols;
}
template <typename Rnd_It, typename Fwd_It>
{
return _symbol_size;
}
template <typename Rnd_It, typename Fwd_It>
uint32_t Encoder<Rnd_It, Fwd_It>::max_repair() const
{
// you can have up to 56403 symbols in a block
// rfc6330 limits you to 992173 repair symbols
// but limits are meant to be broken!
// the limit sould be up to 4294967279 repair symbols 2^32-(_param.S + H)
// but people might misuse the API, and call end_repair(max_repair),
// which would overflow.
// We are sorry for taking away from you that 0.0014% of repair symbols.
auto _param = Parameters (_symbols);
return static_cast<uint32_t> (std::numeric_limits<uint32_t>::max() -
_param.L);
RaptorQ__v1::It::Encoder::Symbol_Iterator<Rnd_It, Fwd_It>
return RaptorQ__v1::It::Encoder::Symbol_Iterator<Rnd_It, Fwd_It> (this, 0);
RaptorQ__v1::It::Encoder::Symbol_Iterator<Rnd_It, Fwd_It>
{
return RaptorQ__v1::It::Encoder::Symbol_Iterator<Rnd_It, Fwd_It> (this,
}
template <typename Rnd_It, typename Fwd_It>
RaptorQ__v1::It::Encoder::Symbol_Iterator<Rnd_It, Fwd_It>
template <typename Rnd_It, typename Fwd_It>
RaptorQ__v1::It::Encoder::Symbol_Iterator<Rnd_It, Fwd_It>
Encoder<Rnd_It, Fwd_It>::end_repair (const uint32_t repair)
return RaptorQ__v1::It::Encoder::Symbol_Iterator<Rnd_It, Fwd_It> (nullptr,
bool Encoder<Rnd_It, Fwd_It>::has_data() const
{ return have_data; }
template <typename Rnd_It, typename Fwd_It>
size_t Encoder<Rnd_It, Fwd_It>::set_data (const Rnd_It &from, const Rnd_It &to)
{
_from = from;
_to = to;
have_data = true;
return static_cast<size_t>(_to - _from) *
sizeof(typename std::iterator_traits<Rnd_It>::value_type);
template <typename Rnd_It, typename Fwd_It>
void Encoder<Rnd_It, Fwd_It>::clear_data()
{
}
template <typename Rnd_It, typename Fwd_It>
bool Encoder<Rnd_It, Fwd_It>::precompute_sync()
{
static RaptorQ__v1::Work_State work = RaptorQ__v1::Work_State::KEEP_WORKING;
if (precomputed.rows() == 0) {
precomputed = encoder.get_precomputed (&work);
if (precomputed.rows() == 0)
return false; // exit was forced.
}
if (have_data)
encoder.generate_symbols (precomputed, &_from, &_to);
return true;
template <typename Rnd_It, typename Fwd_It>
bool Encoder<Rnd_It, Fwd_It>::compute_sync()
{
static RaptorQ__v1::Work_State work = RaptorQ__v1::Work_State::KEEP_WORKING;
if (encoder.ready())
return true;
if (have_data) {
return encoder.generate_symbols (&work, &_from, &_to);
precomputed = encoder.get_precomputed (&work);
return precomputed.rows() != 0;
}
}
template <typename Rnd_It, typename Fwd_It>
void Encoder<Rnd_It, Fwd_It>::compute_thread (
Encoder<Rnd_It, Fwd_It> *obj, bool force_precomputation,
std::promise<Error> p)
{
static RaptorQ__v1::Work_State work = RaptorQ__v1::Work_State::KEEP_WORKING;
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
if (force_precomputation) {
if (obj->precomputed.rows() == 0)
obj->precomputed = obj->encoder.get_precomputed (&work);
if (obj->precomputed.rows() == 0) {
// encoder always works. only possible reason:
p.set_value (Error::EXITING);
return;
}
// if we finished getting data by the time the computation
// finished, update it all.
if (obj->have_data && !obj->encoder.ready())
obj->encoder.generate_symbols (obj->precomputed,
&obj->_from, &obj->_to);
p.set_value (Error::NONE);
} else {
if (obj->encoder.ready()) {
p.set_value (Error::NONE);
return;
}
if (obj->have_data) {
if (obj->encoder.generate_symbols (&work, &obj->_from, &obj->_to)) {
p.set_value (Error::NONE);
return;
} else {
// only possible reason:
p.set_value (Error::EXITING);
return;
}
} else {
if (obj->precomputed.rows() == 0) {
obj->precomputed = obj->encoder.get_precomputed (&work);
if (obj->precomputed.rows() == 0) {
// only possible reason:
p.set_value (Error::EXITING);
return;
}
}
if (obj->have_data) {
// if we finished getting data by the time the computation
// finished, update it all.
obj->encoder.generate_symbols (obj->precomputed,
&obj->_from, &obj->_to);
}
p.set_value (Error::NONE);
return;
}
}
}
template <typename Rnd_It, typename Fwd_It>
std::future<Error> Encoder<Rnd_It, Fwd_It>::precompute()
{
std::promise<Error> p;
auto future = p.get_future();
// only one waiting thread for the encoder
if (waiting.joinable()) {
p.set_value (Error::WORKING);
return p.get_future();
waiting = std::thread (compute_thread, this, true, std::move(p));
return future;
}
template <typename Rnd_It, typename Fwd_It>
std::future<Error> Encoder<Rnd_It, Fwd_It>::compute()
{
std::promise<Error> p;
auto future = p.get_future();
// only one waiting thread for the encoder
if (waiting.joinable()) {
p.set_value (Error::WORKING);
return p.get_future();
}
waiting = std::thread (compute_thread, this, false, std::move(p));
size_t Encoder<Rnd_It, Fwd_It>::encode (Fwd_It &output, const Fwd_It end,
if (have_data) {
if (!encoder.ready()) {
if (precomputed.rows() == 0)
encoder.generate_symbols (precomputed, &_from, &_to);
}
}
///////////////////
//// Decoder
///////////////////
template <typename In_It, typename Fwd_It>
Decoder<In_It, Fwd_It>::~Decoder ()
{
work = RaptorQ__v1::Work_State::ABORT_COMPUTATION;
_cond.notify_all();
// wait threads to exit
do {
std::unique_lock<std::mutex> lock (_mtx);
if (waiting.size() == 0)
break;
_cond.wait (lock);
lock.unlock();
} while (waiting.size() != 0);
}
Decoder<In_It, Fwd_It>::Decoder (const Block_Size symbols,
:_symbols (static_cast<uint16_t> (symbols)), _symbol_size (symbol_size),
_type (type), dec (symbols, symbol_size)
{
IS_INPUT(In_It, "RaptorQ__v1::Decoder");
IS_FORWARD(Fwd_It, "RaptorQ__v1::Decoder");
last_reported.store (0);
symbols_tracker = std::deque<std::atomic<bool>> (2 * _symbols);
for(uint32_t idx = 0; idx < 2 * _symbols; ++idx)
symbols_tracker[idx] = false;
}
template <typename In_It, typename Fwd_It>
uint16_t Decoder<In_It, Fwd_It>::symbols() const
{
return _symbols;
}
template <typename In_It, typename Fwd_It>
RaptorQ__v1::It::Decoder::Symbol_Iterator<In_It, Fwd_It>
return RaptorQ__v1::It::Decoder::Symbol_Iterator<In_It, Fwd_It> (this, 0);
RaptorQ__v1::It::Decoder::Symbol_Iterator<In_It, Fwd_It>
return RaptorQ__v1::It::Decoder::Symbol_Iterator<In_It, Fwd_It> (nullptr,
template <typename In_It, typename Fwd_It>
uint16_t Decoder<In_It, Fwd_It>::needed_symbols() const
{
return dec.needed_symbols();
}
Error Decoder<In_It, Fwd_It>::add_symbol (In_It &from, const In_It to,
auto ret = dec.add_symbol (from, to, esi);
if (ret == Error::NONE && esi < _symbols) {
_cond.notify_all();
}
return ret;
}
template <typename In_It, typename Fwd_It>
switch (_type) {
case Report::PARTIAL_FROM_BEGINNING:
// report the number of symbols that are known, starting from
// the beginning.
last = last_reported.load();
idx = last;
for (; idx < symbols_tracker.size(); idx += 2) {
if (symbols_tracker[idx].load() == true) {
++idx;
if (symbols_tracker[idx].load() == false)
symbols_tracker[idx].store (true);
--idx;
idx /= 2;
if (idx > last) {
while (!last_reported.compare_exchange_weak (last, idx)) {
// expected is now "last_reported.load()"
if (last >= idx) {
// other thread already reported more than us.
// do not report things twice.
if (dec.ready()) {
last_reported.store (_symbols);
return {Error::NONE, _symbols};
}
return {Error::NEED_DATA, 0};
}
// else we can report the new stuff
}
return {Error::NONE, idx};
}
// nothing to report
if (dec.ready()) {
last_reported.store (_symbols);
return {Error::NONE, _symbols};
// report the first available, not yet reported.
// or return {NONE, _symbols} if all have been reported
if (dec.ready())
return {Error::NONE, _symbols};
for (idx = 0; idx < static_cast<uint32_t> (symbols_tracker.size());
idx += 2) {
if (symbols_tracker[idx].load() == true) {
++idx;
if (symbols_tracker[idx].load() == false) {
expected = false;
if (symbols_tracker[idx].
compare_exchange_strong (expected, true)) {
return {Error::NONE, idx / 2};
} // else some other thread raced us, keep trying other
// symbols
for (; idx < symbols_tracker.size(); idx += 2) {
if (symbols_tracker[idx].load() == false) {
idx /= 2;
while (!last_reported.compare_exchange_weak(init, idx))
idx = std::max(init, idx);
return {Error::WORKING, 0};
return {Error::NEED_DATA, 0};
}
}
return {Error::NONE, 0};
}
return {Error::WORKING, 0};
}
template <typename In_It, typename Fwd_It>
void Decoder<In_It, Fwd_It>::waiting_thread (Decoder<In_It, Fwd_It> *obj,
std::promise<std::pair<Error, uint16_t>> p)
{
while (obj->work == RaptorQ__v1::Work_State::KEEP_WORKING) {
bool compute = obj->dec.add_concurrent (obj->_max_threads);
if (compute) {
obj->decode_once();
std::unique_lock<std::mutex> lock (obj->_mtx);
obj->dec.drop_concurrent();
lock.unlock();
obj->_cond.notify_all(); // notify other waiting threads
}
std::unique_lock<std::mutex> lock (obj->_mtx);
// poll() does not actually need to be locked, but we use the
// lock-wait mechanism to signal the arrival of new symbols,
if (res.first == Error::NONE || (obj->dec.end_of_input == true &&
!obj->dec.can_decode() &&
obj->dec.threads() == 0 &&
res.first == Error::NEED_DATA)){
p.set_value (res);
break;
}
obj->_cond.wait (lock);
lock.unlock();
}
if (obj->work != RaptorQ__v1::Work_State::KEEP_WORKING)
p.set_value ({Error::EXITING, 0});
std::unique_lock<std::mutex> lock (obj->_mtx);
RQ_UNUSED (lock);
for (auto th = obj->waiting.begin(); th != obj->waiting.end(); ++th) {
if (std::this_thread::get_id() == th->get_id()) {
th->detach();
obj->waiting.erase (th);
break;
}
}
obj->_cond.notify_all(); // notify exit to destructor
}
template <typename In_It, typename Fwd_It>
std::pair<Error, uint16_t> Decoder<In_It, Fwd_It>::wait_sync ()
{
std::promise<std::pair<Error, uint16_t>> p;
auto fut = p.get_future();
waiting_thread (this, std::move(p));
fut.wait();
return fut.get();
}
template <typename In_It, typename Fwd_It>
std::future<std::pair<Error, uint16_t>> Decoder<In_It, Fwd_It>::wait ()
{
std::promise<std::pair<Error, uint16_t>> p;
auto f = p.get_future();
waiting.emplace_back (waiting_thread, this, std::move(p));
return f;
template <typename In_It, typename Fwd_It>
void Decoder<In_It, Fwd_It>::end_of_input()
{ dec.end_of_input = true; }
template <typename In_It, typename Fwd_It>
bool Decoder<In_It, Fwd_It>::can_decode() const
void Decoder<In_It, Fwd_It>::set_max_concurrency (const uint16_t max_threads)
{ _max_threads = max_threads; }
template <typename In_It, typename Fwd_It>
typename Decoder<In_It, Fwd_It>::Decoder_Result
Decoder<In_It, Fwd_It>::decode_once()
{
auto res = dec.decode (&work);
if (res == Decoder_Result::DECODED) {
std::unique_lock<std::mutex> lock (_mtx);
RQ_UNUSED (lock);
if (_type != Report::COMPLETE) {
uint32_t id = last_reported.load();
for (; id < symbols_tracker.size(); id += 2)
symbols_tracker[id].store (true);
}
last_reported.store(_symbols);
lock.unlock();
}
return res;
}
template <typename In_It, typename Fwd_It>
void Decoder<In_It, Fwd_It>::stop()
{
work = RaptorQ__v1::Work_State::ABORT_COMPUTATION;
_cond.notify_all();
}
template <typename In_It, typename Fwd_It>
std::pair<size_t, size_t> Decoder<In_It, Fwd_It>::decode_bytes (Fwd_It &start,
auto decoded = dec.get_symbols();
uint16_t esi = static_cast<uint16_t> (from_byte /
T element = static_cast<T> (0);
if (skip != 0) {
uint8_t *p = reinterpret_cast<uint8_t *> (&*start);
for (size_t keep = 0; keep < skip; ++keep) {
element += static_cast<T> (*(p++)) << keep * 8;
}
}
while (start != end && esi < _symbols && dec.has_symbol (esi)) {
element += static_cast<T> (static_cast<uint8_t> ((*decoded)(esi, byte)))
<< offset_al * 8;
++offset_al;
if (offset_al == sizeof(T)) {
*start = element;
written += offset_al;
offset_al = 0;
element = static_cast<T> (0);
}
++byte;
if (byte == decoded->cols()) {
byte = 0;
++esi;
}
}
if (start != end && offset_al != 0) {
// we have more stuff in "element", but not enough to fill
// the iterator. Do not overwrite additional data of the iterator.
uint8_t *out = reinterpret_cast<uint8_t *> (&*start);
uint8_t *in = reinterpret_cast<uint8_t *> (&element);
for (size_t idx = 0; idx < offset_al; ++idx, ++out, ++in)
*out = *in;
Error Decoder<In_It, Fwd_It>::decode_symbol (Fwd_It &start, const Fwd_It end,
auto pair = decode_bytes (start_copy, end, esi_byte, 0);
if (pair.first == _symbol_size) {
start = start_copy;
return Error::NONE;
}
return Error::NEED_DATA;