Loading [MathJax]/extensions/TeX/newcommand.js
 Reference documentation for deal.II version 9.3.3
\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\}}
All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Modules Pages
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
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
55namespace 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 &
97 {
98 return *this;
99 }
100 };
101} // namespace Threads
102
103
104namespace 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
187namespace 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
239namespace Threads
240{
241 namespace internal
242 {
261 template <typename RT>
263 {
264 private:
266
267 public:
268 using reference_type = RT &;
269
271 : value()
272 {}
273
274 inline reference_type
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
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
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
608 {}
609
615 void
616 join() const
617 {
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 {
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>
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)
831 -> Thread<decltype(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:
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 {
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.
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.
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
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:
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.
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);
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
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)
1413 -> Task<decltype(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//---------------------------------------------------------------------------
static void initialize_multithreading()
static unsigned int n_threads()
Mutex()=default
Mutex & operator=(const Mutex &)
Mutex(const Mutex &)
bool joinable() const
internal::return_value< RT >::reference_type return_value()
void join() const
Task()=default
Task(const std::function< RT()> &function_object)
std::list< Thread< RT > > threads
ThreadGroup & operator+=(const Thread< RT > &t)
bool valid() const
Thread(const std::function< RT()> &function)
internal::return_value< RT >::reference_type return_value()
std::shared_ptr< internal::ThreadDescriptor< RT > > thread_descriptor
bool operator==(const Thread &t) const
Thread()=default
Thread(const Thread< RT > &t)
#define DEAL_II_NAMESPACE_OPEN
Definition: config.h:402
#define DEAL_II_DEPRECATED_EARLY
Definition: config.h:170
#define DEAL_II_NAMESPACE_CLOSE
Definition: config.h:403
Point< 2 > second
Definition: grid_out.cc:4588
static ::ExceptionBase & ExcNoTask()
TaskGroup & operator+=(const Task< RT > &t)
std::shared_ptr< TaskData > task_data
#define Assert(cond, exc)
Definition: exceptions.h:1465
std::atomic< bool > task_has_finished
std::size_t size() const
internal::return_value< RT > returned_object
std::list< Task< RT > > tasks
TaskData(std::future< RT > &&future)
#define DeclExceptionMsg(Exception, defaulttext)
Definition: exceptions.h:493
static ::ExceptionBase & ExcInternalError()
internal::return_value< RT >::reference_type get()
#define AssertThrow(cond, exc)
Definition: exceptions.h:1575
std::vector< std::pair< ForwardIterator, ForwardIterator > > split_range(const ForwardIterator &begin, const ForwardIterator &end, const unsigned int n_intervals)
Thread< RT > new_thread(const std::function< RT()> &function)
Task< RT > new_task(const std::function< RT()> &function)
std::vector< std::pair< unsigned int, unsigned int > > split_interval(const unsigned int begin, const unsigned int end, const unsigned int n_intervals)
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 call(const std::function< RT()> &function, internal::return_value< RT > &ret_val)
void handle_std_exception(const std::exception &exc)
VectorType::value_type * end(VectorType &V)
VectorType::value_type * begin(VectorType &V)
STL namespace.
std::shared_ptr< return_value< RT > > ret_val
void start(const std::function< RT()> &function)
static void thread_entry_point(const std::function< RT()> &function, std::shared_ptr< return_value< RT > > ret_val)
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)