Merge remote-tracking branch 'upstream/branch-0.32' into increase-pyt…

…hon-thread-benchmark-timeout
rapidsai · Apr 27, 2023 · a463a77 · a463a77
2 parents 0c1f101 + 4b8d835
commit a463a77
Show file tree

Hide file tree

Showing 17 changed files with 54 additions and 175 deletions.
diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml
@@ -28,7 +28,7 @@ concurrency:
 jobs:
   conda-cpp-build:
     secrets: inherit
-    uses: rapidsai/shared-action-workflows/.github/workflows/conda-cpp-build.yaml@branch-23.06
+    uses: rapidsai/shared-action-workflows/.github/workflows/conda-cpp-build.yaml@py-39
     with:
       build_type: ${{ inputs.build_type || 'branch' }}
       branch: ${{ inputs.branch }}
@@ -38,7 +38,7 @@ jobs:
   upload-conda:
     needs: [conda-cpp-build]
     secrets: inherit
-    uses: rapidsai/shared-action-workflows/.github/workflows/conda-upload-packages.yaml@branch-23.06
+    uses: rapidsai/shared-action-workflows/.github/workflows/conda-upload-packages.yaml@py-39
     with:
       build_type: ${{ inputs.build_type || 'branch' }}
       branch: ${{ inputs.branch }}

diff --git a/.github/workflows/pr.yaml b/.github/workflows/pr.yaml
@@ -17,30 +17,30 @@ jobs:
       - conda-cpp-tests
       - conda-python-tests
     secrets: inherit
-    uses: rapidsai/shared-action-workflows/.github/workflows/pr-builder.yaml@branch-23.06
+    uses: rapidsai/shared-action-workflows/.github/workflows/pr-builder.yaml@py-39
   checks:
     secrets: inherit
-    uses: rapidsai/shared-action-workflows/.github/workflows/checks.yaml@branch-23.06
+    uses: rapidsai/shared-action-workflows/.github/workflows/checks.yaml@py-39
     with:
       enable_check_generated_files: false
   conda-cpp-build:
     needs: checks
     secrets: inherit
-    uses: rapidsai/shared-action-workflows/.github/workflows/conda-cpp-build.yaml@branch-23.06
+    uses: rapidsai/shared-action-workflows/.github/workflows/conda-cpp-build.yaml@py-39
     with:
       build_type: pull-request
       matrix_filter: map(select(.ARCH != "arm64"))
   conda-cpp-tests:
     needs: conda-cpp-build
     secrets: inherit
-    uses: rapidsai/shared-action-workflows/.github/workflows/conda-cpp-tests.yaml@branch-23.06
+    uses: rapidsai/shared-action-workflows/.github/workflows/conda-cpp-tests.yaml@py-39
     with:
       build_type: pull-request
       matrix_filter: map(select(.ARCH != "arm64"))
   conda-python-tests:
     needs: conda-cpp-build
     secrets: inherit
-    uses: rapidsai/shared-action-workflows/.github/workflows/conda-python-tests.yaml@branch-23.06
+    uses: rapidsai/shared-action-workflows/.github/workflows/conda-python-tests.yaml@py-39
     with:
       build_type: pull-request
       matrix_filter: map(select(.ARCH != "arm64"))
diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml
@@ -16,7 +16,7 @@ on:
 jobs:
   conda-cpp-tests:
     secrets: inherit
-    uses: rapidsai/shared-action-workflows/.github/workflows/conda-cpp-tests.yaml@branch-23.06
+    uses: rapidsai/shared-action-workflows/.github/workflows/conda-cpp-tests.yaml@py-39
     with:
       build_type: nightly
       branch: ${{ inputs.branch }}
@@ -25,7 +25,7 @@ jobs:
       matrix_filter: map(select(.ARCH != "arm64"))
   conda-python-tests:
     secrets: inherit
-    uses: rapidsai/shared-action-workflows/.github/workflows/conda-python-tests.yaml@branch-23.06
+    uses: rapidsai/shared-action-workflows/.github/workflows/conda-python-tests.yaml@py-39
     with:
       build_type: nightly
       branch: ${{ inputs.branch }}

diff --git a/conda/environments/ucxx-cuda118_arch-x86_64.yml b/conda/environments/ucxx-cuda118_arch-x86_64.yml
@@ -4,7 +4,7 @@ channels:
   - nvidia
 dependencies:
 # Base
-  - python=3.8
+  - python=3.9
   - cudatoolkit=11.8
   - pip
 # RAPIDS

