Resolve "[Alya] Create a new data distributor - Moves instantiation outside read/write"

Ramon Nourequested to merge
88-alya-create-a-new-data-distributor into master

Closes #88 (closed) Closes #125 (closed)

The distributor will read a shared file including [filename] [host] [size] [offset] and distribute the files accordingly. We assume, as it should happen in SLURM that the nodes are in alphabetical order.

Distributors are instantiated per read/write operation. They should be moved to FsData to avoid the overhead. It is also necessary to reduce the creation cost when the distributor reads the data mappings from a file or similar.

Edited by Ramon Nou

Merge request reports

examples/distributors/guided/generate.py

0 → 100644
+40 −0
Changes for examples/distributors/guided/generate.py: 40 added lines, 0 removed lines.
Original line number Diff line number Diff line 
###

#  Copyright 2018-2020, Barcelona Supercomputing Center (BSC), Spain

#  Copyright 2015-2020, Johannes Gutenberg Universitaet Mainz, Germany



#  This software was partially supported by the

#  EC H2020 funded project NEXTGenIO (Project ID: 671951, www.nextgenio.eu).



#  This software was partially supported by the

#  ADA-FS project under the SPPEXA project funded by the DFG.



#  SPDX-License-Identifier: MIT

###

#!/usr/bin/python3

import re

import sys

import collections



file = sys.argv[1]



pattern = re.compile(r".+(read\ )(.*)( host: )(\d+).+(path: )(.+),.+(chunk_start: )(\d+).+(chunk_end: )(\d+)")



d = collections.OrderedDict()



with open(file) as f:

    for line in f:

        result = pattern.match(line)

        if result:

            d[result.group(2)] = 1

keys = sorted(d.keys())

i = 0

for key in keys:

    d[key] = i

    i = i + 1



with open(file) as f:

    for line in f:

        result = pattern.match(line)

        if result:

            for i in range(int(result.group(8)), int(result.group(10))+1):

                print (result.group(6), i, d[result.group(2)])

include/client/logging.hpp

+13 −2
Changes for include/client/logging.hpp: 13 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -47,7 +47,7 @@ 


namespace gkfs::log {



enum class log_level : short {

enum class log_level : unsigned int {

    print_syscalls = 1 << 0,

    print_syscalls_entry = 1 << 1,

    print_info = 1 << 2,
@@ -57,6 +57,7 @@ enum class log_level : short { 
    print_hermes = 1 << 6,

    print_mercury = 1 << 7,

    print_debug = 1 << 8,

    print_trace_reads = 1 << 9,



    // for internal use

    print_none = 0,
@@ -117,6 +118,7 @@ static const auto constexpr warning = log_level::print_warnings; 
static const auto constexpr hermes = log_level::print_hermes;

static const auto constexpr mercury = log_level::print_mercury;

static const auto constexpr debug = log_level::print_debug;

static const auto constexpr trace_reads = log_level::print_trace_reads;

static const auto constexpr none = log_level::print_none;

static const auto constexpr most = log_level::print_most;

static const auto constexpr all = log_level::print_all;
@@ -125,7 +127,8 @@ static const auto constexpr help = log_level::print_help; 
static const auto constexpr level_names = utils::make_array(

        "syscall",

        "syscall", // sycall_entry uses the same name as syscall

        "info", "critical", "error", "warning", "hermes", "mercury", "debug");

        "info", "critical", "error", "warning", "hermes", "mercury", "debug",

        "trace_reads");



inline constexpr auto

lookup_level_name(log_level l) {
@@ -533,6 +536,14 @@ static_buffer::grow(std::size_t size) { 
        }                                                                      \

    } while(0);



#define LOG_TRACE_READS(...)                                                   \

    do {                                                                       \

        if(gkfs::log::get_global_logger()) {                                   \

            gkfs::log::get_global_logger()->log(                               \

                    gkfs::log::trace_reads, __func__, __LINE__, __VA_ARGS__);  \

        }                                                                      \

    } while(0);



#ifdef GKFS_DEBUG_BUILD



#define LOG_SYSCALL(...)                                                       \

include/client/preload_context.hpp

+5 −0
Changes for include/client/preload_context.hpp: 5 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -36,6 +36,7 @@ 
#include <memory>

#include <vector>

#include <string>

#include <config.hpp>



#include <bitset>
@@ -102,6 +103,7 @@ private: 
    mutable std::mutex internal_fds_mutex_;

