Resolve "Increase transfer status to see the status of all the files"

Ramon Nourequested to merge
rnou/38-increase-transfer-status-to-see-the-status-of-all-the-files into main

Closes #38 (closed)

Merge request reports

lib/cargo.hpp

+24 −0
Changes for lib/cargo.hpp: 24 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -125,6 +125,15 @@ public: 
    status() const;





    /**

     * @brief Get all the statuses of the associated transfer.

     * 

     * @return std::vector<transfer_status> 

     */

    [[nodiscard]] std::vector<transfer_status>

    statuses() const;





    /**

     * @brief updates the bw control of the transfer

     *
@@ -165,9 +174,21 @@ class transfer_status { 
    friend transfer_status

    transfer::status() const;

    



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



public:



    transfer_status(std::string name, transfer_state status, float bw, error_code error) noexcept;

    

    /**

     * Get the name of the associated dataset.

     *

     * @return The name of the dataset.

     */

    [[nodiscard]] std::string

    name() const noexcept;



    /**

     * Get the current status of the associated transfer.

     *
@@ -207,11 +228,14 @@ public: 
    bw() const;



private:

    std::string m_name;

    transfer_state m_state;

    float m_bw;

    error_code m_error;

};







/**

 * Request the transfer of a dataset collection.

 *

lib/libcargo.cpp

+57 −1
Changes for lib/libcargo.cpp: 57 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -133,6 +133,53 @@ transfer::status() const { 
    throw std::runtime_error("rpc lookup failed");

}



std::vector<transfer_status>

transfer::statuses() const {

    using proto::statuses_response;



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

    const auto rpc =

            network::rpc_info::create("transfer_statuses", m_srv.address());



    using response_type =

            statuses_response<std::string, transfer_state, float, error_code>;



    if(const auto lookup_rv = rpc_client.lookup(m_srv.address());

       lookup_rv.has_value()) {

        const auto& endp = lookup_rv.value();



        LOGGER_INFO("rpc {:<} body: {{tid: {}}}", rpc, m_id);



        if(const auto call_rv = endp.call(rpc.name(), m_id);

           call_rv.has_value()) {



            const response_type resp{call_rv.value()};

            const auto& v = resp.value();



            LOGGER_EVAL(resp.error_code(), ERROR, INFO,

                        "rpc {:>} body: {{retval: {}}} [op_id: {}]", rpc,

                        resp.error_code(), resp.op_id());



            if(resp.error_code()) {

                throw std::runtime_error(

                        fmt::format("rpc call failed: {}", resp.error_code()));

            }

            // convert vector of tuples to vector of transfer_status

            // (for some reason it asks for a public constructor)



            std::vector<transfer_status> v_statuses;

            for(const auto& [name, s, bw, ec] : v) {

                v_statuses.emplace_back(transfer_status{

                        name, s, bw, ec.value_or(error_code::success)});

            }



            return v_statuses;

        }

    }



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

}



void

transfer::bw_control(std::int16_t bw_control) const {
@@ -170,13 +217,22 @@ transfer::bw_control(std::int16_t bw_control) const { 


transfer_status::transfer_status(transfer_state status, float bw,

                                 error_code error) noexcept

    : m_state(status), m_bw(bw), m_error(error) {}

    : m_name(""), m_state(status), m_bw(bw), m_error(error) {}



transfer_status::transfer_status(std::string name, transfer_state status,

                                 float bw, error_code error) noexcept

    : m_name(name), m_state(status), m_bw(bw), m_error(error) {}



transfer_state

transfer_status::state() const noexcept {

    return m_state;

}



std::string

transfer_status::name() const noexcept {

    return m_name;

}



bool

transfer_status::done() const noexcept {

    return m_state == transfer_state::completed;

spack/packages/cargo/package.py

+1 −1
Changes for spack/packages/cargo/package.py: 1 added line, 1 removed line.
Original line number Diff line number Diff line
@@ -39,7 +39,7 @@ class Cargo(CMakePackage): 
    version("0.3.1", sha256="613485354e24c4b97cb6d045657569f94dc1d9bbb391b5a166f8d18b3595428b")

    version("0.3.2", sha256="ceb6bcb738a35fb41f40b7b1cdd8a806d99995a227980e8ced61dd90418e5960")

    version("0.3.3", sha256="1c4ab215e41905cc359894fa1df9006be16730ddc37c5b1369a9ea759bcb61cd")

    version("0.3.4", branch="rnou/fallocate")

    

    # build variants

    variant('build_type',

            default='Release',

src/proto/mpi/message.hpp

+13 −6
Changes for src/proto/mpi/message.hpp: 13 added lines, 6 removed lines.
Original line number Diff line number Diff line
@@ -120,10 +120,10 @@ class status_message { 
public:

    status_message() = default;



    status_message(std::uint64_t tid, std::uint32_t seqno,

    status_message(std::uint64_t tid, std::uint32_t seqno, std::string name, 

                   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_bw(bw),

        : m_tid(tid), m_seqno(seqno), m_name(name), m_state(state), m_bw(bw),

          m_error_code(error_code) {}



    [[nodiscard]] std::uint64_t
@@ -136,6 +136,11 @@ public: 
        return m_seqno;

    }



    [[nodiscard]] const std::string&

    name() const {

        return m_name;

    }



    [[nodiscard]] cargo::transfer_state

    state() const {

        return m_state;
@@ -160,6 +165,7 @@ private: 


        ar& m_tid;

        ar& m_seqno;

        ar& m_name;

        ar& m_state;

        ar& m_bw;

        ar& m_error_code;
@@ -167,6 +173,7 @@ private: 


    std::uint64_t m_tid{};

    std::uint32_t m_seqno{};

    std::string m_name{};

    cargo::transfer_state m_state{};

    float m_bw{};

    std::optional<cargo::error_code> m_error_code{};
@@ -246,13 +253,13 @@ struct fmt::formatter<cargo::status_message> : formatter<std::string_view> { 
        const auto str =

                s.error_code()

                        ? fmt::format(

                                  "{{tid: {}, seqno: {}, state: {}, bw: {}, "

                                  "{{tid: {}, seqno: {}, name: {}, state: {}, bw: {}, "

                                  "error_code: {}}}",

                                  s.tid(), s.seqno(), s.state(), s.bw(),

                                  s.tid(), s.seqno(), s.name(), s.state(), s.bw(),

                                  *s.error_code())

                        : fmt::format(

                                  "{{tid: {}, seqno: {}, state: {}, bw: {}}}",

                                  s.tid(), s.seqno(), s.state(), s.bw());

                                  "{{tid: {}, seqno: {}, name: {}, state: {}, bw: {}}}",

                                  s.tid(), s.seqno(), s.name(), s.state(), s.bw());

        return formatter<std::string_view>::format(str, ctx);

    }

};

src/proto/rpc/response.hpp

+4 −0
Changes for src/proto/rpc/response.hpp: 4 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -109,6 +109,10 @@ using status_response = 
        response_with_value<std::tuple<Status, Bw, std::optional<Error>>,

                            Error>;



template <typename Name, typename Status, typename Bw, typename Error>

using statuses_response = response_with_value<

        std::vector<std::tuple<Name, Status, Bw, std::optional<Error>>>, Error>;



} // namespace cargo::proto



#endif // CARGO_PROTO_RPC_RESPONSE_HPP