Orocos Real-Time Toolkit  2.6.0
BufferLockFree.hpp
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                         bufs.dequeue( itmp );
00174                         mpool.deallocate( itmp );
00175                     } while ( bufs.enqueue( mitem ) == false );
00176                 }
00177             }
00178             return true;
00179         }
00180 
00181         size_type Push(const std::vector<T>& items)
00182         {
00183             // @todo Make this function more efficient as in BufferLocked.
00184             int towrite  = items.size();
00185             typename std::vector<T>::const_iterator it;
00186             for(  it = items.begin(); it != items.end(); ++it)
00187                 if ( this->Push( *it ) == false ) {
00188                     break; // will only happen in non-circular case !
00189                 }
00190             return towrite - (items.end() - it);
00191         }
00192 
00193 
00194         bool Pop( reference_t item )
00195         {
00196             Item* ipop;
00197             if (bufs.dequeue( ipop ) == false )
00198                 return false;
00199             item = *ipop;
00200             if (mpool.deallocate( ipop ) == false )
00201                 assert(false);
00202             return true;
00203         }
00204 
00205         size_type Pop(std::vector<T>& items )
00206         {
00207             Item* ipop;
00208             items.clear();
00209             while( bufs.dequeue(ipop) ) {
00210                 items.push_back( *ipop );
00211                 if (mpool.deallocate(ipop) == false)
00212                     assert(false);
00213             }
00214             return items.size();
00215         }
00216         
00217         value_t* PopWithoutRelease()
00218     {
00219             Item* ipop;
00220             if (bufs.dequeue( ipop ) == false )
00221                 return 0;
00222         return ipop;
00223     }
00224 
00225     void Release(value_t *item) 
00226     {
00227             if (mpool.deallocate( item ) == false )
00228                 assert(false);  
00229     }
00230     };
00231 }}
00232 
00233 #endif