00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012 #ifndef __deal2__work_stream_h
00013 #define __deal2__work_stream_h
00014
00015
00016 #include <deal.II/base/config.h>
00017 #include <deal.II/base/multithread_info.h>
00018 #include <deal.II/base/template_constraints.h>
00019 #include <deal.II/base/std_cxx1x/function.h>
00020 #include <deal.II/base/std_cxx1x/bind.h>
00021
00022 #if DEAL_II_USE_MT == 1
00023 # include <deal.II/base/thread_management.h>
00024 # include <tbb/pipeline.h>
00025 #endif
00026
00027 #include <vector>
00028 #include <utility>
00029
00030
00031 DEAL_II_NAMESPACE_OPEN
00032
00033
00034
00125 namespace WorkStream
00126 {
00127
00128 #if DEAL_II_USE_MT == 1
00129
00130
00131 namespace internal
00132 {
00137 template <typename Iterator,
00138 typename ScratchData,
00139 typename CopyData>
00140 class IteratorRangeToItemStream : public tbb::filter
00141 {
00142 public:
00156 typedef
00157 std_cxx1x::tuple<std::vector<Iterator>,
00158 ScratchData*,
00159 std::vector<CopyData>,
00160 unsigned int>
00161 ItemType;
00162
00163
00172 IteratorRangeToItemStream (const Iterator &begin,
00173 const Iterator &end,
00174 const unsigned int buffer_size,
00175 const unsigned int chunk_size,
00176 const ScratchData &sample_scratch_data,
00177 const CopyData &sample_copy_data)
00178 :
00179 tbb::filter (true),
00180 remaining_iterator_range (begin, end),
00181 ring_buffer (buffer_size),
00182 n_emitted_items (0),
00183 chunk_size (chunk_size)
00184 {
00185
00186
00187
00188
00189
00190
00191 Threads::TaskGroup<> tasks;
00192 for (unsigned int i=0; i<ring_buffer.size(); ++i)
00193 tasks += Threads::new_task (&IteratorRangeToItemStream::init_buffer_elements,
00194 *this,
00195 i,
00196 std_cxx1x::cref(sample_scratch_data),
00197 std_cxx1x::cref(sample_copy_data));
00198 tasks.join_all ();
00199 }
00200
00204 ~IteratorRangeToItemStream ()
00205 {
00206 for (unsigned int i=0; i<ring_buffer.size(); ++i)
00207 delete std_cxx1x::get<1>(ring_buffer[i]);
00208 }
00209
00214 virtual void * operator () (void *)
00215 {
00216
00217
00218 ItemType *current_item
00219 = &ring_buffer[n_emitted_items % ring_buffer.size()];
00220
00221
00222
00223
00224 std_cxx1x::get<3>(*current_item) = 0;
00225 while ((remaining_iterator_range.first !=
00226 remaining_iterator_range.second)
00227 &&
00228 (std_cxx1x::get<3>(*current_item) < chunk_size))
00229 {
00230 std_cxx1x::get<0>(*current_item)[std_cxx1x::get<3>(*current_item)]
00231 = remaining_iterator_range.first;
00232
00233 ++remaining_iterator_range.first;
00234 ++std_cxx1x::get<3>(*current_item);
00235 }
00236
00237 if (std_cxx1x::get<3>(*current_item) == 0)
00238
00239
00240 return 0;
00241 else
00242 {
00243 ++n_emitted_items;
00244 return current_item;
00245 }
00246 }
00247
00248 private:
00254 std::pair<Iterator,Iterator> remaining_iterator_range;
00255
00259 std::vector<ItemType> ring_buffer;
00260
00266 unsigned int n_emitted_items;
00267
00280 const unsigned int chunk_size;
00281
00287 void init_buffer_elements (const unsigned int element,
00288 const ScratchData &sample_scratch_data,
00289 const CopyData &sample_copy_data)
00290 {
00291 Assert (std_cxx1x::get<1>(ring_buffer[element]) == 0,
00292 ExcInternalError());
00293
00294 std_cxx1x::get<0>(ring_buffer[element])
00295 .resize (chunk_size, remaining_iterator_range.second);
00296 std_cxx1x::get<1>(ring_buffer[element])
00297 = new ScratchData(sample_scratch_data);
00298 std_cxx1x::get<2>(ring_buffer[element])
00299 .resize (chunk_size, sample_copy_data);
00300 }
00301 };
00302
00303
00304
00312 template <typename Iterator,
00313 typename ScratchData,
00314 typename CopyData>
00315 class Worker : public tbb::filter
00316 {
00317 public:
00326 Worker (const std_cxx1x::function<void (const Iterator &,
00327 ScratchData &,
00328 CopyData &)> &worker)
00329 :
00330 tbb::filter ( false),
00331 worker (worker)
00332 {}
00333
00334
00338 void * operator () (void *item)
00339 {
00340
00341 typedef
00342 typename IteratorRangeToItemStream<Iterator,ScratchData,CopyData>::ItemType
00343 ItemType;
00344
00345 ItemType *current_item = reinterpret_cast<ItemType*> (item);
00346
00347
00348
00349
00350 for (unsigned int i=0; i<std_cxx1x::get<3>(*current_item); ++i)
00351 worker (std_cxx1x::get<0>(*current_item)[i],
00352 *std_cxx1x::get<1>(*current_item),
00353 std_cxx1x::get<2>(*current_item)[i]);
00354
00355
00356
00357 return item;
00358 }
00359
00360
00361 private:
00367 const std_cxx1x::function<void (const Iterator &,
00368 ScratchData &,
00369 CopyData &)> worker;
00370 };
00371
00372
00373
00382 template <typename Iterator,
00383 typename ScratchData,
00384 typename CopyData>
00385 class Copier : public tbb::filter
00386 {
00387 public:
00399 Copier (const std_cxx1x::function<void (const CopyData &)> &copier)
00400 :
00401 tbb::filter ( true),
00402 copier (copier)
00403 {}
00404
00405
00409 void * operator () (void *item)
00410 {
00411
00412 typedef
00413 typename IteratorRangeToItemStream<Iterator,ScratchData,CopyData>::ItemType
00414 ItemType;
00415
00416 ItemType *current_item = reinterpret_cast<ItemType*> (item);
00417
00418
00419 for (unsigned int i=0; i<std_cxx1x::get<3>(*current_item); ++i)
00420 copier (std_cxx1x::get<2>(*current_item)[i]);
00421
00422
00423
00424
00425
00426 return 0;
00427 }
00428
00429
00430 private:
00436 const std_cxx1x::function<void (const CopyData &)> copier;
00437 };
00438
00439 }
00440
00441
00442 #endif // DEAL_II_USE_MT
00443
00444
00445
00500 template <typename Worker,
00501 typename Copier,
00502 typename Iterator,
00503 typename ScratchData,
00504 typename CopyData>
00505 void
00506 run (const Iterator &begin,
00507 const typename identity<Iterator>::type &end,
00508 Worker worker,
00509 Copier copier,
00510 const ScratchData &sample_scratch_data,
00511 const CopyData &sample_copy_data,
00512 const unsigned int queue_length = 2*multithread_info.n_default_threads,
00513 const unsigned int chunk_size = 8)
00514 {
00515 Assert (queue_length > 0,
00516 ExcMessage ("The queue length must be at least one, and preferably "
00517 "larger than the number of processors on this system."));
00518 Assert (chunk_size > 0,
00519 ExcMessage ("The chunk_size must be at least one."));
00520
00521
00522
00523
00524
00525 if (!(begin != end))
00526 return;
00527
00528 #if DEAL_II_USE_MT == 1
00529
00530
00531 internal::IteratorRangeToItemStream<Iterator,ScratchData,CopyData>
00532 iterator_range_to_item_stream (begin, end,
00533 queue_length,
00534 chunk_size,
00535 sample_scratch_data,
00536 sample_copy_data);
00537
00538 internal::Worker<Iterator, ScratchData, CopyData> worker_filter (worker);
00539 internal::Copier<Iterator, ScratchData, CopyData> copier_filter (copier);
00540
00541
00542
00543 tbb::pipeline assembly_line;
00544 assembly_line.add_filter (iterator_range_to_item_stream);
00545 assembly_line.add_filter (worker_filter);
00546 assembly_line.add_filter (copier_filter);
00547
00548
00549 assembly_line.run (queue_length);
00550
00551 assembly_line.clear ();
00552
00553 #else
00554
00555
00556
00557 ScratchData scratch_data = sample_scratch_data;
00558 CopyData copy_data = sample_copy_data;
00559
00560 for (Iterator i=begin; i!=end; ++i)
00561 {
00562 worker (i, scratch_data, copy_data);
00563 copier (copy_data);
00564 }
00565 #endif
00566 }
00567
00568
00569
00610 template <typename MainClass,
00611 typename Iterator,
00612 typename ScratchData,
00613 typename CopyData>
00614 void
00615 run (const Iterator &begin,
00616 const typename identity<Iterator>::type &end,
00617 MainClass &main_object,
00618 void (MainClass::*worker) (const Iterator &,
00619 ScratchData &,
00620 CopyData &),
00621 void (MainClass::*copier) (const CopyData &),
00622 const ScratchData &sample_scratch_data,
00623 const CopyData &sample_copy_data,
00624 const unsigned int queue_length = 2*multithread_info.n_default_threads,
00625 const unsigned int chunk_size = 8)
00626 {
00627
00628 run (begin, end,
00629 std_cxx1x::bind (worker,
00630 std_cxx1x::ref (main_object),
00631 std_cxx1x::_1, std_cxx1x::_2, std_cxx1x::_3),
00632 std_cxx1x::bind (copier,
00633 std_cxx1x::ref (main_object),
00634 std_cxx1x::_1),
00635 sample_scratch_data,
00636 sample_copy_data,
00637 queue_length,
00638 chunk_size);
00639 }
00640
00641 }
00642
00643
00644
00645
00646 DEAL_II_NAMESPACE_CLOSE
00647
00648
00649
00650
00651
00652
00653 #endif
00654
00655