request_tag_multi.h

 
#pragma once
 
#include <memory>
#include <mutex>
#include <string>
#include <vector>
 
#include <ucp/api/ucp.h>
 
#include <ucxx/buffer.h>
#include <ucxx/endpoint.h>
#include <ucxx/future.h>
#include <ucxx/request.h>
 
namespace ucxx {
 
class RequestTagMulti;
 
struct BufferRequest {
  std::shared_ptr<Request> request{nullptr};  
  std::shared_ptr<std::string> stringBuffer{nullptr};  
  std::shared_ptr<Buffer> buffer{nullptr};  
 
  BufferRequest();
  ~BufferRequest();
 
  BufferRequest(const BufferRequest&)            = delete;
  BufferRequest& operator=(BufferRequest const&) = delete;
  BufferRequest(BufferRequest&& o)               = delete;
  BufferRequest& operator=(BufferRequest&& o)    = delete;
};
 
typedef std::shared_ptr<BufferRequest> BufferRequestPtr;
 
class RequestTagMulti : public Request {
 private:
  size_t _totalFrames{0};  
  std::mutex
    _completedRequestsMutex{};   
  size_t _completedRequests{0};  
  ucs_status_t _finalStatus{
    UCS_OK};  
 
 public:
  std::vector<BufferRequestPtr> _bufferRequests{};  
  bool _isFilled{false};                            
 
 private:
  RequestTagMulti()                                  = delete;
  RequestTagMulti(const RequestTagMulti&)            = delete;
  RequestTagMulti& operator=(RequestTagMulti const&) = delete;
  RequestTagMulti(RequestTagMulti&& o)               = delete;
  RequestTagMulti& operator=(RequestTagMulti&& o)    = delete;
 
  RequestTagMulti(std::shared_ptr<Endpoint> endpoint,
                  const std::variant<data::TagMultiSend, data::TagMultiReceive> requestData,
                  const std::string operationName,
                  const bool enablePythonFuture);
 
  void recvFrames();
 
  void recvHeader();
 
  void send();
 
 public:
  [[nodiscard]] friend std::shared_ptr<RequestTagMulti> createRequestTagMulti(
    std::shared_ptr<Endpoint> endpoint,
    const std::variant<data::TagMultiSend, data::TagMultiReceive> requestData,
    const bool enablePythonFuture);
 
  virtual ~RequestTagMulti();
 
  void markCompleted(ucs_status_t status, RequestCallbackUserData request);
 
  void recvCallback(ucs_status_t status);
 
  void populateDelayedSubmission() override;
 
  void cancel() override;
};
 
typedef std::shared_ptr<RequestTagMulti> RequestTagMultiPtr;
 
}  // namespace ucxx