request.h

 
#pragma once
 
#include <atomic>
#include <chrono>
#include <memory>
#include <string>
 
#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:
  ucs_status_t _status{UCS_INPROGRESS};            
  std::string _status_msg{};                       
  void* _request{nullptr};                         
  std::shared_ptr<Future> _future{nullptr};        
  RequestCallbackUserFunction _callback{nullptr};  
  RequestCallbackUserData _callbackData{nullptr};  
  std::shared_ptr<Worker> _worker{
    nullptr};  
  std::shared_ptr<Endpoint> _endpoint{
    nullptr};  
  std::string _ownerString{
    "undetermined owner"};           
  data::RequestData _requestData{};  
  std::string _operationName{
    "request_undefined"};          
  std::recursive_mutex _mutex{};   
  bool _enablePythonFuture{true};  
 
  Request(std::shared_ptr<Component> endpointOrWorker,
          const data::RequestData requestData,
          const std::string operationName,
          const bool enablePythonFuture = false);
 
  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();
 
  ucs_status_t getStatus();
 
  void* getFuture();
 
  void checkError();
 
  bool isCompleted();
 
  void callback(void* request, ucs_status_t status);
 
  virtual void populateDelayedSubmission() = 0;
 
  const std::string& getOwnerString() const;
 
  virtual std::shared_ptr<Buffer> getRecvBuffer();
};
 
}  // namespace ucxx