Handle null Cython close requests

python/ucxx/ucxx/_lib/libucxx.pxd

@@ -66,6 +66,7 @@ cdef class UCXRequest:
         bint _enable_python_future
         bint _completed
 
+    cdef Request* _request_ptr(self) except NULL
     cdef shared_ptr[Request] get_ucxx_shared_ptr(self) nogil
 
 

python/ucxx/ucxx/_lib/libucxx.pyx

@@ -947,17 +947,29 @@ cdef class UCXRequest():
         self._enable_python_future = enable_python_future
         self._completed = False
 
-    def __dealloc__(self) -> None:
+    cdef Request* _request_ptr(self) except NULL:
+        cdef Request* request
+
         with nogil:
-            self._request.get().cancel()
-            self._request.reset()
+            request = self._request.get()
 
-    @property
-    def ucxx_ptr(self) -> int:
+        if request == nullptr:
+            raise RuntimeError("UCXRequest does not hold a valid request")
+
+        return request
+
+    def __dealloc__(self) -> None:
         cdef Request* request
 
         with nogil:
             request = self._request.get()
+            if request != nullptr:
+                request.cancel()
+            self._request.reset()
+
+    @property
+    def ucxx_ptr(self) -> int:
+        cdef Request* request = self._request_ptr()
 
         return int(<uintptr_t>request)
 
@@ -967,40 +979,45 @@ cdef class UCXRequest():
     @property
     def completed(self) -> bool:
         cdef bint completed
+        cdef Request* request
 
         if self._completed is True:
             return True
 
+        request = self._request_ptr()
         with nogil:
-            completed = self._request.get().isCompleted()
+            completed = request.isCompleted()
 
         return completed
 
     @property
     def status(self) -> ucs_status_t:
         cdef ucs_status_t status
+        cdef Request* request = self._request_ptr()
 
         with nogil:
-            status = self._request.get().getStatus()
+            status = request.getStatus()
 
         return status
 
     @property
     def future(self) -> object:
         cdef PyObject* future_ptr
+        cdef Request* request = self._request_ptr()
 
         with nogil:
-            future_ptr = <PyObject*>self._request.get().getFuture()
+            future_ptr = <PyObject*>request.getFuture()
 
         return <object>future_ptr
 
     @property
     def recv_buffer(self) -> object:
         cdef shared_ptr[Buffer] buf
         cdef BufferType bufType
+        cdef Request* request = self._request_ptr()
 
         with nogil:
-            buf = self._request.get().getRecvBuffer()
+            buf = request.getRecvBuffer()
             bufType = buf.get().getType() if buf != nullptr else BufferType.Invalid
 
         # If buf == NULL, it's not allocated by the request but rather the user
@@ -1012,8 +1029,10 @@ cdef class UCXRequest():
             return _get_host_buffer(<uintptr_t><void*>buf.get())
 
     def check_error(self) -> None:
+        cdef Request* request = self._request_ptr()
+
         with nogil:
-            self._request.get().checkError()
+            request.checkError()
 
     async def wait_yield(self) -> None:
         while True:
@@ -1331,14 +1350,17 @@ cdef class UCXEndpoint():
 
         return alive
 
-    def close(self) -> None:
+    def close(self) -> Optional[UCXRequest]:
         cdef shared_ptr[Request] req
 
         with nogil:
             req = self._endpoint.get().close(
                 self._enable_python_future
             )
 
+        if req.get() == nullptr:
+            return None
+
         return UCXRequest(<uintptr_t><void*>&req, self._enable_python_future)
 
     def close_blocking(self, uint64_t period=0, uint64_t max_attempts=1) -> None:

python/ucxx/ucxx/_lib/tests/test_endpoint.py

@@ -3,6 +3,7 @@
 
 import multiprocessing as mp
 import os
+import time
 
 import pytest
 
@@ -118,3 +119,22 @@ def test_close_callback(server_close_callback):
     join_processes([client, server], timeout=10)
     terminate_process(client, error_queue=client_error_q)
     terminate_process(server, error_queue=server_error_q)
+
+
+def test_close_returns_none_if_already_closing():
+    ctx = ucx_api.UCXContext()
+    worker = ucx_api.UCXWorker(ctx)
+    ep = ucx_api.UCXEndpoint.create_from_worker_address(
+        worker, worker.address, endpoint_error_handling=True
+    )
+
+    close_request = ep.close()
+    assert close_request is not None
+    assert ep.close() is None
+
+    deadline = time.monotonic() + 5
+    while not close_request.completed and time.monotonic() < deadline:
+        worker.progress()
+
+    assert close_request.completed
+    close_request.check_error()