Rnou/49 refactor some functions (!38) · Merge requests · hpc / cargo

cli/CMakeLists.txt

+19 −74

Changes for cli/CMakeLists.txt: 19 added lines, 74 removed lines.
Original line number	Diff line number	Diff line
		@@ -34,95 +34,40 @@ configure_file(cargoctl.in cargoctl @ONLY)


		################################################################################
		## cargo_ping: A CLI tool to check if a Cargo server is running
		add_executable(cargo_ping)

		target_sources(cargo_ping
		PRIVATE
		ping.cpp
		)

		target_link_libraries(cargo_ping
		PUBLIC
		## Common object library for CLI tools
		add_library(cli_common OBJECT common.cpp)
		target_link_libraries(cli_common PUBLIC
		fmt::fmt
		CLI11::CLI11
		net::rpc_client
		cargo
		)

		################################################################################
		## cargo_shutdown: A CLI tool to shutdown a Cargo server
		add_executable(cargo_shutdown)

		target_sources(cargo_shutdown
		PRIVATE
		shutdown.cpp
		# Helper function to define a CLI tool
		function(add_cargo_cli_tool name source)
		add_executable(${name})
		target_sources(${name} PRIVATE
		${source}
		$<TARGET_OBJECTS:cli_common>
		)

		target_link_libraries(cargo_shutdown
		PUBLIC
		target_link_libraries(${name} PUBLIC
		fmt::fmt
		CLI11::CLI11
		net::rpc_client
		cargo
		)
		install(TARGETS ${name} RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR})
		endfunction()

		################################################################################
		## ccp: A CLI tool to request a Cargo server to copy files between storage tiers
		add_executable(ccp)

		target_sources(ccp
		PRIVATE
		copy.cpp
		)

		target_link_libraries(ccp
		PUBLIC
		fmt::fmt
		CLI11::CLI11
		net::rpc_client
		cargo
		)
		## CLI tool definitions
		add_cargo_cli_tool(cargo_ping ping.cpp)
		add_cargo_cli_tool(cargo_shutdown shutdown.cpp)
		add_cargo_cli_tool(ccp copy.cpp)
		add_cargo_cli_tool(shaping shaping.cpp)
		add_cargo_cli_tool(cargo_ftio ftio.cpp)

		################################################################################
		## shaping: A CLI tool to request a Cargo server to slowdown transfers
		add_executable(shaping)

		target_sources(shaping
		PRIVATE
		shaping.cpp
		)

		target_link_libraries(shaping
		PUBLIC
		fmt::fmt
		CLI11::CLI11
		net::rpc_client
		cargo
		)


		################################################################################
		## ftio: A CLI tool to send the ftio info to a Cargo server
		add_executable(cargo_ftio)

		target_sources(cargo_ftio
		PRIVATE
		ftio.cpp
		)

		target_link_libraries(cargo_ftio
		PUBLIC
		fmt::fmt
		CLI11::CLI11
		net::rpc_client
		cargo
		)


		install(TARGETS cargo_ping cargo_shutdown ccp shaping cargo_ftio
		RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}
		)

		# Installation
		install(PROGRAMS ${CMAKE_CURRENT_BINARY_DIR}/cargoctl
		DESTINATION ${CMAKE_INSTALL_BINDIR})
		No newline at end of file

cli/common.cpp

0 → 100644

+49 −0

Changes for cli/common.cpp: 49 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 "common.hpp"
		#include <fmt/format.h>
		#include <stdexcept>
		#include <CLI/CLI.hpp>

		std::pair<std::string, std::string>
		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);
		}

		void parse_rpc_command_line(int argc, char* argv[], CLI::App& app, std::string& server_address) {
		app.add_option("-s,--server", server_address, "Server address")
		->option_text("ADDRESS")
		->required();
		try {
		app.parse(argc, argv);
		} catch(const CLI::ParseError& ex) {
		std::exit(app.exit(ex));
		}
		}
		No newline at end of file

cli/common.hpp

0 → 100644

+44 −0

