single.hpp

 
#pragma once
 
#include <cstdlib>
#include <memory>
#include <string>
#include <unordered_map>
#include <vector>
 
#include <rapidsmpf/communicator/communicator.hpp>
#include <rapidsmpf/config.hpp>
#include <rapidsmpf/memory/buffer.hpp>
#include <rapidsmpf/memory/buffer_resource.hpp>
#include <rapidsmpf/progress_thread.hpp>
 
namespace rapidsmpf {
 
class Single final : public Communicator {
  public:
    class Future : public Communicator::Future {
        friend class Single;
 
      public:
        ~Future() noexcept override = default;
    };
 
    Single(config::Options options, std::shared_ptr<ProgressThread> progress_thread);
 
    ~Single() noexcept override = default;
 
    [[nodiscard]] constexpr Rank rank() const override {
        return 0;
    }
 
    [[nodiscard]] constexpr Rank nranks() const override {
        return 1;
    }
 
    [[nodiscard]] std::unique_ptr<Communicator::Future> send(
        std::unique_ptr<std::vector<std::uint8_t>> msg, Rank rank, Tag tag
    ) override;
 
    // clang-format off
    // clang-format on
    [[nodiscard]] std::unique_ptr<Communicator::Future> send(
        std::unique_ptr<Buffer> msg, Rank rank, Tag tag
    ) override;
 
    [[nodiscard]] std::unique_ptr<Communicator::Future> recv(
        Rank rank, Tag tag, std::unique_ptr<Buffer> recv_buffer
    ) override;
 
    // clang-format off
    // clang-format on
    [[nodiscard]] std::unique_ptr<Communicator::Future> recv_sync_host_data(
        Rank rank, Tag tag, std::unique_ptr<std::vector<std::uint8_t>> synced_buffer
    ) override;
 
    [[nodiscard]] std::pair<std::unique_ptr<std::vector<std::uint8_t>>, Rank> recv_any(
        Tag tag
    ) override;
 
    [[nodiscard]] std::unique_ptr<std::vector<std::uint8_t>> recv_from(
        Rank src, Tag tag
    ) override;
 
    std::pair<
        std::vector<std::unique_ptr<Communicator::Future>>,
        std::vector<std::size_t>>
    test_some(std::vector<std::unique_ptr<Communicator::Future>>& future_vector) override;
 
    // clang-format off
    // clang-format on
    std::vector<std::size_t> test_some(
        std::unordered_map<std::size_t, std::unique_ptr<Communicator::Future>> const&
            future_map
    ) override;
 
    bool test(std::unique_ptr<Communicator::Future>& future) override;
    std::vector<std::unique_ptr<Buffer>> wait_all(
        std::vector<std::unique_ptr<Communicator::Future>>&& futures
    ) override;
 
    [[nodiscard]] std::unique_ptr<Buffer> wait(
        std::unique_ptr<Communicator::Future> future
    ) override;
 
    [[nodiscard]] std::unique_ptr<Buffer> release_data(
        std::unique_ptr<Communicator::Future> future
    ) override;
 
    [[nodiscard]] std::unique_ptr<std::vector<std::uint8_t>> release_sync_host_data(
        std::unique_ptr<Communicator::Future> future
    ) override;
 
    [[nodiscard]] std::shared_ptr<Communicator::Logger> const& logger() override {
        return logger_;
    }
 
    [[nodiscard]] std::shared_ptr<ProgressThread> const&
    progress_thread() const override {
        return progress_thread_;
    }
 
    [[nodiscard]] std::string str() const override;
 
  private:
    std::shared_ptr<Logger> logger_;
    std::shared_ptr<ProgressThread> progress_thread_;
};
 
 
}  // namespace rapidsmpf