Commit 3bdd5f71 authored by Ramon Nou's avatar Ramon Nou
Browse files

Read replica migration sources by copy

parent d5b9804d

include/common/rpc/rpc_types_thallium.hpp

+2 −1
Changes for include/common/rpc/rpc_types_thallium.hpp: 2 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -308,6 +308,7 @@ struct rpc_read_data_in_t { 
    uint64_t chunk_start;

    uint64_t chunk_end;

    uint64_t total_chunk_size;

    int32_t copy{0};

    uint64_t trace_id{gkfs::trace::next()};

    tl::bulk bulk_handle; // SERIALIZATION OF BULK HANDLE?

                          // Thallium bulk handles generally need to be exposed.
@@ -320,7 +321,7 @@ struct rpc_read_data_in_t { 
    void

    serialize(Archive& ar) {

        ar(path, offset, host_id, host_size, wbitset, chunk_n, chunk_start,

           chunk_end, total_chunk_size, trace_id, bulk_handle);

           chunk_end, total_chunk_size, copy, trace_id, bulk_handle);

    }

};

include/daemon/malleability/rpc/forward_redistribution.hpp

+4 −0
Changes for include/daemon/malleability/rpc/forward_redistribution.hpp: 4 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -51,6 +51,10 @@ forward_metadata(const std::string& key, const std::string& value, 
int

forward_data(const std::string& path, void* buf, const size_t count,

             const uint64_t chnk_id, const uint64_t dest_id);

std::pair<int, size_t>

forward_data_from_source(const std::string& path, void* buf, size_t count,

                         uint64_t chnk_id, uint64_t source_id, int source_copy,

                         uint64_t dest_id);



} // namespace gkfs::malleable::rpc

src/client/rpc/forward_data.cpp

+1 −0
Changes for src/client/rpc/forward_data.cpp: 1 added line, 0 removed lines.
Original line number Diff line number Diff line
@@ -621,6 +621,7 @@ repair_data_copy(const gkfs::rpc::repair_task& task) { 
        read_in.path = task.path;

        read_in.offset = 0;

        read_in.host_id = source->host;

        read_in.copy = task.source_copy;

        read_in.host_size = CTX->hosts().size();

        read_in.wbitset = gkfs::rpc::compress_bitset(std::vector<uint8_t>{1});

        read_in.chunk_n = 1;

src/daemon/malleability/malleable_manager.cpp

+8 −9
Changes for src/daemon/malleability/malleable_manager.cpp: 8 added lines, 9 removed lines.
Original line number Diff line number Diff line
@@ -678,16 +678,15 @@ MalleableManager::redistribute_metadata() { 


int

MalleableManager::do_migration(gkfs::rpc::MigrationJob& job) {

    // The current daemon migration path reads only its local primary chunk

    // storage. Do not let a replica-tagged job silently use that path: it must

    // first gain an explicit source-copy read operation.

    if(gkfs::rpc::is_replica_migration_job(job)) {

        GKFS_DATA->spdlogger()->warn(

                "{}() Replica migration requires a source-copy read path; "

                "rejecting path '{}' chunk {} copy {} -> {}",

                __func__, job.path, job.chunk_id, job.source_copy,

                job.target_copy);

        return -ENOTSUP;

        std::vector<char> buffer(gkfs::config::rpc::chunksize);

        const auto result = gkfs::malleable::rpc::forward_data_from_source(

                job.path, buffer.data(), buffer.size(), job.chunk_id,

                job.source_node, job.source_copy, job.target_node);

        if(result.first == 0) {

            job.bytes_transferred = result.second;

        }

        return result.first == 0 ? 0 : -1;

    }



    // Read chunk data from local storage

src/daemon/malleability/rpc/forward_redistribution.cpp

+52 −0
Changes for src/daemon/malleability/rpc/forward_redistribution.cpp: 52 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -158,5 +158,57 @@ forward_data(const std::string& path, void* buf, const size_t count, 
    }

    return err;

}

std::pair<int, size_t>

forward_data_from_source(const std::string& path, void* buf, const size_t count,

                         const uint64_t chnk_id, const uint64_t source_id,

                         const int source_copy, const uint64_t dest_id) {

    if(source_id == dest_id) {

        return {forward_data(path, buf, count, chnk_id, dest_id), count};

    }



    std::vector<std::pair<void*, std::size_t>> segments{{buf, count}};

    tl::bulk read_bulk;

    try {

        read_bulk = RPC_DATA->client_rpc_engine()->expose(

                segments, tl::bulk_mode::write_only);

    } catch(const std::exception& e) {

        GKFS_DATA->spdlogger()->error(

                "{}() failed to expose source-read buffer: {}", __func__,

                e.what());

        return {EBUSY, 0};

    }



    gkfs::rpc::rpc_read_data_in_t read_in{};

    read_in.path = path;

    read_in.offset = 0;

    read_in.host_id = source_id;

    read_in.host_size = RPC_DATA->hosts_size();

    read_in.wbitset = gkfs::rpc::compress_bitset(std::vector<uint8_t>{1});

    read_in.chunk_n = 1;

    read_in.chunk_start = chnk_id;

    read_in.chunk_end = chnk_id;

    read_in.total_chunk_size = count;

    read_in.copy = source_copy;

    read_in.bulk_handle = read_bulk;



    try {

        auto read_rpc =

                RPC_DATA->client_rpc_engine()->define(gkfs::rpc::tag::read);

        const auto read_out =

                read_rpc.on(RPC_DATA->rpc_endpoints().at(source_id))(read_in)

                        .as<gkfs::rpc::rpc_data_out_t>();

        if(read_out.err != 0 || read_out.io_size == 0 ||

           read_out.io_size > count) {

            return {read_out.err != 0 ? read_out.err : EIO, 0};

        }

        return {forward_data(path, buf, read_out.io_size, chnk_id, dest_id),

                static_cast<size_t>(read_out.io_size)};

    } catch(const std::exception& e) {

        GKFS_DATA->spdlogger()->error(

                "{}() source read failed for '{}' chunk {} from host {}: {}",

                __func__, path, chnk_id, source_id, e.what());

        return {EBUSY, 0};

    }

}



} // namespace gkfs::malleable::rpc
 
 No newline at end of file