RFC.hpp

/*
 * 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

/////////////////////
//
//  These templates are just a wrapper around the
//  functionalities offered by the RaptorQ__v1::Impl namespace
//  So if you want to see what the algorithm looks like,
//  you are in the wrong place
//
/////////////////////

#include "RaptorQ/v1/block_sizes.hpp"
#include "RaptorQ/v1/Interleaver.hpp"
#include "RaptorQ/v1/De_Interleaver.hpp"
#include "RaptorQ/v1/Decoder.hpp"
#include "RaptorQ/v1/Encoder.hpp"
#include "RaptorQ/v1/API_Iterators.hpp"
#include "RaptorQ/v1/Shared_Computation/Decaying_LF.hpp"
#include "RaptorQ/v1/Thread_Pool.hpp"
#include <algorithm>
#include <cassert>
#include <future>
#include <map>
#include <memory>
#include <mutex>
#include <limits>
#include <thread>
#include <tuple>
#include <type_traits>
#include <utility>

namespace RFC6330__v1 {

constexpr uint64_t max_data = RFC6330_max_data;  // ~881 GB

namespace Impl {
template <typename Rnd_It, typename Fwd_It = Rnd_It>
class RAPTORQ_LOCAL Encoder;
template <typename In_It, typename Fwd_It = In_It>
class RAPTORQ_LOCAL Decoder;
} // namespace Impl
#ifdef RQ_HEADER_ONLY
    namespace Impl {
    template <typename Rnd_It, typename Fwd_It = Rnd_It>
    using Encoder = RAPTORQ_API Impl::Encoder<Rnd_It, Fwd_It>
    template <typename In_It, typename Fwd_It>
    using Decoder = RAPTORQ_API Impl::Decoder<In_It, Fwd_It>
#endif

namespace Impl {


template <typename Rnd_It, typename Fwd_It>
class RAPTORQ_LOCAL Encoder
{
public:

    ~Encoder();
    Encoder (const Rnd_It data_from, const Rnd_It data_to,
                                            const uint16_t min_subsymbol_size,
                                            const uint16_t symbol_size,
                                            const size_t max_memory)
        : _mem (max_memory), _data_from (data_from), _data_to (data_to),
                                            _symbol_size (symbol_size),
                                            _min_subsymbol (min_subsymbol_size),
                                            interleave (_data_from,
                                                        _data_to,
                                                        _min_subsymbol,
                                                        _mem,
                                                        _symbol_size)
    {
        IS_RANDOM(Rnd_It, "RFC6330__v1::Encoder");
        IS_FORWARD(Fwd_It, "RFC6330__v1::Encoder");
        auto _alignment = sizeof(typename
                                    std::iterator_traits<Rnd_It>::value_type);
        RQ_UNUSED(_alignment);  // used only for asserts
        assert(_symbol_size >= _alignment &&
                        "RaptorQ: symbol_size must be >= alignment");
        assert((_symbol_size % _alignment) == 0 &&
                        "RaptorQ: symbol_size must be multiple of alignment");
        assert(min_subsymbol_size >= _alignment &&
                        "RaptorQ: minimum subsymbol must be at least aligment");
        assert(min_subsymbol_size <= _symbol_size &&
                    "RaptorQ: minimum subsymbol must be at most symbol_size");
        assert((min_subsymbol_size % _alignment) == 0 &&
                    "RaptorQ: minimum subsymbol must be multiple of alignment");
        assert((_symbol_size % min_subsymbol_size == 0) &&
                    "RaptorQ: symbol size must be multiple of subsymbol size");
        // max size: ~881 GB
        if (static_cast<uint64_t> (data_to - data_from) *
                    sizeof(typename std::iterator_traits<Rnd_It>::value_type)
                                                                > max_data) {
            return;
        }

        _pool_notify = std::make_shared<std::condition_variable>();
        _pool_mtx = std::make_shared<std::mutex>();
        pool_last_reported = -1;
        use_pool = true;
        exiting = false;
    }

    Block_Iterator<Rnd_It, Fwd_It> begin ()
    {
        return Block_Iterator<Rnd_It, Fwd_It> (this,
                                                interleave.get_partition(), 0);
    }
    const Block_Iterator<Rnd_It, Fwd_It> end ()
    {
        auto part = interleave.get_partition();
        return Block_Iterator<Rnd_It, Fwd_It> (this, part,
                            static_cast<uint8_t> (part.num(0) + part.num(1)));
    }

    operator bool() const { return interleave; }
    RFC6330_OTI_Common_Data OTI_Common() const;
    RFC6330_OTI_Scheme_Specific_Data OTI_Scheme_Specific() const;

    // TODO: introduce memory limits on threading ?
    std::future<std::pair<Error, uint8_t>> compute (const Compute flags);

    size_t precompute_max_memory ();
    size_t encode (Fwd_It &output, const Fwd_It end, const uint32_t esi,
                                                            const uint8_t sbn);
    // id: 8-bit sbn + 24 bit esi
    size_t encode (Fwd_It &output, const Fwd_It end, const uint32_t &id);
    void free (const uint8_t sbn);
    uint8_t blocks() const;
    uint32_t block_size (const uint8_t sbn) const; // TODO: eliminate
    uint16_t symbol_size() const;
    uint16_t symbols (const uint8_t sbn) const;
    uint32_t max_repair (const uint8_t sbn) const;
private:

    static void wait_threads (Encoder<Rnd_It, Fwd_It> *obj, const Compute flags,
                                    std::promise<std::pair<Error, uint8_t>> p);

    class Block_Work final : public Impl::Pool_Work {
    public:
        std::weak_ptr<RaptorQ__v1::Impl::Raw_Encoder<Rnd_It, Fwd_It,
                                  RaptorQ__v1::Impl::with_interleaver>> work;
        std::weak_ptr<std::condition_variable> notify;
        std::weak_ptr<std::mutex> lock;

        Work_Exit_Status do_work (RaptorQ__v1::Work_State *state) override;
        ~Block_Work() override;
    };

    // TODO: tagged pointer
    class Enc {
    public:
        Enc (Impl::Interleaver<Rnd_It> *interleaver, const uint8_t sbn)
        {
            enc = std::make_shared<RaptorQ__v1::Impl::Raw_Encoder<Rnd_It,
                                Fwd_It, RaptorQ__v1::Impl::with_interleaver>> (
                                                            interleaver, sbn);
            reported = false;
        }
        std::shared_ptr<RaptorQ__v1::Impl::Raw_Encoder<Rnd_It, Fwd_It,
                                    RaptorQ__v1::Impl::with_interleaver>> enc;
        bool reported;
    };

    std::pair<Error, uint8_t> get_report (const Compute flags);
    std::shared_ptr<std::condition_variable> _pool_notify;
    std::shared_ptr<std::mutex> _pool_mtx;
    std::deque<std::thread> pool_wait;

    std::map<uint8_t, Enc> encoders;
    std::mutex _mtx;

    const size_t _mem;
    const Rnd_It _data_from, _data_to;
    const uint16_t _symbol_size;
    const uint16_t _min_subsymbol;
    Impl::Interleaver<Rnd_It> interleave;
    bool use_pool, exiting;
    int16_t pool_last_reported;

};