diff --git a/conda/recipes/ucxx/conda_build_config.yaml b/conda/recipes/ucxx/conda_build_config.yaml
@@ -11,5 +11,5 @@ cmake:
   - ">=3.23.1,!=3.25.0"
 
 python:
-  - 3.8
+  - 3.9
   - 3.10
diff --git a/cpp/benchmarks/perftest.cpp b/cpp/benchmarks/perftest.cpp
@@ -203,7 +203,7 @@ std::function<void()> getProgressFunction(std::shared_ptr<ucxx::Worker> worker,
   switch (progressMode) {
     case ProgressMode::Polling: return std::bind(std::mem_fn(&ucxx::Worker::progress), worker);
     case ProgressMode::Blocking:
-      return std::bind(std::mem_fn(&ucxx::Worker::progressWorkerEvent), worker);
+      return std::bind(std::mem_fn(&ucxx::Worker::progressWorkerEvent), worker, -1);
     case ProgressMode::Wait: return std::bind(std::mem_fn(&ucxx::Worker::waitProgress), worker);
     default: return []() {};
   }

diff --git a/cpp/examples/basic.cpp b/cpp/examples/basic.cpp
@@ -142,7 +142,7 @@ std::function<void()> getProgressFunction(std::shared_ptr<ucxx::Worker> worker,
   switch (progressMode) {
     case ProgressMode::Polling: return std::bind(std::mem_fn(&ucxx::Worker::progress), worker);
     case ProgressMode::Blocking:
-      return std::bind(std::mem_fn(&ucxx::Worker::progressWorkerEvent), worker);
+      return std::bind(std::mem_fn(&ucxx::Worker::progressWorkerEvent), worker, -1);
     case ProgressMode::Wait: return std::bind(std::mem_fn(&ucxx::Worker::waitProgress), worker);
     default: return []() {};
   }

diff --git a/cpp/include/ucxx/worker.h b/cpp/include/ucxx/worker.h
@@ -220,9 +220,9 @@ class Worker : public Component {
    * @brief Progress worker event while in blocking progress mode.
    *
    * Blocks until a new worker event has happened and the worker notifies the file descriptor
-   * associated with it. Requires blocking progress mode to be initialized with
-   * `initBlockingProgressMode()` before the first call to this method. Additionally ensure
-   * inflight messages pending for cancelation are canceled.
+   * associated with it, or `epollTimeout` has elapsed. Requires blocking progress mode to
+   * be initialized with `initBlockingProgressMode()` before the first call to this method.
+   * Additionally ensure inflight messages pending for cancelation are canceled.
    *
    * @code{.cpp}
    * // worker is `std::shared_ptr<ucxx::Worker>`
@@ -235,10 +235,15 @@ class Worker : public Component {
    * // All events have been progressed.
    * @endcode
    *
+   * @param[in] epollTimeout  timeout in ms when waiting for worker event, or -1 to block
+   *                          indefinitely.
+   *
    * @throws std::ios_base::failure if creating any of the file descriptors or setting their
    *                                statuses.
+   *
+   * @returns `true` if any communication was progressed, `false` otherwise.
    */
-  bool progressWorkerEvent();
+  bool progressWorkerEvent(const int epollTimeout = -1);
 
   /**
    * @brief Signal the worker that an event happened.
@@ -442,10 +447,11 @@ class Worker : public Component {
    * when worker events happen, or in polling mode by continuously calling `progress()`
    * (incurs in high CPU utilization).
    *
-   * @param[in] pollingMode use polling mode if `true`, or blocking mode if `false`.
-   * @param[in] callbackArg argument to be passed to the callback function
+   * @param[in] pollingMode   use polling mode if `true`, or blocking mode if `false`.
+   * @param[in] epollTimeout  timeout in ms when waiting for worker event, or -1 to block
+   *                          indefinitely, only applicable if `pollingMode==true`.
    */
-  void startProgressThread(const bool pollingMode = false);
+  void startProgressThread(const bool pollingMode = false, const int epollTimeout = 1);
 
   /**
    * @brief Stop the progress thread.

diff --git a/cpp/src/worker.cpp b/cpp/src/worker.cpp
@@ -151,7 +151,7 @@ bool Worker::arm()
   return true;
 }
 
-bool Worker::progressWorkerEvent()
+bool Worker::progressWorkerEvent(const int epollTimeout)
 {
   int ret;
   epoll_event ev;
@@ -163,7 +163,7 @@ bool Worker::progressWorkerEvent()
   if ((_epollFileDescriptor == -1) || !arm()) return false;
 
   do {
-    ret = epoll_wait(_epollFileDescriptor, &ev, 1, -1);
+    ret = epoll_wait(_epollFileDescriptor, &ev, 1, epollTimeout);
   } while ((ret == -1) && (errno == EINTR || errno == EAGAIN));
 
   return false;
@@ -244,18 +244,23 @@ void Worker::setProgressThreadStartCallback(std::function<void(void*)> callback,
   _progressThreadStartCallbackArg = callbackArg;
 }
 
-void Worker::startProgressThread(const bool pollingMode)
+void Worker::startProgressThread(const bool pollingMode, const int epollTimeout)
 {
   if (_progressThread) {
     ucxx_warn("Worker progress thread already running");
     return;
   }
 
-  if (!pollingMode) initBlockingProgressMode();
-  auto progressFunction = pollingMode ? std::bind(&Worker::progress, this)
-                                      : std::bind(&Worker::progressWorkerEvent, this);
-  auto signalWorkerFunction =
-    pollingMode ? std::function<void()>{[]() {}} : std::bind(&Worker::signal, this);
+  std::function<bool()> progressFunction;
+  std::function<void()> signalWorkerFunction;
+  if (pollingMode) {
+    progressFunction     = [this]() { return this->progress(); };
+    signalWorkerFunction = []() {};
+  } else {
+    initBlockingProgressMode();
+    progressFunction = [this, epollTimeout]() { return this->progressWorkerEvent(epollTimeout); };
+    signalWorkerFunction = [this]() { return this->signal(); };
+  }
 
   _progressThread = std::make_shared<WorkerProgressThread>(pollingMode,
                                                            progressFunction,

diff --git a/cpp/tests/utils.cpp b/cpp/tests/utils.cpp
@@ -43,7 +43,7 @@ std::function<void()> getProgressFunction(std::shared_ptr<ucxx::Worker> worker,
   if (progressMode == ProgressMode::Polling)
     return std::bind(std::mem_fn(&ucxx::Worker::progress), worker);
   else if (progressMode == ProgressMode::Blocking)
-    return std::bind(std::mem_fn(&ucxx::Worker::progressWorkerEvent), worker);
+    return std::bind(std::mem_fn(&ucxx::Worker::progressWorkerEvent), worker, -1);
   else if (progressMode == ProgressMode::Wait)
     return std::bind(std::mem_fn(&ucxx::Worker::waitProgress), worker);
   else

diff --git a/dependencies.yaml b/dependencies.yaml
@@ -77,10 +77,6 @@ dependencies:
     specific:
       - output_types: conda
         matrices:
-          - matrix:
-              py: "3.8"
-            packages:
-              - python=3.8
           - matrix:
               py: "3.9"
             packages:
@@ -91,7 +87,7 @@ dependencies:
               - python=3.10
           - matrix:
             packages:
-              - python>=3.8,<3.11
+              - python>=3.9,<3.11
   run_python:
     common:
       - output_types: [conda, requirements, pyproject]

diff --git a/docker/README.md b/docker/README.md
diff --git a/docker/conda-build.dockerfile b/docker/conda-build.dockerfile
diff --git a/docker/conda-build.sh b/docker/conda-build.sh
diff --git a/python/pyproject.toml b/python/pyproject.toml
@@ -22,7 +22,7 @@ authors = [
     { name = "NVIDIA Corporation" },
 ]
 license = { text = "BSD-3-Clause" }
-requires-python = ">=3.8"
+requires-python = ">=3.9"
 dependencies = [
     "numpy>=1.21",
     "pynvml>=11.4.1",

diff --git a/python/ucxx/_lib/libucxx.pyx b/python/ucxx/_lib/libucxx.pyx
@@ -500,13 +500,17 @@ cdef class UCXWorker():
 
         return progress_made
 
-    def progress_worker_event(self):
+    def progress_worker_event(self, epoll_timeout=-1):
+        cdef int ucxx_epoll_timeout = epoll_timeout
+
         with nogil:
-            self._worker.get().progressWorkerEvent()
+            self._worker.get().progressWorkerEvent(ucxx_epoll_timeout)
+
+    def start_progress_thread(self, bint polling_mode=False, epoll_timeout=-1):
+        cdef int ucxx_epoll_timeout = epoll_timeout
 
-    def start_progress_thread(self, bint polling_mode=False):
         with nogil:
-            self._worker.get().startProgressThread(polling_mode)
+            self._worker.get().startProgressThread(polling_mode, epoll_timeout=ucxx_epoll_timeout)
 
     def stop_progress_thread(self):
         with nogil: