Resolve "Add a CLI tool for requesting data transfers."

Alberto Mirandarequested to merge
amiranda/19-add-a-cli-tool-for-requesting-data-transfers into main

This MR adds the ccp tool to Cargo so that data transfers can be started from the command line. The tool supports the following arguments:

$ ccp -h
Cargo parallel copy tool
Usage: ccp [OPTIONS]

Options:
  -h,--help                   Print this help message and exit
  -s,--server ADDRESS         Address of the Cargo server (can also be
                              provided via the CCP_SERVER environment
                              variable).
  -i,--input SRC...           Input dataset(s)
  -o,--output DST...          Output dataset(s)
  --if FLAGS                  Flags for input datasets. Accepted values
                                - posix: read data using POSIX (default)
                                - mpio: read data using MPI-IO
  --of FLAGS                  Flags for output datasets. Accepted values
                                - posix: write data using POSIX (default)
                                - mpio: write data using MPI-IO

Once a Cargo server is up and running, ccp can be used as follows:

$ srun -N4 cargo -l ofi+tcp://192.18.0.10:62000 -o cargo.log
[...]
$ ccp -s ofi+tcp://192.18.0.10:62000 -i input0.dat input1.dat -o output0.dat output1.dat

Modes of operation that control how transfers should be executed can be selected through the the --if and --of options. For now, posix and mpio modes are supported. Note however that not all combinations are supported yet, and that performance will depend on whether the underlying file system supports parallel I/O through MPI-IO.

Closes #19 (closed)

Edited by Alberto Miranda

Merge request reports

lib/libcargo.cpp

+6 −0
Changes for lib/libcargo.cpp: 6 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -158,6 +158,12 @@ transfer 
transfer_datasets(const server& srv, const std::vector<dataset>& sources,

                  const std::vector<dataset>& targets) {



    if(sources.size() != targets.size()) {

        throw std::runtime_error(

                "The number of input datasets does not match the number of "

                "output datasets");

    }



    network::client rpc_client{srv.protocol()};

    const auto rpc = network::rpc_info::create(RPC_NAME(), srv.address());

util/CMakeLists.txt

+16 −1
Changes for util/CMakeLists.txt: 16 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -52,6 +52,21 @@ target_link_libraries(cargo_shutdown 
        cargo

)



install(TARGETS cargo_ping cargo_shutdown

add_executable(ccp)



target_sources(ccp

  PRIVATE

    copy.cpp

)



target_link_libraries(ccp

  PUBLIC

    fmt::fmt

    CLI11::CLI11

    net::rpc_client

    cargo

)



install(TARGETS cargo_ping cargo_shutdown ccp

        RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}

)

util/copy.cpp

0 → 100644
+144 −0
Changes for util/copy.cpp: 144 added lines, 0 removed lines.
Original line number Diff line number Diff line 
/******************************************************************************

 * Copyright 2022-2023, Barcelona Supercomputing Center (BSC), Spain

 *

 * This software was partially supported by the EuroHPC-funded project ADMIRE

 *   (Project ID: 956748, https://www.admire-eurohpc.eu).

 *

 * This file is part of Cargo.

 *

 * Cargo is free software: you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation, either version 3 of the License, or

 * (at your option) any later version.

 *

 * Cargo is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 * GNU General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License

 * along with Cargo.  If not, see <https://www.gnu.org/licenses/>.

 *

 * SPDX-License-Identifier: GPL-3.0-or-later

 *****************************************************************************/



#include <fmt/format.h>

#include <cargo.hpp>

#include <fmt_formatters.hpp>

#include <filesystem>

#include <CLI/CLI.hpp>

#include <ranges>



enum class dataset_flags { posix, mpio };



std::map<std::string, dataset_flags> dataset_flags_map{

        {"posix", dataset_flags::posix},

        {"mpio", dataset_flags::mpio}};



struct copy_config {

    std::string progname;

    std::string server_address;

    std::vector<std::filesystem::path> inputs;

    dataset_flags input_flags = dataset_flags::posix;

    std::vector<std::filesystem::path> outputs;

    dataset_flags output_flags = dataset_flags::posix;

};



copy_config

parse_command_line(int argc, char* argv[]) {



    copy_config cfg;



    cfg.progname = std::filesystem::path{argv[0]}.filename().string();



    CLI::App app{"Cargo parallel copy tool", cfg.progname};



    app.add_option("-s,--server", cfg.server_address,

                   "Address of the Cargo server (can also be\n"

                   "provided via the CCP_SERVER environment\n"

                   "variable)")

            ->option_text("ADDRESS")

            ->envname("CCP_SERVER")

            ->required();



    app.add_option("-i,--input", cfg.inputs, "Input dataset(s)")

            ->option_text("SRC...")

            ->required();



    app.add_option("-o,--output", cfg.outputs, "Output dataset(s)")

            ->option_text("DST...")

            ->required();



    app.add_option("--if", cfg.input_flags,

                   "Flags for input datasets. Accepted values\n"

                   "  - posix: read data using POSIX (default)\n"

                   "  - mpio: read data using MPI-IO")

            ->option_text("FLAGS")

            ->transform(CLI::CheckedTransformer(dataset_flags_map,

                                                CLI::ignore_case));



    app.add_option("--of", cfg.output_flags,

                   "Flags for output datasets. Accepted values\n"

                   "  - posix: write data using POSIX (default)\n"

                   "  - mpio: write data using MPI-IO")

            ->option_text("FLAGS")

            ->transform(CLI::CheckedTransformer(dataset_flags_map,

                                                CLI::ignore_case));



    try {

        app.parse(argc, argv);

        return cfg;

    } catch(const CLI::ParseError& ex) {

        std::exit(app.exit(ex));

    }

}



auto

parse_address(const std::string& address) {

    const auto pos = address.find("://");

    if(pos == std::string::npos) {

        throw std::runtime_error(fmt::format("Invalid address: {}", address));

    }



    const auto protocol = address.substr(0, pos);

    return std::make_pair(protocol, address);

}



int

main(int argc, char* argv[]) {



    const auto cfg = parse_command_line(argc, argv);



    try {

        const auto [protocol, address] = parse_address(cfg.server_address);



        cargo::server server{address};

        std::vector<cargo::dataset> inputs;

        std::vector<cargo::dataset> outputs;



        std::transform(cfg.inputs.cbegin(), cfg.inputs.cend(),

                       std::back_inserter(inputs), [&](const auto& src) {

                           return cargo::dataset{

                                   src, cfg.input_flags == dataset_flags::mpio

                                                ? cargo::dataset::type::parallel

                                                : cargo::dataset::type::posix};

                       });



        std::transform(cfg.outputs.cbegin(), cfg.outputs.cend(),

                       std::back_inserter(outputs), [&cfg](const auto& tgt) {

                           return cargo::dataset{

                                   tgt, cfg.output_flags == dataset_flags::mpio

                                                ? cargo::dataset::type::parallel

                                                : cargo::dataset::type::posix};

                       });



        const auto tx = cargo::transfer_datasets(server, inputs, outputs);



        if(const auto st = tx.wait(); st.failed()) {

            throw std::runtime_error(st.error().message());

        }

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

        fmt::print(stderr, "{}: Error: {}\n", cfg.progname, ex.what());

        return EXIT_FAILURE;

    }

}