Optimize small files

Ramon Nourequested to merge
rnou/newdevelop into main

Solves #45 (closed)

Merge request reports

lib/libcargo.cpp

+3 −3
Changes for lib/libcargo.cpp: 3 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -22,9 +22,9 @@ 
 * SPDX-License-Identifier: GPL-3.0-or-later

 *****************************************************************************/



#include <cargo.hpp>

#include <fmt_formatters.hpp>

#include <net/serialization.hpp>

#include "cargo.hpp"

#include "fmt_formatters.hpp"

#include "net/serialization.hpp"

#include <iomanip>

#include <logger/logger.hpp>

#include <net/client.hpp>

src/proto/mpi/message.hpp

+11 −2
Changes for src/proto/mpi/message.hpp: 11 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -28,6 +28,7 @@ 
#include <fmt/format.h>

#include <fmt/ranges.h>

#include <filesystem>

#include <cstddef>

#include <boost/archive/binary_oarchive.hpp>

#include <utility>

#include <optional>
@@ -57,10 +58,11 @@ public: 


    transfer_message(std::uint64_t tid, std::uint32_t seqno,

                     std::vector<std::string> input_path, std::uint32_t i_type,

                     std::vector<std::string> output_path, std::uint32_t o_type)

                     std::vector<std::string> output_path, std::uint32_t o_type,

                     std::vector<std::size_t> sizes)

        : m_tid(tid), m_seqno(seqno), m_input_path(std::move(input_path)),

          m_i_type(i_type), m_output_path(std::move(output_path)),

          m_o_type(o_type) {}

          m_o_type(o_type), m_sizes(std::move(sizes)) {}



    [[nodiscard]] std::uint64_t

    tid() const {
@@ -93,6 +95,11 @@ public: 
        return static_cast<cargo::FSPlugin::type>(m_i_type);

    }



    [[nodiscard]] const std::vector<std::size_t> &

    sizes() const {

        return m_sizes;

    }

    

private:

    template <class Archive>

    void
@@ -105,6 +112,7 @@ private: 
        ar& m_output_path;

        ar& m_i_type;

        ar& m_o_type;

        ar& m_sizes;

    }



    std::uint64_t m_tid{};
@@ -113,6 +121,7 @@ private: 
    std::uint32_t m_i_type{};

    std::vector<std::string> m_output_path;

    std::uint32_t m_o_type{};

    std::vector<std::size_t> m_sizes;

};



class status_message {

src/worker/mpio_read.cpp

+10 −8
Changes for src/worker/mpio_read.cpp: 10 added lines, 8 removed lines.
Original line number Diff line number Diff line
@@ -34,10 +34,11 @@ mpio_read::mpio_read(mpi::communicator workers, 
                     std::filesystem::path input_path,

                     std::filesystem::path output_path,

                     std::uint64_t block_size, FSPlugin::type fs_i_type,

                     FSPlugin::type fs_o_type)

                     FSPlugin::type fs_o_type, std::size_t size, bool single)

    : m_workers(std::move(workers)), m_input_path(std::move(input_path)),

      m_output_path(std::move(output_path)), m_kb_size(std::move(block_size)),

      m_fs_i_type(fs_i_type), m_fs_o_type(fs_o_type) {}

      m_fs_i_type(fs_i_type), m_fs_o_type(fs_o_type), m_file_size(size),

      m_single(single) {}



cargo::error_code

mpio_read::operator()() {
@@ -51,9 +52,8 @@ mpio_read::operator()() { 
        const auto input_file = mpioxx::file::open(

                m_workers, m_input_path, mpioxx::file_open_mode::rdonly);



        mpioxx::offset file_size = input_file.size();

        mpioxx::offset file_size = m_file_size;

        std::size_t block_size = m_kb_size * 1024u;



        // create block type

        MPI_Datatype block_type;

        MPI_Type_contiguous(static_cast<int>(block_size), MPI_BYTE,
@@ -67,9 +67,13 @@ mpio_read::operator()() { 
            ++total_blocks;

        }



        const auto workers_size = m_workers.size();

        const auto workers_rank = m_workers.rank();

        auto workers_size = m_workers.size();

        auto workers_rank = m_workers.rank();



        if(m_single) {

            workers_size = 1;

            workers_rank = 0;

        }

        // create file type

        MPI_Datatype file_type;

        /*
@@ -138,7 +142,6 @@ mpio_read::operator()() { 
        m_workers_rank = workers_rank;

        m_block_size = block_size;





    } catch(const mpioxx::io_error& e) {

        LOGGER_ERROR("{}() failed: {}", e.where(), e.what());

        m_status = make_mpi_error(e.error_code());
@@ -169,7 +172,6 @@ mpio_read::progress(int ongoing_index) { 
    try {

        int index = 0;

        // TODO : FS not defined...



        m_status = error_code::transfer_in_progress;

        for(const auto& file_range :

            all_of(posix_file::file{m_input_path, m_fs_i_type}) |

src/worker/mpio_read.hpp

+4 −1
Changes for src/worker/mpio_read.hpp: 4 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -39,7 +39,7 @@ class mpio_read : public operation { 
public:

    mpio_read(mpi::communicator workers, std::filesystem::path input_path,

              std::filesystem::path output_path, std::uint64_t block_size,

              FSPlugin::type fs_i_type, FSPlugin::type m_fs_o_type);

              FSPlugin::type fs_i_type, FSPlugin::type m_fs_o_type, std::size_t size, bool single);



    cargo::error_code

    operator()() final;
@@ -65,6 +65,7 @@ private: 
    cargo::error_code m_status;

    std::filesystem::path m_input_path{};

    std::filesystem::path m_output_path{};

    

    std::unique_ptr<posix_file::file> m_output_file;

    int m_workers_size;

    int m_workers_rank;
@@ -74,6 +75,8 @@ private: 
    std::uint64_t m_kb_size;

    FSPlugin::type m_fs_i_type;

    FSPlugin::type m_fs_o_type;

    std::size_t m_file_size;

    bool m_single;

};



} // namespace cargo

src/worker/mpio_write.cpp

+9 −3
Changes for src/worker/mpio_write.cpp: 9 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -38,14 +38,20 @@ mpio_write::operator()() { 
    m_status = error_code::transfer_in_progress;

    try {



        const auto workers_size = m_workers.size();

        const auto workers_rank = m_workers.rank();

        auto workers_size = m_workers.size();

        auto workers_rank = m_workers.rank();



        if (m_single) {

            workers_size = 1;

            workers_rank = 0;

        }

        

        std::size_t block_size = m_kb_size * 1024u;

        // We need to open the file and ask size (using fs_plugin)

        m_input_file = std::make_unique<posix_file::file>(

                posix_file::open(m_input_path, O_RDONLY, 0, m_fs_i_type));



        std::size_t file_size = m_input_file->size();

        std::size_t file_size = m_file_size;



        // compute the number of blocks in the file

        int total_blocks = static_cast<int>(file_size / block_size);