git.sesse.net Git - casparcg/blob - tbb30_20100406oss/include/tbb/pipeline.h

   1 /*
   2     Copyright 2005-2010 Intel Corporation.  All Rights Reserved.
   3
   4     This file is part of Threading Building Blocks.
   5
   6     Threading Building Blocks is free software; you can redistribute it
   7     and/or modify it under the terms of the GNU General Public License
   8     version 2 as published by the Free Software Foundation.
   9
  10     Threading Building Blocks is distributed in the hope that it will be
  11     useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  12     of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13     GNU General Public License for more details.
  14
  15     You should have received a copy of the GNU General Public License
  16     along with Threading Building Blocks; if not, write to the Free Software
  17     Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  18
  19     As a special exception, you may use this file as part of a free software
  20     library without restriction.  Specifically, if other files instantiate
  21     templates or use macros or inline functions from this file, or you compile
  22     this file and link it with other files to produce an executable, this
  23     file does not by itself cause the resulting executable to be covered by
  24     the GNU General Public License.  This exception does not however
  25     invalidate any other reasons why the executable file might be covered by
  26     the GNU General Public License.
  27 */
  28
  29 #ifndef __TBB_pipeline_H
  30 #define __TBB_pipeline_H
  31
  32 #include "atomic.h"
  33 #include "task.h"
  34 #include <cstddef>
  35
  36 namespace tbb {
  37
  38 class pipeline;
  39 class filter;
  40
  41 //! @cond INTERNAL
  42 namespace internal {
  43
  44 // The argument for PIPELINE_VERSION should be an integer between 2 and 9
  45 #define __TBB_PIPELINE_VERSION(x) (unsigned char)(x-2)<<1
  46
  47 typedef unsigned long Token;
  48 typedef long tokendiff_t;
  49 class stage_task;
  50 class input_buffer;
  51 class pipeline_root_task;
  52 class pipeline_cleaner;
  53
  54 } // namespace internal
  55
  56 namespace interface5 {
  57     template<typename T, typename U> class filter_t;
  58
  59     namespace internal {
  60         class pipeline_proxy;
  61     }
  62 }
  63
  64 //! @endcond
  65
  66 //! A stage in a pipeline.
  67 /** @ingroup algorithms */
  68 class filter: internal::no_copy {
  69 private:
  70     //! Value used to mark "not in pipeline"
  71     static filter* not_in_pipeline() {return reinterpret_cast<filter*>(intptr_t(-1));}
  72
  73     //! The lowest bit 0 is for parallel vs. serial
  74     static const unsigned char filter_is_serial = 0x1;
  75
  76     //! 4th bit distinguishes ordered vs unordered filters.
  77     /** The bit was not set for parallel filters in TBB 2.1 and earlier,
  78         but is_ordered() function always treats parallel filters as out of order. */
  79     static const unsigned char filter_is_out_of_order = 0x1<<4;
  80
  81     //! 5th bit distinguishes thread-bound and regular filters.
  82     static const unsigned char filter_is_bound = 0x1<<5;
  83
  84     //! 7th bit defines exception propagation mode expected by the application.
  85     static const unsigned char exact_exception_propagation =
  86 #if TBB_USE_CAPTURED_EXCEPTION
  87             0x0;
  88 #else
  89             0x1<<7;
  90 #endif /* TBB_USE_CAPTURED_EXCEPTION */
  91
  92     static const unsigned char current_version = __TBB_PIPELINE_VERSION(5);
  93     static const unsigned char version_mask = 0x7<<1; // bits 1-3 are for version
  94 public:
  95     enum mode {
  96         //! processes multiple items in parallel and in no particular order
  97         parallel = current_version | filter_is_out_of_order,
  98         //! processes items one at a time; all such filters process items in the same order
  99         serial_in_order = current_version | filter_is_serial,
 100         //! processes items one at a time and in no particular order
 101         serial_out_of_order = current_version | filter_is_serial | filter_is_out_of_order,
 102         //! @deprecated use serial_in_order instead
 103         serial = serial_in_order
 104     };
 105 protected:
 106     filter( bool is_serial_ ) :
 107         next_filter_in_pipeline(not_in_pipeline()),
 108         my_input_buffer(NULL),
 109         my_filter_mode(static_cast<unsigned char>((is_serial_ ? serial : parallel) | exact_exception_propagation)),
 110         prev_filter_in_pipeline(not_in_pipeline()),
 111         my_pipeline(NULL),
 112         next_segment(NULL)
 113     {}
 114
 115     filter( mode filter_mode ) :
 116         next_filter_in_pipeline(not_in_pipeline()),
 117         my_input_buffer(NULL),
 118         my_filter_mode(static_cast<unsigned char>(filter_mode | exact_exception_propagation)),
 119         prev_filter_in_pipeline(not_in_pipeline()),
 120         my_pipeline(NULL),
 121         next_segment(NULL)
 122     {}
 123
 124 public:
 125     //! True if filter is serial.
 126     bool is_serial() const {
 127         return bool( my_filter_mode & filter_is_serial );
 128     }
 129
 130     //! True if filter must receive stream in order.
 131     bool is_ordered() const {
 132         return (my_filter_mode & (filter_is_out_of_order|filter_is_serial))==filter_is_serial;
 133     }
 134
 135     //! True if filter is thread-bound.
 136     bool is_bound() const {
 137         return ( my_filter_mode & filter_is_bound )==filter_is_bound;
 138     }
 139
 140     //! Operate on an item from the input stream, and return item for output stream.
 141     /** Returns NULL if filter is a sink. */
 142     virtual void* operator()( void* item ) = 0;
 143
 144     //! Destroy filter.
 145     /** If the filter was added to a pipeline, the pipeline must be destroyed first. */
 146     virtual __TBB_EXPORTED_METHOD ~filter();
 147
 148 #if __TBB_TASK_GROUP_CONTEXT
 149     //! Destroys item if pipeline was cancelled.
 150     /** Required to prevent memory leaks.
 151         Note it can be called concurrently even for serial filters.*/
 152     virtual void finalize( void* /*item*/ ) {};
 153 #endif
 154
 155 private:
 156     //! Pointer to next filter in the pipeline.
 157     filter* next_filter_in_pipeline;
 158
 159     //! Buffer for incoming tokens, or NULL if not required.
 160     /** The buffer is required if the filter is serial or follows a thread-bound one. */
 161     internal::input_buffer* my_input_buffer;
 162
 163     friend class internal::stage_task;
 164     friend class internal::pipeline_root_task;
 165     friend class pipeline;
 166     friend class thread_bound_filter;
 167
 168     //! Storage for filter mode and dynamically checked implementation version.
 169     const unsigned char my_filter_mode;
 170
 171     //! Pointer to previous filter in the pipeline.
 172     filter* prev_filter_in_pipeline;
 173
 174     //! Pointer to the pipeline.
 175     pipeline* my_pipeline;
 176
 177     //! Pointer to the next "segment" of filters, or NULL if not required.
 178     /** In each segment, the first filter is not thread-bound but follows a thread-bound one. */
 179     filter* next_segment;
 180 };
 181
 182 //! A stage in a pipeline served by a user thread.
 183 /** @ingroup algorithms */
 184 class thread_bound_filter: public filter {
 185 public:
 186     enum result_type {
 187         // item was processed
 188         success,
 189         // item is currently not available
 190         item_not_available,
 191         // there are no more items to process
 192         end_of_stream
 193     };
 194 protected:
 195     thread_bound_filter(mode filter_mode):
 196          filter(static_cast<mode>(filter_mode | filter::filter_is_bound | filter::exact_exception_propagation))
 197     {}
 198 public:
 199     //! If a data item is available, invoke operator() on that item.
 200     /** This interface is non-blocking.
 201         Returns 'success' if an item was processed.
 202         Returns 'item_not_available' if no item can be processed now
 203         but more may arrive in the future, or if token limit is reached.
 204         Returns 'end_of_stream' if there are no more items to process. */
 205     result_type __TBB_EXPORTED_METHOD try_process_item();
 206
 207     //! Wait until a data item becomes available, and invoke operator() on that item.
 208     /** This interface is blocking.
 209         Returns 'success' if an item was processed.
 210         Returns 'end_of_stream' if there are no more items to process.
 211         Never returns 'item_not_available', as it blocks until another return condition applies. */
 212     result_type __TBB_EXPORTED_METHOD process_item();
 213
 214 private:
 215     //! Internal routine for item processing
 216     result_type internal_process_item(bool is_blocking);
 217 };
 218
 219 //! A processing pipeling that applies filters to items.
 220 /** @ingroup algorithms */
 221 class pipeline {
 222 public:
 223     //! Construct empty pipeline.
 224     __TBB_EXPORTED_METHOD pipeline();
 225
 226     /** Though the current implementation declares the destructor virtual, do not rely on this
 227         detail.  The virtualness is deprecated and may disappear in future versions of TBB. */
 228     virtual __TBB_EXPORTED_METHOD ~pipeline();
 229
 230     //! Add filter to end of pipeline.
 231     void __TBB_EXPORTED_METHOD add_filter( filter& filter_ );
 232
 233     //! Run the pipeline to completion.
 234     void __TBB_EXPORTED_METHOD run( size_t max_number_of_live_tokens );
 235
 236 #if __TBB_TASK_GROUP_CONTEXT
 237     //! Run the pipeline to completion with user-supplied context.
 238     void __TBB_EXPORTED_METHOD run( size_t max_number_of_live_tokens, tbb::task_group_context& context );
 239 #endif
 240
 241     //! Remove all filters from the pipeline.
 242     void __TBB_EXPORTED_METHOD clear();
 243
 244 private:
 245     friend class internal::stage_task;
 246     friend class internal::pipeline_root_task;
 247     friend class filter;
 248     friend class thread_bound_filter;
 249     friend class internal::pipeline_cleaner;
 250     friend class tbb::interface5::internal::pipeline_proxy;
 251
 252     //! Pointer to first filter in the pipeline.
 253     filter* filter_list;
 254
 255     //! Pointer to location where address of next filter to be added should be stored.
 256     filter* filter_end;
 257
 258     //! task who's reference count is used to determine when all stages are done.
 259     task* end_counter;
 260
 261     //! Number of idle tokens waiting for input stage.
 262     atomic<internal::Token> input_tokens;
 263
 264     //! Global counter of tokens
 265     atomic<internal::Token> token_counter;
 266
 267     //! False until fetch_input returns NULL.
 268     bool end_of_input;
 269
 270     //! True if the pipeline contains a thread-bound filter; false otherwise.
 271     bool has_thread_bound_filters;
 272
 273     //! Remove filter from pipeline.
 274     void remove_filter( filter& filter_ );
 275
 276     //! Not used, but retained to satisfy old export files.
 277     void __TBB_EXPORTED_METHOD inject_token( task& self );
 278
 279 #if __TBB_TASK_GROUP_CONTEXT
 280     //! Does clean up if pipeline is cancelled or exception occured
 281     void clear_filters();
 282 #endif
 283 };
 284
 285 //------------------------------------------------------------------------
 286 // Support for lambda-friendly parallel_pipeline interface
 287 //------------------------------------------------------------------------
 288
 289 namespace interface5 {
 290
 291 namespace internal {
 292     template<typename T, typename U, typename Body> class concrete_filter;
 293 }
 294
 295 class flow_control {
 296     bool is_pipeline_stopped;
 297     flow_control() { is_pipeline_stopped = false; }
 298     template<typename T, typename U, typename Body> friend class internal::concrete_filter;
 299 public:
 300     void stop() { is_pipeline_stopped = true; }
 301 };
 302
 303 //! @cond INTERNAL
 304 namespace internal {
 305
 306 template<typename T, typename U, typename Body>
 307 class concrete_filter: public tbb::filter {
 308     Body my_body;
 309
 310     /*override*/ void* operator()(void* input) {
 311         T* temp_input = (T*)input;
 312         // Call user's operator()() here
 313         void* output = (void*) new U(my_body(*temp_input));
 314         delete temp_input;
 315         return output;
 316     }
 317
 318 public:
 319     concrete_filter(tbb::filter::mode filter_mode, const Body& body) : filter(filter_mode), my_body(body) {}
 320 };
 321
 322 template<typename U, typename Body>
 323 class concrete_filter<void,U,Body>: public filter {
 324     Body my_body;
 325
 326     /*override*/void* operator()(void*) {
 327         flow_control control;
 328         U temp_output = my_body(control);
 329         void* output = control.is_pipeline_stopped ? NULL : (void*) new U(temp_output);
 330         return output;
 331     }
 332 public:
 333     concrete_filter(tbb::filter::mode filter_mode, const Body& body) : filter(filter_mode), my_body(body) {}
 334 };
 335
 336 template<typename T, typename Body>
 337 class concrete_filter<T,void,Body>: public filter {
 338     Body my_body;
 339
 340     /*override*/ void* operator()(void* input) {
 341         T* temp_input = (T*)input;
 342         my_body(*temp_input);
 343         delete temp_input;
 344         return NULL;
 345     }
 346 public:
 347     concrete_filter(tbb::filter::mode filter_mode, const Body& body) : filter(filter_mode), my_body(body) {}
 348 };
 349
 350 template<typename Body>
 351 class concrete_filter<void,void,Body>: public filter {
 352     Body my_body;
 353
 354     /** Override privately because it is always called virtually */
 355     /*override*/ void* operator()(void*) {
 356         flow_control control;
 357         my_body(control);
 358         void* output = control.is_pipeline_stopped ? NULL : (void*)(intptr_t)-1;
 359         return output;
 360     }
 361 public:
 362     concrete_filter(filter::mode filter_mode, const Body& body) : filter(filter_mode), my_body(body) {}
 363 };
 364
 365 //! The class that represents an object of the pipeline for parallel_pipeline().
 366 /** It primarily serves as RAII class that deletes heap-allocated filter instances. */
 367 class pipeline_proxy {
 368     tbb::pipeline my_pipe;
 369 public:
 370     pipeline_proxy( const filter_t<void,void>& filter_chain );
 371     ~pipeline_proxy() {
 372         while( filter* f = my_pipe.filter_list )
 373             delete f; // filter destructor removes it from the pipeline
 374     }
 375     tbb::pipeline* operator->() { return &my_pipe; }
 376 };
 377
 378 //! Abstract base class that represents a node in a parse tree underlying a filter_t.
 379 /** These nodes are always heap-allocated and can be shared by filter_t objects. */
 380 class filter_node: tbb::internal::no_copy {
 381     /** Count must be atomic because it is hidden state for user, but might be shared by threads. */
 382     tbb::atomic<intptr_t> ref_count;
 383 protected:
 384     filter_node() {
 385         ref_count = 0;
 386 #ifdef __TBB_TEST_FILTER_NODE_COUNT
 387         ++(__TBB_TEST_FILTER_NODE_COUNT);
 388 #endif
 389     }
 390 public:
 391     //! Add concrete_filter to pipeline
 392     virtual void add_to( pipeline& ) = 0;
 393     //! Increment reference count
 394     void add_ref() {++ref_count;}
 395     //! Decrement reference count and delete if it becomes zero.
 396     void remove_ref() {
 397         __TBB_ASSERT(ref_count>0,"ref_count underflow");
 398         if( --ref_count==0 )
 399             delete this;
 400     }
 401     virtual ~filter_node() {
 402 #ifdef __TBB_TEST_FILTER_NODE_COUNT
 403         --(__TBB_TEST_FILTER_NODE_COUNT);
 404 #endif
 405     }
 406 };
 407
 408 //! Node in parse tree representing result of make_filter.
 409 template<typename T, typename U, typename Body>
 410 class filter_node_leaf: public filter_node  {
 411     const tbb::filter::mode mode;
 412     const Body& body;
 413     /*override*/void add_to( pipeline& p ) {
 414         concrete_filter<T,U,Body>* f = new concrete_filter<T,U,Body>(mode,body);
 415         p.add_filter( *f );
 416     }
 417 public:
 418     filter_node_leaf( tbb::filter::mode m, const Body& b ) : mode(m), body(b) {}
 419 };
 420
 421 //! Node in parse tree representing join of two filters.
 422 class filter_node_join: public filter_node {
 423     friend class filter_node; // to suppress GCC 3.2 warnings
 424     filter_node& left;
 425     filter_node& right;
 426     /*override*/~filter_node_join() {
 427        left.remove_ref();
 428        right.remove_ref();
 429     }
 430     /*override*/void add_to( pipeline& p ) {
 431         left.add_to(p);
 432         right.add_to(p);
 433     }
 434 public:
 435     filter_node_join( filter_node& x, filter_node& y ) : left(x), right(y) {
 436        left.add_ref();
 437        right.add_ref();
 438     }
 439 };
 440
 441 } // namespace internal
 442 //! @endcond
 443
 444 template<typename T, typename U, typename Body>
 445 filter_t<T,U> make_filter(tbb::filter::mode mode, const Body& body) {
 446     return new internal::filter_node_leaf<T,U,Body>(mode, body);
 447 }
 448
 449 template<typename T, typename V, typename U>
 450 filter_t<T,U> operator& (const filter_t<T,V>& left, const filter_t<V,U>& right) {
 451     __TBB_ASSERT(left.root,"cannot use default-constructed filter_t as left argument of '&'");
 452     __TBB_ASSERT(right.root,"cannot use default-constructed filter_t as right argument of '&'");
 453     return new internal::filter_node_join(*left.root,*right.root);
 454 }
 455
 456 //! Class representing a chain of type-safe pipeline filters
 457 template<typename T, typename U>
 458 class filter_t {
 459     typedef internal::filter_node filter_node;
 460     filter_node* root;
 461     filter_t( filter_node* root_ ) : root(root_) {
 462         root->add_ref();
 463     }
 464     friend class internal::pipeline_proxy;
 465     template<typename T_, typename U_, typename Body>
 466     friend filter_t<T_,U_> make_filter(tbb::filter::mode, const Body& );
 467     template<typename T_, typename V_, typename U_>
 468     friend filter_t<T_,U_> operator& (const filter_t<T_,V_>& , const filter_t<V_,U_>& );
 469 public:
 470     filter_t() : root(NULL) {}
 471     filter_t( const filter_t<T,U>& rhs ) : root(rhs.root) {
 472         if( root ) root->add_ref();
 473     }
 474     template<typename Body>
 475     filter_t( tbb::filter::mode mode, const Body& body ) :
 476         root( new internal::filter_node_leaf<T,U,Body>(mode, body) ) {
 477         root->add_ref();
 478     }
 479
 480     void operator=( const filter_t<T,U>& rhs ) {
 481         // Order of operations below carefully chosen so that reference counts remain correct
 482         // in unlikely event that remove_ref throws exception.
 483         filter_node* old = root;
 484         root = rhs.root;
 485         if( root ) root->add_ref();
 486         if( old ) old->remove_ref();
 487     }
 488     ~filter_t() {
 489         if( root ) root->remove_ref();
 490     }
 491     void clear() {
 492         // Like operator= with filter_t() on right side.
 493         if( root ) {
 494             filter_node* old = root;
 495             root = NULL;
 496             old->remove_ref();
 497         }
 498     }
 499 };
 500
 501 inline internal::pipeline_proxy::pipeline_proxy( const filter_t<void,void>& filter_chain ) : my_pipe() {
 502     __TBB_ASSERT( filter_chain.root, "cannot apply parallel_pipeline to default-constructed filter_t"  );
 503     filter_chain.root->add_to(my_pipe);
 504 }
 505
 506 inline void parallel_pipeline(size_t max_number_of_live_tokens, const filter_t<void,void>& filter_chain
 507 #if __TBB_TASK_GROUP_CONTEXT
 508     , tbb::task_group_context& context
 509 #endif
 510     ) {
 511     internal::pipeline_proxy pipe(filter_chain);
 512     // tbb::pipeline::run() is called via the proxy
 513     pipe->run(max_number_of_live_tokens
 514 #if __TBB_TASK_GROUP_CONTEXT
 515               , context
 516 #endif
 517     );
 518 }
 519
 520 #if __TBB_TASK_GROUP_CONTEXT
 521 inline void parallel_pipeline(size_t max_number_of_live_tokens, const filter_t<void,void>& filter_chain) {
 522     tbb::task_group_context context;
 523     parallel_pipeline(max_number_of_live_tokens, filter_chain, context);
 524 }
 525 #endif // __TBB_TASK_GROUP_CONTEXT
 526
 527 } // interface5
 528
 529 using interface5::flow_control;
 530 using interface5::filter_t;
 531 using interface5::make_filter;
 532 using interface5::parallel_pipeline;
 533
 534 } // tbb
 535
 536 #endif /* __TBB_pipeline_H */