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