From: Bruno Turcksin Date: Tue, 26 Nov 2013 02:16:46 +0000 (+0000) Subject: Replace ring buffer by item_bugger in WorkStream. X-Git-Tag: v8.1.0~167 X-Git-Url: https://gitweb.dealii.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=e07b9a3acc274ebc319936c855728b41e535d4d7;p=dealii.git Replace ring buffer by item_bugger in WorkStream. git-svn-id: https://svn.dealii.org/trunk@31801 0785d39b-7218-0410-832d-ea1e28bc413d --- diff --git a/deal.II/include/deal.II/base/work_stream.h b/deal.II/include/deal.II/base/work_stream.h index 199a082de3..4236b80911 100644 --- a/deal.II/include/deal.II/base/work_stream.h +++ b/deal.II/include/deal.II/base/work_stream.h @@ -280,6 +280,12 @@ namespace WorkStream */ const ScratchData *sample_scratch_data; + /** + * Flag is true if the buffer is used and false if the buffer can be + * used. + */ + bool currently_in_use; + /** * Default constructor. @@ -290,7 +296,8 @@ namespace WorkStream : n_items (0), scratch_data (0), - sample_scratch_data (0) + sample_scratch_data (0), + currently_in_use (false) {} }; @@ -309,23 +316,23 @@ namespace WorkStream : tbb::filter (/*is_serial=*/true), remaining_iterator_range (begin, end), - ring_buffer (buffer_size), + item_buffer (buffer_size), sample_scratch_data (sample_scratch_data), - n_emitted_items (0), chunk_size (chunk_size) { // initialize the elements of the ring buffer - for (unsigned int element=0; element remaining_iterator_range; /** - * A ring buffer that will store items. + * A buffer that will store items. */ - std::vector ring_buffer; + std::vector item_buffer; /** * Pointer to a thread local variable identifying the scratch data objects @@ -421,13 +435,6 @@ namespace WorkStream */ const ScratchData &sample_scratch_data; - /** - * Counter for the number of emitted - * items. Each item may consist of up - * to chunk_size iterator elements. - */ - unsigned int n_emitted_items; - /** * Number of elements of the * iterator range that each @@ -445,21 +452,21 @@ namespace WorkStream /** * Initialize the pointers and vector * elements in the specified entry of - * the ring buffer. + * the item_buffer. */ void init_buffer_elements (const unsigned int element, const CopyData &sample_copy_data) { - Assert (ring_buffer[element].n_items == 0, + Assert (item_buffer[element].n_items == 0, ExcInternalError()); - ring_buffer[element].work_items + item_buffer[element].work_items .resize (chunk_size, remaining_iterator_range.second); - ring_buffer[element].scratch_data + item_buffer[element].scratch_data = &thread_local_scratch; - ring_buffer[element].sample_scratch_data + item_buffer[element].sample_scratch_data = &sample_scratch_data; - ring_buffer[element].copy_datas + item_buffer[element].copy_datas .resize (chunk_size, sample_copy_data); } }; @@ -489,10 +496,12 @@ namespace WorkStream */ Worker (const std_cxx1x::function &worker) + CopyData &)> &worker, + bool copier_exist=true) : tbb::filter (/* is_serial= */ false), - worker (worker) + worker (worker), + copier_exist (copier_exist) {} @@ -587,6 +596,9 @@ namespace WorkStream } } + // if there is no copier, mark current item as usable again + if (copier_exist==false) + current_item->currently_in_use = false; // then return the original pointer @@ -604,6 +616,12 @@ namespace WorkStream const std_cxx1x::function worker; + + /** + * This flag is true if the copier stage exist. If it does not, + * the worker has to free the buffer. Otherwise the copier will do it. + */ + bool copier_exist; }; @@ -672,6 +690,9 @@ namespace WorkStream } } + // mark current item as useable again + current_item->currently_in_use = false; + // return an invalid item since we are at the end of the // pipeline @@ -916,7 +937,7 @@ namespace WorkStream std::pair::const_iterator,typename std::vector::const_iterator> remaining_iterator_range; /** - * A ring buffer that will store items. + * A buffer that will store items. */ std::vector item_buffer; @@ -1238,20 +1259,37 @@ namespace WorkStream sample_scratch_data, sample_copy_data); - internal::Implementation2::Worker worker_filter (worker); - internal::Implementation2::Copier copier_filter (copier); - - // now create a pipeline from these stages - tbb::pipeline assembly_line; - assembly_line.add_filter (iterator_range_to_item_stream); - assembly_line.add_filter (worker_filter); + // Check that the copier exist if (static_cast& >(copier)) - assembly_line.add_filter (copier_filter); + { + internal::Implementation2::Worker worker_filter (worker); + internal::Implementation2::Copier copier_filter (copier); + + // now create a pipeline from these stages + tbb::pipeline assembly_line; + assembly_line.add_filter (iterator_range_to_item_stream); + assembly_line.add_filter (worker_filter); + assembly_line.add_filter (copier_filter); - // and run it - assembly_line.run (queue_length); + // and run it + assembly_line.run (queue_length); - assembly_line.clear (); + assembly_line.clear (); + } + else + { + internal::Implementation2::Worker worker_filter (worker,false); + + // now create a pipeline from these stages + tbb::pipeline assembly_line; + assembly_line.add_filter (iterator_range_to_item_stream); + assembly_line.add_filter (worker_filter); + + // and run it + assembly_line.run (queue_length); + + assembly_line.clear (); + } } #endif }