Reference documentation for deal.II version Git 040c6ad7d4 2020-09-26 18:01:03 +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 
574  template <typename RT = void>
575  class Thread
576  {
577  public:
581  Thread(const std::function<RT()> &function)
582  : thread_descriptor(new internal::ThreadDescriptor<RT>())
583  {
584  // in a second step, start the thread.
585  thread_descriptor->start(function);
586  }
587 
593  Thread() = default;
594 
598  Thread(const Thread<RT> &t)
599  : thread_descriptor(t.thread_descriptor)
600  {}
601 
607  void
608  join() const
609  {
610  if (thread_descriptor)
611  thread_descriptor->join();
612  }
613 
659  {
660  join();
661  return thread_descriptor->ret_val->get();
662  }
663 
668  bool
669  valid() const
670  {
671  return static_cast<bool>(thread_descriptor);
672  }
673 
674 
680  bool
681  operator==(const Thread &t) const
682  {
683  return thread_descriptor == t.thread_descriptor;
684  }
685 
686  private:
692  std::shared_ptr<internal::ThreadDescriptor<RT>> thread_descriptor;
693  };
694 
695 
696  namespace internal
697  {
705  template <typename T>
707  {
708  static T
709  act(T &t)
710  {
711  return t;
712  }
713  };
714 
715 
716 
724  template <typename T>
725  struct maybe_make_ref<T &>
726  {
727  static std::reference_wrapper<T>
728  act(T &t)
729  {
730  return std::ref(t);
731  }
732  };
733  } // namespace internal
734 
735 
736 
737  // ----------- thread starters for functions not taking any parameters
738 
747  template <typename RT>
748  inline Thread<RT>
749  new_thread(const std::function<RT()> &function)
750  {
751  return Thread<RT>(function);
752  }
753 
754 
755 
820  template <typename FunctionObjectType>
821  inline auto
822  new_thread(FunctionObjectType function_object)
824  {
825  using return_type = decltype(function_object());
826  return Thread<return_type>(std::function<return_type()>(function_object));
827  }
828 
829 
830 
837  template <typename RT, typename... Args>
838  inline Thread<RT>
839  new_thread(RT (*fun_ptr)(Args...), typename identity<Args>::type... args)
840  {
841  auto dummy = std::make_tuple(internal::maybe_make_ref<Args>::act(args)...);
842  return new_thread(
843  [dummy, fun_ptr]() -> RT { return std_cxx17::apply(fun_ptr, dummy); });
844  }
845 
846 
847 
853  template <typename RT, typename C, typename... Args>
854  inline Thread<RT>
855  new_thread(RT (C::*fun_ptr)(Args...),
856  typename identity<C>::type &c,
857  typename identity<Args>::type... args)
858  {
859  // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
860  return new_thread(std::function<RT()>(std::bind(
861  fun_ptr, std::ref(c), internal::maybe_make_ref<Args>::act(args)...)));
862  }
863 
869  template <typename RT, typename C, typename... Args>
870  inline Thread<RT>
871  new_thread(RT (C::*fun_ptr)(Args...) const,
872  typename identity<const C>::type &c,
873  typename identity<Args>::type... args)
874  {
875  // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
876  return new_thread(std::function<RT()>(std::bind(
877  fun_ptr, std::cref(c), internal::maybe_make_ref<Args>::act(args)...)));
878  }
879 
880  // ------------------------ ThreadGroup -------------------------------------
881 
889  template <typename RT = void>
891  {
892  public:
896  ThreadGroup &
898  {
899  threads.push_back(t);
900  return *this;
901  }
902 
909  void
910  join_all() const
911  {
912  for (typename std::list<Thread<RT>>::const_iterator t = threads.begin();
913  t != threads.end();
914  ++t)
915  t->join();
916  }
917 
918  private:
922  std::list<Thread<RT>> threads;
923  };
924 
925 
926  namespace internal
927  {
931  template <typename RT, typename Function>
932  void
933  evaluate_and_set_promise(Function &function, std::promise<RT> &promise)
934  {
935  promise.set_value(function());
936  }
937 
938 
946  template <typename Function>
947  void
948  evaluate_and_set_promise(Function &function, std::promise<void> &promise)
949  {
950  function();
951  promise.set_value();
952  }
953  } // namespace internal
954 
955 
956 
983  template <typename RT = void>
984  class Task
985  {
986  public:
998  Task(const std::function<RT()> &function_object)
999  {
1000  if (MultithreadInfo::n_threads() > 1)
1001  task_data = std::make_shared<TaskData>(
1002  std::async(std::launch::async, function_object));
1003  else
1004  {
1005  // Only one thread allowed. So let the task run to completion
1006  // and just emplace a 'ready' future.
1007  //
1008  // The design of std::promise/std::future is unclear, but it
1009  // seems that the intent is to obtain the std::future before
1010  // we set the std::promise. So create the TaskData object at
1011  // the top and then run the task and set the returned
1012  // value. Since everything here happens sequentially, it
1013  // really doesn't matter in which order all of this is
1014  // happening.
1015  std::promise<RT> promise;
1016  task_data = std::make_shared<TaskData>(promise.get_future());
1017  try
1018  {
1019  internal::evaluate_and_set_promise(function_object, promise);
1020  }
1021  catch (...)
1022  {
1023  try
1024  {
1025  // store anything thrown in the promise
1026  promise.set_exception(std::current_exception());
1027  }
1028  catch (...)
1029  {}
1030  // set_exception() may throw too
1031  }
1032  }
1033  }
1034 
1043  Task() = default;
1044 
1076  void
1077  join() const
1078  {
1079  // Make sure we actually have a task that we can wait for.
1080  AssertThrow(joinable(), ExcNoTask());
1081 
1082  task_data->wait();
1083  }
1084 
1097  bool
1098  joinable() const
1099  {
1100  return (task_data != nullptr);
1101  }
1102 
1103 
1155  {
1156  // Make sure we actually have a task that we can wait for.
1157  AssertThrow(joinable(), ExcNoTask());
1158 
1159  // Then return the promised object. If necessary, wait for the promise to
1160  // be set.
1161  return task_data->get();
1162  }
1163 
1164 
1173  DeclExceptionMsg(ExcNoTask,
1174  "The current object is not associated with a task that "
1175  "can be joined. It may have been detached, or you "
1176  "may have already joined it in the past.");
1178  private:
1187  class TaskData
1188  {
1189  public:
1194  TaskData(std::future<RT> &&future)
1195  : future(std::move(future))
1196  , task_has_finished(false)
1197  {}
1198 
1204  void
1206  {
1207  // If we have previously already moved the result, then we don't
1208  // need a lock and can just return.
1209  if (task_has_finished)
1210  return;
1211 
1212  // Else, we need to go under a lock and try again. A different thread
1213  // may have waited and finished the task since then, so we have to try
1214  // a second time. (This is Schmidt's double-checking pattern.)
1215  std::lock_guard<std::mutex> lock(mutex);
1216  if (task_has_finished)
1217  return;
1218  else
1219  {
1220  // Wait for the task to finish and then move its
1221  // result. (We could have made the set_from() function
1222  // that we call here wait for the future to be ready --
1223  // which happens implicitly when it calls future.get() --
1224  // but that would have required putting an explicit
1225  // future.wait() into the implementation of
1226  // internal::return_value<void>::set_from(), which is a
1227  // bit awkward: that class doesn't actually need to set
1228  // anything, and so it looks odd to have the explicit call
1229  // to future.wait() in the set_from() function. Avoid the
1230  // issue by just explicitly calling future.wait() here.)
1231  future.wait();
1232  returned_object.set_from(future);
1233 
1234  // Now we can safely set the flag and return.
1235  task_has_finished = true;
1236  }
1237  }
1238 
1239 
1240 
1242  get()
1243  {
1244  wait();
1245  return returned_object.get();
1246  }
1247 
1248  private:
1253  std::mutex mutex;
1254 
1259  std::future<RT> future;
1260 
1279  std::atomic<bool> task_has_finished;
1280 
1286  };
1287 
1292  std::shared_ptr<TaskData> task_data;
1293  };
1294 
1295 
1296 
1322  template <typename RT>
1323  inline Task<RT>
1324  new_task(const std::function<RT()> &function)
1325  {
1326  return Task<RT>(function);
1327  }
1328 
1329 
1330 
1412  template <typename FunctionObjectType>
1413  inline auto
1414  new_task(FunctionObjectType function_object)
1416  {
1417  using return_type = decltype(function_object());
1419  return new_task(std::function<return_type()>(function_object));
1420  }
1421 
1422 
1423 
1430  template <typename RT, typename... Args>
1431  inline Task<RT>
1432  new_task(RT (*fun_ptr)(Args...), typename identity<Args>::type... args)
1433  {
1434  auto dummy = std::make_tuple(internal::maybe_make_ref<Args>::act(args)...);
1435  return new_task(
1436  [dummy, fun_ptr]() -> RT { return std_cxx17::apply(fun_ptr, dummy); });
1437  }
1438 
1439 
1440 
1447  template <typename RT, typename C, typename... Args>
1448  inline Task<RT>
1449  new_task(RT (C::*fun_ptr)(Args...),
1450  typename identity<C>::type &c,
1451  typename identity<Args>::type... args)
1452  {
1453  // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
1454  return new_task(std::function<RT()>(std::bind(
1455  fun_ptr, std::ref(c), internal::maybe_make_ref<Args>::act(args)...)));
1456  }
1457 
1464  template <typename RT, typename C, typename... Args>
1465  inline Task<RT>
1466  new_task(RT (C::*fun_ptr)(Args...) const,
1467  typename identity<const C>::type &c,
1468  typename identity<Args>::type... args)
1469  {
1470  // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
1471  return new_task(std::function<RT()>(std::bind(
1472  fun_ptr, std::cref(c), internal::maybe_make_ref<Args>::act(args)...)));
1473  }
1474 
1475 
1476  // ------------------------ TaskGroup -------------------------------------
1477 
1490  template <typename RT = void>
1492  {
1493  public:
1497  TaskGroup &
1499  {
1500  tasks.push_back(t);
1501  return *this;
1502  }
1503 
1504 
1512  std::size_t
1513  size() const
1514  {
1515  return tasks.size();
1516  }
1517 
1518 
1525  void
1526  join_all() const
1527  {
1528  for (auto &t : tasks)
1529  t.join();
1530  }
1531 
1532  private:
1536  std::list<Task<RT>> tasks;
1537  };
1538 
1539 } // namespace Threads
1540 
1546 //---------------------------------------------------------------------------
1548 // end of #ifndef dealii_thread_management_h
1549 #endif
1550 //---------------------------------------------------------------------------
void evaluate_and_set_promise(Function &function, std::promise< void > &promise)
std::shared_ptr< internal::ThreadDescriptor< RT > > thread_descriptor
bool valid() const
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:1521
Point< 2 > second
Definition: grid_out.cc:4336
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:1411
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:363
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:362
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()