Add PythonFutureTask

.github/CODEOWNERS

@@ -1,21 +1,22 @@
 #cpp code owners
-cpp/                              @rapidsai/ucxx-cpp-codeowners
+cpp/                                @rapidsai/ucxx-cpp-codeowners
 
 #python code owners
-python/                           @rapidsai/ucxx-python-codeowners
+python/                             @rapidsai/ucxx-python-codeowners
 
 #cmake code owners
-cpp/CMakeLists.txt                @rapidsai/ucxx-cmake-codeowners
-cpp/python/CMakeLists.txt         @rapidsai/ucxx-cmake-codeowners
-cpp/examples/CMakeLists.txt       @rapidsai/ucxx-cmake-codeowners
-cpp/benchmarks/CMakeLists.txt     @rapidsai/ucxx-cmake-codeowners
-cpp/tests/CMakeLists.txt          @rapidsai/ucxx-cmake-codeowners
-python/CMakeLists.txt             @rapidsai/ucxx-cmake-codeowners
-python/ucxx/_lib/CMakeLists.txt   @rapidsai/ucxx-cmake-codeowners
-fetch_rapids.cmake                @rapidsai/ucxx-cmake-codeowners
-**/cmake/                         @rapidsai/ucxx-cmake-codeowners
+cpp/CMakeLists.txt                  @rapidsai/ucxx-cmake-codeowners
+cpp/python/CMakeLists.txt           @rapidsai/ucxx-cmake-codeowners
+cpp/examples/CMakeLists.txt         @rapidsai/ucxx-cmake-codeowners
+cpp/benchmarks/CMakeLists.txt       @rapidsai/ucxx-cmake-codeowners
+cpp/tests/CMakeLists.txt            @rapidsai/ucxx-cmake-codeowners
+python/CMakeLists.txt               @rapidsai/ucxx-cmake-codeowners
+python/ucxx/examples/CMakeLists.txt @rapidsai/ucxx-cmake-codeowners
+python/ucxx/_lib/CMakeLists.txt     @rapidsai/ucxx-cmake-codeowners
+fetch_rapids.cmake                  @rapidsai/ucxx-cmake-codeowners
+**/cmake/                           @rapidsai/ucxx-cmake-codeowners
 
 #build/ops code owners
-.github/                          @rapidsai/ops-codeowners
-ci/                               @rapidsai/ops-codeowners
-conda/                            @rapidsai/ops-codeowners
+.github/                            @rapidsai/ops-codeowners
+ci/                                 @rapidsai/ops-codeowners
+conda/                              @rapidsai/ops-codeowners

ci/test_python.sh

@@ -132,3 +132,6 @@ run_distributed_ucxx_tests      thread          0                           0
 run_distributed_ucxx_tests      thread          0                           1
 run_distributed_ucxx_tests      thread          1                           0
 run_distributed_ucxx_tests      thread          1                           1
+
+rapids-logger "C++ future -> Python future notifier example"
+python -m ucxx.examples.python_future_task_example

cpp/python/CMakeLists.txt

@@ -18,6 +18,7 @@ add_library(
   src/future.cpp
   src/notifier.cpp
   src/python_future.cpp
+  src/python_future_task_collector.cpp
   src/worker.cpp
 )
 

cpp/python/include/ucxx/python/api.h

@@ -8,4 +8,5 @@
 #include <ucxx/python/exception.h>
 #include <ucxx/python/notifier.h>
 #include <ucxx/python/python_future.h>
+#include <ucxx/python/python_future_task_collector.h>
 #include <ucxx/python/worker.h>

cpp/python/include/ucxx/python/constructors.h

