Data replication (client side, synchronous)

Ramon Nourequested to merge
rnou/replication into master

This MR adds support for data replication using one environment variable:

LIBGKFS_NUM_REPL=<num repl> The number of replicas should go from 0 to the number of servers-1. The replicas are guided by the client, so it reduces write performance but we mantain the same level of consistency. On the other hand, it may increase read performance on some corner scenearios. Metadata replication is also implemented The replication environment variable can be set up for each client, independently.

If a server is down, the data will be read from another replica. The metadata management is also done from another replica.

The replication is done in a synchronous way. A new function forward_write is used to sent to the different replicas. The reads are distributed, but this shouldn't produce an performance improvement as the distribution is similar to the original.

In the case of the write, the original is sent to the target servers, and then the replicas are processed. This is done to avoid issues if a server, that should host a replica, is not available.

In order to process the replicas a new method to check that a chunk needs to be processed inside a server is included, a bitset of 1024 is sent (coded in base-64 in a string). This represents 1024-chunks per write-read operation. If that is exceeded the normal hash check per chunk is done in the server. Exceeding this value, will disable the replica capabilities and produce unknown behaviours.

This can be potentially increased.

Finally, most of the operations are replica-aware, but some of them are missing yet. i.e., dirent.

Edited by Ramon Nou

Merge request reports

docs/sphinx/users/running.md

+86 −0
Changes for docs/sphinx/users/running.md: 86 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -218,3 +218,89 @@ the logging subsystem to truncate the file used for logging, rather than append 
For the daemon, the `GKFS_DAEMON_LOG_PATH=<path/to/file>` environment variable can be provided to set the path to the

log file, and the log module can be selected with the `GKFS_DAEMON_LOG_LEVEL={off,critical,err,warn,info,debug,trace}`

environment variable whereas `trace` produces the most trace records while `info` is the default value.



## Miscellaneous



### External functions



GekkoFS allows to use external functions on your client code, via LD_PRELOAD.

Source code needs to be compiled with -fPIC. We include a pfind io500 substitution,

`examples/gfind/gfind.cpp` and a non-mpi version `examples/gfind/sfind.cpp`



### Data distributors



The data distribution can be selected at compilation time, we have 2 distributors available:



#### Simple Hash (Default)



Chunks are distributed randomly to the different GekkoFS servers.



#### Guided Distributor



The guided distributor allows defining a specific distribution of data on a per directory or file basis.

The distribution configurations are defined within a shared file (called `guided_config.txt` henceforth) with the

following format:

`<path> <chunk_number> <host>`



To enable the distributor, the following CMake compilation flags are required:



* `GKFS_USE_GUIDED_DISTRIBUTION` ON

* `GKFS_USE_GUIDED_DISTRIBUTION_PATH` `<path_guided_config.txt>`



To use a custom distribution, a path needs to have the prefix `#` (e.g., `#/mdt-hard 0 0`), in which all the data of all

files in that directory goes to the same place as the metadata.

Note, that a chunk/host configuration is inherited to all children files automatically even if not using the prefix.

In this example, `/mdt-hard/file1` is therefore also using the same distribution as the `/mdt-hard` directory.

If no prefix is used, the Simple Hash distributor is used.



##### Guided configuration file



Creating a guided configuration file is based on an I/O trace file of a previous execution of the application.

For this the `trace_reads` tracing module is used (see above).



The `trace_reads` module enables a `TRACE_READS` level log at the clients writing the I/O information of the client

which is used as the input for a script that creates the guided distributor setting.

Note that capturing the necessary trace records can involve performance degradation.

To capture the I/O of each client within a SLURM environment, i.e., enabling the `trace_reads` module and print its

output to a user-defined path, the following example can be used:

`srun -N 10 -n 320 --export="ALL" /bin/bash -c "export LIBGKFS_LOG=trace_reads;LIBGKFS_LOG_OUTPUT=${HOME}/test/GLOBAL.txt;LD_PRELOAD=${GKFS_PRLD} <app>"`



Then, the `examples/distributors/guided/generate.py` scrpt is used to create the guided distributor configuration file:



* `python examples/distributors/guided/generate.py ~/test/GLOBAL.txt >> guided_config.txt`



Finally, modify `guided_config.txt` to your distribution requirements.



### Metadata Backends



There are two different metadata backends in GekkoFS. The default one uses `rocksdb`, however an alternative based

on `PARALLAX` from `FORTH`

is available. To enable it use the `-DGKFS_ENABLE_PARALLAX:BOOL=ON` option, you can also disable `rocksdb`

with `-DGKFS_ENABLE_ROCKSDB:BOOL=OFF`.



Once it is enabled, `--dbbackend` option will be functional.



### Statistics



GekkoFS daemons are able to output general operations (`--enable-collection`) and data chunk

statistics (`--enable-chunkstats`) to a specified output file via `--output-stats <FILE>`. Prometheus can also be used

