All Classes Namespaces Functions Variables Typedefs Enumerations Friends
endpoint.h
1 
5 #pragma once
6 
7 #include <netdb.h>
8 
9 #include <memory>
10 #include <string>
11 #include <vector>
12 
13 #include <ucp/api/ucp.h>
14 
15 #include <ucxx/address.h>
16 #include <ucxx/component.h>
17 #include <ucxx/exception.h>
18 #include <ucxx/inflight_requests.h>
19 #include <ucxx/listener.h>
20 #include <ucxx/request.h>
21 #include <ucxx/typedefs.h>
22 #include <ucxx/utils/sockaddr.h>
23 #include <ucxx/worker.h>
24 
25 namespace ucxx {
26 
41  void operator()(ucp_ep_params_t* ptr);
42 };
43 
50 class Endpoint : public Component {
51  private:
52  ucp_ep_h _handle{nullptr};
53  ucp_ep_h _originalHandle{nullptr};
55  bool _endpointErrorHandling{true};
56  std::unique_ptr<InflightRequests> _inflightRequests{
57  std::make_unique<InflightRequests>()};
58  std::mutex _mutex{};
61  ucs_status_t _status{UCS_INPROGRESS};
62  std::atomic<bool> _closing{false};
63  EndpointCloseCallbackUserFunction _closeCallback{nullptr};
64  EndpointCloseCallbackUserData _closeCallbackArg{
65  nullptr};
66 
91  Endpoint(std::shared_ptr<Component> workerOrListener, bool endpointErrorHandling);
92 
102  void create(ucp_ep_params_t* params);
103 
115  [[nodiscard]] std::shared_ptr<Request> registerInflightRequest(std::shared_ptr<Request> request);
116 
127  friend void endpointErrorCallback(void* arg, ucp_ep_h ep, ucs_status_t status);
128 
129  public:
130  Endpoint() = delete;
131  Endpoint(const Endpoint&) = delete;
132  Endpoint& operator=(Endpoint const&) = delete;
133  Endpoint(Endpoint&& o) = delete;
134  Endpoint& operator=(Endpoint&& o) = delete;
135 
136  ~Endpoint();
137 
160  friend std::shared_ptr<Endpoint> createEndpointFromHostname(std::shared_ptr<Worker> worker,
161  std::string ipAddress,
162  uint16_t port,
163  bool endpointErrorHandling);
164 
187  friend std::shared_ptr<Endpoint> createEndpointFromConnRequest(std::shared_ptr<Listener> listener,
188  ucp_conn_request_h connRequest,
189  bool endpointErrorHandling);
190 
210  friend std::shared_ptr<Endpoint> createEndpointFromWorkerAddress(std::shared_ptr<Worker> worker,
211  std::shared_ptr<Address> address,
212  bool endpointErrorHandling);
213 
229  [[nodiscard]] ucp_ep_h getHandle();
230 
242  [[nodiscard]] bool isAlive() const;
243 
253  void raiseOnError();
254 
267  void removeInflightRequest(const Request* const request);
268 
280 
290  [[nodiscard]] size_t getCancelingSize() const;
291 
313  size_t cancelInflightRequestsBlocking(uint64_t period = 0, uint64_t maxAttempts = 1);
314 
333  EndpointCloseCallbackUserData closeCallbackArg);
334 
366  [[nodiscard]] std::shared_ptr<Request> amSend(
367  void* buffer,
368  const size_t length,
369  const ucs_memory_type_t memoryType,
370  const std::optional<AmReceiverCallbackInfo> receiverCallbackInfo = std::nullopt,
371  const bool enablePythonFuture = false,
372  RequestCallbackUserFunction callbackFunction = nullptr,
373  RequestCallbackUserData callbackData = nullptr);
374 
396  [[nodiscard]] std::shared_ptr<Request> amRecv(
397  const bool enablePythonFuture = false,
398  RequestCallbackUserFunction callbackFunction = nullptr,
399  RequestCallbackUserData callbackData = nullptr);
400 
423  [[nodiscard]] std::shared_ptr<Request> memPut(
424  void* buffer,
425  size_t length,
426  uint64_t remote_addr,
427  ucp_rkey_h rkey,
428  const bool enablePythonFuture = false,
429  RequestCallbackUserFunction callbackFunction = nullptr,
430  RequestCallbackUserData callbackData = nullptr);
431 
456  [[nodiscard]] std::shared_ptr<Request> memPut(
457  void* buffer,
458  size_t length,
459  std::shared_ptr<ucxx::RemoteKey> remoteKey,
460  uint64_t remoteAddrOffset = 0,
461  const bool enablePythonFuture = false,
462  RequestCallbackUserFunction callbackFunction = nullptr,
463  RequestCallbackUserData callbackData = nullptr);
464 
487  [[nodiscard]] std::shared_ptr<Request> memGet(
488  void* buffer,
489  size_t length,
490  uint64_t remoteAddr,
491  ucp_rkey_h rkey,
492  const bool enablePythonFuture = false,
493  RequestCallbackUserFunction callbackFunction = nullptr,
494  RequestCallbackUserData callbackData = nullptr);
495 
520  [[nodiscard]] std::shared_ptr<Request> memGet(
521  void* buffer,
522  size_t length,
523  std::shared_ptr<ucxx::RemoteKey> remoteKey,
524  uint64_t remoteAddrOffset = 0,
525  const bool enablePythonFuture = false,
526  RequestCallbackUserFunction callbackFunction = nullptr,
527  RequestCallbackUserData callbackData = nullptr);
528 
548  [[nodiscard]] std::shared_ptr<Request> streamSend(void* buffer,
549  size_t length,
550  const bool enablePythonFuture);
551 
572  [[nodiscard]] std::shared_ptr<Request> streamRecv(void* buffer,
573  size_t length,
574  const bool enablePythonFuture);
575 
598  [[nodiscard]] std::shared_ptr<Request> tagSend(
599  void* buffer,
600  size_t length,
601  Tag tag,
602  const bool enablePythonFuture = false,
603  RequestCallbackUserFunction callbackFunction = nullptr,
604  RequestCallbackUserData callbackData = nullptr);
605 
630  [[nodiscard]] std::shared_ptr<Request> tagRecv(
631  void* buffer,
632  size_t length,
633  Tag tag,
634  TagMask tagMask,
635  const bool enablePythonFuture = false,
636  RequestCallbackUserFunction callbackFunction = nullptr,
637  RequestCallbackUserData callbackData = nullptr);
638 
672  [[nodiscard]] std::shared_ptr<Request> tagMultiSend(const std::vector<void*>& buffer,
673  const std::vector<size_t>& size,
674  const std::vector<int>& isCUDA,
675  const Tag tag,
676  const bool enablePythonFuture);
677 
700  [[nodiscard]] std::shared_ptr<Request> tagMultiRecv(const Tag tag,
701  const TagMask tagMask,
702  const bool enablePythonFuture);
703 
725  [[nodiscard]] std::shared_ptr<Request> flush(
726  const bool enablePythonFuture = false,
727  RequestCallbackUserFunction callbackFunction = nullptr,
728  RequestCallbackUserData callbackData = nullptr);
729 
740  [[nodiscard]] std::shared_ptr<Worker> getWorker();
741 
782  [[nodiscard]] std::shared_ptr<Request> close(
783  const bool enablePythonFuture = false,
784  EndpointCloseCallbackUserFunction callbackFunction = nullptr,
785  EndpointCloseCallbackUserData callbackData = nullptr);
786 
810  void closeBlocking(uint64_t period = 0, uint64_t maxAttempts = 1);
811 };
812 
813 } // namespace ucxx
A UCXX component class to prevent early destruction of parent object.
Definition: component.h:17
Component encapsulating a UCP endpoint.
Definition: endpoint.h:50
std::shared_ptr< Request > memPut(void *buffer, size_t length, uint64_t remote_addr, ucp_rkey_h rkey, const bool enablePythonFuture=false, RequestCallbackUserFunction callbackFunction=nullptr, RequestCallbackUserData callbackData=nullptr)
Enqueue a memory put operation.
std::shared_ptr< Request > flush(const bool enablePythonFuture=false, RequestCallbackUserFunction callbackFunction=nullptr, RequestCallbackUserData callbackData=nullptr)
Enqueue a flush operation.
std::shared_ptr< Request > close(const bool enablePythonFuture=false, EndpointCloseCallbackUserFunction callbackFunction=nullptr, EndpointCloseCallbackUserData callbackData=nullptr)
Enqueue a non-blocking endpoint close operation.
bool isAlive() const
Check whether the endpoint is still alive.
size_t cancelInflightRequests()
Cancel inflight requests.
std::shared_ptr< Request > tagSend(void *buffer, size_t length, Tag tag, const bool enablePythonFuture=false, RequestCallbackUserFunction callbackFunction=nullptr, RequestCallbackUserData callbackData=nullptr)
Enqueue a tag send operation.
std::shared_ptr< Request > streamRecv(void *buffer, size_t length, const bool enablePythonFuture)
Enqueue a stream receive operation.
void removeInflightRequest(const Request *const request)
Remove reference to request from internal container.
size_t cancelInflightRequestsBlocking(uint64_t period=0, uint64_t maxAttempts=1)
Cancel inflight requests.
std::shared_ptr< Request > tagMultiRecv(const Tag tag, const TagMask tagMask, const bool enablePythonFuture)
Enqueue a multi-buffer tag receive operation.
std::shared_ptr< Request > amRecv(const bool enablePythonFuture=false, RequestCallbackUserFunction callbackFunction=nullptr, RequestCallbackUserData callbackData=nullptr)
Enqueue an active message receive operation.
void setCloseCallback(EndpointCloseCallbackUserFunction closeCallback, EndpointCloseCallbackUserData closeCallbackArg)
Register a user-defined callback to call when endpoint closes.
friend std::shared_ptr< Endpoint > createEndpointFromWorkerAddress(std::shared_ptr< Worker > worker, std::shared_ptr< Address > address, bool endpointErrorHandling)
Constructor for shared_ptr<ucxx::Endpoint>.
std::shared_ptr< Request > memGet(void *buffer, size_t length, std::shared_ptr< ucxx::RemoteKey > remoteKey, uint64_t remoteAddrOffset=0, const bool enablePythonFuture=false, RequestCallbackUserFunction callbackFunction=nullptr, RequestCallbackUserData callbackData=nullptr)
Enqueue a memory get operation.
std::shared_ptr< Worker > getWorker()
Get ucxx::Worker component from a worker or listener object.
std::shared_ptr< Request > tagMultiSend(const std::vector< void * > &buffer, const std::vector< size_t > &size, const std::vector< int > &isCUDA, const Tag tag, const bool enablePythonFuture)
Enqueue a multi-buffer tag send operation.
std::shared_ptr< Request > streamSend(void *buffer, size_t length, const bool enablePythonFuture)
Enqueue a stream send operation.
friend void endpointErrorCallback(void *arg, ucp_ep_h ep, ucs_status_t status)
The error callback registered at endpoint creation time.
ucp_ep_h getHandle()
Get the underlying ucp_ep_h handle.
std::shared_ptr< Request > amSend(void *buffer, const size_t length, const ucs_memory_type_t memoryType, const std::optional< AmReceiverCallbackInfo > receiverCallbackInfo=std::nullopt, const bool enablePythonFuture=false, RequestCallbackUserFunction callbackFunction=nullptr, RequestCallbackUserData callbackData=nullptr)
Enqueue an active message send operation.
size_t getCancelingSize() const
Check the number of inflight requests being canceled.
void raiseOnError()
Raises an exception if an error occurred.
std::shared_ptr< Request > memGet(void *buffer, size_t length, uint64_t remoteAddr, ucp_rkey_h rkey, const bool enablePythonFuture=false, RequestCallbackUserFunction callbackFunction=nullptr, RequestCallbackUserData callbackData=nullptr)
Enqueue a memory get operation.
std::shared_ptr< Request > tagRecv(void *buffer, size_t length, Tag tag, TagMask tagMask, const bool enablePythonFuture=false, RequestCallbackUserFunction callbackFunction=nullptr, RequestCallbackUserData callbackData=nullptr)
Enqueue a tag receive operation.
friend std::shared_ptr< Endpoint > createEndpointFromHostname(std::shared_ptr< Worker > worker, std::string ipAddress, uint16_t port, bool endpointErrorHandling)
Constructor for shared_ptr<ucxx::Endpoint>.
std::shared_ptr< Request > memPut(void *buffer, size_t length, std::shared_ptr< ucxx::RemoteKey > remoteKey, uint64_t remoteAddrOffset=0, const bool enablePythonFuture=false, RequestCallbackUserFunction callbackFunction=nullptr, RequestCallbackUserData callbackData=nullptr)
Enqueue a memory put operation.
friend std::shared_ptr< Endpoint > createEndpointFromConnRequest(std::shared_ptr< Listener > listener, ucp_conn_request_h connRequest, bool endpointErrorHandling)
Constructor for shared_ptr<ucxx::Endpoint>.
void closeBlocking(uint64_t period=0, uint64_t maxAttempts=1)
Close the endpoint while keeping the object alive.
Base type for a UCXX transfer request.
Definition: request.h:38
Definition: address.h:15
std::function< void(ucs_status_t, std::shared_ptr< void >)> RequestCallbackUserFunction
A user-defined function to execute as part of a ucxx::Request callback.
Definition: typedefs.h:103
std::shared_ptr< void > RequestCallbackUserData
Data for the user-defined function provided to the ucxx::Request callback.
Definition: typedefs.h:111
RequestCallbackUserData EndpointCloseCallbackUserData
Data for the user-defined function provided to endpoint close callback.
Definition: typedefs.h:127
RequestCallbackUserFunction EndpointCloseCallbackUserFunction
A user-defined function to execute after an endpoint closes.
Definition: typedefs.h:119
TagMask
Strong type for a UCP tag mask.
Definition: typedefs.h:66
Tag
Strong type for a UCP tag.
Definition: typedefs.h:58
Deleter for a endpoint parameters object.
Definition: endpoint.h:33
void operator()(ucp_ep_params_t *ptr)
Execute the deletion.