gdbsupport/parallel-for.h

   1 /* Parallel for loops
   2
   3    Copyright (C) 2019-2023 Free Software Foundation, Inc.
   4
   5    This file is part of GDB.
   6
   7    This program is free software; you can redistribute it and/or modify
   8    it under the terms of the GNU General Public License as published by
   9    the Free Software Foundation; either version 3 of the License, or
  10    (at your option) any later version.
  11
  12    This program is distributed in the hope that it will be useful,
  13    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15    GNU General Public License for more details.
  16
  17    You should have received a copy of the GNU General Public License
  18    along with this program.  If not, see <http://www.gnu.org/licenses/>.  */
  19
  20 #ifndef GDBSUPPORT_PARALLEL_FOR_H
  21 #define GDBSUPPORT_PARALLEL_FOR_H
  22
  23 #include <algorithm>
  24 #include <type_traits>
  25 #include "gdbsupport/invoke-result.h"
  26 #include "gdbsupport/thread-pool.h"
  27 #include "gdbsupport/function-view.h"
  28
  29 namespace gdb
  30 {
  31
  32 namespace detail
  33 {
  34
  35 /* This is a helper class that is used to accumulate results for
  36    parallel_for.  There is a specialization for 'void', below.  */
  37 template<typename T>
  38 struct par_for_accumulator
  39 {
  40 public:
  41
  42   explicit par_for_accumulator (size_t n_threads)
  43     : m_futures (n_threads)
  44   {
  45   }
  46
  47   /* The result type that is accumulated.  */
  48   typedef std::vector<T> result_type;
  49
  50   /* Post the Ith task to a background thread, and store a future for
  51      later.  */
  52   void post (size_t i, std::function<T ()> task)
  53   {
  54     m_futures[i]
  55       = gdb::thread_pool::g_thread_pool->post_task (std::move (task));
  56   }
  57
  58   /* Invoke TASK in the current thread, then compute all the results
  59      from all background tasks and put them into a result vector,
  60      which is returned.  */
  61   result_type finish (gdb::function_view<T ()> task)
  62   {
  63     result_type result (m_futures.size () + 1);
  64
  65     result.back () = task ();
  66
  67     for (size_t i = 0; i < m_futures.size (); ++i)
  68       result[i] = m_futures[i].get ();
  69
  70     return result;
  71   }
  72
  73   /* Resize the results to N.  */
  74   void resize (size_t n)
  75   {
  76     m_futures.resize (n);
  77   }
  78
  79 private:
  80
  81   /* A vector of futures coming from the tasks run in the
  82      background.  */
  83   std::vector<gdb::future<T>> m_futures;
  84 };
  85
  86 /* See the generic template.  */
  87 template<>
  88 struct par_for_accumulator<void>
  89 {
  90 public:
  91
  92   explicit par_for_accumulator (size_t n_threads)
  93     : m_futures (n_threads)
  94   {
  95   }
  96
  97   /* This specialization does not compute results.  */
  98   typedef void result_type;
  99
 100   void post (size_t i, std::function<void ()> task)
 101   {
 102     m_futures[i]
 103       = gdb::thread_pool::g_thread_pool->post_task (std::move (task));
 104   }
 105
 106   result_type finish (gdb::function_view<void ()> task)
 107   {
 108     task ();
 109
 110     for (auto &future : m_futures)
 111       {
 112         /* Use 'get' and not 'wait', to propagate any exception.  */
 113         future.get ();
 114       }
 115   }
 116
 117   /* Resize the results to N.  */
 118   void resize (size_t n)
 119   {
 120     m_futures.resize (n);
 121   }
 122
 123 private:
 124
 125   std::vector<gdb::future<void>> m_futures;
 126 };
 127
 128 }
 129
 130 /* A very simple "parallel for".  This splits the range of iterators
 131    into subranges, and then passes each subrange to the callback.  The
 132    work may or may not be done in separate threads.
 133
 134    This approach was chosen over having the callback work on single
 135    items because it makes it simple for the caller to do
 136    once-per-subrange initialization and destruction.
 137
 138    The parameter N says how batching ought to be done -- there will be
 139    at least N elements processed per thread.  Setting N to 0 is not
 140    allowed.
 141
 142    If the function returns a non-void type, then a vector of the
 143    results is returned.  The size of the resulting vector depends on
 144    the number of threads that were used.  */
 145
 146 template<class RandomIt, class RangeFunction>
 147 typename gdb::detail::par_for_accumulator<
 148     typename gdb::invoke_result<RangeFunction, RandomIt, RandomIt>::type
 149   >::result_type
 150 parallel_for_each (unsigned n, RandomIt first, RandomIt last,
 151                    RangeFunction callback,
 152                    gdb::function_view<size_t(RandomIt)> task_size = nullptr)
 153 {
 154   using result_type
 155     = typename gdb::invoke_result<RangeFunction, RandomIt, RandomIt>::type;
 156
 157   /* If enabled, print debug info about how the work is distributed across
 158      the threads.  */
 159   const bool parallel_for_each_debug = false;
 160
 161   size_t n_worker_threads = thread_pool::g_thread_pool->thread_count ();
 162   size_t n_threads = n_worker_threads;
 163   size_t n_elements = last - first;
 164   size_t elts_per_thread = 0;
 165   size_t elts_left_over = 0;
 166   size_t total_size = 0;
 167   size_t size_per_thread = 0;
 168   size_t max_element_size = n_elements == 0 ? 1 : SIZE_MAX / n_elements;
 169
 170   if (n_threads > 1)
 171     {
 172       if (task_size != nullptr)
 173         {
 174           gdb_assert (n == 1);
 175           for (RandomIt i = first; i != last; ++i)
 176             {
 177               size_t element_size = task_size (i);
 178               gdb_assert (element_size > 0);
 179               if (element_size > max_element_size)
 180                 /* We could start scaling here, but that doesn't seem to be
 181                    worth the effort.  */
 182                 element_size = max_element_size;
 183               size_t prev_total_size = total_size;
 184               total_size += element_size;
 185               /* Check for overflow.  */
 186               gdb_assert (prev_total_size < total_size);
 187             }
 188           size_per_thread = total_size / n_threads;
 189         }
 190       else
 191         {
 192           /* Require that there should be at least N elements in a
 193              thread.  */
 194           gdb_assert (n > 0);
 195           if (n_elements / n_threads < n)
 196             n_threads = std::max (n_elements / n, (size_t) 1);
 197           elts_per_thread = n_elements / n_threads;
 198           elts_left_over = n_elements % n_threads;
 199           /* n_elements == n_threads * elts_per_thread + elts_left_over. */
 200         }
 201     }
 202
 203   size_t count = n_threads == 0 ? 0 : n_threads - 1;
 204   gdb::detail::par_for_accumulator<result_type> results (count);
 205
 206   if (parallel_for_each_debug)
 207     {
 208       debug_printf (_("Parallel for: n_elements: %zu\n"), n_elements);
 209       if (task_size != nullptr)
 210         {
 211           debug_printf (_("Parallel for: total_size: %zu\n"), total_size);
 212           debug_printf (_("Parallel for: size_per_thread: %zu\n"), size_per_thread);
 213         }
 214       else
 215         {
 216           debug_printf (_("Parallel for: minimum elements per thread: %u\n"), n);
 217           debug_printf (_("Parallel for: elts_per_thread: %zu\n"), elts_per_thread);
 218         }
 219     }
 220
 221   size_t remaining_size = total_size;
 222   for (int i = 0; i < count; ++i)
 223     {
 224       RandomIt end;
 225       size_t chunk_size = 0;
 226       if (task_size == nullptr)
 227         {
 228           end = first + elts_per_thread;
 229           if (i < elts_left_over)
 230             /* Distribute the leftovers over the worker threads, to avoid having
 231                to handle all of them in a single thread.  */
 232             end++;
 233         }
 234       else
 235         {
 236           RandomIt j;
 237           for (j = first; j < last && chunk_size < size_per_thread; ++j)
 238             {
 239               size_t element_size = task_size (j);
 240               if (element_size > max_element_size)
 241                 element_size = max_element_size;
 242               chunk_size += element_size;
 243             }
 244           end = j;
 245           remaining_size -= chunk_size;
 246         }
 247
 248       /* This case means we don't have enough elements to really
 249          distribute them.  Rather than ever submit a task that does
 250          nothing, we short-circuit here.  */
 251       if (first == end)
 252         end = last;
 253
 254       if (end == last)
 255         {
 256           /* We're about to dispatch the last batch of elements, which
 257              we normally process in the main thread.  So just truncate
 258              the result list here.  This avoids submitting empty tasks
 259              to the thread pool.  */
 260           count = i;
 261           results.resize (count);
 262           break;
 263         }
 264
 265       if (parallel_for_each_debug)
 266         {
 267           debug_printf (_("Parallel for: elements on worker thread %i\t: %zu"),
 268                         i, (size_t)(end - first));
 269           if (task_size != nullptr)
 270             debug_printf (_("\t(size: %zu)"), chunk_size);
 271           debug_printf (_("\n"));
 272         }
 273       results.post (i, [=] ()
 274         {
 275           return callback (first, end);
 276         });
 277       first = end;
 278     }
 279
 280   for (int i = count; i < n_worker_threads; ++i)
 281     if (parallel_for_each_debug)
 282       {
 283         debug_printf (_("Parallel for: elements on worker thread %i\t: 0"), i);
 284         if (task_size != nullptr)
 285           debug_printf (_("\t(size: 0)"));
 286         debug_printf (_("\n"));
 287       }
 288
 289   /* Process all the remaining elements in the main thread.  */
 290   if (parallel_for_each_debug)
 291     {
 292       debug_printf (_("Parallel for: elements on main thread\t\t: %zu"),
 293                     (size_t)(last - first));
 294       if (task_size != nullptr)
 295         debug_printf (_("\t(size: %zu)"), remaining_size);
 296       debug_printf (_("\n"));
 297     }
 298   return results.finish ([=] ()
 299     {
 300       return callback (first, last);
 301     });
 302 }
 303
 304 /* A sequential drop-in replacement of parallel_for_each.  This can be useful
 305    when debugging multi-threading behaviour, and you want to limit
 306    multi-threading in a fine-grained way.  */
 307
 308 template<class RandomIt, class RangeFunction>
 309 typename gdb::detail::par_for_accumulator<
 310     typename gdb::invoke_result<RangeFunction, RandomIt, RandomIt>::type
 311   >::result_type
 312 sequential_for_each (unsigned n, RandomIt first, RandomIt last,
 313                      RangeFunction callback,
 314                      gdb::function_view<size_t(RandomIt)> task_size = nullptr)
 315 {
 316   using result_type = typename gdb::invoke_result<RangeFunction, RandomIt, RandomIt>::type;
 317
 318   gdb::detail::par_for_accumulator<result_type> results (0);
 319
 320   /* Process all the remaining elements in the main thread.  */
 321   return results.finish ([=] ()
 322     {
 323       return callback (first, last);
 324     });
 325 }
 326
 327 }
 328
 329 #endif /* GDBSUPPORT_PARALLEL_FOR_H */