Reference documentation for deal.II version Git 1c2f3d7ba8 2021-05-16 10:03:46 +0200
\(\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 - 2020 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 
22 # include <deal.II/base/exceptions.h>
26 
27 # include <atomic>
28 # include <condition_variable>
29 # include <functional>
30 # include <future>
31 # include <iterator>
32 # include <list>
33 # include <memory>
34 # include <mutex>
35 # include <thread>
36 # include <tuple>
37 # include <utility>
38 # include <vector>
39 
40 
41 
43 
46 
47 
55 namespace Threads
56 {
75  class Mutex : public std::mutex
76  {
77  public:
81  Mutex() = default;
82 
87  Mutex(const Mutex &)
88  : std::mutex()
89  {}
90 
95  Mutex &
96  operator=(const Mutex &)
97  {
98  return *this;
99  }
100  };
101 } // namespace Threads
102 
103 
104 namespace Threads
105 {
120  template <typename ForwardIterator>
121  std::vector<std::pair<ForwardIterator, ForwardIterator>>
122  split_range(const ForwardIterator &begin,
123  const ForwardIterator &end,
124  const unsigned int n_intervals);
125 
134  std::vector<std::pair<unsigned int, unsigned int>>
135  split_interval(const unsigned int begin,
136  const unsigned int end,
137  const unsigned int n_intervals);
138 
148  namespace internal
149  {
165  [[noreturn]] void
166  handle_std_exception(const std::exception &exc);
167 
175  [[noreturn]] void
177  } // namespace internal
178 
183 } // namespace Threads
184 
185 /* ----------- implementation of functions in namespace Threads ---------- */
186 # ifndef DOXYGEN
187 namespace Threads
188 {
189  template <typename ForwardIterator>
190  std::vector<std::pair<ForwardIterator, ForwardIterator>>
191  split_range(const ForwardIterator &begin,
192  const ForwardIterator &end,
193  const unsigned int n_intervals)
194  {
195  using IteratorPair = std::pair<ForwardIterator, ForwardIterator>;
196 
197  // in non-multithreaded mode, we often have the case that this
198  // function is called with n_intervals==1, so have a shortcut here
199  // to handle that case efficiently
200 
201  if (n_intervals == 1)
202  return (std::vector<IteratorPair>(1, IteratorPair(begin, end)));
203 
204  // if more than one interval requested, do the full work
205  const unsigned int n_elements = std::distance(begin, end);
206  const unsigned int n_elements_per_interval = n_elements / n_intervals;
207  const unsigned int residual = n_elements % n_intervals;
208 
209  std::vector<IteratorPair> return_values(n_intervals);
210 
211  return_values[0].first = begin;
212  for (unsigned int i = 0; i < n_intervals; ++i)
213  {
214  if (i != n_intervals - 1)
215  {
216  return_values[i].second = return_values[i].first;
217  // note: the cast is performed to avoid a warning of gcc
218  // that in the library `dist>=0' is checked (dist has a
219  // template type, which here is unsigned if no cast is
220  // performed)
221  std::advance(return_values[i].second,
222  static_cast<signed int>(n_elements_per_interval));
223  // distribute residual in division equally among the first
224  // few subintervals
225  if (i < residual)
226  ++return_values[i].second;
227 
228  return_values[i + 1].first = return_values[i].second;
229  }
230  else
231  return_values[i].second = end;
232  }
233  return return_values;
234  }
235 } // namespace Threads
236 
237 # endif // DOXYGEN
238 
239 namespace Threads
240 {
241  namespace internal
242  {
261  template <typename RT>
263  {
264  private:
265  RT value;
266 
267  public:
268  using reference_type = RT &;
269 
270  inline return_value()
271  : value()
272  {}
273 
274  inline reference_type
275  get()
276  {
277  return value;
278  }
279 
280  inline void
281  set(RT &&v)
282  {
283  value = std::move(v);
284  }
285 
286  inline void
287  set_from(std::future<RT> &v)
288  {
289  value = std::move(v.get());
290  }
291  };
292 
293 
313  template <typename RT>
314  struct return_value<RT &>
315  {
316  private:
317  RT *value;
318 
319  public:
320  using reference_type = RT &;
321 
322  inline return_value()
323  : value(nullptr)
324  {}
325 
326  inline reference_type
327  get() const
328  {
329  return *value;
330  }
331 
332  inline void
333  set(RT &v)
334  {
335  value = &v;
336  }
337 
338  inline void
339  set_from(std::future<RT &> &v)
340  {
341  value = &v.get();
342  }
343  };
344 
345 
364  template <>
365  struct return_value<void>
366  {
367  using reference_type = void;
368 
369  static inline void
370  get()
371  {}
372 
373 
374  inline void
375  set_from(std::future<void> &)
376  {}
377  };
378  } // namespace internal
379 
380 
381 
382  namespace internal
383  {
384  template <typename RT>
385  inline void
386  call(const std::function<RT()> & function,
388  {
389  ret_val.set(function());
390  }
391 
392 
393  inline void
394  call(const std::function<void()> &function, internal::return_value<void> &)
395  {
396  function();
397  }
398  } // namespace internal
399 
400 
401 
402  namespace internal
403  {
414  template <typename RT>
416  {
420  std::thread thread;
421 
430  std::shared_ptr<return_value<RT>> ret_val;
431 
467  std::atomic<bool> thread_is_active;
468 
473 
478  : thread_is_active(false)
479  {}
480 
482  {
483  if (!thread_is_active)
484  return;
485  thread.detach();
486  thread_is_active = false;
487  }
488 
493  void
494  start(const std::function<RT()> &function)
495  {
496  thread_is_active = true;
497  ret_val = std::make_shared<return_value<RT>>();
498  thread = std::thread(thread_entry_point, function, ret_val);
499  }
500 
501 
505  void
507  {
508  // see if the thread hasn't been joined yet. if it has, then
509  // join() is a no-op. use schmidt's double-checking strategy
510  // to use the mutex only when necessary
511  if (thread_is_active == false)
512  return;
513 
514  std::lock_guard<std::mutex> lock(thread_is_active_mutex);
515  if (thread_is_active == true)
516  {
517  Assert(thread.joinable(), ExcInternalError());
518  thread.join();
519  thread_is_active = false;
520  }
521  }
522 
523  private:
527  static void
528  thread_entry_point(const std::function<RT()> & function,
529  std::shared_ptr<return_value<RT>> ret_val)
530  {
531  // call the function in question. since an exception that is
532  // thrown from one of the called functions will not propagate
533  // to the main thread, it will kill the program if not treated
534  // here before we return to the operating system's thread
535  // library
536  try
537  {
538  call(function, *ret_val);
539  }
540  catch (const std::exception &exc)
541  {
543  }
544  catch (...)
545  {
547  }
548  }
549  };
550  } // namespace internal
551 
552 
579  template <typename RT = void>
580  class Thread
581  {
582  public:
587  Thread(const std::function<RT()> &function)
588  : thread_descriptor(new internal::ThreadDescriptor<RT>())
589  {
590  // in a second step, start the thread.
591  thread_descriptor->start(function);
592  }
593 
600  Thread() = default;
601 
606  Thread(const Thread<RT> &t)
607  : thread_descriptor(t.thread_descriptor)
608  {}
609 
615  void
616  join() const
617  {
618  if (thread_descriptor)
619  thread_descriptor->join();
620  }
621 
667  {
668  join();
669  return thread_descriptor->ret_val->get();
670  }
671 
676  bool
677  valid() const
678  {
679  return static_cast<bool>(thread_descriptor);
680  }
681 
682 
688  bool
689  operator==(const Thread &t) const
690  {
691  return thread_descriptor == t.thread_descriptor;
692  }
693 
694  private:
700  std::shared_ptr<internal::ThreadDescriptor<RT>> thread_descriptor;
701  };
702 
703 
704  namespace internal
705  {
713  template <typename T>
715  {
716  static T
717  act(T &t)
718  {
719  return t;
720  }
721  };
722 
723 
724 
732  template <typename T>
733  struct maybe_make_ref<T &>
734  {
735  static std::reference_wrapper<T>
736  act(T &t)
737  {
738  return std::ref(t);
739  }
740  };
741  } // namespace internal
742 
743 
744 
745  // ----------- thread starters for functions not taking any parameters
746 
755  template <typename RT>
756  inline Thread<RT>
757  new_thread(const std::function<RT()> &function)
758  {
759  return Thread<RT>(function);
760  }
761 
762 
763 
828  template <typename FunctionObjectType>
829  inline auto
830  new_thread(FunctionObjectType function_object)
832  {
833  using return_type = decltype(function_object());
834  return Thread<return_type>(std::function<return_type()>(function_object));
835  }
836 
837 
838 
845  template <typename RT, typename... Args>
846  inline Thread<RT>
847  new_thread(RT (*fun_ptr)(Args...), typename identity<Args>::type... args)
848  {
849  auto dummy = std::make_tuple(internal::maybe_make_ref<Args>::act(args)...);
850  return new_thread(
851  [dummy, fun_ptr]() -> RT { return std_cxx17::apply(fun_ptr, dummy); });
852  }
853 
854 
855 
861  template <typename RT, typename C, typename... Args>
862  inline Thread<RT>
863  new_thread(RT (C::*fun_ptr)(Args...),
864  typename identity<C>::type &c,
865  typename identity<Args>::type... args)
866  {
867  // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
868  return new_thread(std::function<RT()>(std::bind(
869  fun_ptr, std::ref(c), internal::maybe_make_ref<Args>::act(args)...)));
870  }
871 
877  template <typename RT, typename C, typename... Args>
878  inline Thread<RT>
879  new_thread(RT (C::*fun_ptr)(Args...) const,
880  typename identity<const C>::type &c,
881  typename identity<Args>::type... args)
882  {
883  // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
884  return new_thread(std::function<RT()>(std::bind(
885  fun_ptr, std::cref(c), internal::maybe_make_ref<Args>::act(args)...)));
886  }
887 
888  // ------------------------ ThreadGroup -------------------------------------
889 
899  template <typename RT = void>
901  {
902  public:
906  ThreadGroup &
908  {
909  threads.push_back(t);
910  return *this;
911  }
912 
919  void
920  join_all() const
921  {
922  for (typename std::list<Thread<RT>>::const_iterator t = threads.begin();
923  t != threads.end();
924  ++t)
925  t->join();
926  }
927 
928  private:
932  std::list<Thread<RT>> threads;
933  };
934 
935 
936  namespace internal
937  {
941  template <typename RT, typename Function>
942  void
943  evaluate_and_set_promise(Function &function, std::promise<RT> &promise)
944  {
945  promise.set_value(function());
946  }
947 
948 
956  template <typename Function>
957  void
958  evaluate_and_set_promise(Function &function, std::promise<void> &promise)
959  {
960  function();
961  promise.set_value();
962  }
963  } // namespace internal
964 
965 
966 
993  template <typename RT = void>
994  class Task
995  {
996  public:
1008  Task(const std::function<RT()> &function_object)
1009  {
1010  if (MultithreadInfo::n_threads() > 1)
1011  task_data = std::make_shared<TaskData>(
1012  std::async(std::launch::async, function_object));
1013  else
1014  {
1015  // Only one thread allowed. So let the task run to completion
1016  // and just emplace a 'ready' future.
1017  //
1018  // The design of std::promise/std::future is unclear, but it
1019  // seems that the intent is to obtain the std::future before
1020  // we set the std::promise. So create the TaskData object at
1021  // the top and then run the task and set the returned
1022  // value. Since everything here happens sequentially, it
1023  // really doesn't matter in which order all of this is
1024  // happening.
1025  std::promise<RT> promise;
1026  task_data = std::make_shared<TaskData>(promise.get_future());
1027  try
1028  {
1029  internal::evaluate_and_set_promise(function_object, promise);
1030  }
1031  catch (...)
1032  {
1033  try
1034  {
1035  // store anything thrown in the promise
1036  promise.set_exception(std::current_exception());
1037  }
1038  catch (...)
1039  {}
1040  // set_exception() may throw too
1041  }
1042  }
1043  }
1044 
1053  Task() = default;
1054 
1086  void
1087  join() const
1088  {
1089  // Make sure we actually have a task that we can wait for.
1090  AssertThrow(joinable(), ExcNoTask());
1091 
1092  task_data->wait();
1093  }
1094 
1107  bool
1108  joinable() const
1109  {
1110  return (task_data != nullptr);
1111  }
1112 
1113 
1165  {
1166  // Make sure we actually have a task that we can wait for.
1167  AssertThrow(joinable(), ExcNoTask());
1168 
1169  // Then return the promised object. If necessary, wait for the promise to
1170  // be set.
1171  return task_data->get();
1172  }
1173 
1174 
1183  DeclExceptionMsg(ExcNoTask,
1184  "The current object is not associated with a task that "
1185  "can be joined. It may have been detached, or you "
1186  "may have already joined it in the past.");
1188  private:
1197  class TaskData
1198  {
1199  public:
1204  TaskData(std::future<RT> &&future)
1205  : future(std::move(future))
1206  , task_has_finished(false)
1207  {}
1208 
1214  void
1216  {
1217  // If we have previously already moved the result, then we don't
1218  // need a lock and can just return.
1219  if (task_has_finished)
1220  return;
1221 
1222  // Else, we need to go under a lock and try again. A different thread
1223  // may have waited and finished the task since then, so we have to try
1224  // a second time. (This is Schmidt's double-checking pattern.)
1225  std::lock_guard<std::mutex> lock(mutex);
1226  if (task_has_finished)
1227  return;
1228  else
1229  {
1230  // Wait for the task to finish and then move its
1231  // result. (We could have made the set_from() function
1232  // that we call here wait for the future to be ready --
1233  // which happens implicitly when it calls future.get() --
1234  // but that would have required putting an explicit
1235  // future.wait() into the implementation of
1236  // internal::return_value<void>::set_from(), which is a
1237  // bit awkward: that class doesn't actually need to set
1238  // anything, and so it looks odd to have the explicit call
1239  // to future.wait() in the set_from() function. Avoid the
1240  // issue by just explicitly calling future.wait() here.)
1241  future.wait();
1242  returned_object.set_from(future);
1243 
1244  // Now we can safely set the flag and return.
1245  task_has_finished = true;
1246  }
1247  }
1248 
1249 
1250 
1252  get()
1253  {
1254  wait();
1255  return returned_object.get();
1256  }
1257 
1258  private:
1263  std::mutex mutex;
1264 
1269  std::future<RT> future;
1270 
1289  std::atomic<bool> task_has_finished;
1290 
1296  };
1297 
1302  std::shared_ptr<TaskData> task_data;
1303  };
1304 
1305 
1306 
1326  template <typename RT>
1327  inline Task<RT>
1328  new_task(const std::function<RT()> &function)
1329  {
1330  return Task<RT>(function);
1331  }
1332 
1333 
1334 
1410  template <typename FunctionObjectType>
1411  inline auto
1412  new_task(FunctionObjectType function_object)
1414  {
1415  using return_type = decltype(function_object());
1417  return new_task(std::function<return_type()>(function_object));
1418  }
1419 
1420 
1421 
1428  template <typename RT, typename... Args>
1429  inline Task<RT>
1430  new_task(RT (*fun_ptr)(Args...), typename identity<Args>::type... args)
1431  {
1432  auto dummy = std::make_tuple(internal::maybe_make_ref<Args>::act(args)...);
1433  return new_task(
1434  [dummy, fun_ptr]() -> RT { return std_cxx17::apply(fun_ptr, dummy); });
1435  }
1436 
1437 
1438 
1445  template <typename RT, typename C, typename... Args>
1446  inline Task<RT>
1447  new_task(RT (C::*fun_ptr)(Args...),
1448  typename identity<C>::type &c,
1449  typename identity<Args>::type... args)
1450  {
1451  // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
1452  return new_task(std::function<RT()>(std::bind(
1453  fun_ptr, std::ref(c), internal::maybe_make_ref<Args>::act(args)...)));
1454  }
1455 
1462  template <typename RT, typename C, typename... Args>
1463  inline Task<RT>
1464  new_task(RT (C::*fun_ptr)(Args...) const,
1465  typename identity<const C>::type &c,
1466  typename identity<Args>::type... args)
1467  {
1468  // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
1469  return new_task(std::function<RT()>(std::bind(
1470  fun_ptr, std::cref(c), internal::maybe_make_ref<Args>::act(args)...)));
1471  }
1472 
1473 
1474  // ------------------------ TaskGroup -------------------------------------
1475 
1488  template <typename RT = void>
1490  {
1491  public:
1495  TaskGroup &
1497  {
1498  tasks.push_back(t);
1499  return *this;
1500  }
1501 
1502 
1510  std::size_t
1511  size() const
1512  {
1513  return tasks.size();
1514  }
1515 
1516 
1523  void
1524  join_all() const
1525  {
1526  for (auto &t : tasks)
1527  t.join();
1528  }
1529 
1530  private:
1534  std::list<Task<RT>> tasks;
1535  };
1536 
1537 } // namespace Threads
1538 
1544 //---------------------------------------------------------------------------
1546 // end of #ifndef dealii_thread_management_h
1547 #endif
1548 //---------------------------------------------------------------------------
void evaluate_and_set_promise(Function &function, std::promise< void > &promise)
std::shared_ptr< internal::ThreadDescriptor< RT > > thread_descriptor
bool valid() const
#define DEAL_II_DEPRECATED_EARLY
Definition: config.h:166
Task< RT > new_task(const std::function< RT()> &function)
Thread(const Thread< RT > &t)
Mutex(const Mutex &)
void evaluate_and_set_promise(Function &function, std::promise< RT > &promise)
std::list< Task< RT > > tasks
internal::return_value< RT >::reference_type return_value()
void join() const
void call(const std::function< void()> &function, internal::return_value< void > &)
std::size_t size() const
std::shared_ptr< return_value< RT > > ret_val
STL namespace.
#define AssertThrow(cond, exc)
Definition: exceptions.h:1575
Point< 2 > second
Definition: grid_out.cc:4588
std::shared_ptr< TaskData > task_data
SymmetricTensor< 2, dim, Number > C(const Tensor< 2, dim, Number > &F)
std::vector< std::pair< unsigned int, unsigned int > > split_interval(const unsigned int begin, const unsigned int end, const unsigned int n_intervals)
Task(const std::function< RT()> &function_object)
Mutex & operator=(const Mutex &)
static const char T
void set_from(std::future< void > &)
void set_from(std::future< RT &> &v)
#define Assert(cond, exc)
Definition: exceptions.h:1465
static void initialize_multithreading()
auto apply(F &&fn, Tuple &&t) -> decltype(apply_impl(std::forward< F >(fn), std::forward< Tuple >(t), std::make_index_sequence< std::tuple_size< typename std::remove_reference< Tuple >::type >::value >()))
Definition: tuple.h:36
#define DeclExceptionMsg(Exception, defaulttext)
Definition: exceptions.h:493
Mutex()=default
std::atomic< bool > task_has_finished
#define DEAL_II_NAMESPACE_CLOSE
Definition: config.h:396
VectorType::value_type * end(VectorType &V)
void set_from(std::future< RT > &v)
bool joinable() const
Thread(const std::function< RT()> &function)
static std::reference_wrapper< T > act(T &t)
internal::return_value< RT >::reference_type return_value()
void advance(std::tuple< I1, I2 > &t, const unsigned int n)
bool operator==(const Thread &t) const
#define DEAL_II_NAMESPACE_OPEN
Definition: config.h:395
VectorType::value_type * begin(VectorType &V)
TaskGroup & operator+=(const Task< RT > &t)
void start(const std::function< RT()> &function)
ThreadGroup & operator+=(const Thread< RT > &t)
TaskData(std::future< RT > &&future)
Thread< RT > new_thread(const std::function< RT()> &function)
static void thread_entry_point(const std::function< RT()> &function, std::shared_ptr< return_value< RT >> ret_val)
static unsigned int n_threads()
void handle_std_exception(const std::exception &exc)
internal::return_value< RT > returned_object
std::vector< std::pair< ForwardIterator, ForwardIterator > > split_range(const ForwardIterator &begin, const ForwardIterator &end, const unsigned int n_intervals)
std::list< Thread< RT > > threads
static ::ExceptionBase & ExcInternalError()