15#ifndef dealii_work_stream_h
16# define dealii_work_stream_h
29# ifdef DEAL_II_WITH_TBB
30# ifdef DEAL_II_TBB_WITH_ONEAPI
31# include <tbb/parallel_pipeline.h>
33# include <tbb/pipeline.h>
172 template <
typename Iterator,
typename ScratchData,
typename CopyData>
187 std::unique_ptr<CopyData> &&q,
207 template <
typename ScratchData>
240# ifdef DEAL_II_WITH_TBB
255 namespace tbb_no_coloring
260 template <
typename Iterator,
typename ScratchData,
typename CopyData>
364 const unsigned int buffer_size,
367 const CopyData &sample_copy_data)
382 item.copy_datas.resize(
chunk_size, sample_copy_data);
383 item.currently_in_use =
false;
409 for (
unsigned int i = 0; i <
item_buffer.size(); ++i)
416 Assert(current_item !=
nullptr,
417 ExcMessage(
"This can't be. There must be a free item!"));
505 template <
typename Worker,
508 typename ScratchData,
511 run(
const Iterator &begin,
515 const ScratchData &sample_scratch_data,
516 const CopyData &sample_copy_data,
517 const unsigned int queue_length,
518 const unsigned int chunk_size)
532 iterator_range_to_item_stream(begin,
538 auto item_generator = [&](tbb::flow_control &fc) -> ItemType * {
539 if (
const auto item = iterator_range_to_item_stream.
get_item())
555 std::function<void(const Iterator &, ScratchData &, CopyData &)>(
558 static_cast<bool>(std::function<void(const CopyData &)>(copier))](
559 ItemType *current_item) {
571 ScratchData *scratch_data =
nullptr;
575 for (
auto &p : current_item->scratch_data->get())
576 if (p.currently_in_use ==
false)
578 scratch_data = p.scratch_data.get();
579 p.currently_in_use =
true;
585 if (scratch_data ==
nullptr)
588 new ScratchData(*current_item->sample_scratch_data);
589 current_item->scratch_data->get().emplace_back(scratch_data,
598 for (
unsigned int i = 0; i < current_item->n_iterators; ++i)
603 worker(current_item->iterators[i],
605 current_item->copy_datas[i]);
607 catch (
const std::exception &exc)
620 for (
auto &p : current_item->scratch_data->get())
621 if (p.scratch_data.get() == scratch_data)
624 p.currently_in_use =
false;
630 if (copier_exists ==
false)
631 current_item->currently_in_use =
false;
644 auto item_copier = [copier = std::function<void(const CopyData &)>(
645 copier)](ItemType *current_item) {
651 for (
unsigned int i = 0; i < current_item->n_iterators; ++i)
655 copier(current_item->copy_datas[i]);
657 catch (
const std::exception &exc)
668 current_item->currently_in_use =
false;
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,
681 auto tbb_worker_filter = tbb::make_filter<ItemType *, ItemType *>(
682# ifdef DEAL_II_TBB_WITH_ONEAPI
683 tbb::filter_mode::parallel,
685 tbb::filter::parallel,
689 auto tbb_copier_filter = tbb::make_filter<ItemType *, void>(
690# ifdef DEAL_II_TBB_WITH_ONEAPI
691 tbb::filter_mode::serial_in_order,
697 tbb::parallel_pipeline(queue_length,
698 tbb_item_stream_filter & tbb_worker_filter &
707# ifdef DEAL_II_WITH_TASKFLOW
715 namespace taskflow_no_coloring
723 template <
typename Worker,
726 typename ScratchData,
729 run(
const Iterator &begin,
733 const ScratchData &sample_scratch_data,
734 const CopyData &sample_copy_data,
735 const unsigned int = 2 *
737 const unsigned int = 8)
741 tf::Taskflow taskflow;
743 using ScratchDataList = std::list<ScratchDataObject<ScratchData>>;
746 thread_safe_scratch_data_list;
748 tf::Task last_copier;
754 unsigned int idx = 0;
756 std::vector<std::unique_ptr<CopyData>> copy_datas;
761 for (Iterator it = begin; it != end; ++it, ++idx)
763 copy_datas.emplace_back();
769 &thread_safe_scratch_data_list,
770 &sample_scratch_data,
774 ScratchData *scratch_data =
nullptr;
776 ScratchDataList &scratch_data_list =
777 thread_safe_scratch_data_list.
get();
780 for (
auto &p : scratch_data_list)
782 if (p.currently_in_use ==
false)
784 scratch_data = p.scratch_data.get();
785 p.currently_in_use =
true;
791 if (scratch_data ==
nullptr)
793 scratch_data_list.emplace_back(
794 std::make_unique<ScratchData>(sample_scratch_data),
797 scratch_data_list.back().scratch_data.get();
802 auto © = copy_datas[idx];
803 copy = std::make_unique<CopyData>(sample_copy_data);
804 worker(it, *scratch_data, *copy.get());
808 for (
auto &p : scratch_data_list)
810 if (p.scratch_data.get() == scratch_data)
812 Assert(p.currently_in_use ==
true,
814 p.currently_in_use =
false;
822 tf::Task copier_task = taskflow
823 .emplace([idx, ©_datas, &copier]() {
824 copier(*copy_datas[idx].get());
825 copy_datas[idx].reset();
830 worker_task.precede(copier_task);
835 if (!last_copier.empty())
836 last_copier.precede(copier_task);
841 last_copier = copier_task;
847 executor.run(taskflow).wait();
863 template <
typename Worker,
866 typename ScratchData,
869 run(
const Iterator &begin,
873 const ScratchData &sample_scratch_data,
874 const CopyData &sample_copy_data)
877 ScratchData scratch_data = sample_scratch_data;
878 CopyData copy_data = sample_copy_data;
882 const bool have_worker =
883 (
static_cast<const std::function<
884 void(
const Iterator &, ScratchData &, CopyData &)
> &>(worker)) !=
886 const bool have_copier =
887 (
static_cast<const std::function<
void(
const CopyData &)
> &>(
891 for (Iterator i = begin; i != end; ++i)
894 worker(i, scratch_data, copy_data);
905 template <
typename Worker,
908 typename ScratchData,
911 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
914 const ScratchData &sample_scratch_data,
915 const CopyData &sample_copy_data)
918 ScratchData scratch_data = sample_scratch_data;
919 CopyData copy_data = sample_copy_data;
923 const bool have_worker =
924 (
static_cast<const std::function<
925 void(
const Iterator &, ScratchData &, CopyData &)
> &>(worker)) !=
927 const bool have_copier =
928 (
static_cast<const std::function<
void(
const CopyData &)
> &>(
932 for (
unsigned int color = 0; color < colored_iterators.size(); ++color)
933 if (colored_iterators[color].size() > 0)
934 for (
auto &it : colored_iterators[color])
937 worker(it, scratch_data, copy_data);
947# ifdef DEAL_II_WITH_TBB
955 namespace tbb_colored
962 template <
typename Iterator,
typename ScratchData,
typename CopyData>
970 const std::function<
void(
const Iterator &, ScratchData &, CopyData &)>
972 const std::function<
void(
const CopyData &)> &
copier,
988 typename std::vector<Iterator>::const_iterator> &range)
1000 ScratchData *scratch_data =
nullptr;
1001 CopyData *copy_data =
nullptr;
1007 for (
typename ScratchAndCopyDataList::iterator p =
1008 scratch_and_copy_data_list.begin();
1009 p != scratch_and_copy_data_list.end();
1011 if (p->currently_in_use ==
false)
1013 scratch_data = p->scratch_data.get();
1014 copy_data = p->copy_data.get();
1015 p->currently_in_use =
true;
1021 if (scratch_data ==
nullptr)
1025 scratch_and_copy_data_list.emplace_back(
1030 scratch_and_copy_data_list.back().scratch_data.get();
1031 copy_data = scratch_and_copy_data_list.back().copy_data.get();
1037 for (
typename std::vector<Iterator>::const_iterator p = range.begin();
1044 worker(*p, *scratch_data, *copy_data);
1048 catch (
const std::exception &exc)
1064 for (
typename ScratchAndCopyDataList::iterator p =
1065 scratch_and_copy_data_list.begin();
1066 p != scratch_and_copy_data_list.end();
1068 if (p->scratch_data.get() == scratch_data)
1071 p->currently_in_use =
false;
1078 ScratchAndCopyDataObjects<Iterator, ScratchData, CopyData>;
1092 const std::function<void(
const Iterator &, ScratchData &, CopyData &)>
1099 const std::function<void(
const CopyData &)>
copier;
1111 template <
typename Worker,
1114 typename ScratchData,
1117 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1120 const ScratchData &sample_scratch_data,
1121 const CopyData &sample_copy_data,
1122 const unsigned int chunk_size)
1125 for (
unsigned int color = 0; color < colored_iterators.size(); ++color)
1126 if (colored_iterators[color].size() > 0)
1129 WorkerAndCopier<Iterator, ScratchData, CopyData>;
1133 sample_scratch_data,
1137 colored_iterators[color].begin(),
1138 colored_iterators[color].end(),
1139 [&worker_and_copier](
1140 const tbb::blocked_range<
1141 typename std::vector<Iterator>::const_iterator> &range) {
1142 worker_and_copier(range);
1153# ifdef DEAL_II_WITH_TASKFLOW
1159 namespace taskflow_colored
1167 template <
typename Worker,
1170 typename ScratchData,
1173 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1176 const ScratchData &sample_scratch_data,
1177 const CopyData &sample_copy_data,
1178 const unsigned int = 2 *
1180 const unsigned int = 8)
1185 ScratchAndCopyDataObjects<Iterator, ScratchData, CopyData>;
1187 using ScratchAndCopyDataList = std::list<ScratchAndCopyDataObjects>;
1190 thread_safe_scratch_and_copy_data_list;
1192 tf::Taskflow taskflow;
1196 tf::Future<void> execution_future;
1198 const bool have_worker =
1199 (
static_cast<const std::function<
1200 void(
const Iterator &, ScratchData &, CopyData &)
> &>(worker)) !=
1202 const bool have_copier =
1203 (
static_cast<const std::function<
void(
const CopyData &)
> &>(
1204 copier)) !=
nullptr;
1211 for (
unsigned int color = 0; color < colored_iterators.size(); ++color)
1213 if (colored_iterators[color].size() > 0)
1217 for (
const Iterator &it : colored_iterators[color])
1224 &thread_safe_scratch_and_copy_data_list,
1225 &sample_scratch_data,
1229 ScratchData *scratch_data =
nullptr;
1230 CopyData *copy_data =
nullptr;
1232 ScratchAndCopyDataList &scratch_and_copy_data_list =
1233 thread_safe_scratch_and_copy_data_list.
get();
1236 for (
typename ScratchAndCopyDataList::iterator p =
1237 scratch_and_copy_data_list.begin();
1238 p != scratch_and_copy_data_list.end();
1241 if (p->currently_in_use ==
false)
1243 scratch_data = p->scratch_data.get();
1244 copy_data = p->copy_data.get();
1245 p->currently_in_use =
true;
1251 if (scratch_data ==
nullptr)
1254 scratch_and_copy_data_list.emplace_back(
1255 std::make_unique<ScratchData>(
1256 sample_scratch_data),
1257 std::make_unique<CopyData>(sample_copy_data),
1259 scratch_data = scratch_and_copy_data_list.back()
1260 .scratch_data.get();
1262 scratch_and_copy_data_list.back().copy_data.get();
1265 worker(it, *scratch_data, *copy_data);
1270 for (
typename ScratchAndCopyDataList::iterator p =
1271 scratch_and_copy_data_list.begin();
1272 p != scratch_and_copy_data_list.end();
1275 if (p->scratch_data.get() == scratch_data)
1277 Assert(p->currently_in_use ==
true,
1279 p->currently_in_use =
false;
1283 .name(
"worker_and_copier");
1287 execution_future.wait();
1288 execution_future = executor.run(std::move(taskflow));
1291 if (colored_iterators.size() > 0)
1292 execution_future.wait();
1349 template <
typename Worker,
1352 typename ScratchData,
1355 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1358 const ScratchData &sample_scratch_data,
1359 const CopyData &sample_copy_data,
1361 const unsigned int chunk_size = 8);
1413 template <
typename Worker,
1416 typename ScratchData,
1423 const ScratchData &sample_scratch_data,
1424 const CopyData &sample_copy_data,
1426 const unsigned int chunk_size = 8)
1429 ExcMessage(
"The queue length must be at least one, and preferably "
1430 "larger than the number of processors on this system."));
1432 Assert(chunk_size > 0,
ExcMessage(
"The chunk_size must be at least one."));
1437 if (!(begin != end))
1442# if defined(DEAL_II_WITH_TBB) || defined(DEAL_II_WITH_TASKFLOW)
1443 if (
static_cast<const std::function<
void(
const CopyData &)
> &>(copier))
1446# if defined(DEAL_II_WITH_TASKFLOW)
1451 sample_scratch_data,
1455# elif defined(DEAL_II_WITH_TBB)
1460 sample_scratch_data,
1482 std::vector<std::vector<Iterator>> all_iterators(1);
1483 for (Iterator p = begin; p != end; ++p)
1484 all_iterators[0].push_back(p);
1489 sample_scratch_data,
1502 begin, end, worker, copier, sample_scratch_data, sample_copy_data);
1514 template <
typename Worker,
1516 typename IteratorRangeType,
1517 typename ScratchData,
1519 typename = std::enable_if_t<has_begin_and_end<IteratorRangeType>>>
1521 run(IteratorRangeType iterator_range,
1524 const ScratchData &sample_scratch_data,
1525 const CopyData &sample_copy_data,
1527 const unsigned int chunk_size = 8)
1530 run(iterator_range.begin(),
1531 iterator_range.end(),
1534 sample_scratch_data,
1545 template <
typename Worker,
1548 typename ScratchData,
1554 const ScratchData &sample_scratch_data,
1555 const CopyData &sample_copy_data,
1557 const unsigned int chunk_size = 8)
1561 iterator_range.
end(),
1564 sample_scratch_data,
1572 template <
typename Worker,
1575 typename ScratchData,
1578 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1581 const ScratchData &sample_scratch_data,
1582 const CopyData &sample_copy_data,
1583 const unsigned int queue_length,
1584 const unsigned int chunk_size)
1587 ExcMessage(
"The queue length must be at least one, and preferably "
1588 "larger than the number of processors on this system."));
1590 Assert(chunk_size > 0,
ExcMessage(
"The chunk_size must be at least one."));
1596# ifdef DEAL_II_WITH_TASKFLOW
1600 sample_scratch_data,
1606# elif defined(DEAL_II_WITH_TBB)
1610 sample_scratch_data,
1624 sample_scratch_data,
1672 template <
typename MainClass,
1674 typename ScratchData,
1679 MainClass &main_object,
1680 void (MainClass::*worker)(const Iterator &, ScratchData &, CopyData &),
1681 void (MainClass::*copier)(const CopyData &),
1682 const ScratchData &sample_scratch_data,
1683 const CopyData &sample_copy_data,
1685 const unsigned
int chunk_size = 8)
1691 [&main_object, worker](
const Iterator &iterator,
1692 ScratchData &scratch_data,
1693 CopyData ©_data) {
1694 (main_object.*worker)(iterator, scratch_data, copy_data);
1696 [&main_object, copier](
const CopyData ©_data) {
1697 (main_object.*copier)(copy_data);
1699 sample_scratch_data,
1706 template <
typename MainClass,
1708 typename ScratchData,
1713 MainClass &main_object,
1714 void (MainClass::*worker)(const Iterator &, ScratchData &, CopyData &),
1715 void (MainClass::*copier)(const CopyData &),
1716 const ScratchData &sample_scratch_data,
1717 const CopyData &sample_copy_data,
1719 const unsigned
int chunk_size = 8)
1725 [&main_object, worker](
const Iterator &iterator,
1726 ScratchData &scratch_data,
1727 CopyData ©_data) {
1728 (main_object.*worker)(iterator, scratch_data, copy_data);
1730 [&main_object, copier](
const CopyData ©_data) {
1731 (main_object.*copier)(copy_data);
1733 sample_scratch_data,
1748 template <
typename MainClass,
1749 typename IteratorRangeType,
1750 typename ScratchData,
1752 typename = std::enable_if_t<has_begin_and_end<IteratorRangeType>>>
1755 IteratorRangeType iterator_range,
1756 MainClass &main_object,
1757 void (MainClass::*worker)(
1758 const typename
std_cxx20::type_identity_t<IteratorRangeType>::iterator &,
1761 void (MainClass::*copier)(const CopyData &),
1762 const ScratchData &sample_scratch_data,
1763 const CopyData &sample_copy_data,
1765 const unsigned
int chunk_size = 8)
1768 run(std::begin(iterator_range),
1769 std::end(iterator_range),
1773 sample_scratch_data,
1784 template <
typename MainClass,
1786 typename ScratchData,
1790 MainClass &main_object,
1791 void (MainClass::*worker)(const Iterator &, ScratchData &, CopyData &),
1792 void (MainClass::*copier)(const CopyData &),
1793 const ScratchData &sample_scratch_data,
1794 const CopyData &sample_copy_data,
1796 const unsigned
int chunk_size = 8)
1799 run(std::begin(iterator_range),
1800 std::end(iterator_range),
1804 sample_scratch_data,
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 CopyData & sample_copy_data
const ScratchData & sample_scratch_data
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
std::vector< ItemType > item_buffer
const unsigned int chunk_size
std::pair< Iterator, Iterator > remaining_iterator_range
const ScratchData & sample_scratch_data
#define DEAL_II_NAMESPACE_OPEN
#define DEAL_II_NAMESPACE_CLOSE
#define Assert(cond, exc)
static ::ExceptionBase & ExcInternalError()
static ::ExceptionBase & ExcMessage(std::string arg1)
void handle_unknown_exception()
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=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=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)
typename type_identity< T >::type type_identity_t
std::unique_ptr< ScratchData > scratch_data
ScratchAndCopyDataObjects()
std::unique_ptr< CopyData > copy_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
const ScratchData * sample_scratch_data
std::vector< Iterator > iterators
std::vector< CopyData > copy_datas
Threads::ThreadLocalStorage< ScratchDataList > * scratch_data