include/tbb/parallel_while.h

   1 /*
   2     Copyright (c) 2005-2019 Intel Corporation
   3
   4     Licensed under the Apache License, Version 2.0 (the "License");
   5     you may not use this file except in compliance with the License.
   6     You may obtain a copy of the License at
   7
   8         http://www.apache.org/licenses/LICENSE-2.0
   9
  10     Unless required by applicable law or agreed to in writing, software
  11     distributed under the License is distributed on an "AS IS" BASIS,
  12     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13     See the License for the specific language governing permissions and
  14     limitations under the License.
  15 */
  16
  17 #ifndef __TBB_parallel_while
  18 #define __TBB_parallel_while
  19
  20 #define __TBB_parallel_while_H_include_area
  21 #include "internal/_warning_suppress_enable_notice.h"
  22
  23 #include "task.h"
  24 #include <new>
  25
  26 namespace tbb {
  27
  28 template<typename Body>
  29 class parallel_while;
  30
  31 //! @cond INTERNAL
  32 namespace internal {
  33
  34     template<typename Stream, typename Body> class while_task;
  35
  36     //! For internal use only.
  37     /** Executes one iteration of a while.
  38         @ingroup algorithms */
  39     template<typename Body>
  40     class while_iteration_task: public task {
  41         const Body& my_body;
  42         typename Body::argument_type my_value;
  43         task* execute() __TBB_override {
  44             my_body(my_value);
  45             return NULL;
  46         }
  47         while_iteration_task( const typename Body::argument_type& value, const Body& body ) :
  48             my_body(body), my_value(value)
  49         {}
  50         template<typename Body_> friend class while_group_task;
  51         friend class tbb::parallel_while<Body>;
  52     };
  53
  54     //! For internal use only
  55     /** Unpacks a block of iterations.
  56         @ingroup algorithms */
  57     template<typename Body>
  58     class while_group_task: public task {
  59         static const size_t max_arg_size = 4;
  60         const Body& my_body;
  61         size_t size;
  62         typename Body::argument_type my_arg[max_arg_size];
  63         while_group_task( const Body& body ) : my_body(body), size(0) {}
  64         task* execute() __TBB_override {
  65             typedef while_iteration_task<Body> iteration_type;
  66             __TBB_ASSERT( size>0, NULL );
  67             task_list list;
  68             task* t;
  69             size_t k=0;
  70             for(;;) {
  71                 t = new( allocate_child() ) iteration_type(my_arg[k],my_body);
  72                 if( ++k==size ) break;
  73                 list.push_back(*t);
  74             }
  75             set_ref_count(int(k+1));
  76             spawn(list);
  77             spawn_and_wait_for_all(*t);
  78             return NULL;
  79         }
  80         template<typename Stream, typename Body_> friend class while_task;
  81     };
  82
  83     //! For internal use only.
  84     /** Gets block of iterations from a stream and packages them into a while_group_task.
  85         @ingroup algorithms */
  86     template<typename Stream, typename Body>
  87     class while_task: public task {
  88         Stream& my_stream;
  89         const Body& my_body;
  90         empty_task& my_barrier;
  91         task* execute() __TBB_override {
  92             typedef while_group_task<Body> block_type;
  93             block_type& t = *new( allocate_additional_child_of(my_barrier) ) block_type(my_body);
  94             size_t k=0;
  95             while( my_stream.pop_if_present(t.my_arg[k]) ) {
  96                 if( ++k==block_type::max_arg_size ) {
  97                     // There might be more iterations.
  98                     recycle_to_reexecute();
  99                     break;
 100                 }
 101             }
 102             if( k==0 ) {
 103                 destroy(t);
 104                 return NULL;
 105             } else {
 106                 t.size = k;
 107                 return &t;
 108             }
 109         }
 110         while_task( Stream& stream, const Body& body, empty_task& barrier ) :
 111             my_stream(stream),
 112             my_body(body),
 113             my_barrier(barrier)
 114         {}
 115         friend class tbb::parallel_while<Body>;
 116     };
 117
 118 } // namespace internal
 119 //! @endcond
 120
 121 //! Parallel iteration over a stream, with optional addition of more work.
 122 /** The Body b has the requirement: \n
 123         "b(v)"                      \n
 124         "b.argument_type"           \n
 125     where v is an argument_type
 126     @ingroup algorithms */
 127 template<typename Body>
 128 class parallel_while: internal::no_copy {
 129 public:
 130     //! Construct empty non-running parallel while.
 131     parallel_while() : my_body(NULL), my_barrier(NULL) {}
 132
 133     //! Destructor cleans up data members before returning.
 134     ~parallel_while() {
 135         if( my_barrier ) {
 136             my_barrier->destroy(*my_barrier);
 137             my_barrier = NULL;
 138         }
 139     }
 140
 141     //! Type of items
 142     typedef typename Body::argument_type value_type;
 143
 144     //! Apply body.apply to each item in the stream.
 145     /** A Stream s has the requirements \n
 146          "S::value_type"                \n
 147          "s.pop_if_present(value) is convertible to bool */
 148     template<typename Stream>
 149     void run( Stream& stream, const Body& body );
 150
 151     //! Add a work item while running.
 152     /** Should be executed only by body.apply or a thread spawned therefrom. */
 153     void add( const value_type& item );
 154
 155 private:
 156     const Body* my_body;
 157     empty_task* my_barrier;
 158 };
 159
 160 template<typename Body>
 161 template<typename Stream>
 162 void parallel_while<Body>::run( Stream& stream, const Body& body ) {
 163     using namespace internal;
 164     empty_task& barrier = *new( task::allocate_root() ) empty_task();
 165     my_body = &body;
 166     my_barrier = &barrier;
 167     my_barrier->set_ref_count(2);
 168     while_task<Stream,Body>& w = *new( my_barrier->allocate_child() ) while_task<Stream,Body>( stream, body, barrier );
 169     my_barrier->spawn_and_wait_for_all(w);
 170     my_barrier->destroy(*my_barrier);
 171     my_barrier = NULL;
 172     my_body = NULL;
 173 }
 174
 175 template<typename Body>
 176 void parallel_while<Body>::add( const value_type& item ) {
 177     __TBB_ASSERT(my_barrier,"attempt to add to parallel_while that is not running");
 178     typedef internal::while_iteration_task<Body> iteration_type;
 179     iteration_type& i = *new( task::allocate_additional_child_of(*my_barrier) ) iteration_type(item,*my_body);
 180     task::self().spawn( i );
 181 }
 182
 183 } // namespace
 184
 185 #include "internal/_warning_suppress_disable_notice.h"
 186 #undef __TBB_parallel_while_H_include_area
 187
 188 #endif /* __TBB_parallel_while */