Reference documentation for deal.II version GIT relicensing-43-g3fc2fae7c3 2024-03-03 21:10: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\}}\)
Loading...
Searching...
No Matches
thread_management.h
Go to the documentation of this file.
1// ------------------------------------------------------------------------
2//
3// SPDX-License-Identifier: LGPL-2.1-or-later
4// Copyright (C) 2000 - 2023 by the deal.II authors
5//
6// This file is part of the deal.II library.
7//
8// Part of the source code is dual licensed under Apache-2.0 WITH
9// LLVM-exception OR LGPL-2.1-or-later. Detailed license information
10// governing the source code and code contributions can be found in
11// LICENSE.md and CONTRIBUTING.md at the top level directory of deal.II.
12//
13// ------------------------------------------------------------------------
14
15#ifndef dealii_thread_management_h
16#define dealii_thread_management_h
17
18
19#include <deal.II/base/config.h>
20
23#include <deal.II/base/mutex.h>
25
26#include <atomic>
27#include <functional>
28#include <future>
29#include <list>
30#include <memory>
31#include <thread>
32#include <tuple>
33#include <utility>
34#include <vector>
35
36#ifdef DEAL_II_HAVE_CXX20
37# include <concepts>
38#endif
39
40
41#ifdef DEAL_II_WITH_TBB
42# include <tbb/task_group.h>
43#endif
44
46
59namespace Threads
60{
75 template <typename ForwardIterator>
76 std::vector<std::pair<ForwardIterator, ForwardIterator>>
77 split_range(const ForwardIterator &begin,
78 const ForwardIterator &end,
79 const unsigned int n_intervals);
80
89 std::vector<std::pair<unsigned int, unsigned int>>
90 split_interval(const unsigned int begin,
91 const unsigned int end,
92 const unsigned int n_intervals);
93
103 namespace internal
104 {
120 [[noreturn]] void
121 handle_std_exception(const std::exception &exc);
122
130 [[noreturn]] void
132 } // namespace internal
133
138} // namespace Threads
139
140/* ----------- implementation of functions in namespace Threads ---------- */
141#ifndef DOXYGEN
142namespace Threads
143{
144 template <typename ForwardIterator>
145 std::vector<std::pair<ForwardIterator, ForwardIterator>>
146 split_range(const ForwardIterator &begin,
147 const ForwardIterator &end,
148 const unsigned int n_intervals)
149 {
150 using IteratorPair = std::pair<ForwardIterator, ForwardIterator>;
151
152 // in non-multithreaded mode, we often have the case that this
153 // function is called with n_intervals==1, so have a shortcut here
154 // to handle that case efficiently
155
156 if (n_intervals == 1)
157 return (std::vector<IteratorPair>(1, IteratorPair(begin, end)));
158
159 // if more than one interval requested, do the full work
160 const unsigned int n_elements = std::distance(begin, end);
161 const unsigned int n_elements_per_interval = n_elements / n_intervals;
162 const unsigned int residual = n_elements % n_intervals;
163
164 std::vector<IteratorPair> return_values(n_intervals);
165
166 return_values[0].first = begin;
167 for (unsigned int i = 0; i < n_intervals; ++i)
168 {
169 if (i != n_intervals - 1)
170 {
171 return_values[i].second = return_values[i].first;
172 // note: the cast is performed to avoid a warning of gcc
173 // that in the library `dist>=0' is checked (dist has a
174 // template type, which here is unsigned if no cast is
175 // performed)
176 std::advance(return_values[i].second,
177 static_cast<signed int>(n_elements_per_interval));
178 // distribute residual in division equally among the first
179 // few subintervals
180 if (i < residual)
181 ++return_values[i].second;
182
183 return_values[i + 1].first = return_values[i].second;
184 }
185 else
186 return_values[i].second = end;
187 }
188 return return_values;
189 }
190} // namespace Threads
191
192#endif // DOXYGEN
193
194namespace Threads
195{
196 namespace internal
197 {
216 template <typename RT>
218 {
219 private:
222
223 public:
224 using reference_type = RT &;
225
227 : value()
228 , value_is_initialized(false)
229 {}
230
231 inline reference_type
233 {
234 Assert(
237 "You cannot read the return value of a thread or task "
238 "if that value has not been set. This happens, for example, if "
239 "a task or thread threw an exception."));
240 return value;
241 }
242
243 inline void
244 set(RT &&v)
245 {
246 value = std::move(v);
247 }
248
254 inline void
255 set_from(std::future<RT> &v)
256 {
257 // Get the value from the std::future object. If the future holds
258 // an exception, then the assignment fails, we exit the function via the
259 // exception right away, and value_is_initialized is not set to true --
260 // that's something we can check later on.
261 value = std::move(v.get());
263 }
264 };
265
266
286 template <typename RT>
287 struct return_value<RT &>
288 {
289 private:
290 RT *value;
292
293 public:
294 using reference_type = RT &;
295
297 : value(nullptr)
298 , value_is_initialized(false)
299 {}
300
301 inline reference_type
302 get() const
303 {
304 Assert(
305 value_is_initialized,
307 "You cannot read the return value of a thread or task "
308 "if that value has not been set. This happens, for example, if "
309 "a task or thread threw an exception."));
310 return *value;
311 }
312
313 inline void
314 set(RT &v)
315 {
316 value = &v;
317 }
318
324 inline void
325 set_from(std::future<RT &> &v)
326 {
327 // Get the value from the std::future object. If the future holds
328 // an exception, then the assignment fails, we exit the function via the
329 // exception right away, and value_is_initialized is not set to true --
330 // that's something we can check later on.
331 value = &v.get();
332 value_is_initialized = true;
333 }
334 };
335
336
355 template <>
356 struct return_value<void>
357 {
358 using reference_type = void;
359
360 static inline void
362 {}
363
364
371 inline void
372 set_from(std::future<void> &)
373 {}
374 };
375 } // namespace internal
376
377
378
379 namespace internal
380 {
388 template <typename T>
390 {
391 static T
392 act(T &t)
393 {
394 return t;
395 }
396 };
397
398
399
407 template <typename T>
408 struct maybe_make_ref<T &>
409 {
410 static std::reference_wrapper<T>
411 act(T &t)
412 {
413 return std::ref(t);
414 }
415 };
416
417
418
425 template <typename RT, typename Function>
427 (std::invocable<Function> &&
428 std::convertible_to<std::invoke_result_t<Function>, RT>))
430 {
431 promise.set_value(function());
432 }
433
434
444 template <typename Function>
445 DEAL_II_CXX20_REQUIRES((std::invocable<Function>))
447 std::promise<void> &promise)
448 {
449 function();
450 promise.set_value();
451 }
452 } // namespace internal
453
454
455
482 template <typename RT = void>
483 class Task
484 {
485 public:
497 Task(const std::function<RT()> &function_object)
498 {
500 {
501#ifdef DEAL_II_WITH_TBB
502 // Create a promise object and from it extract a future that
503 // we can use to refer to the outcome of the task. For reasons
504 // explained below, we can't just create a std::promise object,
505 // but have to make do with a pointer to such an object.
506 std::unique_ptr<std::promise<RT>> promise =
507 std::make_unique<std::promise<RT>>();
508 task_data =
509 std::make_shared<TaskData>(std::move(promise->get_future()));
510
511 // Then start the task, using a task_group object (for just this one
512 // task) that is associated with the TaskData object. Note that we
513 // have to *copy* the function object being executed so that it is
514 // guaranteed to live on the called thread as well -- the copying is
515 // facilitated by capturing the 'function_object' variable by value.
516 //
517 // We also have to *move* the promise object into the new task's
518 // memory space because promises can not be copied and we can't refer
519 // to it by reference because it's a local variable of the current
520 // (surrounding) function that may go out of scope before the promise
521 // is ultimately set. This leads to a conundrum: if we had just
522 // declared 'promise' as an object of type std::promise, then we could
523 // capture it in the lambda function via
524 // [..., promise=std::move(promise)]() {...}
525 // and set the promise in the body of the lambda. But setting a
526 // promise is a non-const operation on the promise, and so we would
527 // actually have to declare the lambda function as 'mutable' because
528 // by default, lambda captures are 'const'. That is, we would have
529 // to write
530 // [..., promise=std::move(promise)]() mutable {...}
531 // But this leads to other problems: It turns out that the
532 // tbb::task_group::run() function cannot take mutable lambdas as
533 // argument :-(
534 //
535 // We work around this issue by not declaring the 'promise' variable
536 // as an object of type std::promise, but as a pointer to such an
537 // object. This pointer we can move, and the *pointer* itself can
538 // be 'const' (meaning we can leave the lambda as non-mutable)
539 // even though we modify the object *pointed to*. One would think
540 // that a std::unique_ptr would be the right choice for this, but
541 // that's not true: the resulting lambda function can then be
542 // non-mutable, but the lambda function object is not copyable
543 // and at least some TBB variants require that as well. So
544 // instead we move the std::unique_ptr used above into a
545 // std::shared_ptr to be stored within the lambda function object.
546 task_data->task_group->run(
547 [function_object,
548 promise =
549 std::shared_ptr<std::promise<RT>>(std::move(promise))]() {
550 try
551 {
552 internal::evaluate_and_set_promise(function_object, *promise);
553 }
554 catch (...)
555 {
556 try
557 {
558 // store anything thrown in the promise
559 promise->set_exception(std::current_exception());
560 }
561 catch (...)
562 {
563 // set_exception() may throw too. But ignore this on
564 // the task.
565 }
566 }
567 });
568
569#else
570 // If no threading library is supported, just fall back onto C++11
571 // facilities. The problem with this is that the standard does
572 // not actually say what std::async should do. The first
573 // argument to that function can be std::launch::async or
574 // std::launch::deferred, or both. The *intent* of the standard's
575 // authors was probably that if one sets it to
576 // std::launch::async | std::launch::deferred,
577 // that the task is run in a thread pool. But at least as of
578 // 2021, GCC doesn't do that: It just runs it on a new thread.
579 // If one chooses std::launch::deferred, it runs the task on
580 // the same thread but only when one calls join() on the task's
581 // std::future object. In the former case, this leads to
582 // oversubscription, in the latter case to undersubscription of
583 // resources. We choose oversubscription here.
584 //
585 // The issue illustrates why relying on external libraries
586 // with task schedulers is the way to go.
587 task_data = std::make_shared<TaskData>(
588 std::async(std::launch::async | std::launch::deferred,
589 function_object));
590#endif
591 }
592 else
593 {
594 // Only one thread allowed. So let the task run to completion
595 // and just emplace a 'ready' future.
596 //
597 // The design of std::promise/std::future is unclear, but it
598 // seems that the intent is to obtain the std::future before
599 // we set the std::promise. So create the TaskData object at
600 // the top and then run the task and set the returned
601 // value. Since everything here happens sequentially, it
602 // really doesn't matter in which order all of this is
603 // happening.
604 std::promise<RT> promise;
605 task_data = std::make_shared<TaskData>(promise.get_future());
606 try
607 {
608 internal::evaluate_and_set_promise(function_object, promise);
609 }
610 catch (...)
611 {
612 try
613 {
614 // store anything thrown in the promise
615 promise.set_exception(std::current_exception());
616 }
617 catch (...)
618 {
619 // set_exception() may throw too. But ignore this on
620 // the task.
621 }
622 }
623 }
624 }
625
634 Task() = default;
635
648 Task(const Task &other) = default;
649
663 Task(Task &&other) noexcept = default;
664
678 Task &
679 operator=(const Task &other) = default;
680
695 Task &
696 operator=(Task &&other) noexcept = default;
697
729 void
730 join() const
731 {
732 // Make sure we actually have a task that we can wait for.
734
735 task_data->wait();
736 }
737
750 bool
751 joinable() const
752 {
753 return (task_data != nullptr);
754 }
755
756
808 {
809 // Make sure we actually have a task that we can wait for.
811
812 // Then return the promised object. If necessary, wait for the promise to
813 // be set.
814 return task_data->get();
815 }
816
817
827 "The current object is not associated with a task that "
828 "can be joined. It may have been detached, or you "
829 "may have already joined it in the past.");
831 private:
841 {
842 public:
847 TaskData(std::future<RT> &&future) noexcept
848 : future(std::move(future))
849 , task_has_finished(false)
850#ifdef DEAL_II_WITH_TBB
851 , task_group(std::make_unique<tbb::task_group>())
852#endif
853 {}
854
859 TaskData(const TaskData &) = delete;
860
865 TaskData(TaskData &&) = delete;
866
871 TaskData &
872 operator=(const TaskData &) = delete;
873
878 TaskData &
879 operator=(TaskData &&) = delete;
880
888 ~TaskData() noexcept
889 {
890 // Explicitly wait for the results to be ready. This class stores
891 // a std::future object, and we could just let the compiler generate
892 // the destructor which would then call the destructor of std::future
893 // which *may* block until the future is ready. As explained in
894 // https://en.cppreference.com/w/cpp/thread/future/~future
895 // this is only a *may*, not a *must*. (The standard does not
896 // appear to say anything about it at all.) As a consequence,
897 // let's be explicit about waiting.
898 //
899 // One of the corner cases we have to worry about is that if a task
900 // ends by throwing an exception, then wait() will re-throw that
901 // exception on the thread that calls it, the first time around
902 // someone calls wait() (or the return_value() function of the
903 // surrounding class). So if we get to this constructor and an exception
904 // is thrown by wait(), then that means that the last Task object
905 // referring to a task is going out of scope with nobody having
906 // ever checked the return value of the task itself. In that case,
907 // one could argue that they would also not have cared about whether
908 // an exception is thrown, and that we should simply ignore the
909 // exception. This is what we do here. It is also the simplest solution,
910 // because we don't know what one should do with the exception to begin
911 // with: destructors aren't allowed to throw exceptions, so we can't
912 // just rethrow it here if one had been triggered.
913 try
914 {
915 wait();
916 }
917 catch (...)
918 {}
919 }
920
926 void
928 {
929 // If we have previously already moved the result, then we don't
930 // need a lock and can just return.
932 return;
933
934 // Else, we need to go under a lock and try again. A different thread
935 // may have waited and finished the task since then, so we have to try
936 // a second time. (This is Schmidt's double-checking pattern.)
937 std::lock_guard<std::mutex> lock(mutex);
939 return;
940 else
941 {
942#ifdef DEAL_II_WITH_TBB
943 // If we build on the TBB, then we can't just wait for the
944 // std::future object to get ready. Apparently the TBB happily
945 // enqueues a task into an arena and then just sits on it without
946 // ever executing it unless someone expresses an interest in the
947 // task. The way to avoid this is to add the task to a
948 // tbb::task_group, and then here wait for the single task
949 // associated with that task group.
950 //
951 // If we get here, we know for a fact that atomically
952 // (because under a lock), no other thread has so far
953 // determined that we are finished and removed the
954 // 'task_group' object. So we know that the pointer is
955 // still valid. But we also know that, because below we
956 // set the task_has_finished flag to 'true', that no other
957 // thread will ever get back to this point and query the
958 // 'task_group' object, so we can delete it.
959 task_group->wait();
960 task_group.reset();
961#endif
962
963 // Wait for the task to finish and then move its
964 // result. (We could have made the set_from() function
965 // that we call here wait for the future to be ready --
966 // which happens implicitly when it calls future.get() --
967 // but that would have required putting an explicit
968 // future.wait() into the implementation of
969 // internal::return_value<void>::set_from(), which is a
970 // bit awkward: that class doesn't actually need to set
971 // anything, and so it looks odd to have the explicit call
972 // to future.wait() in the set_from() function. Avoid the
973 // issue by just explicitly calling future.wait() here.)
974 future.wait();
975
976 // Acquire the returned object. If the task ended in an
977 // exception, `set_from` will call `std::future::get`, which
978 // will throw an exception. This leaves `returned_object` in
979 // an undefined state, but moreover we would bypass setting
980 // `task_has_finished=true` below. So catch the exception
981 // for just long enough that we can set that flag, and then
982 // re-throw it:
983 try
984 {
985 returned_object.set_from(future);
986 }
987 catch (...)
988 {
989 task_has_finished = true;
990 throw;
991 }
992
993 // If we got here, the task has ended without an exception and
994 // we can safely set the flag and return.
995 task_has_finished = true;
996 }
997 }
998
999
1000
1003 {
1004 wait();
1005 return returned_object.get();
1006 }
1007
1008 private:
1013 std::mutex mutex;
1014
1019 std::future<RT> future;
1020
1039 std::atomic<bool> task_has_finished;
1040
1046
1047#ifdef DEAL_II_WITH_TBB
1055 std::unique_ptr<tbb::task_group> task_group;
1056
1057 friend class Task<RT>;
1058#endif
1059 };
1060
1065 std::shared_ptr<TaskData> task_data;
1066 };
1067
1068
1069
1089 template <typename RT>
1090 inline Task<RT>
1091 new_task(const std::function<RT()> &function)
1092 {
1093 return Task<RT>(function);
1094 }
1095
1096
1097
1175 template <typename FunctionObjectType>
1176 DEAL_II_CXX20_REQUIRES((std::invocable<FunctionObjectType>))
1177 inline auto new_task(FunctionObjectType function_object)
1178 -> Task<decltype(function_object())>
1179 {
1180 using return_type = decltype(function_object());
1182 return new_task(std::function<return_type()>(function_object));
1183 }
1184
1185
1186
1193 template <typename RT, typename... Args>
1194 inline Task<RT>
1195 new_task(RT (*fun_ptr)(Args...), std_cxx20::type_identity_t<Args>... args)
1196 {
1197 auto dummy = std::make_tuple(internal::maybe_make_ref<Args>::act(args)...);
1198 return new_task(
1199 [dummy, fun_ptr]() -> RT { return std::apply(fun_ptr, dummy); });
1200 }
1201
1202
1203
1244 template <
1245 typename FunctionObject,
1246 typename... Args,
1247 typename = std::enable_if_t<std::is_invocable_v<FunctionObject, Args...>>,
1248 typename = std::enable_if_t<std::is_function_v<FunctionObject> == false>,
1249 typename =
1250 std::enable_if_t<std::is_member_pointer_v<FunctionObject> == false>,
1251 typename = std::enable_if_t<std::is_pointer_v<FunctionObject> == false>>
1252 inline Task<std::invoke_result_t<FunctionObject, Args...>>
1253 new_task(const FunctionObject &fun, Args &&...args)
1254 {
1255 using RT = std::invoke_result_t<FunctionObject, Args...>;
1256 auto dummy = std::make_tuple(std::forward<Args>(args)...);
1257 return new_task([dummy, fun]() -> RT { return std::apply(fun, dummy); });
1258 }
1259
1260
1261
1268 template <typename RT, typename C, typename... Args>
1269 inline Task<RT>
1270 new_task(RT (C::*fun_ptr)(Args...),
1271 std_cxx20::type_identity_t<C> &c,
1272 std_cxx20::type_identity_t<Args>... args)
1273 {
1274 // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
1275 return new_task(std::function<RT()>(std::bind(
1276 fun_ptr, std::ref(c), internal::maybe_make_ref<Args>::act(args)...)));
1277 }
1278
1285 template <typename RT, typename C, typename... Args>
1286 inline Task<RT>
1287 new_task(RT (C::*fun_ptr)(Args...) const,
1288 std_cxx20::type_identity_t<const C> &c,
1289 std_cxx20::type_identity_t<Args>... args)
1290 {
1291 // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
1292 return new_task(std::function<RT()>(std::bind(
1293 fun_ptr, std::cref(c), internal::maybe_make_ref<Args>::act(args)...)));
1294 }
1295
1296
1297 // ------------------------ TaskGroup -------------------------------------
1298
1311 template <typename RT = void>
1313 {
1314 public:
1318 TaskGroup &
1320 {
1321 tasks.push_back(t);
1322 return *this;
1323 }
1324
1325
1333 std::size_t
1334 size() const
1335 {
1336 return tasks.size();
1337 }
1338
1353 std::vector<RT>
1355 {
1356 std::vector<RT> results;
1357 results.reserve(size());
1358 for (auto &t : tasks)
1359 results.emplace_back(std::move(t.return_value()));
1360 return results;
1361 }
1362
1363
1370 void
1371 join_all() const
1372 {
1373 for (auto &t : tasks)
1374 t.join();
1375 }
1376
1377 private:
1381 std::list<Task<RT>> tasks;
1382 };
1383
1384} // namespace Threads
1385
1392#endif
static void initialize_multithreading()
static unsigned int n_threads()
TaskGroup & operator+=(const Task< RT > &t)
std::vector< RT > return_values()
std::size_t size() const
std::list< Task< RT > > tasks
TaskData(std::future< RT > &&future) noexcept
TaskData(const TaskData &)=delete
TaskData & operator=(const TaskData &)=delete
std::atomic< bool > task_has_finished
internal::return_value< RT > returned_object
std::unique_ptr< tbb::task_group > task_group
TaskData & operator=(TaskData &&)=delete
TaskData(TaskData &&)=delete
internal::return_value< RT >::reference_type get()
Task(Task &&other) noexcept=default
std::shared_ptr< TaskData > task_data
bool joinable() const
internal::return_value< RT >::reference_type return_value()
void join() const
Task & operator=(Task &&other) noexcept=default
Task(const Task &other)=default
Task()=default
Task(const std::function< RT()> &function_object)
Task & operator=(const Task &other)=default
#define DEAL_II_NAMESPACE_OPEN
Definition config.h:502
#define DEAL_II_CXX20_REQUIRES(condition)
Definition config.h:177
#define DEAL_II_NAMESPACE_CLOSE
Definition config.h:503
Point< 2 > second
Definition grid_out.cc:4614
static ::ExceptionBase & ExcNoTask()
#define Assert(cond, exc)
#define DeclExceptionMsg(Exception, defaulttext)
Definition exceptions.h:494
static ::ExceptionBase & ExcMessage(std::string arg1)
#define AssertThrow(cond, exc)
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)
std::vector< std::pair< unsigned int, unsigned int > > split_interval(const unsigned int begin, const unsigned int end, const unsigned int n_intervals)
std::promise< RT > & promise
void evaluate_and_set_promise(Function &function, std::promise< void > &promise)
void handle_std_exception(const std::exception &exc)
const Iterator const std_cxx20::type_identity_t< Iterator > & end
Definition parallel.h:610
const Iterator & begin
Definition parallel.h:609
typename type_identity< T >::type type_identity_t
Definition type_traits.h:95
STL namespace.
static std::reference_wrapper< T > act(T &t)
void set_from(std::future< RT & > &v)
void set_from(std::future< RT > &v)