Reference documentation for deal.II version GIT c14369f203 2023-10-01 07:40:02+00:00
\(\newcommand{\dealvcentcolon}{\mathrel{\mathop{:}}}\) \(\newcommand{\dealcoloneq}{\dealvcentcolon\mathrel{\mkern-1.2mu}=}\) \(\newcommand{\jump}[1]{\left[\!\left[ #1 \right]\!\right]}\) \(\newcommand{\average}[1]{\left\{\!\left\{ #1 \right\}\!\right\}}\)
thread_management.h
Go to the documentation of this file.
1 // ---------------------------------------------------------------------
2 //
3 // Copyright (C) 2000 - 2023 by the deal.II authors
4 //
5 // This file is part of the deal.II library.
6 //
7 // The deal.II library is free software; you can use it, redistribute
8 // it, and/or modify it under the terms of the GNU Lesser General
9 // Public License as published by the Free Software Foundation; either
10 // version 2.1 of the License, or (at your option) any later version.
11 // The full text of the license can be found in the file LICENSE.md at
12 // the top level directory of deal.II.
13 //
14 // ---------------------------------------------------------------------
15 
16 #ifndef dealii_thread_management_h
17 #define dealii_thread_management_h
18 
19 
20 #include <deal.II/base/config.h>
21 
24 #include <deal.II/base/mutex.h>
26 
27 #include <atomic>
28 #include <functional>
29 #include <future>
30 #include <list>
31 #include <memory>
32 #include <thread>
33 #include <tuple>
34 #include <utility>
35 #include <vector>
36 
37 #ifdef DEAL_II_HAVE_CXX20
38 # include <concepts>
39 #endif
40 
41 
42 #ifdef DEAL_II_WITH_TBB
43 # include <tbb/task_group.h>
44 #endif
45 
47 
60 namespace Threads
61 {
76  template <typename ForwardIterator>
77  std::vector<std::pair<ForwardIterator, ForwardIterator>>
78  split_range(const ForwardIterator &begin,
79  const ForwardIterator &end,
80  const unsigned int n_intervals);
81 
90  std::vector<std::pair<unsigned int, unsigned int>>
91  split_interval(const unsigned int begin,
92  const unsigned int end,
93  const unsigned int n_intervals);
94 
104  namespace internal
105  {
121  [[noreturn]] void
122  handle_std_exception(const std::exception &exc);
123 
131  [[noreturn]] void
133  } // namespace internal
134 
139 } // namespace Threads
140 
141 /* ----------- implementation of functions in namespace Threads ---------- */
142 #ifndef DOXYGEN
143 namespace Threads
144 {
145  template <typename ForwardIterator>
146  std::vector<std::pair<ForwardIterator, ForwardIterator>>
147  split_range(const ForwardIterator &begin,
148  const ForwardIterator &end,
149  const unsigned int n_intervals)
150  {
151  using IteratorPair = std::pair<ForwardIterator, ForwardIterator>;
152 
153  // in non-multithreaded mode, we often have the case that this
154  // function is called with n_intervals==1, so have a shortcut here
155  // to handle that case efficiently
156 
157  if (n_intervals == 1)
158  return (std::vector<IteratorPair>(1, IteratorPair(begin, end)));
159 
160  // if more than one interval requested, do the full work
161  const unsigned int n_elements = std::distance(begin, end);
162  const unsigned int n_elements_per_interval = n_elements / n_intervals;
163  const unsigned int residual = n_elements % n_intervals;
164 
165  std::vector<IteratorPair> return_values(n_intervals);
166 
167  return_values[0].first = begin;
168  for (unsigned int i = 0; i < n_intervals; ++i)
169  {
170  if (i != n_intervals - 1)
171  {
172  return_values[i].second = return_values[i].first;
173  // note: the cast is performed to avoid a warning of gcc
174  // that in the library `dist>=0' is checked (dist has a
175  // template type, which here is unsigned if no cast is
176  // performed)
177  std::advance(return_values[i].second,
178  static_cast<signed int>(n_elements_per_interval));
179  // distribute residual in division equally among the first
180  // few subintervals
181  if (i < residual)
182  ++return_values[i].second;
183 
184  return_values[i + 1].first = return_values[i].second;
185  }
186  else
187  return_values[i].second = end;
188  }
189  return return_values;
190  }
191 } // namespace Threads
192 
193 #endif // DOXYGEN
194 
195 namespace Threads
196 {
197  namespace internal
198  {
217  template <typename RT>
219  {
220  private:
221  RT value;
223 
224  public:
225  using reference_type = RT &;
226 
227  inline return_value()
228  : value()
229  , value_is_initialized(false)
230  {}
231 
232  inline reference_type
233  get()
234  {
235  Assert(
237  ExcMessage(
238  "You cannot read the return value of a thread or task "
239  "if that value has not been set. This happens, for example, if "
240  "a task or thread threw an exception."));
241  return value;
242  }
243 
244  inline void
245  set(RT &&v)
246  {
247  value = std::move(v);
248  }
249 
255  inline void
256  set_from(std::future<RT> &v)
257  {
258  // Get the value from the std::future object. If the future holds
259  // an exception, then the assignment fails, we exit the function via the
260  // exception right away, and value_is_initialized is not set to true --
261  // that's something we can check later on.
262  value = std::move(v.get());
263  value_is_initialized = true;
264  }
265  };
266 
267 
287  template <typename RT>
288  struct return_value<RT &>
289  {
290  private:
291  RT *value;
293 
294  public:
295  using reference_type = RT &;
296 
297  inline return_value()
298  : value(nullptr)
299  , value_is_initialized(false)
300  {}
301 
302  inline reference_type
303  get() const
304  {
305  Assert(
306  value_is_initialized,
307  ExcMessage(
308  "You cannot read the return value of a thread or task "
309  "if that value has not been set. This happens, for example, if "
310  "a task or thread threw an exception."));
311  return *value;
312  }
313 
314  inline void
315  set(RT &v)
316  {
317  value = &v;
318  }
319 
325  inline void
326  set_from(std::future<RT &> &v)
327  {
328  // Get the value from the std::future object. If the future holds
329  // an exception, then the assignment fails, we exit the function via the
330  // exception right away, and value_is_initialized is not set to true --
331  // that's something we can check later on.
332  value = &v.get();
333  value_is_initialized = true;
334  }
335  };
336 
337 
356  template <>
357  struct return_value<void>
358  {
359  using reference_type = void;
360 
361  static inline void
362  get()
363  {}
364 
365 
372  inline void
373  set_from(std::future<void> &)
374  {}
375  };
376  } // namespace internal
377 
378 
379 
380  namespace internal
381  {
389  template <typename T>
391  {
392  static T
393  act(T &t)
394  {
395  return t;
396  }
397  };
398 
399 
400 
408  template <typename T>
409  struct maybe_make_ref<T &>
410  {
411  static std::reference_wrapper<T>
412  act(T &t)
413  {
414  return std::ref(t);
415  }
416  };
417 
418 
419 
426  template <typename RT, typename Function>
428  (std::invocable<Function> &&
429  std::convertible_to<std::invoke_result_t<Function>, RT>))
430  void evaluate_and_set_promise(Function &function, std::promise<RT> &promise)
431  {
432  promise.set_value(function());
433  }
434 
435 
445  template <typename Function>
446  DEAL_II_CXX20_REQUIRES((std::invocable<Function>))
448  std::promise<void> &promise)
449  {
450  function();
451  promise.set_value();
452  }
453  } // namespace internal
454 
455 
456 
483  template <typename RT = void>
484  class Task
485  {
486  public:
498  Task(const std::function<RT()> &function_object)
499  {
500  if (MultithreadInfo::n_threads() > 1)
501  {
502 #ifdef DEAL_II_WITH_TBB
503  // Create a promise object and from it extract a future that
504  // we can use to refer to the outcome of the task. For reasons
505  // explained below, we can't just create a std::promise object,
506  // but have to make do with a pointer to such an object.
507  std::unique_ptr<std::promise<RT>> promise =
508  std::make_unique<std::promise<RT>>();
509  task_data =
510  std::make_shared<TaskData>(std::move(promise->get_future()));
511 
512  // Then start the task, using a task_group object (for just this one
513  // task) that is associated with the TaskData object. Note that we
514  // have to *copy* the function object being executed so that it is
515  // guaranteed to live on the called thread as well -- the copying is
516  // facilitated by capturing the 'function_object' variable by value.
517  //
518  // We also have to *move* the promise object into the new task's
519  // memory space because promises can not be copied and we can't refer
520  // to it by reference because it's a local variable of the current
521  // (surrounding) function that may go out of scope before the promise
522  // is ultimately set. This leads to a conundrum: if we had just
523  // declared 'promise' as an object of type std::promise, then we could
524  // capture it in the lambda function via
525  // [..., promise=std::move(promise)]() {...}
526  // and set the promise in the body of the lambda. But setting a
527  // promise is a non-const operation on the promise, and so we would
528  // actually have to declare the lambda function as 'mutable' because
529  // by default, lambda captures are 'const'. That is, we would have
530  // to write
531  // [..., promise=std::move(promise)]() mutable {...}
532  // But this leads to other problems: It turns out that the
533  // tbb::task_group::run() function cannot take mutable lambdas as
534  // argument :-(
535  //
536  // We work around this issue by not declaring the 'promise' variable
537  // as an object of type std::promise, but as a pointer to such an
538  // object. This pointer we can move, and the *pointer* itself can
539  // be 'const' (meaning we can leave the lambda as non-mutable)
540  // even though we modify the object *pointed to*. One would think
541  // that a std::unique_ptr would be the right choice for this, but
542  // that's not true: the resulting lambda function can then be
543  // non-mutable, but the lambda function object is not copyable
544  // and at least some TBB variants require that as well. So
545  // instead we move the std::unique_ptr used above into a
546  // std::shared_ptr to be stored within the lambda function object.
547  task_data->task_group.run(
548  [function_object,
549  promise =
550  std::shared_ptr<std::promise<RT>>(std::move(promise))]() {
551  try
552  {
553  internal::evaluate_and_set_promise(function_object, *promise);
554  }
555  catch (...)
556  {
557  try
558  {
559  // store anything thrown in the promise
560  promise->set_exception(std::current_exception());
561  }
562  catch (...)
563  {
564  // set_exception() may throw too. But ignore this on
565  // the task.
566  }
567  }
568  });
569 
570 #else
571  // If no threading library is supported, just fall back onto C++11
572  // facilities. The problem with this is that the standard does
573  // not actually say what std::async should do. The first
574  // argument to that function can be std::launch::async or
575  // std::launch::deferred, or both. The *intent* of the standard's
576  // authors was probably that if one sets it to
577  // std::launch::async | std::launch::deferred,
578  // that the task is run in a thread pool. But at least as of
579  // 2021, GCC doesn't do that: It just runs it on a new thread.
580  // If one chooses std::launch::deferred, it runs the task on
581  // the same thread but only when one calls join() on the task's
582  // std::future object. In the former case, this leads to
583  // oversubscription, in the latter case to undersubscription of
584  // resources. We choose oversubscription here.
585  //
586  // The issue illustrates why relying on external libraries
587  // with task schedulers is the way to go.
588  task_data = std::make_shared<TaskData>(
589  std::async(std::launch::async | std::launch::deferred,
590  function_object));
591 #endif
592  }
593  else
594  {
595  // Only one thread allowed. So let the task run to completion
596  // and just emplace a 'ready' future.
597  //
598  // The design of std::promise/std::future is unclear, but it
599  // seems that the intent is to obtain the std::future before
600  // we set the std::promise. So create the TaskData object at
601  // the top and then run the task and set the returned
602  // value. Since everything here happens sequentially, it
603  // really doesn't matter in which order all of this is
604  // happening.
605  std::promise<RT> promise;
606  task_data = std::make_shared<TaskData>(promise.get_future());
607  try
608  {
609  internal::evaluate_and_set_promise(function_object, promise);
610  }
611  catch (...)
612  {
613  try
614  {
615  // store anything thrown in the promise
616  promise.set_exception(std::current_exception());
617  }
618  catch (...)
619  {
620  // set_exception() may throw too. But ignore this on
621  // the task.
622  }
623  }
624  }
625  }
626 
635  Task() = default;
636 
668  void
669  join() const
670  {
671  // Make sure we actually have a task that we can wait for.
673 
674  task_data->wait();
675  }
676 
689  bool
690  joinable() const
691  {
692  return (task_data != nullptr);
693  }
694 
695 
747  {
748  // Make sure we actually have a task that we can wait for.
750 
751  // Then return the promised object. If necessary, wait for the promise to
752  // be set.
753  return task_data->get();
754  }
755 
756 
766  "The current object is not associated with a task that "
767  "can be joined. It may have been detached, or you "
768  "may have already joined it in the past.");
770  private:
779  class TaskData
780  {
781  public:
786  TaskData(std::future<RT> &&future) noexcept
787  : future(std::move(future))
788  , task_has_finished(false)
789  {}
790 
795  TaskData(const TaskData &) = delete;
796 
801  TaskData(TaskData &&) = delete;
802 
807  TaskData &
808  operator=(const TaskData &) = delete;
809 
814  TaskData &
815  operator=(TaskData &&) = delete;
816 
824  ~TaskData() noexcept
825  {
826  // Explicitly wait for the results to be ready. This class stores
827  // a std::future object, and we could just let the compiler generate
828  // the destructor which would then call the destructor of std::future
829  // which *may* block until the future is ready. As explained in
830  // https://en.cppreference.com/w/cpp/thread/future/~future
831  // this is only a *may*, not a *must*. (The standard does not
832  // appear to say anything about it at all.) As a consequence,
833  // let's be explicit about waiting.
834  //
835  // One of the corner cases we have to worry about is that if a task
836  // ends by throwing an exception, then wait() will re-throw that
837  // exception on the thread that calls it, the first time around
838  // someone calls wait() (or the return_value() function of the
839  // surrounding class). So if we get to this constructor and an exception
840  // is thrown by wait(), then that means that the last Task object
841  // referring to a task is going out of scope with nobody having
842  // ever checked the return value of the task itself. In that case,
843  // one could argue that they would also not have cared about whether
844  // an exception is thrown, and that we should simply ignore the
845  // exception. This is what we do here. It is also the simplest solution,
846  // because we don't know what one should do with the exception to begin
847  // with: destructors aren't allowed to throw exceptions, so we can't
848  // just rethrow it here if one had been triggered.
849  try
850  {
851  wait();
852  }
853  catch (...)
854  {}
855  }
856 
862  void
864  {
865  // If we have previously already moved the result, then we don't
866  // need a lock and can just return.
867  if (task_has_finished)
868  return;
869 
870  // Else, we need to go under a lock and try again. A different thread
871  // may have waited and finished the task since then, so we have to try
872  // a second time. (This is Schmidt's double-checking pattern.)
873  std::lock_guard<std::mutex> lock(mutex);
874  if (task_has_finished)
875  return;
876  else
877  {
878 #ifdef DEAL_II_WITH_TBB
879  // If we build on the TBB, then we can't just wait for the
880  // std::future object to get ready. Apparently the TBB happily
881  // enqueues a task into an arena and then just sits on it without
882  // ever executing it unless someone expresses an interest in the
883  // task. The way to avoid this is to add the task to a
884  // tbb::task_group, and then here wait for the single task
885  // associated with that task group.
886  task_group.wait();
887 #endif
888 
889  // Wait for the task to finish and then move its
890  // result. (We could have made the set_from() function
891  // that we call here wait for the future to be ready --
892  // which happens implicitly when it calls future.get() --
893  // but that would have required putting an explicit
894  // future.wait() into the implementation of
895  // internal::return_value<void>::set_from(), which is a
896  // bit awkward: that class doesn't actually need to set
897  // anything, and so it looks odd to have the explicit call
898  // to future.wait() in the set_from() function. Avoid the
899  // issue by just explicitly calling future.wait() here.)
900  future.wait();
901 
902  // Acquire the returned object. If the task ended in an
903  // exception, `set_from` will call `std::future::get`, which
904  // will throw an exception. This leaves `returned_object` in
905  // an undefined state, but moreover we would bypass setting
906  // `task_has_finished=true` below. So catch the exception
907  // for just long enough that we can set that flag, and then
908  // re-throw it:
909  try
910  {
911  returned_object.set_from(future);
912  }
913  catch (...)
914  {
915  task_has_finished = true;
916  throw;
917  }
918 
919  // If we got here, the task has ended without an exception and
920  // we can safely set the flag and return.
921  task_has_finished = true;
922  }
923  }
924 
925 
926 
928  get()
929  {
930  wait();
931  return returned_object.get();
932  }
933 
934  private:
939  std::mutex mutex;
940 
945  std::future<RT> future;
946 
965  std::atomic<bool> task_has_finished;
966 
972 
973 #ifdef DEAL_II_WITH_TBB
977  tbb::task_group task_group;
978 
979  friend class Task<RT>;
980 #endif
981  };
982 
987  std::shared_ptr<TaskData> task_data;
988  };
989 
990 
991 
1011  template <typename RT>
1012  inline Task<RT>
1013  new_task(const std::function<RT()> &function)
1014  {
1015  return Task<RT>(function);
1016  }
1017 
1018 
1019 
1097  template <typename FunctionObjectType>
1098  DEAL_II_CXX20_REQUIRES((std::invocable<FunctionObjectType>))
1099  inline auto new_task(FunctionObjectType function_object)
1100  -> Task<decltype(function_object())>
1101  {
1102  using return_type = decltype(function_object());
1104  return new_task(std::function<return_type()>(function_object));
1105  }
1106 
1107 
1108 
1115  template <typename RT, typename... Args>
1116  inline Task<RT>
1117  new_task(RT (*fun_ptr)(Args...), std_cxx20::type_identity_t<Args>... args)
1118  {
1119  auto dummy = std::make_tuple(internal::maybe_make_ref<Args>::act(args)...);
1120  return new_task(
1121  [dummy, fun_ptr]() -> RT { return std::apply(fun_ptr, dummy); });
1122  }
1123 
1124 
1125 
1132  template <typename RT, typename C, typename... Args>
1133  inline Task<RT>
1134  new_task(RT (C::*fun_ptr)(Args...),
1137  {
1138  // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
1139  return new_task(std::function<RT()>(std::bind(
1140  fun_ptr, std::ref(c), internal::maybe_make_ref<Args>::act(args)...)));
1141  }
1142 
1149  template <typename RT, typename C, typename... Args>
1150  inline Task<RT>
1151  new_task(RT (C::*fun_ptr)(Args...) const,
1154  {
1155  // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
1156  return new_task(std::function<RT()>(std::bind(
1157  fun_ptr, std::cref(c), internal::maybe_make_ref<Args>::act(args)...)));
1158  }
1159 
1160 
1161  // ------------------------ TaskGroup -------------------------------------
1162 
1175  template <typename RT = void>
1177  {
1178  public:
1182  TaskGroup &
1184  {
1185  tasks.push_back(t);
1186  return *this;
1187  }
1188 
1189 
1197  std::size_t
1198  size() const
1199  {
1200  return tasks.size();
1201  }
1202 
1217  std::vector<RT>
1219  {
1220  std::vector<RT> results;
1221  results.reserve(size());
1222  for (auto &t : tasks)
1223  results.emplace_back(std::move(t.return_value()));
1224  return results;
1225  }
1226 
1227 
1234  void
1235  join_all() const
1236  {
1237  for (auto &t : tasks)
1238  t.join();
1239  }
1240 
1241  private:
1245  std::list<Task<RT>> tasks;
1246  };
1247 
1248 } // namespace Threads
1249 
1256 #endif
static void initialize_multithreading()
static unsigned int n_threads()
std::size_t size() const
std::list< Task< RT > > tasks
std::vector< RT > return_values()
TaskGroup & operator+=(const Task< RT > &t)
TaskData(std::future< RT > &&future) noexcept
TaskData(const TaskData &)=delete
std::future< RT > future
std::atomic< bool > task_has_finished
internal::return_value< RT > returned_object
TaskData & operator=(const TaskData &)=delete
TaskData(TaskData &&)=delete
TaskData & operator=(TaskData &&)=delete
internal::return_value< RT >::reference_type get()
std::shared_ptr< TaskData > task_data
bool joinable() const
internal::return_value< RT >::reference_type return_value()
void join() const
Task()=default
Task(const std::function< RT()> &function_object)
#define DEAL_II_NAMESPACE_OPEN
Definition: config.h:477
#define DEAL_II_CXX20_REQUIRES(condition)
Definition: config.h:166
#define DEAL_II_NAMESPACE_CLOSE
Definition: config.h:478
Point< 2 > second
Definition: grid_out.cc:4615
static ::ExceptionBase & ExcNoTask()
#define Assert(cond, exc)
Definition: exceptions.h:1616
#define DeclExceptionMsg(Exception, defaulttext)
Definition: exceptions.h:490
static ::ExceptionBase & ExcMessage(std::string arg1)
#define AssertThrow(cond, exc)
Definition: exceptions.h:1705
std::vector< std::pair< unsigned int, unsigned int > > split_interval(const unsigned int begin, const unsigned int end, const unsigned int n_intervals)
std::vector< std::pair< ForwardIterator, ForwardIterator > > split_range(const ForwardIterator &begin, const ForwardIterator &end, const unsigned int n_intervals)
Task< RT > new_task(const std::function< RT()> &function)
void apply(const Kokkos::TeamPolicy< MemorySpace::Default::kokkos_space::execution_space >::member_type &team_member, const Kokkos::View< Number *, MemorySpace::Default::kokkos_space > shape_data, const ViewTypeIn in, ViewTypeOut out)
static const char T
SymmetricTensor< 2, dim, Number > C(const Tensor< 2, dim, Number > &F)
void evaluate_and_set_promise(Function &function, std::promise< RT > &promise)
void handle_std_exception(const std::exception &exc)
Definition: mutex.h:32
VectorType::value_type * begin(VectorType &V)
VectorType::value_type * end(VectorType &V)
typename type_identity< T >::type type_identity_t
Definition: type_traits.h:96
static std::reference_wrapper< T > act(T &t)
void set_from(std::future< RT & > &v)
void set_from(std::future< void > &)
void set_from(std::future< RT > &v)
void advance(std::tuple< I1, I2 > &t, const unsigned int n)