BufferLockFree.hpp
Go to the documentation of this file.
00001 /***************************************************************************
00002   tag: Peter Soetens  Thu Jan 13 10:24:51 CET 2005  BufferLockFree.hpp
00003 
00004                         BufferLockFree.hpp -  description
00005                            -------------------
00006     begin                : Thu January 13 2005
00007     copyright            : (C) 2005 Peter Soetens
00008     email                : peter.soetens@mech.kuleuven.ac.be
00009 
00010  ***************************************************************************
00011  *   This library is free software; you can redistribute it and/or         *
00012  *   modify it under the terms of the GNU General Public                   *
00013  *   License as published by the Free Software Foundation;                 *
00014  *   version 2 of the License.                                             *
00015  *                                                                         *
00016  *   As a special exception, you may use this file as part of a free       *
00017  *   software library without restriction.  Specifically, if other files   *
00018  *   instantiate templates or use macros or inline functions from this     *
00019  *   file, or you compile this file and link it with other files to        *
00020  *   produce an executable, this file does not by itself cause the         *
00021  *   resulting executable to be covered by the GNU General Public          *
00022  *   License.  This exception does not however invalidate any other        *
00023  *   reasons why the executable file might be covered by the GNU General   *
00024  *   Public License.                                                       *
00025  *                                                                         *
00026  *   This library is distributed in the hope that it will be useful,       *
00027  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
00028  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU     *
00029  *   Lesser General Public License for more details.                       *
00030  *                                                                         *
00031  *   You should have received a copy of the GNU General Public             *
00032  *   License along with this library; if not, write to the Free Software   *
00033  *   Foundation, Inc., 59 Temple Place,                                    *
00034  *   Suite 330, Boston, MA  02111-1307  USA                                *
00035  *                                                                         *
00036  ***************************************************************************/
00037 
00038 #ifndef ORO_BUFFER_LOCK_FREE_HPP
00039 #define ORO_BUFFER_LOCK_FREE_HPP
00040 
00041 #include "../os/oro_arch.h"
00042 #include "../os/CAS.hpp"
00043 #include "BufferInterface.hpp"
00044 #include "../internal/AtomicMWSRQueue.hpp"
00045 #include "../internal/TsPool.hpp"
00046 #include <vector>
00047 
00048 #ifdef ORO_PRAGMA_INTERFACE
00049 #pragma interface
00050 #endif
00051 
00052 namespace RTT
00053 { namespace base {
00054 
00055 
00056     using os::CAS;
00057 
00067     template< class T>
00068     class BufferLockFree
00069         : public BufferInterface<T>
00070     {
00071     public:
00072         typedef typename BufferInterface<T>::reference_t reference_t;
00073         typedef typename BufferInterface<T>::param_t param_t;
00074         typedef typename BufferInterface<T>::size_type size_type;
00075         typedef T value_t;
00076     private:
00077         typedef T Item;
00078         internal::AtomicMWSRQueue<Item*> bufs;
00079         // is mutable because of reference counting.
00080         mutable internal::TsPool<Item> mpool;
00081         const bool mcircular;
00082     public:
00087         BufferLockFree( unsigned int bufsize, const T& initial_value = T(), bool circular = false)
00088             : bufs( bufsize ), mpool(bufsize + 1), mcircular(circular)
00089         {
00090             mpool.data_sample( initial_value );
00091         }
00092 
00093         ~BufferLockFree() {
00094             // free all items still in the buffer.
00095             clear();
00096         }
00097 
00098         virtual void data_sample( const T& sample )
00099         {
00100             mpool.data_sample(sample);
00101         }
00102 
00103         virtual T data_sample() const
00104         {
00105             T result = T();
00106             Item* mitem = mpool.allocate();
00107             if (mitem != 0) {
00108                 result = *mitem;
00109                 mpool.deallocate( mitem );
00110             }
00111             return result;
00112         }
00113 
00114 
00115         size_type capacity() const
00116         {
00117             return bufs.capacity();
00118         }
00119 
00120         size_type size() const
00121         {
00122             return bufs.size();
00123         }
00124 
00125         bool empty() const
00126         {
00127             return bufs.isEmpty();
00128         }
00129 
00130         bool full() const
00131         {
00132             return bufs.isFull();
00133         }
00134 
00135         void clear()
00136         {
00137             Item* item;
00138             while ( bufs.dequeue(item) )
00139                 mpool.deallocate( item );
00140         }
00141 
00142         bool Push( param_t item)
00143         {
00144             if ( capacity() == (size_type)bufs.size() ) {
00145                 if (!mcircular)
00146                     return false;
00147                 // we will recover below in case of circular
00148             }
00149             Item* mitem = mpool.allocate();
00150             if ( mitem == 0 ) { // queue full ( rare but possible in race with PopWithoutRelease )
00151                 if (!mcircular)
00152                     return false;
00153                 else {
00154                     if (bufs.dequeue( mitem ) == false )
00155                         return false; // assert(false) ???
00156                     // we keep mitem to write item to next
00157                 }
00158             }
00159 
00160             // copy over.
00161             *mitem = item;
00162             if (bufs.enqueue( mitem ) == false ) {
00163                 //got memory, but buffer is full
00164                 //this can happen, as the memory pool is
00165                 //bigger than the buffer
00166                 if (!mcircular) {
00167                     mpool.deallocate( mitem );
00168                     return false;
00169                 } else {
00170                     // pop & deallocate until we have free space.
00171                     Item* itmp = 0;
00172                     do {
00173                         if ( bufs.dequeue( itmp ) ) {
00174                             mpool.deallocate( itmp );
00175                         } else {
00176                             // Both operations, enqueue() and dequeue() failed on the buffer:
00177                             // We could free the allocated pool item return false here,
00178                             // but in fact this can only happen during massive concurrent
00179                             // access to the circular buffer or in the trivial case that
00180                             // the buffer size is zero. So just keep on trying...
00181                         }
00182                     } while ( bufs.enqueue( mitem ) == false );
00183                 }
00184             }
00185             return true;
00186         }
00187 
00188         size_type Push(const std::vector<T>& items)
00189         {
00190             // @todo Make this function more efficient as in BufferLocked.
00191             int towrite  = items.size();
00192             typename std::vector<T>::const_iterator it;
00193             for(  it = items.begin(); it != items.end(); ++it)
00194                 if ( this->Push( *it ) == false ) {
00195                     break; // will only happen in non-circular case !
00196                 }
00197             return towrite - (items.end() - it);
00198         }
00199 
00200 
00201         bool Pop( reference_t item )
00202         {
00203             Item* ipop;
00204             if (bufs.dequeue( ipop ) == false )
00205                 return false;
00206             item = *ipop;
00207             if (mpool.deallocate( ipop ) == false )
00208                 assert(false);
00209             return true;
00210         }
00211 
00212         size_type Pop(std::vector<T>& items )
00213         {
00214             Item* ipop;
00215             items.clear();
00216             while( bufs.dequeue(ipop) ) {
00217                 items.push_back( *ipop );
00218                 if (mpool.deallocate(ipop) == false)
00219                     assert(false);
00220             }
00221             return items.size();
00222         }
00223         
00224         value_t* PopWithoutRelease()
00225         {
00226             Item* ipop;
00227             if (bufs.dequeue( ipop ) == false )
00228                 return 0;
00229             return ipop;
00230         }
00231 
00232         void Release(value_t *item) 
00233         {
00234             if (mpool.deallocate( item ) == false )
00235                 assert(false);  
00236         }
00237     };
00238 }}
00239 
00240 #endif


rtt
Author(s): RTT Developers
autogenerated on Fri Sep 9 2016 04:01:50