From: Wolfgang Bangerth Date: Wed, 8 May 2024 04:20:36 +0000 (+0530) Subject: Restructure WorkStream slightly. X-Git-Tag: v9.6.0-rc1~307^2 X-Git-Url: https://gitweb.dealii.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=8579043890d71f760aa743fd8a4029e4509b313f;p=dealii.git Restructure WorkStream slightly. --- diff --git a/include/deal.II/base/work_stream.h b/include/deal.II/base/work_stream.h index f0c7c73ea5..d5651fe1e3 100644 --- a/include/deal.II/base/work_stream.h +++ b/include/deal.II/base/work_stream.h @@ -481,13 +481,13 @@ namespace WorkStream ScratchData, CopyData>::ItemType; - // Create the three stages of the pipeline: + // Define the three stages of the pipeline: // // ----- Stage 1 ----- // // The first stage is the one that provides us with chunks of data - // to work on (the stream of "items"). This stage runs sequentially. + // to work on (the stream of "items"). This stage will run sequentially. IteratorRangeToItemStream iterator_range_to_item_stream(begin, end, @@ -495,33 +495,22 @@ namespace WorkStream chunk_size, sample_scratch_data, sample_copy_data); - auto tbb_item_stream_filter = tbb::make_filter( -# ifdef DEAL_II_TBB_WITH_ONEAPI - tbb::filter_mode::serial_in_order, -# else - tbb::filter::serial, -# endif - [&](tbb::flow_control &fc) -> ItemType * { - if (const auto item = iterator_range_to_item_stream.get_item()) - return item; - else - { - fc.stop(); - return nullptr; - } - }); + auto item_generator = [&](tbb::flow_control &fc) -> ItemType * { + if (const auto item = iterator_range_to_item_stream.get_item()) + return item; + else + { + fc.stop(); + return nullptr; + } + }; // // ----- Stage 2 ----- // // The second stage is the one that does the actual work. This is the // stage that runs in parallel - auto tbb_worker_filter = tbb::make_filter( -# ifdef DEAL_II_TBB_WITH_ONEAPI - tbb::filter_mode::parallel, -# else - tbb::filter::parallel, -# endif + auto item_worker = [worker = std::function( worker), @@ -560,7 +549,7 @@ namespace WorkStream current_item->scratch_data->get().emplace_back(scratch_data, true); } - } + }; // then call the worker function on each element of the chunk we // were given. since these worker functions are called on separate @@ -605,52 +594,66 @@ namespace WorkStream // Then return the original pointer // to the now modified object. The copier will work on it next. return current_item; - }); - + }; // // ----- Stage 3 ----- // // The last stage is the one that copies data from the CopyData objects // to the final destination. This stage runs sequentially again. - auto tbb_copier_filter = tbb::make_filter( + auto item_copier = [copier = std::function( + copier)](ItemType *current_item) { + if (copier) + { + // Initiate copying data. For the same reasons as in the worker + // class above, catch exceptions rather than letting them + // propagate into unknown territories: + for (unsigned int i = 0; i < current_item->n_iterators; ++i) + { + try + { + copier(current_item->copy_datas[i]); + } + catch (const std::exception &exc) + { + Threads::internal::handle_std_exception(exc); + } + catch (...) + { + Threads::internal::handle_unknown_exception(); + } + } + } + // mark current item as usable again + current_item->currently_in_use = false; + }; + + + // Now we just have to set up the pipeline and run it: + auto tbb_item_stream_filter = tbb::make_filter( # ifdef DEAL_II_TBB_WITH_ONEAPI tbb::filter_mode::serial_in_order, # else tbb::filter::serial, # endif - [copier = std::function(copier)]( - ItemType *current_item) { - if (copier) - { - // Initiate copying data. For the same reasons as in the worker - // class above, catch exceptions rather than letting them - // propagate into unknown territories: - for (unsigned int i = 0; i < current_item->n_iterators; ++i) - { - try - { - copier(current_item->copy_datas[i]); - } - catch (const std::exception &exc) - { - Threads::internal::handle_std_exception(exc); - } - catch (...) - { - Threads::internal::handle_unknown_exception(); - } - } - } - // mark current item as usable again - current_item->currently_in_use = false; - }); + item_generator); + auto tbb_worker_filter = tbb::make_filter( +# ifdef DEAL_II_TBB_WITH_ONEAPI + tbb::filter_mode::parallel, +# else + tbb::filter::parallel, +# endif + item_worker); + + auto tbb_copier_filter = tbb::make_filter( +# ifdef DEAL_II_TBB_WITH_ONEAPI + tbb::filter_mode::serial_in_order, +# else + tbb::filter::serial, +# endif + item_copier); - // - // ----- The pipeline ----- - // - // Now create a pipeline from these stages and execute it: tbb::parallel_pipeline(queue_length, tbb_item_stream_filter & tbb_worker_filter & tbb_copier_filter);