15#ifndef dealii_work_stream_h
16# define dealii_work_stream_h
30# ifdef DEAL_II_WITH_TBB
31# ifdef DEAL_II_TBB_WITH_ONEAPI
32# include <tbb/parallel_pipeline.h>
34# include <tbb/pipeline.h>
36# include <tbb/blocked_range.h>
39# ifdef DEAL_II_WITH_TASKFLOW
40# include <taskflow/taskflow.hpp>
179 template <
typename Iterator,
typename ScratchData,
typename CopyData>
194 std::unique_ptr<CopyData> &&q,
214 template <
typename ScratchData>
247# ifdef DEAL_II_WITH_TBB
262 namespace tbb_no_coloring
267 template <
typename Iterator,
typename ScratchData,
typename CopyData>
371 const unsigned int buffer_size,
374 const CopyData &sample_copy_data)
389 item.copy_datas.resize(
chunk_size, sample_copy_data);
390 item.currently_in_use =
false;
416 for (
unsigned int i = 0; i <
item_buffer.size(); ++i)
423 Assert(current_item !=
nullptr,
424 ExcMessage(
"This can't be. There must be a free item!"));
512 template <
typename Worker,
515 typename ScratchData,
518 run(
const Iterator &begin,
522 const ScratchData &sample_scratch_data,
523 const CopyData &sample_copy_data,
524 const unsigned int queue_length,
525 const unsigned int chunk_size)
539 iterator_range_to_item_stream(begin,
545 auto item_generator = [&](tbb::flow_control &fc) -> ItemType * {
546 if (
const auto item = iterator_range_to_item_stream.
get_item())
562 std::function<void(const Iterator &, ScratchData &, CopyData &)>(
565 static_cast<bool>(std::function<void(const CopyData &)>(copier))](
566 ItemType *current_item) {
578 ScratchData *scratch_data =
nullptr;
582 for (
auto &p : current_item->scratch_data->get())
583 if (p.currently_in_use ==
false)
585 scratch_data = p.scratch_data.get();
586 p.currently_in_use =
true;
592 if (scratch_data ==
nullptr)
595 new ScratchData(*current_item->sample_scratch_data);
596 current_item->scratch_data->get().emplace_back(scratch_data,
605 for (
unsigned int i = 0; i < current_item->n_iterators; ++i)
610 worker(current_item->iterators[i],
612 current_item->copy_datas[i]);
614 catch (
const std::exception &exc)
627 for (
auto &p : current_item->scratch_data->get())
628 if (p.scratch_data.get() == scratch_data)
631 p.currently_in_use =
false;
637 if (copier_exists ==
false)
638 current_item->currently_in_use =
false;
651 auto item_copier = [copier = std::function<void(const CopyData &)>(
652 copier)](ItemType *current_item) {
658 for (
unsigned int i = 0; i < current_item->n_iterators; ++i)
662 copier(current_item->copy_datas[i]);
664 catch (
const std::exception &exc)
675 current_item->currently_in_use =
false;
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,
688 auto tbb_worker_filter = tbb::make_filter<ItemType *, ItemType *>(
689# ifdef DEAL_II_TBB_WITH_ONEAPI
690 tbb::filter_mode::parallel,
692 tbb::filter::parallel,
696 auto tbb_copier_filter = tbb::make_filter<ItemType *, void>(
697# ifdef DEAL_II_TBB_WITH_ONEAPI
698 tbb::filter_mode::serial_in_order,
704 tbb::parallel_pipeline(queue_length,
705 tbb_item_stream_filter & tbb_worker_filter &
262 namespace tbb_no_coloring {
…}
714# ifdef DEAL_II_WITH_TASKFLOW
722 namespace taskflow_no_coloring
730 template <
typename Worker,
733 typename ScratchData,
736 run(
const Iterator &begin,
740 const ScratchData &sample_scratch_data,
741 const CopyData &sample_copy_data,
742 const unsigned int = 2 *
744 const unsigned int chunk_size = 8)
748 tf::Taskflow taskflow;
750 using ScratchDataList = std::list<ScratchDataObject<ScratchData>>;
753 thread_safe_scratch_data_list;
755 tf::Task last_copier;
764 std::vector<CopyData> copy_datas;
766 CopyChunk(
const unsigned int chunk_size,
767 const CopyData &sample_copy_data)
768 : copy_datas(chunk_size, sample_copy_data)
776 unsigned int idx = 0;
781 std::vector<std::unique_ptr<CopyChunk>> copy_chunks;
783 std::vector<Iterator> chunk;
784 chunk.reserve(chunk_size);
786 unsigned int total_chunks = 0;
787 unsigned int chunk_counter = 0;
792 for (Iterator it = begin; it != end;)
794 for (; (chunk_counter < chunk_size) && (it != end);
795 ++it, ++chunk_counter)
796 chunk.emplace_back(it);
804 &thread_safe_scratch_data_list,
805 &sample_scratch_data,
809 ScratchData *scratch_data =
nullptr;
811 ScratchDataList &scratch_data_list =
812 thread_safe_scratch_data_list.
get();
818 for (
auto &p : scratch_data_list)
820 if (p.currently_in_use ==
false)
822 scratch_data = p.scratch_data.get();
823 p.currently_in_use =
true;
829 if (scratch_data ==
nullptr)
831 scratch_data_list.emplace_back(
832 std::make_unique<ScratchData>(sample_scratch_data),
835 scratch_data_list.back().scratch_data.get();
841 std::make_unique<CopyChunk>(chunk_counter,
843 auto copy_chunk = copy_chunks[idx].get();
845 for (
auto &it : chunk)
847 worker(it, *scratch_data, copy_chunk->copy_datas[i]);
853 for (
auto &p : scratch_data_list)
855 if (p.scratch_data.get() == scratch_data)
857 Assert(p.currently_in_use ==
true,
859 p.currently_in_use =
false;
867 tf::Task copier_task =
869 .emplace([idx, ©_chunks, &copier]() {
870 auto copy_chunk = copy_chunks[idx].get();
871 for (
auto ©_data : copy_chunk->copy_datas)
876 copy_chunks[idx].reset();
881 worker_task.precede(copier_task);
886 if (!last_copier.empty())
887 last_copier.precede(copier_task);
892 last_copier = copier_task;
897 copy_chunks.resize(total_chunks);
901 executor.run(taskflow).wait();
722 namespace taskflow_no_coloring {
…}
917 template <
typename Worker,
920 typename ScratchData,
923 run(
const Iterator &begin,
927 const ScratchData &sample_scratch_data,
928 const CopyData &sample_copy_data)
931 ScratchData scratch_data = sample_scratch_data;
932 CopyData copy_data = sample_copy_data;
936 const bool have_worker =
937 (
static_cast<const std::function<
938 void(
const Iterator &, ScratchData &, CopyData &)
> &>(worker)) !=
940 const bool have_copier =
941 (
static_cast<const std::function<
void(
const CopyData &)
> &>(
945 for (Iterator i = begin; i != end; ++i)
948 worker(i, scratch_data, copy_data);
959 template <
typename Worker,
962 typename ScratchData,
965 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
968 const ScratchData &sample_scratch_data,
969 const CopyData &sample_copy_data)
972 ScratchData scratch_data = sample_scratch_data;
973 CopyData copy_data = sample_copy_data;
977 const bool have_worker =
978 (
static_cast<const std::function<
979 void(
const Iterator &, ScratchData &, CopyData &)
> &>(worker)) !=
981 const bool have_copier =
982 (
static_cast<const std::function<
void(
const CopyData &)
> &>(
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])
991 worker(it, scratch_data, copy_data);
965 run(
const std::vector<std::vector<Iterator>> &colored_iterators, {
…}
912 namespace sequential {
…}
1001# ifdef DEAL_II_WITH_TBB
1009 namespace tbb_colored
1016 template <
typename Iterator,
typename ScratchData,
typename CopyData>
1024 const std::function<
void(
const Iterator &, ScratchData &, CopyData &)>
1026 const std::function<
void(
const CopyData &)> &
copier,
1042 typename std::vector<Iterator>::const_iterator> &range)
1054 ScratchData *scratch_data =
nullptr;
1055 CopyData *copy_data =
nullptr;
1061 for (
typename ScratchAndCopyDataList::iterator p =
1062 scratch_and_copy_data_list.begin();
1063 p != scratch_and_copy_data_list.end();
1065 if (p->currently_in_use ==
false)
1067 scratch_data = p->scratch_data.get();
1068 copy_data = p->copy_data.get();
1069 p->currently_in_use =
true;
1075 if (scratch_data ==
nullptr)
1079 scratch_and_copy_data_list.emplace_back(
1084 scratch_and_copy_data_list.back().scratch_data.get();
1085 copy_data = scratch_and_copy_data_list.back().copy_data.get();
1091 for (
typename std::vector<Iterator>::const_iterator p = range.begin();
1098 worker(*p, *scratch_data, *copy_data);
1102 catch (
const std::exception &exc)
1118 for (
typename ScratchAndCopyDataList::iterator p =
1119 scratch_and_copy_data_list.begin();
1120 p != scratch_and_copy_data_list.end();
1122 if (p->scratch_data.get() == scratch_data)
1125 p->currently_in_use =
false;
1132 ScratchAndCopyDataObjects<Iterator, ScratchData, CopyData>;
1146 const std::function<void(
const Iterator &, ScratchData &, CopyData &)>
1153 const std::function<void(
const CopyData &)>
copier;
1165 template <
typename Worker,
1168 typename ScratchData,
1171 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1174 const ScratchData &sample_scratch_data,
1175 const CopyData &sample_copy_data,
1176 const unsigned int chunk_size)
1179 for (
unsigned int color = 0; color < colored_iterators.size(); ++color)
1180 if (colored_iterators[color].
size() > 0)
1183 WorkerAndCopier<Iterator, ScratchData, CopyData>;
1187 sample_scratch_data,
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);
1171 run(
const std::vector<std::vector<Iterator>> &colored_iterators, {
…}
1009 namespace tbb_colored {
…}
1207# ifdef DEAL_II_WITH_TASKFLOW
1213 namespace taskflow_colored
1221 template <
typename Worker,
1224 typename ScratchData,
1227 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1230 const ScratchData &sample_scratch_data,
1231 const CopyData &sample_copy_data,
1232 const unsigned int = 2 *
1234 const unsigned int chunk_size = 8)
1239 ScratchAndCopyDataObjects<Iterator, ScratchData, CopyData>;
1241 using ScratchAndCopyDataList = std::list<ScratchAndCopyDataObjects>;
1244 thread_safe_scratch_and_copy_data_list;
1246 tf::Taskflow taskflow;
1250 tf::Future<void> execution_future;
1252 const bool have_worker =
1253 (
static_cast<const std::function<
1254 void(
const Iterator &, ScratchData &, CopyData &)
> &>(worker)) !=
1256 const bool have_copier =
1257 (
static_cast<const std::function<
void(
const CopyData &)
> &>(
1258 copier)) !=
nullptr;
1260 std::vector<Iterator> chunk;
1261 chunk.reserve(chunk_size);
1263 unsigned int chunk_counter = 0;
1269 for (
unsigned int color = 0; color < colored_iterators.size(); ++color)
1271 if (colored_iterators[color].
size() > 0)
1275 for (
auto it = colored_iterators[color].begin();
1276 it != colored_iterators[color].end();)
1278 for (; (chunk_counter < chunk_size) &&
1279 (it != colored_iterators[color].end());
1280 ++it, ++chunk_counter)
1281 chunk.emplace_back(*it);
1286 &thread_safe_scratch_and_copy_data_list,
1287 &sample_scratch_data,
1291 ScratchData *scratch_data =
nullptr;
1292 CopyData *copy_data =
nullptr;
1294 ScratchAndCopyDataList &scratch_and_copy_data_list =
1295 thread_safe_scratch_and_copy_data_list.
get();
1298 for (
typename ScratchAndCopyDataList::iterator p =
1299 scratch_and_copy_data_list.begin();
1300 p != scratch_and_copy_data_list.end();
1303 if (p->currently_in_use ==
false)
1305 scratch_data = p->scratch_data.get();
1306 copy_data = p->copy_data.get();
1307 p->currently_in_use =
true;
1313 if (scratch_data ==
nullptr)
1316 scratch_and_copy_data_list.emplace_back(
1317 std::make_unique<ScratchData>(sample_scratch_data),
1318 std::make_unique<CopyData>(sample_copy_data),
1320 scratch_data = scratch_and_copy_data_list.back()
1321 .scratch_data.get();
1323 scratch_and_copy_data_list.back().copy_data.get();
1326 for (
const Iterator &it : chunk)
1329 worker(it, *scratch_data, *copy_data);
1335 for (
typename ScratchAndCopyDataList::iterator p =
1336 scratch_and_copy_data_list.begin();
1337 p != scratch_and_copy_data_list.end();
1340 if (p->scratch_data.get() == scratch_data)
1342 Assert(p->currently_in_use ==
true,
1344 p->currently_in_use =
false;
1348 .name(
"worker_and_copier");
1355 execution_future.wait();
1356 execution_future = executor.run(std::move(taskflow));
1359 if (colored_iterators.size() > 0)
1360 execution_future.wait();
1227 run(
const std::vector<std::vector<Iterator>> &colored_iterators, {
…}
1213 namespace taskflow_colored {
…}
1417 template <
typename Worker,
1420 typename ScratchData,
1423 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1426 const ScratchData &sample_scratch_data,
1427 const CopyData &sample_copy_data,
1429 const unsigned int chunk_size = 8);
1481 template <
typename Worker,
1484 typename ScratchData,
1491 const ScratchData &sample_scratch_data,
1492 const CopyData &sample_copy_data,
1494 const unsigned int chunk_size = 8)
1497 ExcMessage(
"The queue length must be at least one, and preferably "
1498 "larger than the number of processors on this system."));
1500 Assert(chunk_size > 0,
ExcMessage(
"The chunk_size must be at least one."));
1505 if (!(begin != end))
1510# if defined(DEAL_II_WITH_TBB) || defined(DEAL_II_WITH_TASKFLOW)
1511 if (
static_cast<const std::function<
void(
const CopyData &)
> &>(copier))
1514# if defined(DEAL_II_WITH_TASKFLOW)
1519 sample_scratch_data,
1523# elif defined(DEAL_II_WITH_TBB)
1528 sample_scratch_data,
1550 std::vector<std::vector<Iterator>> all_iterators(1);
1551 for (Iterator p = begin; p != end; ++p)
1552 all_iterators[0].push_back(p);
1557 sample_scratch_data,
1570 begin, end, worker, copier, sample_scratch_data, sample_copy_data);
1582 template <
typename Worker,
1584 typename IteratorRangeType,
1585 typename ScratchData,
1587 typename = std::enable_if_t<has_begin_and_end<IteratorRangeType>>>
1589 run(IteratorRangeType iterator_range,
1592 const ScratchData &sample_scratch_data,
1593 const CopyData &sample_copy_data,
1595 const unsigned int chunk_size = 8)
1598 run(iterator_range.begin(),
1599 iterator_range.end(),
1602 sample_scratch_data,
1589 run(IteratorRangeType iterator_range, {
…}
1613 template <
typename Worker,
1616 typename ScratchData,
1622 const ScratchData &sample_scratch_data,
1623 const CopyData &sample_copy_data,
1625 const unsigned int chunk_size = 8)
1629 iterator_range.
end(),
1632 sample_scratch_data,
1640 template <
typename Worker,
1643 typename ScratchData,
1646 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1649 const ScratchData &sample_scratch_data,
1650 const CopyData &sample_copy_data,
1651 const unsigned int queue_length,
1652 const unsigned int chunk_size)
1655 ExcMessage(
"The queue length must be at least one, and preferably "
1656 "larger than the number of processors on this system."));
1658 Assert(chunk_size > 0,
ExcMessage(
"The chunk_size must be at least one."));
1664# ifdef DEAL_II_WITH_TASKFLOW
1668 sample_scratch_data,
1674# elif defined(DEAL_II_WITH_TBB)
1678 sample_scratch_data,
1692 sample_scratch_data,
1646 run(
const std::vector<std::vector<Iterator>> &colored_iterators, {
…}
1740 template <
typename MainClass,
1742 typename ScratchData,
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,
1753 const unsigned
int chunk_size = 8)
1759 [&main_object, worker](
const Iterator &iterator,
1760 ScratchData &scratch_data,
1761 CopyData ©_data) {
1762 (main_object.*worker)(iterator, scratch_data, copy_data);
1764 [&main_object, copier](
const CopyData ©_data) {
1765 (main_object.*copier)(copy_data);
1767 sample_scratch_data,
1774 template <
typename MainClass,
1776 typename ScratchData,
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,
1787 const unsigned
int chunk_size = 8)
1793 [&main_object, worker](
const Iterator &iterator,
1794 ScratchData &scratch_data,
1795 CopyData ©_data) {
1796 (main_object.*worker)(iterator, scratch_data, copy_data);
1798 [&main_object, copier](
const CopyData ©_data) {
1799 (main_object.*copier)(copy_data);
1801 sample_scratch_data,
1816 template <
typename MainClass,
1817 typename IteratorRangeType,
1818 typename ScratchData,
1820 typename = std::enable_if_t<has_begin_and_end<IteratorRangeType>>>
1823 IteratorRangeType iterator_range,
1824 MainClass &main_object,
1825 void (MainClass::*worker)(
1826 const typename
std_cxx20::type_identity_t<IteratorRangeType>::iterator &,
1829 void (MainClass::*copier)(const CopyData &),
1830 const ScratchData &sample_scratch_data,
1831 const CopyData &sample_copy_data,
1833 const unsigned
int chunk_size = 8)
1836 run(std::begin(iterator_range),
1837 std::end(iterator_range),
1841 sample_scratch_data,
1852 template <
typename MainClass,
1854 typename ScratchData,
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,
1864 const unsigned
int chunk_size = 8)
1867 run(std::begin(iterator_range),
1868 std::end(iterator_range),
1872 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 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)
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