parallel_block_compressor.hpp

Go to the documentation of this file.

/*********************************************************************/
// dar - disk archive - a backup/restoration program
// Copyright (C) 2002-2025 Denis Corbin
//
// This program is free software; you can redistribute it and/or
// modify it under the terms of the GNU General Public License
// as published by the Free Software Foundation; either version 2
// of the License, or (at your option) any later version.
//
// This program 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
// along with this program; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
//
// to contact the author, see the AUTHOR file
/*********************************************************************/
 
 
 
#ifndef PARALLEL_BLOCK_COMPRESSOR_HPP
#define PARALLEL_BLOCK_COMPRESSOR_HPP
 
#include "../my_config.h"
 
#include "infinint.hpp"
#include "crypto_segment.hpp"
#include "heap.hpp"
#include "compress_module.hpp"
#include "proto_compressor.hpp"
 
#include <libthreadar/libthreadar.hpp>
 
namespace libdar
{
 
 
 
 
    enum class compressor_block_flags { data = 0, eof_die = 1, error = 2, worker_error = 3 };
 
    // the following classes hold the subthreads of class parallel_block_compressor
    // and are defined just after it below
 
    class zip_below_read;
    class zip_below_write;
    class zip_worker;
 
 
    //
    // paralle_compressor class, which holds the sub-threads
    //
    //
 
    class parallel_block_compressor: public proto_compressor
    {
    public:
 
    parallel_block_compressor(U_I num_workers,
                  std::unique_ptr<compress_module> block_zipper,
                  generic_file & compressed_side,
                  U_I uncompressed_bs = default_uncompressed_block_size);
        // compressed_side is not owned by the object and will remains
            // after the objet destruction
 
    parallel_block_compressor(const parallel_block_compressor & ref) = delete;
    parallel_block_compressor(parallel_block_compressor && ref) noexcept = delete;
    parallel_block_compressor & operator = (const parallel_block_compressor & ref) = delete;
    parallel_block_compressor & operator = (parallel_block_compressor && ref) noexcept = delete;
    ~parallel_block_compressor();
 
        // inherited from proto_compressor
 
    virtual compression get_algo() const override { return suspended? compression::none : zipper->get_algo(); };
    virtual void suspend_compression() override;
    virtual void resume_compression() override;
    virtual bool is_compression_suspended() const override { return suspended; };
 
        // inherited from generic file
 
    virtual bool skippable(skippability direction, const infinint & amount) override;
        virtual bool skip(const infinint & pos) override;
        virtual bool skip_to_eof() override;
        virtual bool skip_relative(S_I x) override;
    virtual bool truncatable(const infinint & pos) const override;
        virtual infinint get_position() const override;
 
    protected :
    virtual void inherited_read_ahead(const infinint & amount) override;
        virtual U_I inherited_read(char *a, U_I size) override;
        virtual void inherited_write(const char *a, U_I size) override;
    virtual void inherited_truncate(const infinint & pos) override;
    virtual void inherited_sync_write() override;
    virtual void inherited_flush_read() override { stop_read_threads(); reof = false; };
    virtual void inherited_terminate() override;
 
    private:
 
        // the local fields
 
    U_I num_w;                                             
    std::unique_ptr<compress_module> zipper;               
    generic_file *compressed;                              
    U_I uncompressed_block_size;                           
    bool suspended;                                        
    bool running_threads;                                  
    std::unique_ptr<crypto_segment> curwrite;              
    std::deque<std::unique_ptr<crypto_segment> > lus_data; 
    std::deque<signed int> lus_flags;                      
    bool reof;                                             
 
 
        // inter-thread data structure
 
    std::shared_ptr<libthreadar::ratelier_scatter<crypto_segment> > disperse;
    std::shared_ptr<libthreadar::ratelier_gather<crypto_segment> > rassemble;
    std::shared_ptr<heap<crypto_segment> > tas;
 
 
        // the subthreads
 
    std::unique_ptr<zip_below_read> reader;
    std::unique_ptr<zip_below_write> writer;
    std::deque<std::unique_ptr<zip_worker> > travailleurs;
 
 
        // private methods
 
    void send_flag_to_workers(compressor_block_flags flag);
    void stop_threads();
    void stop_read_threads();
    void stop_write_threads();
    void run_threads();
    void run_read_threads();
    void run_write_threads();
    compressor_block_flags purge_ratelier_up_to_non_data();
 
 
        // static methods
 
    static U_I get_ratelier_size(U_I num_workers) { return num_workers + num_workers/2; };
    static U_I get_heap_size(U_I num_workers);
    };
 
 
 
    //
    // zip_below_write class/sub-thread
    //
    //
 
    class zip_below_write: public libthreadar::thread
    {
    public:
    zip_below_write(const std::shared_ptr<libthreadar::ratelier_gather<crypto_segment> > & source,
            generic_file *dest,
            const std::shared_ptr<heap<crypto_segment> > & xtas,
            U_I num_workers);
 
    ~zip_below_write() { cancel(); join(); };
 
 
    bool exception_pending() const { return error; };
 
    void reset();
 
    protected:
    virtual void inherited_run() override;
 
    private:
    std::shared_ptr<libthreadar::ratelier_gather<crypto_segment> > src;
    generic_file *dst;
    std::shared_ptr<heap<crypto_segment> > tas;
    U_I num_w;
    bool error;
    U_I ending;
    std::deque<std::unique_ptr<crypto_segment> > data;
    std::deque<signed int> flags;
    libthreadar::mutex get_pos; 
    infinint current_position;
 
    void work();
    void pop_front() { tas->put(std::move(data.front())); data.pop_front(); flags.pop_front(); };
    };
 
 
 
 
    //
    // zip_below_read class/sub-thread
    //
    //
 
 
 
    class zip_below_read: public libthreadar::thread
    {
    public:
    zip_below_read(generic_file *source,
               const std::shared_ptr<libthreadar::ratelier_scatter<crypto_segment> > & dest,
               const std::shared_ptr<heap<crypto_segment> > & xtas,
               U_I num_workers);
 
    ~zip_below_read() { cancel(); join(); };
 
    void do_stop() { should_i_stop = true; };
 
    void reset();
 
    protected:
    virtual void inherited_run() override;
 
    private:
    generic_file *src;
    const std::shared_ptr<libthreadar::ratelier_scatter<crypto_segment> > & dst;
    const std::shared_ptr<heap<crypto_segment> > & tas;
    U_I num_w;
    std::unique_ptr<crypto_segment> ptr;
    bool should_i_stop;
 
 
    void work();
    void push_flag_to_all_workers(compressor_block_flags flag);
    };
 
 
 
 
    //
    // zip_worker class/sub-thread
    //
    //
 
 
 
    class zip_worker: public libthreadar::thread
    {
    public:
    zip_worker(std::shared_ptr<libthreadar::ratelier_scatter <crypto_segment> > & read_side,
           std::shared_ptr<libthreadar::ratelier_gather <crypto_segment> > & write_size,
           std::unique_ptr<compress_module> && ptr,
           bool compress);
 
    ~zip_worker() { cancel(); join(); };
 
    protected:
    virtual void inherited_run() override;
 
    private:
    std::shared_ptr<libthreadar::ratelier_scatter <crypto_segment> > & reader;
    std::shared_ptr<libthreadar::ratelier_gather <crypto_segment> > & writer;
    std::unique_ptr<compress_module> compr;
    bool do_compress;
    bool error;
    std::unique_ptr<crypto_segment> transit;
    unsigned int transit_slot;
 
    void work();
    };
 
 
 
} // end of namespace
 
 
#endif