instead or in addition to the output file. It must be enabled at compile time via the CMake

argument `-DGKFS_ENABLE_PROMETHEUS` and the daemon argument `--enable-prometheus`. The corresponding statistics are then

pushed to the Prometheus instance.



### Advanced experimental features



#### Rename



`-DGKFS_RENAME_SUPPORT` allows the application to rename files.

This is an experimental feature, and some scenarios may not work properly.

Support for fstat in renamed files is included.



This is disabled by default.



#### Replication



The user can enable the data replication feature by setting the replication environment variable:

`LIBGKFS_NUM_REPL=<num repl>`.

The number of replicas should go from `0` to the `number of servers - 1`. The replication environment variable can be

set up for each client independently.
 
 No newline at end of file

include/client/rpc/forward_data.hpp

+8 −3
Changes for include/client/rpc/forward_data.hpp: 8 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -30,6 +30,9 @@ 
#ifndef GEKKOFS_CLIENT_FORWARD_DATA_HPP

#define GEKKOFS_CLIENT_FORWARD_DATA_HPP



#include <string>

#include <memory>

#include <set>

namespace gkfs::rpc {



struct ChunkStat {
@@ -43,14 +46,16 @@ struct ChunkStat { 


std::pair<int, ssize_t>

forward_write(const std::string& path, const void* buf, off64_t offset,

              size_t write_size);

              size_t write_size, const int8_t num_copy = 0);



std::pair<int, ssize_t>

forward_read(const std::string& path, void* buf, off64_t offset,

             size_t read_size);

             size_t read_size, const int8_t num_copies,

             std::set<int8_t>& failed);



int

forward_truncate(const std::string& path, size_t current_size, size_t new_size);

forward_truncate(const std::string& path, size_t current_size, size_t new_size,

                 const int8_t num_copies);



std::pair<int, ChunkStat>

forward_get_chunk_stat();

include/client/rpc/forward_metadata.hpp

+11 −9
Changes for include/client/rpc/forward_metadata.hpp: 11 added lines, 9 removed lines.
Original line number Diff line number Diff line
@@ -50,10 +50,10 @@ class Metadata; 
namespace rpc {



int

forward_create(const std::string& path, mode_t mode);

forward_create(const std::string& path, mode_t mode, const int copy);



int

forward_stat(const std::string& path, std::string& attr);

forward_stat(const std::string& path, std::string& attr, const int copy);



#ifdef HAS_RENAME

int
@@ -62,22 +62,24 @@ forward_rename(const std::string& oldpath, const std::string& newpath, 
#endif // HAS_RENAME



int

forward_remove(const std::string& path);

forward_remove(const std::string& path, const int8_t num_copies);



int

forward_decr_size(const std::string& path, size_t length);

forward_decr_size(const std::string& path, size_t length, const int copy);



int

forward_update_metadentry(

        const std::string& path, const gkfs::metadata::Metadata& md,

        const gkfs::metadata::MetadentryUpdateFlags& md_flags);

forward_update_metadentry(const std::string& path,

                          const gkfs::metadata::Metadata& md,

                          const gkfs::metadata::MetadentryUpdateFlags& md_flags,

                          const int copy);



std::pair<int, off64_t>

forward_update_metadentry_size(const std::string& path, size_t size,

                               off64_t offset, bool append_flag);

                               off64_t offset, bool append_flag,

                               const int num_copies);



std::pair<int, off64_t>

forward_get_metadentry_size(const std::string& path);

forward_get_metadentry_size(const std::string& path, const int copy);



std::pair<int, std::shared_ptr<gkfs::filemap::OpenDir>>

forward_get_dirents(const std::string& path);

include/client/rpc/rpc_types.hpp

+30 −16
Changes for include/client/rpc/rpc_types.hpp: 30 added lines, 16 removed lines.
Original line number Diff line number Diff line
@@ -1469,11 +1469,11 @@ struct write_data { 


    public:

        input(const std::string& path, int64_t offset, uint64_t host_id,

              uint64_t host_size, uint64_t chunk_n, uint64_t chunk_start,

              uint64_t chunk_end, uint64_t total_chunk_size,

              const hermes::exposed_memory& buffers)

              uint64_t host_size, const std::string& wbitset, uint64_t chunk_n,

              uint64_t chunk_start, uint64_t chunk_end,

              uint64_t total_chunk_size, const hermes::exposed_memory& buffers)

            : m_path(path), m_offset(offset), m_host_id(host_id),

              m_host_size(host_size), m_chunk_n(chunk_n),

              m_host_size(host_size), m_wbitset(wbitset), m_chunk_n(chunk_n),

              m_chunk_start(chunk_start), m_chunk_end(chunk_end),

              m_total_chunk_size(total_chunk_size), m_buffers(buffers) {}
@@ -1512,6 +1512,11 @@ struct write_data { 
            return m_chunk_n;

        }



