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 (const Block_Size symbols, const size_t symbol_size);
uint16_t symbols() const;
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
size_t set_data (const Rnd_It &from, const Rnd_It &to);
void clear_data();
std::future<Error> precompute();
size_t encode (Fwd_It &output, const Fwd_It end, const uint32_t id);
Raw_Encoder<Rnd_It, Fwd_It, without_interleaver> encoder;
DenseMtx precomputed;
std::thread waiting;
static uint16_t calc_symbols (const Rnd_It data_from, const Rnd_It data_to,
const size_t symbol_size);
static void compute_thread (Encoder<Rnd_It, Fwd_It> *obj,
};
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);
bool can_decode() const;
void stop();
uint16_t needed_symbols() const;
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> poll();
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);
// return number of bytes written
std::pair<size_t, size_t> decode_bytes (Fwd_It &start, const Fwd_It end,
const size_t from_byte, const size_t skip);
std::atomic<uint32_t> last_reported;
const Report _type;
RaptorQ__v1::Work_State work;
Raw_Decoder<In_It> dec;
// 2* symbols. Actually tracks available and reported symbols.
// each symbol gets 2 bool: 1= available, 2=reported
std::deque<std::atomic<bool>> symbols_tracker;
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,
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
{
}
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);
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);
} else {
if (precomputed.rows() != 0)
return true;
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
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()
{
// 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()
{
// 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));
return future;
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,
const size_t symbol_size, const Report type)
:_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;
work = RaptorQ__v1::Work_State::KEEP_WORKING;
}
template <typename In_It, typename Fwd_It>
uint16_t Decoder<In_It, Fwd_It>::symbols() const
{
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,
_symbols);
template <typename In_It, typename Fwd_It>
uint16_t Decoder<In_It, Fwd_It>::needed_symbols() const
{
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) {
symbols_tracker [2 * esi].store (true);
_cond.notify_all();
}
return ret;
}
template <typename In_It, typename Fwd_It>
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
uint32_t idx;
uint32_t last;
bool expected = false;
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;
} else {
break;
}
}
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::WORKING, 0};
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};
}
return {Error::WORKING, 0};
return {Error::NEED_DATA, 0};
case Report::PARTIAL_ANY:
// 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
}
}
}
if (dec.ready())
return {Error::NONE, _symbols};
return {Error::WORKING, 0};
return {Error::NEED_DATA, 0};
case Report::COMPLETE:
idx = init * 2;
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);
if (dec.threads() > 0)
return {Error::WORKING, 0};
return {Error::NEED_DATA, 0};
}
}
last_reported.store (_symbols);
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,
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,
// so that we retry only when we get new data.
auto res = obj->poll();
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;
}
}
lock.unlock();
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,
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
const Fwd_It end,
const size_t from_byte,
const size_t skip)
{
using T = typename std::iterator_traits<Fwd_It>::value_type;
if (skip >= sizeof(T) || from_byte >=
static_cast<size_t> (_symbols * _symbol_size)) {
return {0, 0};
}
auto decoded = dec.get_symbols();
uint16_t esi = static_cast<uint16_t> (from_byte /
static_cast<size_t> (_symbol_size));
uint16_t byte = static_cast<uint16_t> (from_byte %
static_cast<size_t> (_symbol_size));
size_t offset_al = skip;
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;
}
}
size_t written = 0;
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;
++start;
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;
written += offset_al;
}
return {written, offset_al};
Error Decoder<In_It, Fwd_It>::decode_symbol (Fwd_It &start, const Fwd_It end,
auto start_copy = start;
size_t esi_byte = esi * _symbol_size;
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;