common/scope_buffer.hpp

   1 //  Stethoscope shared memory buffer implementation
   2 //  This file is part of SuperCollider
   3 //
   4 //  Copyright (C) 2011 Jakob Leben
   5 //
   6 //  This program is free software; you can redistribute it and/or modify
   7 //  it under the terms of the GNU General Public License as published by
   8 //  the Free Software Foundation; either version 2 of the License, or
   9 //  (at your option) any later version.
  10 //
  11 //  This program is distributed in the hope that it will be useful,
  12 //  but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 //  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 //  GNU General Public License for more details.
  15 //
  16 //  You should have received a copy of the GNU General Public License
  17 //  along with this program; see the file COPYING.  If not, write to
  18 //  the Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19 //  Boston, MA 02111-1307, USA.
  20
  21 #ifndef SC_SCOPE_BUFFER_HPP
  22 #define SC_SCOPE_BUFFER_HPP
  23
  24 #include <boost/interprocess/offset_ptr.hpp>
  25 #include <boost/atomic.hpp>
  26
  27 extern "C" {
  28 #include "tlsf.h"
  29 }
  30
  31 namespace detail_server_shm {
  32
  33 using boost::interprocess::offset_ptr;
  34 using boost::atomic;
  35
  36 struct scope_buffer_writer;
  37 struct scope_buffer_reader;
  38
  39 class scope_buffer_pool
  40 {
  41 public:
  42
  43         void init (void * pool, size_t size_of_pool)
  44         {
  45                 pool_ = (char*)pool;
  46                 memset(pool_, 0, size_of_pool);
  47                 init_memory_pool(size_of_pool, pool_);
  48         }
  49
  50         void * allocate (size_t bytes)
  51         {
  52                 return malloc_ex(bytes, pool_);
  53         }
  54
  55         void deallocate (void * ptr)
  56         {
  57                 free_ex(ptr, pool_);
  58         }
  59
  60 private:
  61         friend class server_shared_memory;
  62         char * pool_;
  63 };
  64
  65 class scope_buffer
  66 {
  67         friend class scope_buffer_writer;
  68         friend class scope_buffer_reader;
  69
  70         typedef offset_ptr<float> sh_float_ptr;
  71
  72         enum status
  73         {
  74                 free = 0,
  75                 initialized
  76         };
  77
  78         atomic<int> _status;
  79
  80         uint _size;
  81         uint _channels;
  82         sh_float_ptr _data;
  83
  84         /*
  85         Reader/writer synchronization mechanism:
  86
  87         _stage, _in and _out are indexes into _state - an array of 3 equal data regions.
  88
  89         _out denotes the region where the writer writes.
  90         _in denotes the region where the reader reads.
  91         _stage denotes the region where data is exchanged between the writer and the reader.
  92
  93         After the writer is done writing, it sets the changed flag of the _in region,
  94         and swaps _in with _stage.
  95
  96         The reader polls the changed flag of the _stage region. If it is set, it swaps _out
  97         with _stage, reads the new _out region, and unsets its changed flag.
  98         */
  99
 100         atomic<int> _stage;
 101         int _in;
 102         int _out;
 103
 104         struct data_desc {
 105                 data_desc(): data(0), frames(0), changed(false) {}
 106                 sh_float_ptr data;
 107                 uint frames;
 108                 atomic<bool> changed;
 109         } _state [3];
 110
 111 public:
 112
 113         scope_buffer():
 114                 _status(free),
 115                 _stage(0),
 116                 _in(1),
 117                 _out(2)
 118         {}
 119
 120 private:
 121
 122         // writer interface
 123
 124         bool allocate( scope_buffer_pool & pool, uint channels, uint size )
 125         {
 126                 bool available = _status.load( boost::memory_order_relaxed ) == free;
 127                 if( !available ) return false;
 128
 129                 _size = size;
 130                 _channels = channels;
 131
 132                 uint asset_size = channels * size;
 133                 _data = (float*)pool.allocate( asset_size * 3 * sizeof(float) );
 134                 if (_data == NULL)
 135                         return false;
 136
 137                 _state[0].data = _data;
 138                 _state[1].data = _data + asset_size;
 139                 _state[2].data = _data + asset_size + asset_size;
 140
 141                 _status.store( initialized, boost::memory_order_release );
 142
 143                 return true;
 144         }
 145
 146         void release( scope_buffer_pool & pool )
 147         {
 148                 bool allocated = _status.load( boost::memory_order_relaxed ) != free;
 149                 if( !allocated ) return;
 150
 151                 pool.deallocate( _data.get() );
 152
 153                 _status.store( free, boost::memory_order_release );
 154         }
 155
 156         float * write_address() { return _state[_in].data.get(); }
 157
 158         void push( uint frames )
 159         {
 160                 _state[_in].frames = frames;
 161                 _state[_in].changed.store( true, boost::memory_order_relaxed );
 162                 _in = _stage.exchange( _in, boost::memory_order_release );
 163         }
 164
 165         // reader interface
 166
 167         bool valid()
 168         {
 169                 return _status.load( boost::memory_order_acquire ) == initialized;
 170         }
 171
 172         float * read_address() { return _state[_out].data.get(); }
 173
 174         uint pull()
 175         {
 176                 int stage = _stage.load( boost::memory_order_relaxed );
 177                 bool changed = _state[stage].changed.load( boost::memory_order_relaxed );
 178
 179                 if( changed )
 180                 {
 181                         _state[_out].changed.store( false, boost::memory_order_relaxed );
 182                         _out = _stage.exchange( _out, boost::memory_order_acquire );
 183                 }
 184
 185                 return _state[_out].frames;
 186         }
 187 };
 188
 189 struct scope_buffer_writer
 190 {
 191         scope_buffer *buffer;
 192
 193         scope_buffer_writer( scope_buffer *buffer = 0 ):
 194                 buffer(buffer)
 195         {}
 196
 197         scope_buffer_writer( scope_buffer *buf, scope_buffer_pool & pool, uint channels, uint size ):
 198                 buffer(buf)
 199         {
 200                 if( !buffer->allocate( pool, channels, size ) )
 201                         buffer = 0;
 202         }
 203
 204         bool valid()
 205         {
 206                 return buffer != 0;
 207         }
 208
 209         float *data()
 210         {
 211                 return buffer->write_address();
 212         }
 213
 214         uint max_size()
 215         {
 216                 return buffer->_size;
 217         }
 218
 219         void push( uint frames )
 220         {
 221                 buffer->push( frames );
 222         }
 223
 224         void release( scope_buffer_pool & pool )
 225         {
 226                 buffer->release( pool );
 227         }
 228 };
 229
 230 // FIXME: how do we ensure that scope_buffer data members used in the reader
 231 // are consistent among themselves at all times???
 232
 233 class scope_buffer_reader
 234 {
 235         scope_buffer *buffer;
 236
 237 public:
 238         scope_buffer_reader( scope_buffer *buffer_ = 0 ):
 239                 buffer(buffer_)
 240         {}
 241
 242         bool valid()
 243         {
 244                 // places an acquire memory ordering fence
 245                 return (buffer && buffer->valid());
 246         }
 247
 248         bool pull( uint &frames )
 249         {
 250                 uint new_frames = buffer->pull();
 251
 252                 if( new_frames )
 253                         frames = new_frames;
 254
 255                 return new_frames != 0;
 256         }
 257
 258         float *data()
 259         {
 260                 return buffer->read_address();
 261         }
 262
 263         uint max_frames()
 264         {
 265                 return buffer->_size;
 266         }
 267
 268         uint channels()
 269         {
 270                 return buffer->_channels;
 271         }
 272 };
 273
 274 } /* namespace detail_server_shm */
 275
 276 #endif