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