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 class scope_buffer_writer;
  37 class 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         unsigned int _size;
  81         unsigned int _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                 unsigned int 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, unsigned int channels, unsigned int 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                 unsigned int 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( unsigned int 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         unsigned int 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 class scope_buffer_writer
 190 {
 191 public:
 192
 193         scope_buffer *buffer;
 194
 195         scope_buffer_writer( scope_buffer *buffer = 0 ):
 196                 buffer(buffer)
 197         {}
 198
 199         scope_buffer_writer( scope_buffer *buf, scope_buffer_pool & pool, unsigned int channels, unsigned int size ):
 200                 buffer(buf)
 201         {
 202                 if( !buffer->allocate( pool, channels, size ) )
 203                         buffer = 0;
 204         }
 205
 206         bool valid()
 207         {
 208                 return buffer != 0;
 209         }
 210
 211         float *data()
 212         {
 213                 return buffer->write_address();
 214         }
 215
 216         unsigned int max_size()
 217         {
 218                 return buffer->_size;
 219         }
 220
 221         void push( unsigned int frames )
 222         {
 223                 buffer->push( frames );
 224         }
 225
 226         void release( scope_buffer_pool & pool )
 227         {
 228                 buffer->release( pool );
 229         }
 230 };
 231
 232 // FIXME: how do we ensure that scope_buffer data members used in the reader
 233 // are consistent among themselves at all times???
 234
 235 class scope_buffer_reader
 236 {
 237         scope_buffer *buffer;
 238
 239 public:
 240         scope_buffer_reader( scope_buffer *buffer_ = 0 ):
 241                 buffer(buffer_)
 242         {}
 243
 244         bool valid()
 245         {
 246                 // places an acquire memory ordering fence
 247                 return (buffer && buffer->valid());
 248         }
 249
 250         bool pull( unsigned int &frames )
 251         {
 252                 unsigned int new_frames = buffer->pull();
 253
 254                 if( new_frames )
 255                         frames = new_frames;
 256
 257                 return new_frames != 0;
 258         }
 259
 260         float *data()
 261         {
 262                 return buffer->read_address();
 263         }
 264
 265         unsigned int max_frames()
 266         {
 267                 return buffer->_size;
 268         }
 269
 270         unsigned int channels()
 271         {
 272                 return buffer->_channels;
 273         }
 274 };
 275
 276 } /* namespace detail_server_shm */
 277
 278 #endif