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 chunk_size = 8)
741 tf::Taskflow taskflow;
743 using ScratchDataList = std::list<ScratchDataObject<ScratchData>>;
746 thread_safe_scratch_data_list;
748 tf::Task last_copier;
757 std::vector<CopyData> copy_datas;
759 CopyChunk(
const unsigned int chunk_size,
760 const CopyData &sample_copy_data)
761 : copy_datas(chunk_size, sample_copy_data)
769 unsigned int idx = 0;
774 std::vector<std::unique_ptr<CopyChunk>> copy_chunks;
776 std::vector<Iterator> chunk;
777 chunk.reserve(chunk_size);
779 unsigned int total_chunks = 0;
780 unsigned int chunk_counter = 0;
785 for (Iterator it = begin; it != end;)
787 for (; (chunk_counter < chunk_size) && (it != end);
788 ++it, ++chunk_counter)
789 chunk.emplace_back(it);
797 &thread_safe_scratch_data_list,
798 &sample_scratch_data,
802 ScratchData *scratch_data =
nullptr;
804 ScratchDataList &scratch_data_list =
805 thread_safe_scratch_data_list.
get();
811 for (
auto &p : scratch_data_list)
813 if (p.currently_in_use ==
false)
815 scratch_data = p.scratch_data.get();
816 p.currently_in_use =
true;
822 if (scratch_data ==
nullptr)
824 scratch_data_list.emplace_back(
825 std::make_unique<ScratchData>(sample_scratch_data),
828 scratch_data_list.back().scratch_data.get();
834 std::make_unique<CopyChunk>(chunk_counter,
836 auto copy_chunk = copy_chunks[idx].get();
838 for (
auto &it : chunk)
840 worker(it, *scratch_data, copy_chunk->copy_datas[i]);
846 for (
auto &p : scratch_data_list)
848 if (p.scratch_data.get() == scratch_data)
850 Assert(p.currently_in_use ==
true,
852 p.currently_in_use =
false;
860 tf::Task copier_task =
862 .emplace([idx, ©_chunks, &copier]() {
863 auto copy_chunk = copy_chunks[idx].get();
864 for (
auto ©_data : copy_chunk->copy_datas)
869 copy_chunks[idx].reset();
874 worker_task.precede(copier_task);
879 if (!last_copier.empty())
880 last_copier.precede(copier_task);
885 last_copier = copier_task;
890 copy_chunks.resize(total_chunks);
894 executor.run(taskflow).wait();
910 template <
typename Worker,
913 typename ScratchData,
916 run(
const Iterator &begin,
920 const ScratchData &sample_scratch_data,
921 const CopyData &sample_copy_data)
924 ScratchData scratch_data = sample_scratch_data;
925 CopyData copy_data = sample_copy_data;
929 const bool have_worker =
930 (
static_cast<const std::function<
931 void(
const Iterator &, ScratchData &, CopyData &)
> &>(worker)) !=
933 const bool have_copier =
934 (
static_cast<const std::function<
void(
const CopyData &)
> &>(
938 for (Iterator i = begin; i != end; ++i)
941 worker(i, scratch_data, copy_data);
952 template <
typename Worker,
955 typename ScratchData,
958 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
961 const ScratchData &sample_scratch_data,
962 const CopyData &sample_copy_data)
965 ScratchData scratch_data = sample_scratch_data;
966 CopyData copy_data = sample_copy_data;
970 const bool have_worker =
971 (
static_cast<const std::function<
972 void(
const Iterator &, ScratchData &, CopyData &)
> &>(worker)) !=
974 const bool have_copier =
975 (
static_cast<const std::function<
void(
const CopyData &)
> &>(
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])
984 worker(it, scratch_data, copy_data);
994# ifdef DEAL_II_WITH_TBB
1002 namespace tbb_colored
1009 template <
typename Iterator,
typename ScratchData,
typename CopyData>
1017 const std::function<
void(
const Iterator &, ScratchData &, CopyData &)>
1019 const std::function<
void(
const CopyData &)> &
copier,
1035 typename std::vector<Iterator>::const_iterator> &range)
1047 ScratchData *scratch_data =
nullptr;
1048 CopyData *copy_data =
nullptr;
1054 for (
typename ScratchAndCopyDataList::iterator p =
1055 scratch_and_copy_data_list.begin();
1056 p != scratch_and_copy_data_list.end();
1058 if (p->currently_in_use ==
false)
1060 scratch_data = p->scratch_data.get();
1061 copy_data = p->copy_data.get();
1062 p->currently_in_use =
true;
1068 if (scratch_data ==
nullptr)
1072 scratch_and_copy_data_list.emplace_back(
1077 scratch_and_copy_data_list.back().scratch_data.get();
1078 copy_data = scratch_and_copy_data_list.back().copy_data.get();
1084 for (
typename std::vector<Iterator>::const_iterator p = range.begin();
1091 worker(*p, *scratch_data, *copy_data);
1095 catch (
const std::exception &exc)
1111 for (
typename ScratchAndCopyDataList::iterator p =
1112 scratch_and_copy_data_list.begin();
1113 p != scratch_and_copy_data_list.end();
1115 if (p->scratch_data.get() == scratch_data)
1118 p->currently_in_use =
false;
1125 ScratchAndCopyDataObjects<Iterator, ScratchData, CopyData>;
1139 const std::function<void(
const Iterator &, ScratchData &, CopyData &)>
1146 const std::function<void(
const CopyData &)>
copier;
1158 template <
typename Worker,
1161 typename ScratchData,
1164 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1167 const ScratchData &sample_scratch_data,
1168 const CopyData &sample_copy_data,
1169 const unsigned int chunk_size)
1172 for (
unsigned int color = 0; color < colored_iterators.size(); ++color)
1173 if (colored_iterators[color].
size() > 0)
1176 WorkerAndCopier<Iterator, ScratchData, CopyData>;
1180 sample_scratch_data,
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);
1200# ifdef DEAL_II_WITH_TASKFLOW
1206 namespace taskflow_colored
1214 template <
typename Worker,
1217 typename ScratchData,
1220 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1223 const ScratchData &sample_scratch_data,
1224 const CopyData &sample_copy_data,
1225 const unsigned int = 2 *
1227 const unsigned int chunk_size = 8)
1232 ScratchAndCopyDataObjects<Iterator, ScratchData, CopyData>;
1234 using ScratchAndCopyDataList = std::list<ScratchAndCopyDataObjects>;
1237 thread_safe_scratch_and_copy_data_list;
1239 tf::Taskflow taskflow;
1243 tf::Future<void> execution_future;
1245 const bool have_worker =
1246 (
static_cast<const std::function<
1247 void(
const Iterator &, ScratchData &, CopyData &)
> &>(worker)) !=
1249 const bool have_copier =
1250 (
static_cast<const std::function<
void(
const CopyData &)
> &>(
1251 copier)) !=
nullptr;
1253 std::vector<Iterator> chunk;
1254 chunk.reserve(chunk_size);
1256 unsigned int chunk_counter = 0;
1262 for (
unsigned int color = 0; color < colored_iterators.size(); ++color)
1264 if (colored_iterators[color].
size() > 0)
1268 for (
auto it = colored_iterators[color].begin();
1269 it != colored_iterators[color].end();)
1271 for (; (chunk_counter < chunk_size) &&
1272 (it != colored_iterators[color].end());
1273 ++it, ++chunk_counter)
1274 chunk.emplace_back(*it);
1279 &thread_safe_scratch_and_copy_data_list,
1280 &sample_scratch_data,
1284 ScratchData *scratch_data =
nullptr;
1285 CopyData *copy_data =
nullptr;
1287 ScratchAndCopyDataList &scratch_and_copy_data_list =
1288 thread_safe_scratch_and_copy_data_list.
get();
1291 for (
typename ScratchAndCopyDataList::iterator p =
1292 scratch_and_copy_data_list.begin();
1293 p != scratch_and_copy_data_list.end();
1296 if (p->currently_in_use ==
false)
1298 scratch_data = p->scratch_data.get();
1299 copy_data = p->copy_data.get();
1300 p->currently_in_use =
true;
1306 if (scratch_data ==
nullptr)
1309 scratch_and_copy_data_list.emplace_back(
1310 std::make_unique<ScratchData>(sample_scratch_data),
1311 std::make_unique<CopyData>(sample_copy_data),
1313 scratch_data = scratch_and_copy_data_list.back()
1314 .scratch_data.get();
1316 scratch_and_copy_data_list.back().copy_data.get();
1319 for (
const Iterator &it : chunk)
1322 worker(it, *scratch_data, *copy_data);
1328 for (
typename ScratchAndCopyDataList::iterator p =
1329 scratch_and_copy_data_list.begin();
1330 p != scratch_and_copy_data_list.end();
1333 if (p->scratch_data.get() == scratch_data)
1335 Assert(p->currently_in_use ==
true,
1337 p->currently_in_use =
false;
1341 .name(
"worker_and_copier");
1348 execution_future.wait();
1349 execution_future = executor.run(std::move(taskflow));
1352 if (colored_iterators.size() > 0)
1353 execution_future.wait();
1410 template <
typename Worker,
1413 typename ScratchData,
1416 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1419 const ScratchData &sample_scratch_data,
1420 const CopyData &sample_copy_data,
1422 const unsigned int chunk_size = 8);
1474 template <
typename Worker,
1477 typename ScratchData,
1484 const ScratchData &sample_scratch_data,
1485 const CopyData &sample_copy_data,
1487 const unsigned int chunk_size = 8)
1490 ExcMessage(
"The queue length must be at least one, and preferably "
1491 "larger than the number of processors on this system."));
1493 Assert(chunk_size > 0,
ExcMessage(
"The chunk_size must be at least one."));
1498 if (!(begin != end))
1503# if defined(DEAL_II_WITH_TBB) || defined(DEAL_II_WITH_TASKFLOW)
1504 if (
static_cast<const std::function<
void(
const CopyData &)
> &>(copier))
1507# if defined(DEAL_II_WITH_TASKFLOW)
1512 sample_scratch_data,
1516# elif defined(DEAL_II_WITH_TBB)
1521 sample_scratch_data,
1543 std::vector<std::vector<Iterator>> all_iterators(1);
1544 for (Iterator p = begin; p != end; ++p)
1545 all_iterators[0].push_back(p);
1550 sample_scratch_data,
1563 begin, end, worker, copier, sample_scratch_data, sample_copy_data);
1575 template <
typename Worker,
1577 typename IteratorRangeType,
1578 typename ScratchData,
1580 typename = std::enable_if_t<has_begin_and_end<IteratorRangeType>>>
1582 run(IteratorRangeType iterator_range,
1585 const ScratchData &sample_scratch_data,
1586 const CopyData &sample_copy_data,
1588 const unsigned int chunk_size = 8)
1591 run(iterator_range.begin(),
1592 iterator_range.end(),
1595 sample_scratch_data,
1606 template <
typename Worker,
1609 typename ScratchData,
1615 const ScratchData &sample_scratch_data,
1616 const CopyData &sample_copy_data,
1618 const unsigned int chunk_size = 8)
1622 iterator_range.
end(),
1625 sample_scratch_data,
1633 template <
typename Worker,
1636 typename ScratchData,
1639 run(
const std::vector<std::vector<Iterator>> &colored_iterators,
1642 const ScratchData &sample_scratch_data,
1643 const CopyData &sample_copy_data,
1644 const unsigned int queue_length,
1645 const unsigned int chunk_size)
1648 ExcMessage(
"The queue length must be at least one, and preferably "
1649 "larger than the number of processors on this system."));
1651 Assert(chunk_size > 0,
ExcMessage(
"The chunk_size must be at least one."));
1657# ifdef DEAL_II_WITH_TASKFLOW
1661 sample_scratch_data,
1667# elif defined(DEAL_II_WITH_TBB)
1671 sample_scratch_data,
1685 sample_scratch_data,
1733 template <
typename MainClass,
1735 typename ScratchData,
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,
1746 const unsigned
int chunk_size = 8)
1752 [&main_object, worker](
const Iterator &iterator,
1753 ScratchData &scratch_data,
1754 CopyData ©_data) {
1755 (main_object.*worker)(iterator, scratch_data, copy_data);
1757 [&main_object, copier](
const CopyData ©_data) {
1758 (main_object.*copier)(copy_data);
1760 sample_scratch_data,
1767 template <
typename MainClass,
1769 typename ScratchData,
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,
1780 const unsigned
int chunk_size = 8)
1786 [&main_object, worker](
const Iterator &iterator,
1787 ScratchData &scratch_data,
1788 CopyData ©_data) {
1789 (main_object.*worker)(iterator, scratch_data, copy_data);
1791 [&main_object, copier](
const CopyData ©_data) {
1792 (main_object.*copier)(copy_data);
1794 sample_scratch_data,
1809 template <
typename MainClass,
1810 typename IteratorRangeType,
1811 typename ScratchData,
1813 typename = std::enable_if_t<has_begin_and_end<IteratorRangeType>>>
1816 IteratorRangeType iterator_range,
1817 MainClass &main_object,
1818 void (MainClass::*worker)(
1819 const typename
std_cxx20::type_identity_t<IteratorRangeType>::iterator &,
1822 void (MainClass::*copier)(const CopyData &),
1823 const ScratchData &sample_scratch_data,
1824 const CopyData &sample_copy_data,
1826 const unsigned
int chunk_size = 8)
1829 run(std::begin(iterator_range),
1830 std::end(iterator_range),
1834 sample_scratch_data,
1845 template <
typename MainClass,
1847 typename ScratchData,
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,
1857 const unsigned
int chunk_size = 8)
1860 run(std::begin(iterator_range),
1861 std::end(iterator_range),
1865 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