bw shaping workflow (!16) · Merge requests · hpc / cargo

cli/CMakeLists.txt

+18 −1

Changes for cli/CMakeLists.txt: 18 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -84,7 +84,24 @@ target_link_libraries(ccp
		cargo
		)

		install(TARGETS cargo_ping cargo_shutdown ccp
		################################################################################
		## 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
		)

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

cli/shaping.cpp

0 → 100644

+114 −0

Changes for cli/shaping.cpp: 114 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 <filesystem>
		#include <CLI/CLI.hpp>
		#include <net/client.hpp>
		#include <net/endpoint.hpp>

		struct shaping_config {
		std::string progname;
		std::string server_address;
		std::int64_t tid;
		std::int16_t shaping;
		};

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

		shaping_config cfg;

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

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

		app.add_option("-s,--server", cfg.server_address, "Server address")
		->option_text("ADDRESS")
		->required();

		app.add_option("-i,--tid", cfg.tid, "transfer id")
		->option_text("integer")
		->required();

		app.add_option("-b,--bw", cfg.shaping, "bw shaping")
		->option_text("integer")
		->required();


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

		shaping_config cfg = parse_command_line(argc, argv);

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

		if(const auto result = rpc_client.lookup(address); result.has_value()) {
		const auto& endpoint = result.value();
		const auto retval = endpoint.call("bw_shaping", cfg.tid, cfg.shaping);

		if(retval.has_value()) {

		auto error_code = int{retval.value()};

		fmt::print("bw_shaping RPC was successful!\n");
		fmt::print(" (server replied with: {})\n", error_code);
		return EXIT_SUCCESS;
		}

		fmt::print(stderr, "bw_shaping RPC failed\n");
		return EXIT_FAILURE;

		} else {
		fmt::print(stderr, "Failed to lookup address: {}\n", address);
		return EXIT_FAILURE;
		}
		} catch(const std::exception& ex) {
		fmt::print(stderr, "Error: {}\n", ex.what());
		return EXIT_FAILURE;
		}
		}

lib/cargo.hpp

+6 −1

