00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021 #ifndef __TBB_concurrent_queue_H
00022 #define __TBB_concurrent_queue_H
00023
00024 #include "_concurrent_queue_internal.h"
00025
00026 namespace tbb {
00027
00028 namespace strict_ppl {
00029
00031
00034 template<typename T, typename A = cache_aligned_allocator<T> >
00035 class concurrent_queue: public internal::concurrent_queue_base_v3<T> {
00036 template<typename Container, typename Value> friend class internal::concurrent_queue_iterator;
00037
00039 typedef typename A::template rebind<char>::other page_allocator_type;
00040 page_allocator_type my_allocator;
00041
00043 virtual void *allocate_block( size_t n ) {
00044 void *b = reinterpret_cast<void*>(my_allocator.allocate( n ));
00045 if( !b )
00046 internal::throw_exception(internal::eid_bad_alloc);
00047 return b;
00048 }
00049
00051 virtual void deallocate_block( void *b, size_t n ) {
00052 my_allocator.deallocate( reinterpret_cast<char*>(b), n );
00053 }
00054
00055 public:
00057 typedef T value_type;
00058
00060 typedef T& reference;
00061
00063 typedef const T& const_reference;
00064
00066 typedef size_t size_type;
00067
00069 typedef ptrdiff_t difference_type;
00070
00072 typedef A allocator_type;
00073
00075 explicit concurrent_queue(const allocator_type& a = allocator_type()) :
00076 my_allocator( a )
00077 {
00078 }
00079
00081 template<typename InputIterator>
00082 concurrent_queue( InputIterator begin, InputIterator end, const allocator_type& a = allocator_type()) :
00083 my_allocator( a )
00084 {
00085 for( ; begin != end; ++begin )
00086 this->internal_push(&*begin);
00087 }
00088
00090 concurrent_queue( const concurrent_queue& src, const allocator_type& a = allocator_type()) :
00091 internal::concurrent_queue_base_v3<T>(), my_allocator( a )
00092 {
00093 this->assign( src );
00094 }
00095
00097 ~concurrent_queue();
00098
00100 void push( const T& source ) {
00101 this->internal_push( &source );
00102 }
00103
00105
00107 bool try_pop( T& result ) {
00108 return this->internal_try_pop( &result );
00109 }
00110
00112 size_type unsafe_size() const {return this->internal_size();}
00113
00115 bool empty() const {return this->internal_empty();}
00116
00118 void clear() ;
00119
00121 allocator_type get_allocator() const { return this->my_allocator; }
00122
00123 typedef internal::concurrent_queue_iterator<concurrent_queue,T> iterator;
00124 typedef internal::concurrent_queue_iterator<concurrent_queue,const T> const_iterator;
00125
00126
00127
00128
00129 iterator unsafe_begin() {return iterator(*this);}
00130 iterator unsafe_end() {return iterator();}
00131 const_iterator unsafe_begin() const {return const_iterator(*this);}
00132 const_iterator unsafe_end() const {return const_iterator();}
00133 } ;
00134
00135 template<typename T, class A>
00136 concurrent_queue<T,A>::~concurrent_queue() {
00137 clear();
00138 this->internal_finish_clear();
00139 }
00140
00141 template<typename T, class A>
00142 void concurrent_queue<T,A>::clear() {
00143 while( !empty() ) {
00144 T value;
00145 this->internal_try_pop(&value);
00146 }
00147 }
00148
00149 }
00150
00152
00157 template<typename T, class A = cache_aligned_allocator<T> >
00158 class concurrent_bounded_queue: public internal::concurrent_queue_base_v3 {
00159 template<typename Container, typename Value> friend class internal::concurrent_queue_iterator;
00160
00162 typedef typename A::template rebind<char>::other page_allocator_type;
00163 page_allocator_type my_allocator;
00164
00165 typedef typename concurrent_queue_base_v3::padded_page<T> padded_page;
00166
00168 class destroyer: internal::no_copy {
00169 T& my_value;
00170 public:
00171 destroyer( T& value ) : my_value(value) {}
00172 ~destroyer() {my_value.~T();}
00173 };
00174
00175 T& get_ref( page& p, size_t index ) {
00176 __TBB_ASSERT( index<items_per_page, NULL );
00177 return (&static_cast<padded_page*>(static_cast<void*>(&p))->last)[index];
00178 }
00179
00180 virtual void copy_item( page& dst, size_t index, const void* src ) {
00181 new( &get_ref(dst,index) ) T(*static_cast<const T*>(src));
00182 }
00183
00184 virtual void copy_page_item( page& dst, size_t dindex, const page& src, size_t sindex ) {
00185 new( &get_ref(dst,dindex) ) T( get_ref( const_cast<page&>(src), sindex ) );
00186 }
00187
00188 virtual void assign_and_destroy_item( void* dst, page& src, size_t index ) {
00189 T& from = get_ref(src,index);
00190 destroyer d(from);
00191 *static_cast<T*>(dst) = from;
00192 }
00193
00194 virtual page *allocate_page() {
00195 size_t n = sizeof(padded_page) + (items_per_page-1)*sizeof(T);
00196 page *p = reinterpret_cast<page*>(my_allocator.allocate( n ));
00197 if( !p )
00198 internal::throw_exception(internal::eid_bad_alloc);
00199 return p;
00200 }
00201
00202 virtual void deallocate_page( page *p ) {
00203 size_t n = sizeof(padded_page) + items_per_page*sizeof(T);
00204 my_allocator.deallocate( reinterpret_cast<char*>(p), n );
00205 }
00206
00207 public:
00209 typedef T value_type;
00210
00212 typedef A allocator_type;
00213
00215 typedef T& reference;
00216
00218 typedef const T& const_reference;
00219
00221
00223 typedef std::ptrdiff_t size_type;
00224
00226 typedef std::ptrdiff_t difference_type;
00227
00229 explicit concurrent_bounded_queue(const allocator_type& a = allocator_type()) :
00230 concurrent_queue_base_v3( sizeof(T) ), my_allocator( a )
00231 {
00232 }
00233
00235 concurrent_bounded_queue( const concurrent_bounded_queue& src, const allocator_type& a = allocator_type()) :
00236 concurrent_queue_base_v3( sizeof(T) ), my_allocator( a )
00237 {
00238 assign( src );
00239 }
00240
00242 template<typename InputIterator>
00243 concurrent_bounded_queue( InputIterator begin, InputIterator end, const allocator_type& a = allocator_type()) :
00244 concurrent_queue_base_v3( sizeof(T) ), my_allocator( a )
00245 {
00246 for( ; begin != end; ++begin )
00247 internal_push_if_not_full(&*begin);
00248 }
00249
00251 ~concurrent_bounded_queue();
00252
00254 void push( const T& source ) {
00255 internal_push( &source );
00256 }
00257
00259
00260 void pop( T& destination ) {
00261 internal_pop( &destination );
00262 }
00263
00265
00267 bool try_push( const T& source ) {
00268 return internal_push_if_not_full( &source );
00269 }
00270
00272
00274 bool try_pop( T& destination ) {
00275 return internal_pop_if_present( &destination );
00276 }
00277
00279
00282 size_type size() const {return internal_size();}
00283
00285 bool empty() const {return internal_empty();}
00286
00288 size_type capacity() const {
00289 return my_capacity;
00290 }
00291
00293
00295 void set_capacity( size_type new_capacity ) {
00296 internal_set_capacity( new_capacity, sizeof(T) );
00297 }
00298
00300 allocator_type get_allocator() const { return this->my_allocator; }
00301
00303 void clear() ;
00304
00305 typedef internal::concurrent_queue_iterator<concurrent_bounded_queue,T> iterator;
00306 typedef internal::concurrent_queue_iterator<concurrent_bounded_queue,const T> const_iterator;
00307
00308
00309
00310
00311 iterator unsafe_begin() {return iterator(*this);}
00312 iterator unsafe_end() {return iterator();}
00313 const_iterator unsafe_begin() const {return const_iterator(*this);}
00314 const_iterator unsafe_end() const {return const_iterator();}
00315
00316 };
00317
00318 template<typename T, class A>
00319 concurrent_bounded_queue<T,A>::~concurrent_bounded_queue() {
00320 clear();
00321 internal_finish_clear();
00322 }
00323
00324 template<typename T, class A>
00325 void concurrent_bounded_queue<T,A>::clear() {
00326 while( !empty() ) {
00327 T value;
00328 internal_pop_if_present(&value);
00329 }
00330 }
00331
00332 namespace deprecated {
00333
00335
00340 template<typename T, class A = cache_aligned_allocator<T> >
00341 class concurrent_queue: public concurrent_bounded_queue<T,A> {
00342 #if !__TBB_TEMPLATE_FRIENDS_BROKEN
00343 template<typename Container, typename Value> friend class internal::concurrent_queue_iterator;
00344 #endif
00345
00346 public:
00348 explicit concurrent_queue(const A& a = A()) :
00349 concurrent_bounded_queue<T,A>( a )
00350 {
00351 }
00352
00354 concurrent_queue( const concurrent_queue& src, const A& a = A()) :
00355 concurrent_bounded_queue<T,A>( src, a )
00356 {
00357 }
00358
00360 template<typename InputIterator>
00361 concurrent_queue( InputIterator b , InputIterator e , const A& a = A()) :
00362 concurrent_bounded_queue<T,A>( b, e, a )
00363 {
00364 }
00365
00367
00369 bool push_if_not_full( const T& source ) {
00370 return this->try_push( source );
00371 }
00372
00374
00378 bool pop_if_present( T& destination ) {
00379 return this->try_pop( destination );
00380 }
00381
00382 typedef typename concurrent_bounded_queue<T,A>::iterator iterator;
00383 typedef typename concurrent_bounded_queue<T,A>::const_iterator const_iterator;
00384
00385
00386
00387
00388 iterator begin() {return this->unsafe_begin();}
00389 iterator end() {return this->unsafe_end();}
00390 const_iterator begin() const {return this->unsafe_begin();}
00391 const_iterator end() const {return this->unsafe_end();}
00392 };
00393
00394 }
00395
00396
00397 #if TBB_DEPRECATED
00398 using deprecated::concurrent_queue;
00399 #else
00400 using strict_ppl::concurrent_queue;
00401 #endif
00402
00403 }
00404
00405 #endif