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);
enum class Enc_State : uint8_t {
INIT_ERROR = 1,
NEED_DATA = 2,
FULL = 3
};
Raw_Encoder<Rnd_It, Fwd_It, without_interleaver> encoder;
DenseMtx precomputed;
std::thread waiting;
};
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>
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,
: _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");
// check for proper initialization
uint16_t idx;
for (idx = 0; idx < (*blocks).size(); ++idx) {
if ((*blocks)[idx] == symbols)
break;
}
// check that the user did not try some cast trickery,
// and maximum size is ssize_t::max. But ssize_t is not standard,
// so we search the maximum ourselves.
if (idx == (*blocks).size() || symbol_size >= std::pow (2,
(sizeof(size_t) == 4 ? 31 : 63))) {
_state = Enc_State::INIT_ERROR;
}
_state = Enc_State::NEED_DATA;
template <typename Rnd_It, typename Fwd_It>
Encoder<Rnd_It, Fwd_It>::operator bool() const
{ return _state != Enc_State::INIT_ERROR; }
template <typename Rnd_It, typename Fwd_It>
uint16_t Encoder<Rnd_It, Fwd_It>::symbols() const
{
if (_state == Enc_State::INIT_ERROR)
return 0;
if (_state == Enc_State::INIT_ERROR)
return 0;
}
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.
if (_state == Enc_State::INIT_ERROR)
return 0;
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
{
if (_state == Enc_State::INIT_ERROR)
return false;
return _state == Enc_State::FULL;
}
template <typename Rnd_It, typename Fwd_It>
size_t Encoder<Rnd_It, Fwd_It>::set_data (const Rnd_It &from, const Rnd_It &to)
{
if (_state == Enc_State::INIT_ERROR)
return 0;
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()
{
if (_state == Enc_State::INIT_ERROR)
return;
_state = Enc_State::NEED_DATA;
template <typename Rnd_It, typename Fwd_It>
void Encoder<Rnd_It, Fwd_It>::stop()
{ encoder.stop(); }
template <typename Rnd_It, typename Fwd_It>
bool Encoder<Rnd_It, Fwd_It>::precompute_sync()
{
if (_state == Enc_State::INIT_ERROR)
return false;
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.
}
template <typename Rnd_It, typename Fwd_It>
bool Encoder<Rnd_It, Fwd_It>::compute_sync()
{
if (_state == Enc_State::INIT_ERROR)
return false;
static RaptorQ__v1::Work_State work = RaptorQ__v1::Work_State::KEEP_WORKING;
if (encoder.ready())
return true;
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;
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->_state == Enc_State::FULL && !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->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 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()
{
if (_state == Enc_State::INIT_ERROR) {
p.set_value (Error::INITIALIZATION);
return 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()
{
if (_state == Enc_State::INIT_ERROR) {
p.set_value (Error::INITIALIZATION);
return 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));
return future;
size_t Encoder<Rnd_It, Fwd_It>::encode (Fwd_It &output, const Fwd_It end,
if (_state == Enc_State::INIT_ERROR)
return 0;
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");
// check for proper initialization
uint16_t idx;
for (idx = 0; idx < (*blocks).size(); ++idx) {
if ((*blocks)[idx] == symbols)
break;
}
// check that the user did not try some cast trickery,
// and maximum size is ssize_t::max. But ssize_t is not standard,
// so we search the maximum ourselves.
if (idx == (*blocks).size() || symbol_size >= std::pow (2,
(sizeof(size_t) == 4 ? 31 : 63))) {
return;
}
if (type != Report::PARTIAL_FROM_BEGINNING &&
type != Report::PARTIAL_ANY &&
type != Report::COMPLETE) {
return; // no cast trickey plz
}
last_reported.store (0);
symbols_tracker = std::deque<std::atomic<bool>> (2 * _symbols);
symbols_tracker[idx] = false;
work = RaptorQ__v1::Work_State::KEEP_WORKING;
template <typename Rnd_It, typename Fwd_It>
Decoder<Rnd_It, Fwd_It>::operator bool() const
{ return symbols_tracker.size() > 0; }
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,
if (symbols_tracker.size() == 0)
return Error::INITIALIZATION;
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>
if (symbols_tracker.size() == 0)
return {Error::INITIALIZATION, 0};
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
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 ()
if (symbols_tracker.size() == 0)
return {Error::INITIALIZATION, 0};
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 ()
{
if (symbols_tracker.size() == 0) {
p.set_value ({Error::INITIALIZATION, 0});
return p.get_future();
}
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()
{
if (symbols_tracker.size() != 0)
dec.end_of_input = true;
}
template <typename In_It, typename Fwd_It>
bool Decoder<In_It, Fwd_It>::can_decode() const
{
if (symbols_tracker.size() == 0)
return false;
return dec.can_decode();
}
void Decoder<In_It, Fwd_It>::set_max_concurrency (const uint16_t max_threads)
{
if (symbols_tracker.size() != 0)
_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()
{
if (symbols_tracker.size() == 0)
return Decoder<In_It, Fwd_It>::Decoder_Result::STOPPED;
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,
const Fwd_It end,
const size_t from_byte,
const size_t skip)
{
using T = typename std::iterator_traits<Fwd_It>::value_type;
if (symbols_tracker.size() == 0 || skip >= sizeof(T) || from_byte >=
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
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,
if (symbols_tracker.size() == 0)
return Error::INITIALIZATION;
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;