@@ -19,6 +19,9 @@ namespace python {
 
 std::shared_ptr<::ucxx::Future> createFuture(std::shared_ptr<::ucxx::Notifier> notifier);
 
+std::shared_ptr<::ucxx::Future> createFutureWithEventLoop(
+  PyObject* asyncioEventLoop, std::shared_ptr<::ucxx::Notifier> notifier);
+
 std::shared_ptr<::ucxx::Notifier> createNotifier();
 
 std::shared_ptr<::ucxx::Worker> createWorker(std::shared_ptr<ucxx::Context> context,

cpp/python/include/ucxx/python/future.h

@@ -21,7 +21,7 @@ namespace python {
  *
  * @returns The Python asyncio future object.
  */
-PyObject* create_python_future();
+PyObject* create_python_future() noexcept;
 
 /**
  * @brief Set the result of a Python future.
@@ -37,7 +37,7 @@ PyObject* create_python_future();
  *
  * @returns The result of the call to `_asyncio.Future.set_result()`.
  */
-PyObject* future_set_result(PyObject* future, PyObject* value);
+PyObject* future_set_result(PyObject* future, PyObject* value) noexcept;
 
 /**
  * @brief Set the exception of a Python future.
@@ -53,7 +53,61 @@ PyObject* future_set_result(PyObject* future, PyObject* value);
  *
  * @returns The result of the call to `_asyncio.Future.set_result()`.
  */
-PyObject* future_set_exception(PyObject* future, PyObject* exception, const char* message);
+PyObject* future_set_exception(PyObject* future, PyObject* exception, const char* message) noexcept;
+
+/**
+ * @brief Create a Python asyncio future with associated event loop.
+ *
+ * Create Python asyncio future associated with the event loop passed via the `event_loop`
+ * argument, effectively equal to calling `loop.create_future()` directly in Python.
+ *
+ * Note that this call will take the Python GIL and requires that the current thread have
+ * an asynchronous event loop set.
+ *
+ * @param[in] event_loop  the Python asyncio event loop to which the future will belong to.
+ *
+ * @returns The Python asyncio future object.
+ */
+PyObject* create_python_future_with_event_loop(PyObject* event_loop) noexcept;
+
+/**
+ * @brief Set the result of a Python future with associated event loop.
+ *
+ * Schedule setting the result of a Python future in the given event loop using the
+ * threadsafe method `event_loop.call_soon_threadsafe`. The event loop given must be the
+ * same specified when creating the future object with `create_python_future`.
+ *
+ * Note that this may be called from any thread and will take the Python GIL to run.
+ *
+ * @param[in] future  Python object containing the `_asyncio.Future` object.
+ * @param[in] value   Python object containing an arbitrary value to set the future result
+ *                    to.
+ *
+ * @returns The result of the call to `_asyncio.Future.set_result()`.
+ */
+PyObject* future_set_result_with_event_loop(PyObject* event_loop,
+                                            PyObject* future,
+                                            PyObject* value) noexcept;
+
+/**
+ * @brief Set the exception of a Python future with associated event loop.
+ *
+ * Schedule setting an exception of a Python future in the given event loop using the
+ * threadsafe method `event_loop.call_soon_threadsafe`. The event loop given must be the
+ * same specified when creating the future object with `create_python_future`.
+ *
+ * Note that this may be called from any thread and will take the Python GIL to run.
+ *
+ * @param[in] future    Python object containing the `_asyncio.Future` object.
+ * @param[in] exception a Python exception derived of the `Exception` class.
+ * @param[in] message   human-readable error message for the exception.
+ *
+ * @returns The result of the call to `_asyncio.Future.set_result()`.
+ */
+PyObject* future_set_exception_with_event_loop(PyObject* event_loop,
+                                               PyObject* future,
+                                               PyObject* exception,
+                                               const char* message) noexcept;
 
 }  // namespace python
 

cpp/python/include/ucxx/python/python_future.h

@@ -21,7 +21,8 @@ namespace python {
 
 class Future : public ::ucxx::Future {
  private:
-  PyObject* _handle{create_python_future()};  ///< The handle to the Python future
+  PyObject* _asyncioEventLoop{nullptr};  ///< The asyncio event loop the Python future belongs to.
+  PyObject* _handle{nullptr};            ///< The handle to the Python future
 
   /**
    * @brief Construct a future that may be notified from a notifier thread.
@@ -32,9 +33,12 @@ class Future : public ::ucxx::Future {
    * This class may also be used to set the result or exception from any thread, but that
    * currently requires explicitly taking the GIL before calling `set()`.
    *
+   * @param[in] asyncioEventLoop pointer to a valid Python object containing the event loop
+   *                             that the application is using, to which the future will
+   *                             belong to.
    * @param[in] notifier  notifier object running on a separate thread.
    */
-  explicit Future(std::shared_ptr<::ucxx::Notifier> notifier);
+  explicit Future(PyObject* asyncioEventLoop, std::shared_ptr<::ucxx::Notifier> notifier);
 
  public:
   Future()                         = delete;
@@ -56,6 +60,23 @@ class Future : public ::ucxx::Future {
    */
   friend std::shared_ptr<::ucxx::Future> createFuture(std::shared_ptr<::ucxx::Notifier> notifier);
 
+  /**
+   * @brief Constructor of `shared_ptr<ucxx::python::Future>`.
+   *
+   * The constructor for a `shared_ptr<ucxx::python::Future>` object. The default
+   * constructor is made private to ensure all UCXX objects are shared pointers and correct
+   * lifetime management.
+   *
+   * @param[in] asyncioEventLoop  pointer to a valid Python object containing the event loop
+   *                              that the application is using, to which the future will
+   *                              belong to.
+   * @param[in] notifier          notifier object running on a separate thread.
+   *
+   * @returns The `shared_ptr<ucxx::python::Worker>` object
+   */
+  friend std::shared_ptr<::ucxx::Future> createFutureWithEventLoop(
+    PyObject* asyncioEventLoop, std::shared_ptr<::ucxx::Notifier> notifier);
+
   /**
    * @brief Virtual destructor.
    *