Marsyas  0.5.0-beta1
/Users/jleben/code/marsyas/src/marsyas/realtime/realvec_queue.h
Go to the documentation of this file.
00001 /*
00002 ** Copyright (C) 1998-2013 George Tzanetakis <gtzan@cs.uvic.ca>
00003 **
00004 ** This program is free software; you can redistribute it and/or modify
00005 ** it under the terms of the GNU General Public License as published by
00006 ** the Free Software Foundation; either version 2 of the License, or
00007 ** (at your option) any later version.
00008 **
00009 ** This program is distributed in the hope that it will be useful,
00010 ** but WITHOUT ANY WARRANTY; without even the implied warranty of
00011 ** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
00012 ** GNU General Public License for more details.
00013 **
00014 ** You should have received a copy of the GNU General Public License
00015 ** along with this program; if not, write to the Free Software
00016 ** Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
00017 */
00018 
00019 #ifndef MARSYAS_REALVEC_QUEUE_INCLUDED
00020 #define MARSYAS_REALVEC_QUEUE_INCLUDED
00021 
00022 #include <marsyas/realvec.h>
00023 #include <marsyas/types.h>
00024 
00025 #include <cassert>
00026 #include <atomic>
00027 #include <algorithm>
00028 
00029 namespace Marsyas {
00030 
00031 using namespace std;
00032 
00033 class realvec_queue_producer;
00034 class realvec_queue_consumer;
00035 
00062 template <typename T>
00063 T clipped(const T& lower, const T& n, const T& upper) {
00064   return std::max(lower, std::min(n, upper));
00065 }
00066 
00067 class realvec_queue
00068 {
00069   typedef atomic<mrs_natural> atomic_size_t;
00070 
00071   realvec m_buffer;
00072   atomic_size_t m_read_position;
00073   atomic_size_t m_write_position;
00074   atomic_size_t m_capacity;
00075 
00076 public:
00077   friend class realvec_queue_producer;
00078   friend class realvec_queue_consumer;
00079 
00083   realvec_queue():
00084     m_read_position(0),
00085     m_write_position(0),
00086     m_capacity(0)
00087   {}
00088 
00094   realvec_queue(mrs_natural observations, mrs_natural samples):
00095     m_buffer(observations, samples),
00096     m_read_position(0),
00097     m_write_position(0),
00098     m_capacity( samples )
00099   {}
00100 
00101   realvec_queue(mrs_natural observations, mrs_natural samples, mrs_natural capacity):
00102     m_buffer(observations, samples),
00103     m_read_position(0),
00104     m_write_position(0),
00105     m_capacity( clipped((mrs_natural) 0, capacity, samples) )
00106   {}
00107 
00115   void resize(mrs_natural observations, mrs_natural samples, bool clear = true)
00116   {
00117     resize(observations, samples, samples, clear);
00118   }
00119 
00120   void resize(mrs_natural observations, mrs_natural samples, mrs_natural capacity, bool clear = true)
00121   {
00122     if (clear)
00123       m_buffer.create(observations, samples);
00124     else
00125       m_buffer.stretch(observations, samples);
00126 
00127     m_read_position = m_write_position = 0;
00128     m_capacity = clipped((mrs_natural)0, capacity, samples);
00129   }
00130 
00134   void clear()
00135   {
00136     m_read_position = m_write_position = 0;
00137   }
00138 
00139   mrs_natural capacity()
00140   {
00141     return m_capacity.load( memory_order_relaxed );
00142   }
00143 
00144   mrs_natural set_capacity( mrs_natural capacity )
00145   {
00146     capacity = clipped((mrs_natural)0, capacity, samples());
00147     m_capacity.store( capacity, memory_order_relaxed );
00148     return capacity;
00149   }
00150 
00154   mrs_natural observations() { return m_buffer.getRows(); }
00155 
00159   mrs_natural samples() { return m_buffer.getCols(); }
00160 
00164   mrs_natural write_capacity()
00165   {
00166     mrs_natural read_pos = m_read_position.load(memory_order_relaxed);
00167     mrs_natural write_pos = m_write_position.load(memory_order_relaxed);
00168     mrs_natural capacity = m_capacity.load(memory_order_relaxed);
00169     mrs_natural size = samples();
00170 
00171     if (capacity > 0 )
00172       --capacity; // Should never write the last sample.
00173 
00174     mrs_natural written;
00175     if (write_pos >= read_pos)
00176       written = write_pos - read_pos;
00177     else
00178       written = size - (read_pos - write_pos);
00179 
00180     if (written >= capacity)
00181       return 0;
00182     else
00183       return capacity - written;
00184   }
00185 
00189   mrs_natural read_capacity()
00190   {
00191     mrs_natural read_pos = m_read_position.load(memory_order_relaxed);
00192     mrs_natural write_pos = m_write_position.load(memory_order_acquire);
00193     mrs_natural available;
00194     if (write_pos >= read_pos)
00195       available = write_pos - read_pos;
00196     else
00197       available = samples() - (read_pos - write_pos);
00198     return available;
00199   }
00200 };
00201 
00209 class realvec_queue_producer
00210 {
00211   realvec_queue & m_queue;
00212   mrs_natural m_capacity;
00213   mrs_natural m_position;
00214 
00215 public:
00224   realvec_queue_producer(realvec_queue & destination, mrs_natural capacity):
00225     m_queue( destination ),
00226     m_position( destination.m_write_position.load(memory_order_relaxed) )
00227   {
00228     assert(capacity >= 0);
00229     if (destination.write_capacity() < capacity)
00230       m_capacity = 0;
00231     else
00232       m_capacity = capacity;
00233   }
00234 
00239   ~realvec_queue_producer()
00240   {
00241     if (m_capacity > 0) {
00242       mrs_natural position = (m_position + m_capacity) % m_queue.samples();
00243       m_queue.m_write_position.store(position, memory_order_release);
00244     }
00245   }
00246 
00250   mrs_natural capacity() { return m_capacity; }
00251 
00264   bool reserve( mrs_natural capacity )
00265   {
00266     assert(capacity >= 0);
00267     if (capacity <= m_capacity)
00268       return true;
00269     if (m_queue.write_capacity() < capacity)
00270       return false;
00271     m_capacity = capacity;
00272     return true;
00273   }
00274 
00282   mrs_real & operator() ( mrs_natural observation, mrs_natural sample )
00283   {
00284     assert(sample >= 0);
00285     assert(sample < m_capacity);
00286     sample = (m_position + sample) % m_queue.samples();
00287     return m_queue.m_buffer(observation, sample);
00288   }
00289 };
00290 
00298 class realvec_queue_consumer
00299 {
00300   realvec_queue & m_queue;
00301   mrs_natural m_capacity;
00302   mrs_natural m_position;
00303 
00304 public:
00313   realvec_queue_consumer(realvec_queue & source, mrs_natural capacity):
00314     m_queue(source),
00315     m_position( source.m_read_position.load(memory_order_relaxed) )
00316   {
00317     assert(capacity >= 0);
00318     if (source.read_capacity() < capacity)
00319       m_capacity = 0;
00320     else
00321       m_capacity = capacity;
00322   }
00323 
00328   ~realvec_queue_consumer()
00329   {
00330     if (m_capacity > 0) {
00331       mrs_natural position = (m_position + m_capacity) % m_queue.samples();
00332       m_queue.m_read_position.store(position, memory_order_release);
00333     }
00334   }
00335 
00339   mrs_natural capacity() { return m_capacity; }
00340 
00353   bool reserve( mrs_natural capacity )
00354   {
00355     assert(capacity >= 0);
00356     if (capacity <= m_capacity)
00357       return true;
00358     if (m_queue.read_capacity() < capacity)
00359       return false;
00360     m_capacity = capacity;
00361     return true;
00362   }
00363 
00371   mrs_real & operator() ( mrs_natural observation, mrs_natural sample )
00372   {
00373     assert(sample >= 0);
00374     assert(sample < m_capacity);
00375     sample = (m_position + sample) % m_queue.samples();
00376     return m_queue.m_buffer(observation, sample);
00377   }
00378 };
00379 
00380 } // namespace Marsyas
00381 
00382 #endif // MARSYAS_REALVEC_QUEUE_INCLUDED