Changes for cli/common.hpp: 44 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
		*****************************************************************************/

		#ifndef CARGO_CLI_COMMON_HPP
		#define CARGO_CLI_COMMON_HPP

		#include <string>
		#include <utility>

		namespace CLI {
		class App;
		}

		// Parses a server address string into protocol and address.
		// Throws a runtime_error if the address is invalid.
		std::pair<std::string, std::string>
		parse_address(const std::string& address);


		void parse_rpc_command_line(int argc, char* argv[], CLI::App& app, std::string& server_address);


		#endif // CARGO_CLI_COMMON_HPP
		No newline at end of file

cli/copy.cpp

+95 −34

Changes for cli/copy.cpp: 95 added lines, 34 removed lines.
Original line number	Diff line number	Diff line
		@@ -23,11 +23,17 @@
		*****************************************************************************/

		#include <fmt/format.h>
		#include <cargo.hpp>
		#include <fmt_formatters.hpp>
		#include <fmt/chrono.h>
		#include <cargo/cargo.hpp>
		#include <cargo/fmt_formatters.hpp>
		#include "../src/parallel_request.hpp" // Include for request_status formatter
		#include <filesystem>
		#include <CLI/CLI.hpp>
		#include <ranges>
		#include <thread>
		#include <cstdio> // For fflush

		#include "common.hpp"

		enum class dataset_flags { posix, parallel, none, gekkofs, hercules, expand, dataclay };

		@@ -47,8 +53,49 @@ struct copy_config {
		cargo::dataset::type input_flags = cargo::dataset::type::posix;
		std::vector<std::filesystem::path> outputs;
		cargo::dataset::type output_flags = cargo::dataset::type::posix;
		bool show_progress = false;
		bool dry_run = false;
		};

		// This is a free function now, defined in the global namespace or a utility namespace if preferred
		std::string format_bytes(std::size_t bytes) {
		if (bytes < 1024) return fmt::format("{} B", bytes);
		double kb = bytes / 1024.0;
		if (kb < 1024.0) return fmt::format("{:.2f} KB", kb);
		double mb = kb / 1024.0;
		if (mb < 1024.0) return fmt::format("{:.2f} MB", mb);
		double gb = mb / 1024.0;
		return fmt::format("{:.2f} GB", gb);
		}

		void display_progress(const cargo::transfer_status& st) {
		int bar_width = 50;
		float progress = 0.0f;
		if (st.total_bytes() > 0) {
		progress = static_cast<float>(st.bytes_transferred()) / st.total_bytes();
		}

		int pos = static_cast<int>(bar_width * progress);

		fmt::print(stderr, "\r[");
		for (int i = 0; i < bar_width; ++i) {
		if (i < pos) fmt::print(stderr, "=");
		else if (i == pos) fmt::print(stderr, ">");
		else fmt::print(stderr, " ");
		}

		std::string rate_str;
		if (st.bw() > 0) {
		rate_str = fmt::format("{:.2f} MB/s", st.bw());
		} else {
		rate_str = "N/A";
		}

		fmt::print(stderr, "] {:3.0f}% ({}/s) ", progress * 100.0, rate_str);
		fflush(stderr);
		}


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

		@@ -57,40 +104,36 @@ parse_command_line(int argc, char* argv[]) {
		cfg.progname = std::filesystem::path{argv[0]}.filename().string();

		CLI::App app{"Cargo parallel copy tool", cfg.progname};
		app.formatter(std::make_shared<CLI::Formatter>());

		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)")
		"Address of the Cargo server.")
		->option_text("ADDRESS")
		->envname("CCP_SERVER")
		->required();

		app.add_option("-i,--input", cfg.inputs, "Input dataset(s)")
		app.add_option("-i,--input", cfg.inputs, "Input dataset(s).")
		->option_text("SRC...")
		->required();

		app.add_option("-o,--output", cfg.outputs, "Output dataset(s)")
		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"
		" - parallel: read data using MPI-IO\n"
		" - dataclay: read data using DATACLAY\n"
		" - gekkofs: read data using gekkofs user library\n")
		->option_text("FLAGS")
		app.add_flag("-p,--progress", cfg.show_progress, "Show transfer progress bar.");
		app.add_flag("--dry-run", cfg.dry_run, "Plan the transfer and report stats without executing.");

		std::string if_help = "Input dataset type. Accepted values:\n"
		" posix, parallel, gekkofs, hercules, expand, dataclay, none";
		app.add_option("--if", cfg.input_flags, if_help)
		->option_text("TYPE")
		->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"
		" - parallel: write data using MPI-IO\n"
		" - dataclay: write data using DATACLAY\n"
		" - gekkofs: write data using gekkofs user library\n")
		->option_text("FLAGS")
		std::string of_help = "Output dataset type. Accepted values:\n"
		" posix, parallel, gekkofs, hercules, expand, dataclay, none";
		app.add_option("--of", cfg.output_flags, of_help)
		->option_text("TYPE")
		->transform(CLI::CheckedTransformer(dataset_flags_map,
		CLI::ignore_case));

		@@ -102,17 +145,6 @@ parse_command_line(int argc, char* argv[]) {
		}
		}

		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[]) {

		@@ -137,13 +169,42 @@ main(int argc, char* argv[]) {
		tgt, cfg.output_flags};
		});

		if (cfg.dry_run) {
		const auto [file_count, total_size] = cargo::plan_transfer_datasets(server, inputs, outputs);
		fmt::print("Dry Run Plan:\n");
		fmt::print(" - Files to transfer: {}\n", file_count);
		fmt::print(" - Total data size: {}\n", format_bytes(total_size));
		return EXIT_SUCCESS;
		}

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

		if(const auto st = tx.wait(); st.failed()) {
		throw std::runtime_error(st.error().message());
		fmt::print("Started transfer with ID: {}\n", tx.id());

		cargo::transfer_status st = tx.status();
		if (cfg.show_progress) {
		st = tx.status();
		while(!st.done() && !st.failed()) {
		display_progress(st);
		std::this_thread::sleep_for(std::chrono::milliseconds(200));
		st = tx.status();
		}
		fmt::print(stderr, "\r\33[2K"); // Clear the progress bar line
		} else {
		st = tx.wait();
		}

		if(st.failed()) {
		fmt::print(stderr, "Transfer failed: {}\n", st.error().message());
		return EXIT_FAILURE;
		} else {
		fmt::print("Transfer completed successfully in {:.2f}s.\n",
		std::chrono::duration_cast<std::chrono::duration<double>>(st.elapsed_time()).count());
		}

		} catch(const std::exception& ex) {
		fmt::print(stderr, "{}: Error: {}\n", cfg.progname, ex.what());
		return EXIT_FAILURE;
		}
		return EXIT_SUCCESS;
		}
		No newline at end of file

