Upgrade libcoro dependency (#570)

The semaphore fixes we need are now in main. Another change is that the thread pool is now managed as a unique_ptr to avoid reference cycles.

Authors:
  - Lawrence Mitchell (https://github.com/wence-)

Approvers:
  - Mads R. B. Kristensen (https://github.com/madsbk)

URL: #570

Author: wence-

cmake/thirdparty/get_libcoro.cmake

@@ -14,11 +14,10 @@ function(find_and_configure_libcoro)
     GLOBAL_TARGETS libcoro
     BUILD_EXPORT_SET rapidsmpf-exports
     CPM_ARGS
-    GIT_REPOSITORY https://github.com/wence-/libcoro
-    # We need a version that includes https://github.com/jbaldwin/libcoro/pull/371,
-    # https://github.com/jbaldwin/libcoro/pull/384, https://github.com/jbaldwin/libcoro/pull/389,
-    # and https://github.com/jbaldwin/libcoro/pull/399
-    GIT_TAG 4a1f369b2f9c9131b842813fd1b50520a6af3a36
+    GIT_REPOSITORY https://github.com/jbaldwin/libcoro
+    # We need a version that includes all PRs up to https://github.com/jbaldwin/libcoro/pull/399 and
+    # https://github.com/jbaldwin/libcoro/pull/400
+    GIT_TAG 7e0ce982405fb26b6ca8af97f40a8eaa2b78c4fa
     GIT_SHALLOW FALSE
     OPTIONS "LIBCORO_FEATURE_NETWORKING OFF"
             "LIBCORO_EXTERNAL_DEPENDENCIES OFF"

cpp/include/rapidsmpf/streaming/core/channel.hpp

@@ -172,8 +172,8 @@ class Channel {
      * @param executor The thread pool used to process remaining messages.
      * @return A coroutine representing the completion of the shutdown drain.
      */
-    Node drain(std::shared_ptr<coro::thread_pool> executor) {
-        return rb_.shutdown_drain(std::move(executor));
+    Node drain(std::unique_ptr<coro::thread_pool>& executor) {
+        return rb_.shutdown_drain(executor);
     }
 
     /**

cpp/include/rapidsmpf/streaming/core/context.hpp

@@ -30,15 +30,15 @@ class Context {
      * @param options Configuration options.
      * @param comm Shared pointer to a communicator.
      * @param progress_thread Shared pointer to a progress thread.
-     * @param executor Shared pointer to a coroutine thread pool.
+     * @param executor Unique pointer to a coroutine thread pool.
      * @param br Shared pointer to a buffer resource.
      * @param statistics Shared pointer to a statistics collector.
      */
     Context(
         config::Options options,
         std::shared_ptr<Communicator> comm,
         std::shared_ptr<ProgressThread> progress_thread,
-        std::shared_ptr<coro::thread_pool> executor,
+        std::unique_ptr<coro::thread_pool> executor,
         BufferResource* br,
         std::shared_ptr<Statistics> statistics
     );
@@ -84,9 +84,9 @@ class Context {
     /**
      * @brief Returns the coroutine thread pool.
      *
-     * @return Shared pointer to the thread pool.
+     * @return Reference to unique pointer to the thread pool.
      */
-    std::shared_ptr<coro::thread_pool> executor();
+    std::unique_ptr<coro::thread_pool>& executor();
 
     /**
      * @brief Returns the buffer resource.
@@ -106,7 +106,7 @@ class Context {
     config::Options options_;
     std::shared_ptr<Communicator> comm_;
     std::shared_ptr<ProgressThread> progress_thread_;
-    std::shared_ptr<coro::thread_pool> executor_;
+    std::unique_ptr<coro::thread_pool> executor_;
     BufferResource* br_;
     std::shared_ptr<Statistics> statistics_;
 };

cpp/src/streaming/core/context.cpp

@@ -14,7 +14,7 @@ Context::Context(
     config::Options options,
     std::shared_ptr<Communicator> comm,
     std::shared_ptr<ProgressThread> progress_thread,
-    std::shared_ptr<coro::thread_pool> executor,
+    std::unique_ptr<coro::thread_pool> executor,
     BufferResource* br,
     std::shared_ptr<Statistics> statistics
 )
@@ -41,7 +41,7 @@ Context::Context(
           options,
           comm,
           std::make_shared<ProgressThread>(comm->logger(), statistics),
-          coro::thread_pool::make_shared(
+          coro::thread_pool::make_unique(
               coro::thread_pool::options{
                   .thread_count = options.get<std::uint32_t>(
                       "num_streaming_threads",
@@ -75,7 +75,7 @@ std::shared_ptr<ProgressThread> Context::progress_thread() {
     return progress_thread_;
 }
 
-std::shared_ptr<coro::thread_pool> Context::executor() {
+std::unique_ptr<coro::thread_pool>& Context::executor() {
     return executor_;
 }