[QUESTION] parallel_for

[pfeatherstone]

Hi.
What is the recommend way of doing something like parallel_for in a fork join manner?
thank you

[tzcnt]

Using spawn_many with a range iota | transform, or spawn_group with a regular for loop. I've been meaning to put together a parallel_for API, but it'll end up being a wrapper over one of those anyway.

If you want to range over an index only, and have the task capture other references, then you need to avoid the dangling capture problem. There are 3 solutions:

1. Construct, dispatch, and await the tasks in the same expression so the lambda is lifetime-extended (as in the skynet example above).
2. Create the lambda as a named lvalue first so its lifetime is guaranteed to exceed that of the tasks.
3. Pass the indexes into an outer capturing non-coroutine lambda that returns the task from an inner non-capturing coroutine which takes everything by argument. That looks like lines 57-61 here

If you want to dispatch functions instead of coroutines then use spawn_func_many. The most complete example set is in test_spawn_many / test_spawn_func_many

Unless you're looking to fork/join from a synchronous context - then it's post_bulk_waitable.

[dixlorenz]

I don't like spawning possibly hundreds of thousands of equal tasks (feels like it would have quite some overhead being allocated, put in some queue, dequeued...) I use the following code to spawn (by default) 1 worker task per cpu and then use an atomic to get the next index to process. The callable can also take begin/end indices for chunk processing.

   namespace Detail {
      template <typename Process, typename... Args>
      concept SyncProcess =
         std::invocable<Process &, Args...> && std::same_as<std::invoke_result_t<Process &, Args...>, void>;

      template <typename Process, typename... Args>
      concept AsyncProcess = std::invocable<Process &, Args...> &&
                             tmc::traits::is_awaitable<std::invoke_result_t<Process &, Args...>>;

      template <typename Process, typename... Args>
      concept ParallelProcess = SyncProcess<Process, Args...> || AsyncProcess<Process, Args...>;

      template <typename Process>
      concept IndexProcess = ParallelProcess<Process, std::size_t>;

      template <typename Process>
      concept ChunkProcess = ParallelProcess<Process, std::size_t, std::size_t>;

      template <typename Process>
      concept ParallelForProcess = IndexProcess<Process> || ChunkProcess<Process>;
   } // namespace Detail

   struct ParallelForOptions {
      std::size_t num_tasks  = std::thread::hardware_concurrency();
      std::size_t chunk_size = 64;
   };

   template <ExecutorLike Executor, typename Process>
      requires Detail::ParallelForProcess<Process>
   tmc::task<void> parallel_for(
      Executor &&executor, std::size_t n, Process process, ParallelForOptions options = {})
   {
      if (n == 0)
         co_return;

      if (options.num_tasks == 0)
         options.num_tasks = 1;

      if (options.chunk_size == 0)
         options.chunk_size = 1;

      const std::size_t numTasks = std::min(options.num_tasks, n);
      const std::size_t chunk_size =
         numTasks > 1 ? std::min(options.chunk_size, std::max<std::size_t>(1, (n - 1) / (numTasks - 1)))
                      : options.chunk_size;

      auto sg = tmc::spawn_group();

      std::atomic<std::size_t> next{ 0 };

      for (std::size_t taskIndex = 0; taskIndex < numTasks; ++taskIndex) {
         auto worker = [](std::atomic<std::size_t> &next,
                          Process                  &process,
                          std::size_t               n,
                          std::size_t               chunk_size) -> tmc::task<void> {
            while (true) {
               const std::size_t begin = next.fetch_add(chunk_size, std::memory_order_relaxed);

               if (begin >= n)
                  break;

               const std::size_t end = std::min(begin + chunk_size, n);
               if constexpr (Detail::ChunkProcess<Process>) {
                  if constexpr (Detail::AsyncProcess<Process, std::size_t, std::size_t>)
                     co_await std::invoke(process, begin, end);
                  else
                     std::invoke(process, begin, end);
               } else {
                  for (std::size_t i = begin; i < end; ++i) {
                     if constexpr (Detail::AsyncProcess<Process, std::size_t>)
                        co_await std::invoke(process, i);
                     else
                        std::invoke(process, i);
                  }
               }
            }
            co_return;
         };

         sg.add(worker(next, process, n, chunk_size));
      }

      co_await std::move(sg).run_on(executor);
   }