Commit 714fe3aa authored by Ramon Nou's avatar Ramon Nou
Browse files

fix(proxy): eliminate thread_local bulk buffers and segments in Argobots RPC handlers

parent fd011880

src/proxy/rpc/forward_data.cpp

+2 −13
Changes for src/proxy/rpc/forward_data.cpp: 2 added lines, 13 removed lines.
Original line number Diff line number Diff line
@@ -57,25 +57,14 @@ using namespace std; 


namespace gkfs::rpc {



namespace {



std::vector<std::pair<void*, std::size_t>>&

single_bulk_segment(void* buffer, std::size_t size) {

    thread_local std::vector<std::pair<void*, std::size_t>> segments;

    segments.resize(1);

    segments[0] = std::make_pair(buffer, size);

    return segments;

}



} // namespace



std::pair<int, ssize_t>

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

              const size_t write_size) {

    using namespace gkfs::utils::arithmetic;



    // Expose local buffer

    auto& segments = single_bulk_segment(buf, write_size);

    std::vector<std::pair<void*, std::size_t>> segments;

    segments.emplace_back(std::make_pair(buf, write_size));



    auto engine = PROXY_DATA->client_rpc_engine();

    tl::bulk bulk_handle;

src/proxy/rpc/srv_data.cpp

+6 −44
Changes for src/proxy/rpc/srv_data.cpp: 6 added lines, 44 removed lines.
Original line number Diff line number Diff line
@@ -67,42 +67,6 @@ sample_proxy_trace_log() { 
}

} // namespace



namespace {



// Proxy data RPCs need a local staging buffer because the client bulk handle

// belongs to the remote process. Keep one buffer per worker thread to avoid a

// heap allocation for every read/write request. Release unusually large buffers

// after the request so one outlier does not pin excessive memory indefinitely.

constexpr size_t max_reusable_bulk_buffer = 64 * 1024 * 1024;



vector<char>&

reusable_bulk_buffer(size_t size) {

    thread_local vector<char> buffer;

    if(size > buffer.capacity()) {

        PROXY_DATA->record_staging_buffer_growth();

    }

    buffer.resize(size);

    return buffer;

}



vector<pair<void*, size_t>>&

single_bulk_segment(void* buffer, size_t size) {

    thread_local vector<pair<void*, size_t>> segments;

    PROXY_DATA->record_bulk_descriptor_setup();

    segments.resize(1);

    segments[0] = make_pair(buffer, size);

    return segments;

}



void

release_oversized_bulk_buffer(vector<char>& buffer) {

    if(buffer.capacity() > max_reusable_bulk_buffer) {

        vector<char>().swap(buffer);

    }

}



} // namespace



void

proxy_rpc_srv_write(const tl::request& req,

                    const gkfs::rpc::rpc_client_proxy_write_in_t& in) {
@@ -131,10 +95,11 @@ proxy_rpc_srv_write(const tl::request& req, 
                "{}() Got RPC with path '{}' bulk_size '{}' == write_size '{}'",

                __func__, in.path, bulk_size, in.write_size);



        auto& bulk_buf = reusable_bulk_buffer(bulk_size);

        std::vector<char> bulk_buf(bulk_size);

        PROXY_DATA->record_staged_bytes(bulk_size);

        {

            auto& segments = single_bulk_segment(bulk_buf.data(), bulk_size);

            std::vector<std::pair<void*, std::size_t>> segments;

            segments.emplace_back(std::make_pair(bulk_buf.data(), bulk_size));



            auto engine = PROXY_DATA->server_ipc_engine();

            tl::bulk local_bulk =
@@ -150,8 +115,6 @@ proxy_rpc_srv_write(const tl::request& req, 
            out.io_size = daemon_out.second;

        }



        release_oversized_bulk_buffer(bulk_buf);



        PROXY_DATA->log()->debug("{}() Sending output err '{}' io_size '{}'",

                                 __func__, out.err, out.io_size);
@@ -200,7 +163,7 @@ proxy_rpc_srv_read(const tl::request& req, 
                "{}() Got RPC with path '{}' bulk_size '{}' == read_size '{}'",

                __func__, in.path, bulk_size, in.read_size);



        auto& bulk_buf = reusable_bulk_buffer(bulk_size);

        std::vector<char> bulk_buf(bulk_size);

        PROXY_DATA->record_staged_bytes(bulk_size);

        {

            // Forward request to daemon (pulls data into bulk_buf)
@@ -217,7 +180,8 @@ proxy_rpc_srv_read(const tl::request& req, 
            }



            // Push data to client

            auto& segments = single_bulk_segment(bulk_buf.data(), bulk_size);

            std::vector<std::pair<void*, std::size_t>> segments;

            segments.emplace_back(std::make_pair(bulk_buf.data(), bulk_size));



            auto engine = PROXY_DATA->server_ipc_engine();

            tl::bulk local_bulk =
@@ -229,8 +193,6 @@ proxy_rpc_srv_read(const tl::request& req, 
            out.io_size = daemon_out.second;

        }



        release_oversized_bulk_buffer(bulk_buf);



        PROXY_DATA->log()->debug("{}() Sending output err '{}' io_size '{}'",

                                 __func__, out.err, out.io_size);