    bool internal_fds_must_relocate_;

    std::bitset<MAX_USER_FDS> protected_fds_;

    std::string hostname;



public:

    static PreloadContext*
@@ -210,6 +212,9 @@ public: 


    void

    unprotect_user_fds();



    std::string

    get_hostname();

};



} // namespace preload

include/common/rpc/distributor.hpp

+57 −3
Changes for include/common/rpc/distributor.hpp: 57 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -29,11 +29,17 @@ 
#ifndef GEKKOFS_RPC_DISTRIBUTOR_HPP

#define GEKKOFS_RPC_DISTRIBUTOR_HPP



#include "../include/config.hpp"

#include <vector>

#include <string>

#include <numeric>

#include <unordered_map>

#include <fstream>

#include <boost/icl/interval_map.hpp>



namespace gkfs::rpc {

namespace gkfs {



namespace rpc {



using chunkid_t = unsigned int;

using host_t = unsigned int;
@@ -45,6 +51,11 @@ public: 


    virtual host_t

    locate_data(const std::string& path, const chunkid_t& chnk_id) const = 0;

    // TODO: We need to pass hosts_size in the server side, because the number

    // of servers are not defined (in startup)

    virtual host_t

    locate_data(const std::string& path, const chunkid_t& chnk_id,

                unsigned int hosts_size) = 0;



    virtual host_t

    locate_file_metadata(const std::string& path) const = 0;
@@ -57,11 +68,13 @@ public: 
class SimpleHashDistributor : public Distributor {

private:

    host_t localhost_;

    unsigned int hosts_size_;

    unsigned int hosts_size_{0};

    std::vector<host_t> all_hosts_;

    std::hash<std::string> str_hash;



public:

    SimpleHashDistributor();



    SimpleHashDistributor(host_t localhost, unsigned int hosts_size);



    host_t
@@ -71,6 +84,10 @@ public: 
    locate_data(const std::string& path,

                const chunkid_t& chnk_id) const override;



    host_t

    locate_data(const std::string& path, const chunkid_t& chnk_id,

                unsigned int host_size);



    host_t

    locate_file_metadata(const std::string& path) const override;
@@ -123,6 +140,43 @@ public: 
    locate_directory_metadata(const std::string& path) const override;

};



} // namespace gkfs::rpc

class GuidedDistributor : public Distributor {

private:

    host_t localhost_;

    unsigned int hosts_size_{0};

    std::vector<host_t> all_hosts_;

    std::hash<std::string> str_hash;

    std::unordered_map<std::string,

                       boost::icl::interval_map<chunkid_t, unsigned int>>

            map_interval;

    std::vector<std::string> prefix_list; // Should not be very long

    bool

    init_guided();



public:

    GuidedDistributor();



    GuidedDistributor(host_t localhost, unsigned int hosts_size);



    host_t

    localhost() const override;



    host_t

    locate_data(const std::string& path,

                const chunkid_t& chnk_id) const override;



    host_t

    locate_data(const std::string& path, const chunkid_t& chnk_id,

                unsigned int host_size);



    host_t

    locate_file_metadata(const std::string& path) const override;



    std::vector<host_t>

    locate_directory_metadata(const std::string& path) const override;

};



} // namespace rpc

} // namespace gkfs



#endif // GEKKOFS_RPC_LOCATOR_HPP

include/common/cmake_configure.hpp.in

+3 −0
Changes for include/common/cmake_configure.hpp.in: 3 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -16,5 +16,8 @@ 


#cmakedefine01 CREATE_CHECK_PARENTS

#cmakedefine01 LOG_SYSCALLS

#cmakedefine GKFS_USE_GUIDED_DISTRIBUTION

#define GKFS_USE_GUIDED_DISTRIBUTION_PATH "@GKFS_USE_GUIDED_DISTRIBUTION_PATH@"



#endif //FS_CMAKE_CONFIGURE_H

// clang-format on