Changes for lib/cargo.hpp: 6 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -26,6 +26,7 @@
		#ifndef CARGO_HPP
		#define CARGO_HPP

		#include <cstdint>
		#include <string>
		#include <vector>
		#include <chrono>
		@@ -149,7 +150,7 @@ class transfer_status {
		friend transfer_status
		transfer::status() const;

		transfer_status(transfer_state status, error_code error) noexcept;
		transfer_status(transfer_state status, float bw, error_code error) noexcept;

		public:
		/**
		@@ -187,8 +188,12 @@ public:
		[[nodiscard]] error_code
		error() const;

		[[nodiscard]] float
		bw() const;

		private:
		transfer_state m_state;
		float m_bw;
		error_code m_error;
		};

lib/libcargo.cpp

+10 −5

Changes for lib/libcargo.cpp: 10 added lines, 5 removed lines.
Original line number	Diff line number	Diff line
		@@ -93,7 +93,7 @@ transfer::status() const {
		network::client rpc_client{m_srv.protocol()};
		const auto rpc =
		network::rpc_info::create("transfer_status", m_srv.address());
		using response_type = status_response<transfer_state, error_code>;
		using response_type = status_response<transfer_state, float, error_code>;

		if(const auto lookup_rv = rpc_client.lookup(m_srv.address());
		lookup_rv.has_value()) {
		@@ -105,7 +105,7 @@ transfer::status() const {
		call_rv.has_value()) {

		const response_type resp{call_rv.value()};
		const auto& [s, ec] = resp.value();
		const auto& [s, bw, ec] = resp.value();

		LOGGER_EVAL(resp.error_code(), INFO, ERROR,
		"rpc {:>} body: {{retval: {}}} [op_id: {}]", rpc,
		@@ -116,16 +116,16 @@ transfer::status() const {
		fmt::format("rpc call failed: {}", resp.error_code()));
		}

		return transfer_status{s, ec.value_or(error_code::success)};
		return transfer_status{s, bw, ec.value_or(error_code::success)};
		}
		}

		throw std::runtime_error("rpc lookup failed");
		}

		transfer_status::transfer_status(transfer_state status,
		transfer_status::transfer_status(transfer_state status, float bw,
		error_code error) noexcept
		: m_state(status), m_error(error) {}
		: m_state(status), m_bw(bw), m_error(error) {}

		transfer_state
		transfer_status::state() const noexcept {
		@@ -142,6 +142,11 @@ transfer_status::failed() const noexcept {
		return m_state == transfer_state::failed;
		}

		float
		transfer_status::bw() const {
		return m_bw;
		}

		error_code
		transfer_status::error() const {
		switch(m_state) {

src/proto/mpi/message.hpp

+72 −8

Changes for src/proto/mpi/message.hpp: 72 added lines, 8 removed lines.
Original line number	Diff line number	Diff line
		@@ -35,7 +35,14 @@

		namespace cargo {

		enum class tag : int { pread, pwrite, sequential, status, shutdown };
		enum class tag : int {
		pread,
		pwrite,
		sequential,
		bw_shaping,
		status,
		shutdown
		};

		class transfer_message {

		@@ -95,10 +102,10 @@ public:
		status_message() = default;

		status_message(std::uint64_t tid, std::uint32_t seqno,
		cargo::transfer_state state,
		cargo::transfer_state state, float bw,
		std::optional<cargo::error_code> error_code = std::nullopt)
		: m_tid(tid), m_seqno(seqno), m_state(state), m_error_code(error_code) {
		}
		: m_tid(tid), m_seqno(seqno), m_state(state), m_bw(bw),
		m_error_code(error_code) {}

		[[nodiscard]] std::uint64_t
		tid() const {
		@@ -115,6 +122,12 @@ public:
		return m_state;
		}

		[[nodiscard]] float
		bw() const {
		return m_bw;
		}


		[[nodiscard]] std::optional<cargo::error_code>
		error_code() const {
		return m_error_code;
		@@ -129,15 +142,52 @@ private:
		ar& m_tid;
		ar& m_seqno;
		ar& m_state;
		ar& m_bw;
		ar& m_error_code;
		}

		std::uint64_t m_tid{};
		std::uint32_t m_seqno{};
		cargo::transfer_state m_state{};
		float m_bw{};
		std::optional<cargo::error_code> m_error_code{};
		};

		class shaper_message {

		friend class boost::serialization::access;

		public:
		shaper_message() = default;

		shaper_message(std::uint64_t tid, std::int16_t shaping)
		: m_tid(tid), m_shaping(shaping) {}

		[[nodiscard]] std::uint64_t
		tid() const {
		return m_tid;
		}

		[[nodiscard]] std::int16_t
		shaping() const {
		return m_shaping;
		}

		private:
		template <class Archive>
		void
		serialize(Archive& ar, const unsigned int version) {
		(void) version;

		ar& m_tid;
		ar& m_shaping;
		}

		std::uint64_t m_tid{};
		std::uint16_t m_shaping{};
		};


		class shutdown_message {

		friend class boost::serialization::access;
		@@ -176,12 +226,26 @@ struct fmt::formatter<cargo::status_message> : formatter<std::string_view> {
		format(const cargo::status_message& s, FormatContext& ctx) const {
		const auto str =
		s.error_code()
		? fmt::format("{{tid: {}, seqno: {}, state: {}, "
		? fmt::format(
		"{{tid: {}, seqno: {}, state: {}, bw: {}, "
		"error_code: {}}}",
		s.tid(), s.seqno(), s.state(),
		s.tid(), s.seqno(), s.state(), s.bw(),
		*s.error_code())
		: fmt::format("{{tid: {}, seqno: {}, state: {}}}",
		s.tid(), s.seqno(), s.state());
		: fmt::format(
		"{{tid: {}, seqno: {}, state: {}, bw: {}}}",
		s.tid(), s.seqno(), s.state(), s.bw());
		return formatter<std::string_view>::format(str, ctx);
		}
		};

		template <>
		struct fmt::formatter<cargo::shaper_message> : formatter<std::string_view> {
		// parse is inherited from formatter<string_view>.
		template <typename FormatContext>
		auto
		format(const cargo::shaper_message& s, FormatContext& ctx) const {
		const auto str =
		fmt::format("{{tid: {}, shaping: {}}}", s.tid(), s.shaping());
		return formatter<std::string_view>::format(str, ctx);
		}
		};

bw shaping workflow

Merge request reports