        std::string

        wbitset() const {

            return m_wbitset;

        }



        uint64_t

        chunk_start() const {

            return m_chunk_start;
@@ -1535,15 +1540,16 @@ struct write_data { 
        explicit input(const rpc_write_data_in_t& other)

            : m_path(other.path), m_offset(other.offset),

              m_host_id(other.host_id), m_host_size(other.host_size),

              m_chunk_n(other.chunk_n), m_chunk_start(other.chunk_start),

              m_chunk_end(other.chunk_end),

              m_wbitset(other.wbitset), m_chunk_n(other.chunk_n),

              m_chunk_start(other.chunk_start), m_chunk_end(other.chunk_end),

              m_total_chunk_size(other.total_chunk_size),

              m_buffers(other.bulk_handle) {}



        explicit operator rpc_write_data_in_t() {

            return {m_path.c_str(),      m_offset,          m_host_id,

                    m_host_size,    m_chunk_n,          m_chunk_start,

                    m_chunk_end,    m_total_chunk_size, hg_bulk_t(m_buffers)};

                    m_host_size,         m_wbitset.c_str(), m_chunk_n,

                    m_chunk_start,       m_chunk_end,       m_total_chunk_size,

                    hg_bulk_t(m_buffers)};

        }



    private:
@@ -1551,6 +1557,7 @@ struct write_data { 
        int64_t m_offset;

        uint64_t m_host_id;

        uint64_t m_host_size;

        std::string m_wbitset;

        uint64_t m_chunk_n;

        uint64_t m_chunk_start;

        uint64_t m_chunk_end;
@@ -1647,11 +1654,11 @@ struct read_data { 


    public:

        input(const std::string& path, int64_t offset, uint64_t host_id,

              uint64_t host_size, uint64_t chunk_n, uint64_t chunk_start,

              uint64_t chunk_end, uint64_t total_chunk_size,

              const hermes::exposed_memory& buffers)

              uint64_t host_size, const std::string& wbitset, uint64_t chunk_n,

              uint64_t chunk_start, uint64_t chunk_end,

              uint64_t total_chunk_size, const hermes::exposed_memory& buffers)

            : m_path(path), m_offset(offset), m_host_id(host_id),

              m_host_size(host_size), m_chunk_n(chunk_n),

              m_host_size(host_size), m_wbitset(wbitset), m_chunk_n(chunk_n),

              m_chunk_start(chunk_start), m_chunk_end(chunk_end),

              m_total_chunk_size(total_chunk_size), m_buffers(buffers) {}
@@ -1685,6 +1692,11 @@ struct read_data { 
            return m_host_size;

        }



        std::string

        wbitset() const {

            return m_wbitset;

        }



        uint64_t

        chunk_n() const {

            return m_chunk_n;
@@ -1713,15 +1725,16 @@ struct read_data { 
        explicit input(const rpc_read_data_in_t& other)

            : m_path(other.path), m_offset(other.offset),

              m_host_id(other.host_id), m_host_size(other.host_size),

              m_chunk_n(other.chunk_n), m_chunk_start(other.chunk_start),

              m_chunk_end(other.chunk_end),

              m_wbitset(other.wbitset), m_chunk_n(other.chunk_n),

              m_chunk_start(other.chunk_start), m_chunk_end(other.chunk_end),

              m_total_chunk_size(other.total_chunk_size),

              m_buffers(other.bulk_handle) {}



        explicit operator rpc_read_data_in_t() {

            return {m_path.c_str(),      m_offset,          m_host_id,

                    m_host_size,    m_chunk_n,          m_chunk_start,

                    m_chunk_end,    m_total_chunk_size, hg_bulk_t(m_buffers)};

                    m_host_size,         m_wbitset.c_str(), m_chunk_n,

                    m_chunk_start,       m_chunk_end,       m_total_chunk_size,

                    hg_bulk_t(m_buffers)};

        }



    private:
@@ -1729,6 +1742,7 @@ struct read_data { 
        int64_t m_offset;

        uint64_t m_host_id;

        uint64_t m_host_size;

        std::string m_wbitset;

        uint64_t m_chunk_n;

        uint64_t m_chunk_start;

        uint64_t m_chunk_end;

include/client/env.hpp

+1 −1
Changes for include/client/env.hpp: 1 added line, 1 removed line.
Original line number Diff line number Diff line
@@ -52,7 +52,7 @@ static constexpr auto HOSTS_FILE = ADD_PREFIX("HOSTS_FILE"); 
#ifdef GKFS_ENABLE_FORWARDING

static constexpr auto FORWARDING_MAP_FILE = ADD_PREFIX("FORWARDING_MAP_FILE");

#endif



static constexpr auto NUM_REPL = ADD_PREFIX("NUM_REPL");

} // namespace gkfs::env



#undef ADD_PREFIX