request.h

 
#pragma once
 
#include <atomic>
#include <chrono>
#include <memory>
#include <string>
#include <utility>
 
#include <ucp/api/ucp.h>
 
#include <ucxx/component.h>
#include <ucxx/endpoint.h>
#include <ucxx/future.h>
#include <ucxx/request_data.h>
#include <ucxx/typedefs.h>
 
#define ucxx_trace_req_f(_owner, _req, _handle, _name, _message, ...)          \
  ucxx_trace_req("ucxx::Request: %p on %s, UCP handle: %p, op: %s, " _message, \
                 (_req),                                                       \
                 (_owner),                                                     \
                 (_handle),                                                    \
                 (_name),                                                      \
                 ##__VA_ARGS__)
 
namespace ucxx {
 
class Request : public Component {
 protected:
  struct Attributes {
    ucs_memory_type memoryType{UCS_MEMORY_TYPE_UNKNOWN};  
    std::string debugString{};                            
  };
 
  ucs_status_t _status{UCS_INPROGRESS};      
  std::string _status_msg{};                 
  void* _request{nullptr};                   
  std::shared_ptr<Future> _future{nullptr};  
  std::shared_ptr<Worker> _worker{
    nullptr};  
  std::shared_ptr<Endpoint> _endpoint{
    nullptr};  
  std::string _ownerString{
    "undetermined owner"};           
  std::recursive_mutex _mutex{};     
  data::RequestData _requestData{};  
  std::string _operationName{
    "request_undefined"};          
  bool _enablePythonFuture{true};  
  RequestCallbackUserFunction _callback{nullptr};  
  RequestCallbackUserData _callbackData{nullptr};  
  Attributes _requestAttr{};  
 
  Request(std::shared_ptr<Component> endpointOrWorker,
          const data::RequestData requestData,
          std::string operationName,
          const bool enablePythonFuture                = false,
          RequestCallbackUserFunction callbackFunction = nullptr,
          RequestCallbackUserData callbackData         = nullptr);
 
  void process();
 
  void setStatus(ucs_status_t status);
 
 public:
  Request()                          = delete;
  Request(const Request&)            = delete;
  Request& operator=(Request const&) = delete;
  Request(Request&& o)               = delete;
  Request& operator=(Request&& o)    = delete;
 
  virtual ~Request();
 
  virtual void cancel();
 
  [[nodiscard]] ucs_status_t getStatus();
 
  [[nodiscard]] void* getFuture();
 
  void checkError();
 
  [[nodiscard]] bool isCompleted();
 
  void callback(void* request, ucs_status_t status);
 
  void populateDelayedSubmission();
 
  [[nodiscard]] const std::string& getOwnerString() const;
 
  [[nodiscard]] virtual std::shared_ptr<Buffer> getRecvBuffer();
 
  [[nodiscard]] virtual std::string getRecvHeader();
 
  [[nodiscard]] Attributes queryAttributes();
 
 protected:
  virtual void populateDelayedSubmissionImpl() = 0;
 
  void publishRequest(void* request);
};
 
}  // namespace ucxx