Loading [MathJax]/extensions/TeX/AMSsymbols.js
 deal.II version GIT relicensing-3041-g9c4075ddf4 2025-04-07 16:30:00+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\}}\)
All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Modules Pages Concepts
work_stream.h
Go to the documentation of this file.
1// ------------------------------------------------------------------------
2//
3// SPDX-License-Identifier: LGPL-2.1-or-later
4// Copyright (C) 2009 - 2024 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_work_stream_h
16# define dealii_work_stream_h
17
18
19# include <deal.II/base/config.h>
20
29
30# ifdef DEAL_II_WITH_TBB
31# ifdef DEAL_II_TBB_WITH_ONEAPI
32# include <tbb/parallel_pipeline.h>
33# else
34# include <tbb/pipeline.h>
35# endif
36# include <tbb/blocked_range.h>
37# endif
38
39# ifdef DEAL_II_WITH_TASKFLOW
40# include <taskflow/taskflow.hpp>
41# endif
42
43# include <functional>
44# include <iterator>
45# include <list>
46# include <memory>
47# include <utility>
48# include <vector>
49
51
52
53
166namespace WorkStream
167{
172 namespace internal
173 {
179 template <typename Iterator, typename ScratchData, typename CopyData>
181 {
182 std::unique_ptr<ScratchData> scratch_data;
183 std::unique_ptr<CopyData> copy_data;
185
192
193 ScratchAndCopyDataObjects(std::unique_ptr<ScratchData> &&p,
194 std::unique_ptr<CopyData> &&q,
195 const bool in_use)
196 : scratch_data(std::move(p))
197 , copy_data(std::move(q))
198 , currently_in_use(in_use)
199 {}
200
201 // Provide a copy constructor that actually doesn't copy the
202 // internal state. This makes handling ScratchAndCopyDataObjects
203 // easier to handle with STL containers.
207 };
208
214 template <typename ScratchData>
216 {
217 std::unique_ptr<ScratchData> scratch_data;
219
224 : currently_in_use(false)
225 {}
226
227 ScratchDataObject(std::unique_ptr<ScratchData> &&p, const bool in_use)
228 : scratch_data(std::move(p))
229 , currently_in_use(in_use)
230 {}
231
232 ScratchDataObject(ScratchData *p, const bool in_use)
233 : scratch_data(p)
234 , currently_in_use(in_use)
235 {}
236
237 // Provide a copy constructor that actually doesn't copy the
238 // internal state. This makes handling ScratchAndCopyDataObjects
239 // easier to handle with STL containers.
243
244 ScratchDataObject(ScratchDataObject &&o) noexcept = default;
245 };
246
247# ifdef DEAL_II_WITH_TBB
262 namespace tbb_no_coloring
263 {
267 template <typename Iterator, typename ScratchData, typename CopyData>
269 {
270 public:
277 struct ItemType
278 {
283 using ScratchDataList = std::list<ScratchDataObject<ScratchData>>;
284
289 std::vector<Iterator> iterators;
290
296 std::vector<CopyData> copy_datas;
297
303 unsigned int n_iterators;
304
337
342 const ScratchData *sample_scratch_data;
343
349
350
356 : n_iterators(0)
357 , scratch_data(nullptr)
358 , sample_scratch_data(nullptr)
359 , currently_in_use(false)
360 {}
361 };
362
363
369 IteratorRangeToItemStream(const Iterator &begin,
370 const Iterator &end,
371 const unsigned int buffer_size,
372 const unsigned int chunk_size,
373 const ScratchData &sample_scratch_data,
374 const CopyData &sample_copy_data)
375 : remaining_iterator_range(begin, end)
376 , item_buffer(buffer_size)
379 {
380 // initialize the elements of the ring buffer
381 for (auto &item : item_buffer)
382 {
383 Assert(item.n_iterators == 0, ExcInternalError());
384
385 item.iterators.resize(chunk_size,
387 item.scratch_data = &thread_local_scratch;
388 item.sample_scratch_data = &sample_scratch_data;
389 item.copy_datas.resize(chunk_size, sample_copy_data);
390 item.currently_in_use = false;
391 }
392 }
393
394
398 ItemType *
400 {
401 // find first unused item. we know that there must be one
402 // because we have set the maximal number of tokens in flight
403 // and have set the ring buffer to have exactly this size. so
404 // if this function is called, we know that less than the
405 // maximal number of items in currently in flight
406 //
407 // note that we need not lock access to this array since
408 // the current stage is run sequentially and we can therefore
409 // enter the following block only once at any given time.
410 // thus, there can be no race condition between checking that
411 // a flag is false and setting it to true. (there may be
412 // another thread where we release items and set 'false'
413 // flags to 'true', but that too does not produce any
414 // problems)
415 ItemType *current_item = nullptr;
416 for (unsigned int i = 0; i < item_buffer.size(); ++i)
417 if (item_buffer[i].currently_in_use == false)
418 {
420 current_item = &item_buffer[i];
421 break;
422 }
423 Assert(current_item != nullptr,
424 ExcMessage("This can't be. There must be a free item!"));
425
426 // initialize the next item. it may
427 // consist of at most chunk_size
428 // elements
429 current_item->n_iterators = 0;
430 while ((remaining_iterator_range.first !=
431 remaining_iterator_range.second) &&
432 (current_item->n_iterators < chunk_size))
433 {
434 current_item->iterators[current_item->n_iterators] =
436
438 ++current_item->n_iterators;
439 }
440
441 if (current_item->n_iterators == 0)
442 // there were no items
443 // left. terminate the pipeline
444 return nullptr;
445 else
446 return current_item;
447 }
448
449 private:
454 std::pair<Iterator, Iterator> remaining_iterator_range;
455
459 std::vector<ItemType> item_buffer;
460
493
499 const ScratchData &sample_scratch_data;
500
507 const unsigned int chunk_size;
508 };
509
510
511
512 template <typename Worker,
513 typename Copier,
514 typename Iterator,
515 typename ScratchData,
516 typename CopyData>
517 void
518 run(const Iterator &begin,
520 Worker worker,
521 Copier copier,
522 const ScratchData &sample_scratch_data,
523 const CopyData &sample_copy_data,
524 const unsigned int queue_length,
525 const unsigned int chunk_size)
526 {
527 using ItemType = typename IteratorRangeToItemStream<Iterator,
528 ScratchData,
529 CopyData>::ItemType;
530
531 // Define the three stages of the pipeline:
532
533 //
534 // ----- Stage 1 -----
535 //
536 // The first stage is the one that provides us with chunks of data
537 // to work on (the stream of "items"). This stage will run sequentially.
539 iterator_range_to_item_stream(begin,
540 end,
541 queue_length,
542 chunk_size,
543 sample_scratch_data,
544 sample_copy_data);
545 auto item_generator = [&](tbb::flow_control &fc) -> ItemType * {
546 if (const auto item = iterator_range_to_item_stream.get_item())
547 return item;
548 else
549 {
550 fc.stop();
551 return nullptr;
552 }
553 };
554
555 //
556 // ----- Stage 2 -----
557 //
558 // The second stage is the one that does the actual work. This is the
559 // stage that runs in parallel
560 auto item_worker =
561 [worker =
562 std::function<void(const Iterator &, ScratchData &, CopyData &)>(
563 worker),
564 copier_exists =
565 static_cast<bool>(std::function<void(const CopyData &)>(copier))](
566 ItemType *current_item) {
567 // we need to find an unused scratch data object in the list that
568 // corresponds to the current thread and then mark it as used. if
569 // we can't find one, create one
570 //
571 // as discussed in the discussion of the documentation of the
572 // IteratorRangeToItemStream::scratch_data variable, there is no
573 // need to synchronize access to this variable using a mutex
574 // as long as we have no yield-point in between. this means that
575 // we can't take an iterator into the list now and expect it to
576 // still be valid after calling the worker, but we at least do
577 // not have to lock the following section
578 ScratchData *scratch_data = nullptr;
579 {
580 // see if there is an unused object. if so, grab it and mark
581 // it as used
582 for (auto &p : current_item->scratch_data->get())
583 if (p.currently_in_use == false)
584 {
585 scratch_data = p.scratch_data.get();
586 p.currently_in_use = true;
587
588 break;
589 }
590
591 // if no object was found, create one and mark it as used
592 if (scratch_data == nullptr)
593 {
594 scratch_data =
595 new ScratchData(*current_item->sample_scratch_data);
596 current_item->scratch_data->get().emplace_back(scratch_data,
597 true);
598 }
599 };
600
601 // then call the worker function on each element of the chunk we
602 // were given. since these worker functions are called on separate
603 // threads, nothing good can happen if they throw an exception and
604 // we are best off catching it and showing an error message
605 for (unsigned int i = 0; i < current_item->n_iterators; ++i)
606 {
607 try
608 {
609 if (worker)
610 worker(current_item->iterators[i],
611 *scratch_data,
612 current_item->copy_datas[i]);
613 }
614 catch (const std::exception &exc)
615 {
617 }
618 catch (...)
619 {
621 }
622 }
623
624 // finally mark the scratch object as unused again. as above, there
625 // is no need to lock anything here since the object we work on
626 // is thread-local
627 for (auto &p : current_item->scratch_data->get())
628 if (p.scratch_data.get() == scratch_data)
629 {
630 Assert(p.currently_in_use == true, ExcInternalError());
631 p.currently_in_use = false;
632
633 break;
634 }
635
636 // if there is no copier, mark current item as usable again
637 if (copier_exists == false)
638 current_item->currently_in_use = false;
639
640
641 // Then return the original pointer
642 // to the now modified object. The copier will work on it next.
643 return current_item;
644 };
645
646 //
647 // ----- Stage 3 -----
648 //
649 // The last stage is the one that copies data from the CopyData objects
650 // to the final destination. This stage runs sequentially again.
651 auto item_copier = [copier = std::function<void(const CopyData &)>(
652 copier)](ItemType *current_item) {
653 if (copier)
654 {
655 // Initiate copying data. For the same reasons as in the worker
656 // class above, catch exceptions rather than letting them
657 // propagate into unknown territories:
658 for (unsigned int i = 0; i < current_item->n_iterators; ++i)
659 {
660 try
661 {
662 copier(current_item->copy_datas[i]);
663 }
664 catch (const std::exception &exc)
665 {
667 }
668 catch (...)
669 {
671 }
672 }
673 }
674 // mark current item as usable again
675 current_item->currently_in_use = false;
676 };
677
678
679 // Now we just have to set up the pipeline and run it:
680 auto tbb_item_stream_filter = tbb::make_filter<void, ItemType *>(
681# ifdef DEAL_II_TBB_WITH_ONEAPI
682 tbb::filter_mode::serial_in_order,
683# else
684 tbb::filter::serial,
685# endif
686 item_generator);
687
688 auto tbb_worker_filter = tbb::make_filter<ItemType *, ItemType *>(
689# ifdef DEAL_II_TBB_WITH_ONEAPI
690 tbb::filter_mode::parallel,
691# else
692 tbb::filter::parallel,
693# endif
694 item_worker);
695
696 auto tbb_copier_filter = tbb::make_filter<ItemType *, void>(
697# ifdef DEAL_II_TBB_WITH_ONEAPI
698 tbb::filter_mode::serial_in_order,
699# else
700 tbb::filter::serial,
701# endif
702 item_copier);
703
704 tbb::parallel_pipeline(queue_length,
705 tbb_item_stream_filter & tbb_worker_filter &
706 tbb_copier_filter);
707 }
708
709 } // namespace tbb_no_coloring
710# endif // DEAL_II_WITH_TBB
711
712
713
714# ifdef DEAL_II_WITH_TASKFLOW
722 namespace taskflow_no_coloring
723 {
730 template <typename Worker,
731 typename Copier,
732 typename Iterator,
733 typename ScratchData,
734 typename CopyData>
735 void
736 run(const Iterator &begin,
738 Worker worker,
739 Copier copier,
740 const ScratchData &sample_scratch_data,
741 const CopyData &sample_copy_data,
742 const unsigned int /*queue_length*/ = 2 *
744 const unsigned int chunk_size = 8)
745
746 {
747 tf::Executor &executor = MultithreadInfo::get_taskflow_executor();
748 tf::Taskflow taskflow;
749
750 using ScratchDataList = std::list<ScratchDataObject<ScratchData>>;
751
753 thread_safe_scratch_data_list;
754
755 tf::Task last_copier;
756
757
758 // A collection of chunk_size copy objects which each represent the
759 // contribution of a single worker. Chunk_size workers are thus chunked
760 // together by having their outputs stored in this object in a single
761 // task.
762 struct CopyChunk
763 {
764 std::vector<CopyData> copy_datas;
765
766 CopyChunk(const unsigned int chunk_size,
767 const CopyData &sample_copy_data)
768 : copy_datas(chunk_size, sample_copy_data)
769 {}
770 };
771
772 // idx is used to connect each worker to its copier as communication
773 // between tasks is not supported. It does this by providing a unique
774 // index in the vector of pointers copy_datas at which the copy data
775 // object where the work done by work task #idx is stored
776 unsigned int idx = 0;
777
778 // A collection of handles to a CopyChunk object for each chunk. The
779 // actual data will be allocated only when a worker arrives at this
780 // chunk and will be freed as soon as the result has been copied.
781 std::vector<std::unique_ptr<CopyChunk>> copy_chunks;
782
783 std::vector<Iterator> chunk;
784 chunk.reserve(chunk_size);
785
786 unsigned int total_chunks = 0;
787 unsigned int chunk_counter = 0;
788
789 // Generate a static task graph. Here we generate a task for each cell
790 // that will be worked on. The tasks are not executed until all of them
791 // are created, this code runs sequentially.
792 for (Iterator it = begin; it != end;)
793 {
794 for (; (chunk_counter < chunk_size) && (it != end);
795 ++it, ++chunk_counter)
796 chunk.emplace_back(it);
797 ++total_chunks;
798 // Create a worker task.
799 auto worker_task =
800 taskflow
801 .emplace([chunk,
802 idx,
803 chunk_counter,
804 &thread_safe_scratch_data_list,
805 &sample_scratch_data,
806 &sample_copy_data,
807 &copy_chunks,
808 &worker]() {
809 ScratchData *scratch_data = nullptr;
810
811 ScratchDataList &scratch_data_list =
812 thread_safe_scratch_data_list.get();
813 // We need to find an unused scratch data object in the list
814 // that corresponds to the current thread and then mark it as
815 // used. if we can't find one, create one. There is no need to
816 // synchronize access to this variable using a mutex since
817 // each object is local to its own thread.
818 for (auto &p : scratch_data_list)
819 {
820 if (p.currently_in_use == false)
821 {
822 scratch_data = p.scratch_data.get();
823 p.currently_in_use = true;
824 break;
825 }
826 }
827 // If no element in the list was found, create
828 // one and mark it as used.
829 if (scratch_data == nullptr)
830 {
831 scratch_data_list.emplace_back(
832 std::make_unique<ScratchData>(sample_scratch_data),
833 true);
834 scratch_data =
835 scratch_data_list.back().scratch_data.get();
836 }
837
838 // Create a unique copy chunk object where this
839 // worker's work will be stored.
840 copy_chunks[idx] =
841 std::make_unique<CopyChunk>(chunk_counter,
842 sample_copy_data);
843 auto copy_chunk = copy_chunks[idx].get();
844 unsigned int i = 0;
845 for (auto &it : chunk)
846 {
847 worker(it, *scratch_data, copy_chunk->copy_datas[i]);
848 ++i;
849 }
850
851 // Find our currently used scratch data and
852 // mark it as unused.
853 for (auto &p : scratch_data_list)
854 {
855 if (p.scratch_data.get() == scratch_data)
856 {
857 Assert(p.currently_in_use == true,
859 p.currently_in_use = false;
860 }
861 }
862 })
863 .name("worker");
864
865 // Create a copier task. This task is a separate object from the
866 // worker task.
867 tf::Task copier_task =
868 taskflow
869 .emplace([idx, &copy_chunks, &copier]() {
870 auto copy_chunk = copy_chunks[idx].get();
871 for (auto &copy_data : copy_chunk->copy_datas)
872 {
873 copier(copy_data);
874 }
875 // Finally free the memory.
876 copy_chunks[idx].reset();
877 })
878 .name("copy");
879
880 // Ensure the copy task runs after the worker task.
881 worker_task.precede(copier_task);
882
883 // Ensure that only one copy task can run at a time. The code
884 // below makes each copy task wait until the previous one has
885 // finished before it can start
886 if (!last_copier.empty())
887 last_copier.precede(copier_task);
888
889 // Keep a handle to the last copier. Tasks in taskflow are
890 // basically handles to internally stored data, so this does not
891 // perform a copy:
892 last_copier = copier_task;
893 ++idx;
894 chunk_counter = 0;
895 chunk.clear();
896 }
897 copy_chunks.resize(total_chunks);
898 // Now we run all the tasks in the task graph. They will be run in
899 // parallel and are eligible to run when their dependencies established
900 // above are met.
901 executor.run(taskflow).wait();
902 }
903 } // namespace taskflow_no_coloring
904# endif
905
912 namespace sequential
913 {
917 template <typename Worker,
918 typename Copier,
919 typename Iterator,
920 typename ScratchData,
921 typename CopyData>
922 void
923 run(const Iterator &begin,
925 Worker worker,
926 Copier copier,
927 const ScratchData &sample_scratch_data,
928 const CopyData &sample_copy_data)
929 {
930 // need to copy the sample since it is marked const
931 ScratchData scratch_data = sample_scratch_data;
932 CopyData copy_data = sample_copy_data; // NOLINT
933
934 // Optimization: Check if the functions are not the zero function. To
935 // check zero-ness, create a C++ function out of it:
936 const bool have_worker =
937 (static_cast<const std::function<
938 void(const Iterator &, ScratchData &, CopyData &)> &>(worker)) !=
939 nullptr;
940 const bool have_copier =
941 (static_cast<const std::function<void(const CopyData &)> &>(
942 copier)) != nullptr;
943
944 // Finally loop over all items and perform the necessary work:
945 for (Iterator i = begin; i != end; ++i)
946 {
947 if (have_worker)
948 worker(i, scratch_data, copy_data);
949 if (have_copier)
950 copier(copy_data);
951 }
952 }
953
954
955
959 template <typename Worker,
960 typename Copier,
961 typename Iterator,
962 typename ScratchData,
963 typename CopyData>
964 void
965 run(const std::vector<std::vector<Iterator>> &colored_iterators,
966 Worker worker,
967 Copier copier,
968 const ScratchData &sample_scratch_data,
969 const CopyData &sample_copy_data)
970 {
971 // need to copy the sample since it is marked const
972 ScratchData scratch_data = sample_scratch_data;
973 CopyData copy_data = sample_copy_data; // NOLINT
974
975 // Optimization: Check if the functions are not the zero function. To
976 // check zero-ness, create a C++ function out of it:
977 const bool have_worker =
978 (static_cast<const std::function<
979 void(const Iterator &, ScratchData &, CopyData &)> &>(worker)) !=
980 nullptr;
981 const bool have_copier =
982 (static_cast<const std::function<void(const CopyData &)> &>(
983 copier)) != nullptr;
984
985 // Finally loop over all items and perform the necessary work:
986 for (unsigned int color = 0; color < colored_iterators.size(); ++color)
987 if (colored_iterators[color].size() > 0)
988 for (auto &it : colored_iterators[color])
989 {
990 if (have_worker)
991 worker(it, scratch_data, copy_data);
992 if (have_copier)
993 copier(copy_data);
994 }
995 }
996
997 } // namespace sequential
998
999
1000
1001# ifdef DEAL_II_WITH_TBB
1009 namespace tbb_colored
1010 {
1016 template <typename Iterator, typename ScratchData, typename CopyData>
1018 {
1019 public:
1024 const std::function<void(const Iterator &, ScratchData &, CopyData &)>
1025 &worker,
1026 const std::function<void(const CopyData &)> &copier,
1027 const ScratchData &sample_scratch_data,
1028 const CopyData &sample_copy_data)
1029 : worker(worker)
1030 , copier(copier)
1033 {}
1034
1035
1040 void
1041 operator()(const tbb::blocked_range<
1042 typename std::vector<Iterator>::const_iterator> &range)
1043 {
1044 // we need to find an unused scratch and corresponding copy
1045 // data object in the list that corresponds to the current
1046 // thread and then mark it as used. If we can't find one,
1047 // create one as discussed in the discussion of the documentation
1048 // of the IteratorRangeToItemStream::scratch_data variable,
1049 // there is no need to synchronize access to this variable
1050 // using a mutex as long as we have no yield-point in between.
1051 // This means that we can't take an iterator into the list
1052 // now and expect it to still be valid after calling the worker,
1053 // but we at least do not have to lock the following section.
1054 ScratchData *scratch_data = nullptr;
1055 CopyData *copy_data = nullptr;
1056 {
1057 ScratchAndCopyDataList &scratch_and_copy_data_list = data.get();
1058
1059 // see if there is an unused object. if so, grab it and mark
1060 // it as used
1061 for (typename ScratchAndCopyDataList::iterator p =
1062 scratch_and_copy_data_list.begin();
1063 p != scratch_and_copy_data_list.end();
1064 ++p)
1065 if (p->currently_in_use == false)
1066 {
1067 scratch_data = p->scratch_data.get();
1068 copy_data = p->copy_data.get();
1069 p->currently_in_use = true;
1070 break;
1071 }
1072
1073 // if no element in the list was found, create one and mark it as
1074 // used
1075 if (scratch_data == nullptr)
1076 {
1077 Assert(copy_data == nullptr, ExcInternalError());
1078
1079 scratch_and_copy_data_list.emplace_back(
1080 std::make_unique<ScratchData>(sample_scratch_data),
1081 std::make_unique<CopyData>(sample_copy_data),
1082 true);
1083 scratch_data =
1084 scratch_and_copy_data_list.back().scratch_data.get();
1085 copy_data = scratch_and_copy_data_list.back().copy_data.get();
1086 }
1087 }
1088
1089 // then call the worker and copier functions on each
1090 // element of the chunk we were given.
1091 for (typename std::vector<Iterator>::const_iterator p = range.begin();
1092 p != range.end();
1093 ++p)
1094 {
1095 try
1096 {
1097 if (worker)
1098 worker(*p, *scratch_data, *copy_data);
1099 if (copier)
1100 copier(*copy_data);
1101 }
1102 catch (const std::exception &exc)
1103 {
1105 }
1106 catch (...)
1107 {
1109 }
1110 }
1111
1112 // finally mark the scratch object as unused again. as above, there
1113 // is no need to lock anything here since the object we work on
1114 // is thread-local
1115 {
1116 ScratchAndCopyDataList &scratch_and_copy_data_list = data.get();
1117
1118 for (typename ScratchAndCopyDataList::iterator p =
1119 scratch_and_copy_data_list.begin();
1120 p != scratch_and_copy_data_list.end();
1121 ++p)
1122 if (p->scratch_data.get() == scratch_data)
1123 {
1124 Assert(p->currently_in_use == true, ExcInternalError());
1125 p->currently_in_use = false;
1126 }
1127 }
1128 }
1129
1130 private:
1131 using ScratchAndCopyDataObjects = typename internal::
1132 ScratchAndCopyDataObjects<Iterator, ScratchData, CopyData>;
1133
1138 using ScratchAndCopyDataList = std::list<ScratchAndCopyDataObjects>;
1139
1141
1146 const std::function<void(const Iterator &, ScratchData &, CopyData &)>
1148
1153 const std::function<void(const CopyData &)> copier;
1154
1158 const ScratchData &sample_scratch_data;
1159 const CopyData &sample_copy_data;
1160 };
1161
1165 template <typename Worker,
1166 typename Copier,
1167 typename Iterator,
1168 typename ScratchData,
1169 typename CopyData>
1170 void
1171 run(const std::vector<std::vector<Iterator>> &colored_iterators,
1172 Worker worker,
1173 Copier copier,
1174 const ScratchData &sample_scratch_data,
1175 const CopyData &sample_copy_data,
1176 const unsigned int chunk_size)
1177 {
1178 // loop over the various colors of what we're given
1179 for (unsigned int color = 0; color < colored_iterators.size(); ++color)
1180 if (colored_iterators[color].size() > 0)
1181 {
1182 using WorkerAndCopier = internal::tbb_colored::
1183 WorkerAndCopier<Iterator, ScratchData, CopyData>;
1184
1185 WorkerAndCopier worker_and_copier(worker,
1186 copier,
1187 sample_scratch_data,
1188 sample_copy_data);
1189
1191 colored_iterators[color].begin(),
1192 colored_iterators[color].end(),
1193 [&worker_and_copier](
1194 const tbb::blocked_range<
1195 typename std::vector<Iterator>::const_iterator> &range) {
1196 worker_and_copier(range);
1197 },
1198 chunk_size);
1199 }
1200 }
1201
1202 } // namespace tbb_colored
1203# endif // DEAL_II_WITH_TBB
1204
1205
1206
1207# ifdef DEAL_II_WITH_TASKFLOW
1213 namespace taskflow_colored
1214 {
1221 template <typename Worker,
1222 typename Copier,
1223 typename Iterator,
1224 typename ScratchData,
1225 typename CopyData>
1226 void
1227 run(const std::vector<std::vector<Iterator>> &colored_iterators,
1228 Worker worker,
1229 Copier copier,
1230 const ScratchData &sample_scratch_data,
1231 const CopyData &sample_copy_data,
1232 const unsigned int /*queue_length*/ = 2 *
1234 const unsigned int chunk_size = 8)
1235
1236 {
1237 tf::Executor &executor = MultithreadInfo::get_taskflow_executor();
1238 using ScratchAndCopyDataObjects = typename internal::
1239 ScratchAndCopyDataObjects<Iterator, ScratchData, CopyData>;
1240
1241 using ScratchAndCopyDataList = std::list<ScratchAndCopyDataObjects>;
1242
1244 thread_safe_scratch_and_copy_data_list;
1245
1246 tf::Taskflow taskflow;
1247
1248 // Create a "future" object which eventually contains the execution
1249 // result of a taskflow graph and can be used to yield execution
1250 tf::Future<void> execution_future;
1251
1252 const bool have_worker =
1253 (static_cast<const std::function<
1254 void(const Iterator &, ScratchData &, CopyData &)> &>(worker)) !=
1255 nullptr;
1256 const bool have_copier =
1257 (static_cast<const std::function<void(const CopyData &)> &>(
1258 copier)) != nullptr;
1259
1260 std::vector<Iterator> chunk;
1261 chunk.reserve(chunk_size);
1262
1263 unsigned int chunk_counter = 0;
1264 // Generate a static task graph. Here we generate a task for each cell
1265 // that will be worked on. The tasks are not executed until all of them
1266 // are created, this code runs sequentially. Cells have been grouped
1267 // into "colors" data from cells in the same color are safe to copy in
1268 // parallel so copying need not be sequential.
1269 for (unsigned int color = 0; color < colored_iterators.size(); ++color)
1270 // Ignore color blocks which are empty.
1271 if (colored_iterators[color].size() > 0)
1272 {
1273 // For each cell queue up a combined worker and copier task. These
1274 // are not yet run.
1275 for (auto it = colored_iterators[color].begin();
1276 it != colored_iterators[color].end();)
1277 {
1278 for (; (chunk_counter < chunk_size) &&
1279 (it != colored_iterators[color].end());
1280 ++it, ++chunk_counter)
1281 chunk.emplace_back(*it);
1282 taskflow
1283 .emplace([chunk,
1284 have_worker,
1285 have_copier,
1286 &thread_safe_scratch_and_copy_data_list,
1287 &sample_scratch_data,
1288 &sample_copy_data,
1289 &worker,
1290 &copier]() {
1291 ScratchData *scratch_data = nullptr;
1292 CopyData *copy_data = nullptr;
1293
1294 ScratchAndCopyDataList &scratch_and_copy_data_list =
1295 thread_safe_scratch_and_copy_data_list.get();
1296 // See if there is an unused object. if so, grab it
1297 // and mark it as used.
1298 for (typename ScratchAndCopyDataList::iterator p =
1299 scratch_and_copy_data_list.begin();
1300 p != scratch_and_copy_data_list.end();
1301 ++p)
1302 {
1303 if (p->currently_in_use == false)
1304 {
1305 scratch_data = p->scratch_data.get();
1306 copy_data = p->copy_data.get();
1307 p->currently_in_use = true;
1308 break;
1309 }
1310 }
1311 // If no element in the list was found, create one and
1312 // mark it as used.
1313 if (scratch_data == nullptr)
1314 {
1315 Assert(copy_data == nullptr, ExcInternalError());
1316 scratch_and_copy_data_list.emplace_back(
1317 std::make_unique<ScratchData>(sample_scratch_data),
1318 std::make_unique<CopyData>(sample_copy_data),
1319 true);
1320 scratch_data = scratch_and_copy_data_list.back()
1321 .scratch_data.get();
1322 copy_data =
1323 scratch_and_copy_data_list.back().copy_data.get();
1324 }
1325
1326 for (const Iterator &it : chunk)
1327 {
1328 if (have_worker)
1329 worker(it, *scratch_data, *copy_data);
1330 if (have_copier)
1331 copier(*copy_data);
1332 }
1333
1334 // Mark objects as free to be used again.
1335 for (typename ScratchAndCopyDataList::iterator p =
1336 scratch_and_copy_data_list.begin();
1337 p != scratch_and_copy_data_list.end();
1338 ++p)
1339 {
1340 if (p->scratch_data.get() == scratch_data)
1341 {
1342 Assert(p->currently_in_use == true,
1344 p->currently_in_use = false;
1345 }
1346 }
1347 })
1348 .name("worker_and_copier");
1349 chunk.clear();
1350 chunk_counter = 0;
1351 }
1352
1353 if (color > 0)
1354 // Wait for the previous color to finish executing
1355 execution_future.wait();
1356 execution_future = executor.run(std::move(taskflow));
1357 }
1358 // Wait for our final execution to finish
1359 if (colored_iterators.size() > 0)
1360 execution_future.wait();
1361 }
1362 } // namespace taskflow_colored
1363# endif // DEAL_II_WITH_TASKFLOW
1364
1365
1366 } // namespace internal
1367
1368
1369
1417 template <typename Worker,
1418 typename Copier,
1419 typename Iterator,
1420 typename ScratchData,
1421 typename CopyData>
1422 void
1423 run(const std::vector<std::vector<Iterator>> &colored_iterators,
1424 Worker worker,
1425 Copier copier,
1426 const ScratchData &sample_scratch_data,
1427 const CopyData &sample_copy_data,
1428 const unsigned int queue_length = 2 * MultithreadInfo::n_threads(),
1429 const unsigned int chunk_size = 8);
1430
1431
1481 template <typename Worker,
1482 typename Copier,
1483 typename Iterator,
1484 typename ScratchData,
1485 typename CopyData>
1486 void
1487 run(const Iterator &begin,
1489 Worker worker,
1490 Copier copier,
1491 const ScratchData &sample_scratch_data,
1492 const CopyData &sample_copy_data,
1493 const unsigned int queue_length = 2 * MultithreadInfo::n_threads(),
1494 const unsigned int chunk_size = 8)
1495 {
1496 Assert(queue_length > 0,
1497 ExcMessage("The queue length must be at least one, and preferably "
1498 "larger than the number of processors on this system."));
1499 (void)queue_length; // removes -Wunused-parameter warning in optimized mode
1500 Assert(chunk_size > 0, ExcMessage("The chunk_size must be at least one."));
1501 (void)chunk_size; // removes -Wunused-parameter warning in optimized mode
1502
1503 // If no work then skip. (only use operator!= for iterators since we may
1504 // not have an equality comparison operator)
1505 if (!(begin != end))
1506 return;
1507
1509 {
1510# if defined(DEAL_II_WITH_TBB) || defined(DEAL_II_WITH_TASKFLOW)
1511 if (static_cast<const std::function<void(const CopyData &)> &>(copier))
1512 {
1513 // If we have a copier, run the algorithm:
1514# if defined(DEAL_II_WITH_TASKFLOW)
1516 end,
1517 worker,
1518 copier,
1519 sample_scratch_data,
1520 sample_copy_data,
1521 queue_length,
1522 chunk_size);
1523# elif defined(DEAL_II_WITH_TBB)
1525 end,
1526 worker,
1527 copier,
1528 sample_scratch_data,
1529 sample_copy_data,
1530 queue_length,
1531 chunk_size);
1532# endif
1533 }
1534 else
1535 {
1536 // There is no copier function. in this case, we have an
1537 // embarrassingly parallel problem where we can
1538 // essentially apply parallel_for. because parallel_for
1539 // requires subdividing the range for which operator- is
1540 // necessary between iterators, it is often inefficient to
1541 // apply it directly to cell ranges and similar iterator
1542 // types for which operator- is expensive or, in fact,
1543 // nonexistent. rather, in that case, we simply copy the
1544 // iterators into a large array and use operator- on
1545 // iterators to this array of iterators.
1546 //
1547 // instead of duplicating code, this is essentially the
1548 // same situation we have in the colored implementation below, so we
1549 // just defer to that place
1550 std::vector<std::vector<Iterator>> all_iterators(1);
1551 for (Iterator p = begin; p != end; ++p)
1552 all_iterators[0].push_back(p);
1553
1554 run(all_iterators,
1555 worker,
1556 copier,
1557 sample_scratch_data,
1558 sample_copy_data,
1559 queue_length,
1560 chunk_size);
1561 }
1562
1563 // exit this function to not run the sequential version below:
1564 return;
1565# endif
1566 }
1567
1568 // no TBB or Taskflow installed or we are requested to run sequentially:
1570 begin, end, worker, copier, sample_scratch_data, sample_copy_data);
1571 }
1572
1573
1574
1582 template <typename Worker,
1583 typename Copier,
1584 typename IteratorRangeType,
1585 typename ScratchData,
1586 typename CopyData,
1587 typename = std::enable_if_t<has_begin_and_end<IteratorRangeType>>>
1588 void
1589 run(IteratorRangeType iterator_range,
1590 Worker worker,
1591 Copier copier,
1592 const ScratchData &sample_scratch_data,
1593 const CopyData &sample_copy_data,
1594 const unsigned int queue_length = 2 * MultithreadInfo::n_threads(),
1595 const unsigned int chunk_size = 8)
1596 {
1597 // Call the function above
1598 run(iterator_range.begin(),
1599 iterator_range.end(),
1600 worker,
1601 copier,
1602 sample_scratch_data,
1603 sample_copy_data,
1604 queue_length,
1605 chunk_size);
1606 }
1607
1608
1609
1613 template <typename Worker,
1614 typename Copier,
1615 typename Iterator,
1616 typename ScratchData,
1617 typename CopyData>
1618 void
1619 run(const IteratorRange<Iterator> &iterator_range,
1620 Worker worker,
1621 Copier copier,
1622 const ScratchData &sample_scratch_data,
1623 const CopyData &sample_copy_data,
1624 const unsigned int queue_length = 2 * MultithreadInfo::n_threads(),
1625 const unsigned int chunk_size = 8)
1626 {
1627 // Call the function above
1628 run(iterator_range.begin(),
1629 iterator_range.end(),
1630 worker,
1631 copier,
1632 sample_scratch_data,
1633 sample_copy_data,
1634 queue_length,
1635 chunk_size);
1636 }
1637
1638
1639
1640 template <typename Worker,
1641 typename Copier,
1642 typename Iterator,
1643 typename ScratchData,
1644 typename CopyData>
1645 void
1646 run(const std::vector<std::vector<Iterator>> &colored_iterators,
1647 Worker worker,
1648 Copier copier,
1649 const ScratchData &sample_scratch_data,
1650 const CopyData &sample_copy_data,
1651 const unsigned int queue_length,
1652 const unsigned int chunk_size)
1653 {
1654 Assert(queue_length > 0,
1655 ExcMessage("The queue length must be at least one, and preferably "
1656 "larger than the number of processors on this system."));
1657 (void)queue_length; // removes -Wunused-parameter warning in optimized mode
1658 Assert(chunk_size > 0, ExcMessage("The chunk_size must be at least one."));
1659 (void)chunk_size; // removes -Wunused-parameter warning in optimized mode
1660
1661
1663 {
1664# ifdef DEAL_II_WITH_TASKFLOW
1665 internal::taskflow_colored::run(colored_iterators,
1666 worker,
1667 copier,
1668 sample_scratch_data,
1669 sample_copy_data,
1670 chunk_size);
1671
1672 // exit this function to not run the sequential version below:
1673 return;
1674# elif defined(DEAL_II_WITH_TBB)
1675 internal::tbb_colored::run(colored_iterators,
1676 worker,
1677 copier,
1678 sample_scratch_data,
1679 sample_copy_data,
1680 chunk_size);
1681
1682 // exit this function to not run the sequential version below:
1683 return;
1684# endif
1685 }
1686
1687 // run all colors sequentially:
1688 {
1689 internal::sequential::run(colored_iterators,
1690 worker,
1691 copier,
1692 sample_scratch_data,
1693 sample_copy_data);
1694 }
1695 }
1696
1697
1698
1740 template <typename MainClass,
1741 typename Iterator,
1742 typename ScratchData,
1743 typename CopyData>
1744 void
1745 run(const Iterator &begin,
1747 MainClass &main_object,
1748 void (MainClass::*worker)(const Iterator &, ScratchData &, CopyData &),
1749 void (MainClass::*copier)(const CopyData &),
1750 const ScratchData &sample_scratch_data,
1751 const CopyData &sample_copy_data,
1752 const unsigned int queue_length = 2 * MultithreadInfo::n_threads(),
1753 const unsigned int chunk_size = 8)
1754 {
1755 // forward to the other function
1756 run(
1757 begin,
1758 end,
1759 [&main_object, worker](const Iterator &iterator,
1760 ScratchData &scratch_data,
1761 CopyData &copy_data) {
1762 (main_object.*worker)(iterator, scratch_data, copy_data);
1763 },
1764 [&main_object, copier](const CopyData &copy_data) {
1765 (main_object.*copier)(copy_data);
1766 },
1767 sample_scratch_data,
1768 sample_copy_data,
1769 queue_length,
1770 chunk_size);
1771 }
1772
1773
1774 template <typename MainClass,
1775 typename Iterator,
1776 typename ScratchData,
1777 typename CopyData>
1778 void
1781 MainClass &main_object,
1782 void (MainClass::*worker)(const Iterator &, ScratchData &, CopyData &),
1783 void (MainClass::*copier)(const CopyData &),
1784 const ScratchData &sample_scratch_data,
1785 const CopyData &sample_copy_data,
1786 const unsigned int queue_length = 2 * MultithreadInfo::n_threads(),
1787 const unsigned int chunk_size = 8)
1788 {
1789 // forward to the other function
1790 run(
1791 begin,
1792 end,
1793 [&main_object, worker](const Iterator &iterator,
1794 ScratchData &scratch_data,
1795 CopyData &copy_data) {
1796 (main_object.*worker)(iterator, scratch_data, copy_data);
1797 },
1798 [&main_object, copier](const CopyData &copy_data) {
1799 (main_object.*copier)(copy_data);
1800 },
1801 sample_scratch_data,
1802 sample_copy_data,
1803 queue_length,
1804 chunk_size);
1805 }
1806
1807
1808
1816 template <typename MainClass,
1817 typename IteratorRangeType,
1818 typename ScratchData,
1819 typename CopyData,
1820 typename = std::enable_if_t<has_begin_and_end<IteratorRangeType>>>
1821 void
1823 IteratorRangeType iterator_range,
1824 MainClass &main_object,
1825 void (MainClass::*worker)(
1826 const typename std_cxx20::type_identity_t<IteratorRangeType>::iterator &,
1827 ScratchData &,
1828 CopyData &),
1829 void (MainClass::*copier)(const CopyData &),
1830 const ScratchData &sample_scratch_data,
1831 const CopyData &sample_copy_data,
1832 const unsigned int queue_length = 2 * MultithreadInfo::n_threads(),
1833 const unsigned int chunk_size = 8)
1834 {
1835 // Call the function above
1836 run(std::begin(iterator_range),
1837 std::end(iterator_range),
1838 main_object,
1839 worker,
1840 copier,
1841 sample_scratch_data,
1842 sample_copy_data,
1843 queue_length,
1844 chunk_size);
1845 }
1846
1847
1848
1852 template <typename MainClass,
1853 typename Iterator,
1854 typename ScratchData,
1855 typename CopyData>
1856 void
1858 MainClass &main_object,
1859 void (MainClass::*worker)(const Iterator &, ScratchData &, CopyData &),
1860 void (MainClass::*copier)(const CopyData &),
1861 const ScratchData &sample_scratch_data,
1862 const CopyData &sample_copy_data,
1863 const unsigned int queue_length = 2 * MultithreadInfo::n_threads(),
1864 const unsigned int chunk_size = 8)
1865 {
1866 // Call the function above
1867 run(std::begin(iterator_range),
1868 std::end(iterator_range),
1869 main_object,
1870 worker,
1871 copier,
1872 sample_scratch_data,
1873 sample_copy_data,
1874 queue_length,
1875 chunk_size);
1876 }
1877
1878} // namespace WorkStream
1879
1880
1881
1883
1884
1885
1886//---------------------------- work_stream.h ---------------------------
1887// end of #ifndef dealii_work_stream_h
1888#endif
1889//---------------------------- work_stream.h ---------------------------
IteratorOverIterators end() const
IteratorOverIterators begin()
static unsigned int n_threads()
static tf::Executor & get_taskflow_executor()
A class that provides a separate storage location on each thread that accesses the object.
std::list< ScratchAndCopyDataObjects > ScratchAndCopyDataList
typename internal::ScratchAndCopyDataObjects< Iterator, ScratchData, CopyData > ScratchAndCopyDataObjects
WorkerAndCopier(const std::function< void(const Iterator &, ScratchData &, CopyData &)> &worker, const std::function< void(const CopyData &)> &copier, const ScratchData &sample_scratch_data, const CopyData &sample_copy_data)
Threads::ThreadLocalStorage< ScratchAndCopyDataList > data
void operator()(const tbb::blocked_range< typename std::vector< Iterator >::const_iterator > &range)
const std::function< void(const Iterator &, ScratchData &, CopyData &)> worker
const std::function< void(const CopyData &)> copier
IteratorRangeToItemStream(const Iterator &begin, const Iterator &end, const unsigned int buffer_size, const unsigned int chunk_size, const ScratchData &sample_scratch_data, const CopyData &sample_copy_data)
Threads::ThreadLocalStorage< typename ItemType::ScratchDataList > thread_local_scratch
#define DEAL_II_NAMESPACE_OPEN
Definition config.h:35
#define DEAL_II_NAMESPACE_CLOSE
Definition config.h:36
#define Assert(cond, exc)
static ::ExceptionBase & ExcInternalError()
static ::ExceptionBase & ExcMessage(std::string arg1)
std::size_t size
Definition mpi.cc:745
void handle_std_exception(const std::exception &exc)
void run(const Iterator &begin, const std_cxx20::type_identity_t< Iterator > &end, Worker worker, Copier copier, const ScratchData &sample_scratch_data, const CopyData &sample_copy_data)
void run(const std::vector< std::vector< Iterator > > &colored_iterators, Worker worker, Copier copier, const ScratchData &sample_scratch_data, const CopyData &sample_copy_data, const unsigned int=2 *MultithreadInfo::n_threads(), const unsigned int chunk_size=8)
void run(const Iterator &begin, const std_cxx20::type_identity_t< Iterator > &end, Worker worker, Copier copier, const ScratchData &sample_scratch_data, const CopyData &sample_copy_data, const unsigned int=2 *MultithreadInfo::n_threads(), const unsigned int chunk_size=8)
void run(const std::vector< std::vector< Iterator > > &colored_iterators, Worker worker, Copier copier, const ScratchData &sample_scratch_data, const CopyData &sample_copy_data, const unsigned int chunk_size)
void run(const Iterator &begin, const std_cxx20::type_identity_t< Iterator > &end, Worker worker, Copier copier, const ScratchData &sample_scratch_data, const CopyData &sample_copy_data, const unsigned int queue_length, const unsigned int chunk_size)
void run(const std::vector< std::vector< Iterator > > &colored_iterators, Worker worker, Copier copier, const ScratchData &sample_scratch_data, const CopyData &sample_copy_data, const unsigned int queue_length=2 *MultithreadInfo::n_threads(), const unsigned int chunk_size=8)
void parallel_for(Iterator x_begin, Iterator x_end, const Functor &functor, const unsigned int grainsize)
Definition parallel.h:122
typename type_identity< T >::type type_identity_t
Definition type_traits.h:95
STL namespace.
std::unique_ptr< ScratchData > scratch_data
ScratchAndCopyDataObjects(const ScratchAndCopyDataObjects &)
ScratchAndCopyDataObjects(std::unique_ptr< ScratchData > &&p, std::unique_ptr< CopyData > &&q, const bool in_use)
ScratchDataObject(std::unique_ptr< ScratchData > &&p, const bool in_use)
ScratchDataObject(ScratchDataObject &&o) noexcept=default
ScratchDataObject(ScratchData *p, const bool in_use)
ScratchDataObject(const ScratchDataObject &)
std::unique_ptr< ScratchData > scratch_data
std::list< ScratchDataObject< ScratchData > > ScratchDataList
Threads::ThreadLocalStorage< ScratchDataList > * scratch_data