cli/ftio.cpp

+10 −37

Changes for cli/ftio.cpp: 10 added lines, 37 removed lines.
Original line number	Diff line number	Diff line
		@@ -23,15 +23,15 @@
		*****************************************************************************/

		#include <fmt/format.h>
		#include <cargo.hpp>
		#include <cargo/cargo.hpp>
		#include <cargo/fmt_formatters.hpp>
		#include <filesystem>
		#include <CLI/CLI.hpp>
		#include <net/client.hpp>
		#include <net/endpoint.hpp>
		#include "common.hpp"

		struct ftio_config {
		std::string progname;
		std::string server_address;
		float confidence;
		float probability;
		float period;
		@@ -40,18 +40,14 @@ struct ftio_config {
		bool resume{false};
		};

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

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

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

		CLI::App app{"Cargo ftio client", cfg.progname};

		app.add_option("-s,--server", cfg.server_address, "Server address")
		->option_text("ADDRESS")
		->required();
		CLI::App app{"Cargo ftio client", progname};

		app.add_option("-c,--conf", cfg.confidence, "confidence")
		->option_text("float")
		@@ -75,33 +71,10 @@ parse_command_line(int argc, char* argv[]) {

		app.add_flag("--resume", cfg.resume, "Trigger stage operation to resume, only pause or resume will take into account. Others parameters not used.");

		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[]) {

		ftio_config cfg = parse_command_line(argc, argv);
		parse_rpc_command_line(argc, argv, app, server_address);

		try {
		const auto [protocol, address] = parse_address(cfg.server_address);
		const auto [protocol, address] = parse_address(server_address);
		network::client rpc_client{protocol};

		if(const auto result = rpc_client.lookup(address); result.has_value()) {

Rnou/49 refactor some functions

Merge request reports