Malleability extension -> mutate (shrink, expand) and random slicing + cutshift capabilities

Ramon Nourequested to merge
rnou/rslice into master

Malleability extension -> mutate (shrink, expand) and random slicing + cutshift capabilities Malleability is easy (+ , - markers), daemons xpanding automatically create + line in host file Malleability tool automatically removes shrink servers Also GKFS_EXPAND_ON_DEMAND is available --> Chunks are only transferred if clients ask for them (small penalty)

Solves some bugs also.

Merge request reports

examples/gfind/sfind.cpp

+12 −4
Changes for examples/gfind/sfind.cpp: 12 added lines, 4 removed lines.
Original line number Diff line number Diff line
@@ -302,11 +302,15 @@ worker_routine(void* arg) { 


                if(n > 0 && entries && !data->opt->just_count) {

                    char* ptr = reinterpret_cast<char*>(entries);

                    int bytes_processed = 0;

                    size_t bytes_processed = 0;

                    while(bytes_processed < n) {

                        struct dirent_extended* temp =

                                reinterpret_cast<struct dirent_extended*>(ptr);

                        if(temp->d_reclen == 0)

                        const auto remaining =

                                static_cast<size_t>(n) - bytes_processed;

                        if(temp->d_reclen <

                                   offsetof(struct dirent_extended, d_name) + 1 ||

                           temp->d_reclen > remaining)

                            break;



                        local_found++;
@@ -360,11 +364,15 @@ worker_routine(void* arg) { 
            }



            char* ptr = reinterpret_cast<char*>(entries);

            int bytes_processed = 0;

            size_t bytes_processed = 0;

            while(bytes_processed < n) {

                struct dirent_extended* temp =

                        reinterpret_cast<struct dirent_extended*>(ptr);

                if(temp->d_reclen == 0)

                const auto remaining =

                        static_cast<size_t>(n) - bytes_processed;

                if(temp->d_reclen <

                           offsetof(struct dirent_extended, d_name) + 1 ||

                   temp->d_reclen > remaining)

                    break;



                if(temp->d_type != 1) {

include/client/rpc/forward_malleability.hpp

+6 −8
Changes for include/client/rpc/forward_malleability.hpp: 6 added lines, 8 removed lines.
Original line number Diff line number Diff line
@@ -45,23 +45,21 @@ 
namespace gkfs::malleable::rpc {



int

forward_expand_start(int old_server_conf, int new_server_conf);

forward_mutate_start(int old_server_conf, int new_server_conf,

                     const std::string& new_hosts_file);



int

forward_expand_status();

forward_mutate_status();



int

forward_expand_finalize();

forward_mutate_finalize();



int

forward_shrink_start(int old_server_conf, int new_server_conf,

                     const std::string& new_hosts_file);

forward_mutate_reload(const std::string& hosts_file);



int

forward_shrink_status();

forward_mutate_shutdown_removed(const std::string& hostfile);



int

forward_shrink_finalize();

} // namespace gkfs::malleable::rpc



#endif // GEKKOFS_CLIENT_FORWARD_MALLEABILITY_HPP

include/client/rpc/utils.hpp

+10 −2
Changes for include/client/rpc/utils.hpp: 10 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -37,6 +37,7 @@ 
#include <cstdint>

#include <cstring>

#include <type_traits>

#include <limits>



namespace gkfs::rpc {
@@ -54,8 +55,9 @@ namespace gkfs::rpc { 
 */

template <typename OutputOrErr>

std::vector<std::tuple<const std::string, unsigned char, size_t, time_t>>

decompress_and_parse_entries(const OutputOrErr& out,

                             const void* compressed_buffer) {

decompress_and_parse_entries(

        const OutputOrErr& out, const void* compressed_buffer,

        std::size_t buffer_size = std::numeric_limits<std::size_t>::max()) {

    if(out.err != 0) {

        throw std::runtime_error("Server returned an error: " +

                                 std::to_string(out.err));
@@ -63,6 +65,12 @@ decompress_and_parse_entries(const OutputOrErr& out, 
    if(out.dirents_size == 0) {

        return {};

    }

    if(out.dirents_size > buffer_size) {

        throw std::runtime_error(

                "Server returned dirents payload larger than exposed client buffer: " +

                std::to_string(out.dirents_size) + " > " +

                std::to_string(buffer_size));

    }



    const char* p = nullptr;

    const char* end = nullptr;

include/client/env.hpp

+5 −6
Changes for include/client/env.hpp: 5 added lines, 6 removed lines.
Original line number Diff line number Diff line
@@ -83,8 +83,6 @@ static constexpr auto PROTECT_FILES_CONSUMER = 
        ADD_PREFIX("PROTECT_FILES_CONSUMER");

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

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

static constexpr auto USE_DIRENTS_COMPRESSION =

        ADD_PREFIX("USE_DIRENTS_COMPRESSION");



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

static constexpr auto PROXY_PID_FILE = ADD_PREFIX("PROXY_PID_FILE");
@@ -95,10 +93,6 @@ static constexpr auto WRITE_SIZE_THRESHOLD = 
        ADD_PREFIX("WRITE_SIZE_CACHE_THRESHOLD");

} // namespace cache



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

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

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

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

static constexpr auto CREATE_WRITE_OPTIMIZATION =

        ADD_PREFIX("CREATE_WRITE_OPTIMIZATION");

static constexpr auto READ_INLINE_PREFETCH = ADD_PREFIX("READ_INLINE_PREFETCH");
@@ -107,6 +101,11 @@ static constexpr auto METADATA_BATCH = ADD_PREFIX("METADATA_BATCH"); 
static constexpr auto METADATA_BATCH_THRESHOLD =

        ADD_PREFIX("METADATA_BATCH_THRESHOLD");

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

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

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

static constexpr auto FUSE_NEGATIVE_TIMEOUT =

        ADD_PREFIX("FUSE_NEGATIVE_TIMEOUT");

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



// Libfabric interface pinning (consumed by libfabric at HG_init() time)

// OFI_INTERFACE is used with the GKFS_ prefix (e.g., LIBGKFS_OFI_INTERFACE)

include/client/user_functions.hpp

+38 −33
Changes for include/client/user_functions.hpp: 38 added lines, 33 removed lines.
Original line number Diff line number Diff line
@@ -157,54 +157,59 @@ gkfs_msync(void* addr, size_t length, int flags); 
namespace malleable {



/**

 * @brief Start an expansion of the file system

 * @param old_server_conf old number of nodes

 * @param new_server_conf new number of nodes

 * @brief Start a mutation of the file system cluster topology.

 *

 * MARKER-BASED WORKFLOW:

 *   This function uses a marker-based single-hostfile approach. The workspace

 *   hostfile (LIBGKFS_HOSTS_FILE) contains markers that define the before/after

 *   state:

 *     - Normal lines (no prefix)  = active daemons

 *     - Lines with '-' prefix     = to-be-removed daemons (user writes

 * manually)

 *     - Lines with '+' prefix     = to-be-added daemons (auto-written by

 * daemons with GKFS_DAEMON_EXPAND=ON)

 *

 *   The CLI auto-discovers before/after state from markers in

 * LIBGKFS_HOSTS_FILE. No separate --new-hosts-file is needed.

 *

 * SHRINK: User writes '-' on N daemons in LIBGKFS_HOSTS_FILE, then calls mutate

 *         start. The daemon drops the -marked nodes after redistribution.

 *

 * EXPAND: New daemons start with GKFS_DAEMON_EXPAND=ON (auto-write '+'

 * prefix), then user calls mutate start. The daemon promotes + nodes after

 * redistribution.

 *

 * MUTATE: Combination of shrink + expand in one operation.

 *

 * @param old_server_conf old number of nodes (active + removing, auto-detected

 * if -1)

 * @param new_server_conf new number of nodes (active + adding, auto-detected if

 * -1)

 * @param new_hosts_file path to workspace hostfile with markers

 * (LIBGKFS_HOSTS_FILE)

 * @return error code

 */

int

expand_start(int old_server_conf, int new_server_conf);



/**

 * @brief Check for the current status of the expansion process

 * @return 0 when finished, positive numbers indicate how many daemons

 * are still redistributing data

 */

int

expand_status();



/**

 * @brief Finalize the expansion process

 * @return error code

 */

int

expand_finalize();



/**

 * @brief Start a shrinking of the file system

 * @param old_server_conf old number of nodes

 * @param new_server_conf new number of nodes

 * @param new_hosts_file path to hostfile containing only the surviving nodes

 * @return error code

 */

int

shrink_start(int old_server_conf, int new_server_conf,

mutate_start(int old_server_conf, int new_server_conf,

             const std::string& new_hosts_file);



/**

 * @brief Check for the current status of the shrinking process

 * @brief Check for the current status of the mutate process

 * @return 0 when finished, positive numbers indicate how many daemons

 * are still redistributing data

 */

int

shrink_status();

mutate_status();



/**

 * @brief Finalize the shrinking process

 * @brief Finalize the mutate process.

 *         Rewrites the workspace hostfile clean (promotes + to active, removes

 * - lines).

 * @return error code

 */

int

shrink_finalize();

mutate_finalize();



} // namespace malleable

} // namespace gkfs