src/ptlib/unix/macosaudio/ringbuffer.cxx

   1 #include <assert.h>
   2 #include <stdlib.h>
   3 #include <new> // so I can throw std::bad_alloc
   4 #include <string.h>
   5 #include "ringbuffer.h"
   6
   7 class AutoCriticalRegion
   8 {
   9     MPCriticalRegionID m_mutex;
  10 public:
  11     AutoCriticalRegion(MPCriticalRegionID mutex)
  12             : m_mutex(mutex)
  13         {
  14             MPEnterCriticalRegion(m_mutex, kDurationForever);
  15         }
  16     ~AutoCriticalRegion()
  17         {
  18             MPExitCriticalRegion(m_mutex);
  19         }
  20 };
  21
  22 JRingBuffer::JRingBuffer(int bs)
  23         : m_in(0), m_out(-1), m_buffersize(bs), m_shutdown(0), m_reserved(0),
  24           m_mutex(0), m_readtoken(0)
  25 {
  26     if ((m_buffer = new unsigned char[m_buffersize]) == 0) {
  27         throw std::bad_alloc();
  28         return;
  29     }
  30     m_bufferend = m_buffer + m_buffersize;
  31     if (MPCreateCriticalRegion(&m_mutex) == noErr) {
  32         if (MPCreateBinarySemaphore(&m_readtoken) == noErr) {
  33             return;
  34         }
  35         MPDeleteCriticalRegion(m_mutex);
  36         m_mutex = 0;
  37     }
  38     // fall through here on failures
  39     delete[] m_buffer;
  40     m_buffer = 0;
  41     throw std::bad_alloc();
  42 }
  43
  44 // make sure no one is using it when it is destructed!
  45 JRingBuffer::~JRingBuffer() {
  46     if (m_buffer) {
  47         delete[] m_buffer;
  48         m_buffer = 0;
  49     }
  50     if (m_mutex != kInvalidID)
  51         MPDeleteCriticalRegion(m_mutex);
  52     if (m_readtoken != kInvalidID)
  53         MPDeleteSemaphore(m_readtoken);
  54 }
  55
  56 // returns 0 on success, -1 on failure
  57 int JRingBuffer::Write(void *buf, int len)
  58 {
  59     int wasempty = 0, overfill = 0;
  60     if (len == 0) return 0;
  61     // It is a programming error to stuff more than buffersize in at once.
  62     assert(len <= m_buffersize);
  63
  64     AutoCriticalRegion mutex(m_mutex);
  65
  66     if (len >= m_buffersize) {
  67         len = m_buffersize;
  68         // a quickie optimization: if we are filling the buffer, just start
  69         // from the beginning
  70         m_in = 0;
  71         m_out = -1;
  72     }
  73     wasempty = IsEmpty();
  74     overfill = (CanHold() < len);
  75
  76     // OK, are we going to wrap?
  77     if ((m_buffersize - m_in) <= len) {
  78         // yes, copy part and wrap
  79         int part = m_buffersize - m_in;
  80         memmove(m_buffer + m_in, buf, part);
  81         m_in = 0;
  82         buf = (void *)((char *)buf + part);
  83         len -= part;
  84     }
  85     // not wrapping (anymore)
  86     if (len)
  87         memmove(m_buffer + m_in, buf, len);
  88     m_in += len;
  89     if (overfill)
  90         m_out = m_in;
  91     // if empty, signal the CV, we just dumped in some data
  92     if (wasempty) {
  93         m_out = 0;
  94         MPSignalSemaphore(m_readtoken);
  95     }
  96     return 0;
  97 }
  98
  99 // Get a pointer to a reserved buffer
 100 void *JRingBuffer::ReserveToWrite(int &len)
 101 {
 102     if (len <= 0) return NULL;
 103     // It is a programming error to stuff more than buffersize in at once.
 104     assert(len <= m_buffersize);
 105     if (len > m_buffersize) len = m_buffersize;
 106
 107     MPEnterCriticalRegion(m_mutex, kDurationForever);
 108
 109     return ReserveToWrite_locked(len);
 110 }
 111
 112 // Call with the mutex LOCKED.
 113 void *JRingBuffer::ReserveToWrite_locked(int &len)
 114 {
 115     // OK, are we going to wrap?
 116     if ((m_buffersize - m_in) <= len) {
 117         len = m_buffersize - m_in;
 118     }
 119     m_reserved = len;
 120     return &m_buffer[m_in];
 121     // MUTEX IS STILL LOCKED.
 122 }
 123
 124 int JRingBuffer::CommitFinal(int wrotelen)
 125 {
 126     int wasempty = IsEmpty(), overfill = CanHold() < wrotelen;
 127     assert(m_reserved);
 128     assert(wrotelen >= 0);
 129     assert(m_reserved >= wrotelen);
 130     if (wrotelen > 0) {
 131         m_in += wrotelen;
 132         if (m_in == m_buffersize) m_in = 0;
 133         if (overfill)
 134             m_out = m_in;
 135         if (wasempty) {
 136             m_out = 0;
 137             MPSignalSemaphore(m_readtoken);
 138         }
 139     }
 140     MPExitCriticalRegion(m_mutex);
 141     return 0;
 142 }
 143
 144 void *JRingBuffer::CommitMore(int wrotelen, int &morelen)
 145 {
 146     int wasempty = IsEmpty(), overfill = CanHold() < wrotelen;
 147     assert(m_reserved);
 148     assert(wrotelen >= 0);
 149     assert(m_reserved >= wrotelen);
 150     if (wrotelen != 0) {
 151         m_in += wrotelen;
 152         if (m_in == m_buffersize) m_in = 0;
 153         if (overfill)
 154             m_out = m_in;
 155         if (wasempty) {
 156             m_out = 0;
 157             MPSignalSemaphore(m_readtoken);
 158             // Interesting question:  should we drop the mutex and relock?
 159             // I suspect not, since the writer is probably a time-critical
 160             // callback.
 161         }
 162     }
 163     return ReserveToWrite_locked(morelen);
 164 }
 165
 166 int JRingBuffer::ReadTimed(void *buf, int len, long timeout, int flag)
 167 {
 168     OSStatus oops;
 169     AbsoluteTime expiry;
 170     if (flag == kAsync) timeout = 0;
 171
 172     expiry = AddDurationToAbsolute((Duration)timeout, UpTime());
 173
 174     int total_read = 0;
 175 #ifndef NDEBUG // this is used only in an assert
 176     int origlen = len;
 177 #endif
 178
 179     while (len > 0) {
 180         // Obtain the read token
 181         // First, recalculate the timeout (we may have waited before)
 182         if (timeout != aLongTime) {
 183             timeout = AbsoluteToDuration(
 184                           SubAbsoluteFromAbsolute(expiry,
 185                                                   UpTime()));
 186
 187         }
 188         if (IsShutdown()) {
 189             total_read = -1;
 190             break;
 191         }
 192         oops = MPWaitOnSemaphore(m_readtoken, (Duration)timeout);
 193
 194         if (oops != noErr) {
 195             if (flag == kAsync)
 196                 total_read = 0;
 197             else
 198                 total_read = -1;
 199             break;
 200         }
 201
 202         // lock the queue
 203         AutoCriticalRegion mutex(m_mutex);
 204
 205         // While there is data and we need some...
 206         while (!IsShutdown() && !IsEmpty() && len > 0) {
 207             // copy out a contiguous chunk.
 208             int avail = Contains();
 209             // only read what's needed
 210             if (avail > len) avail = len;
 211             // now, are we wrapping?
 212             if (avail >= m_buffersize - m_out)
 213                 avail = m_buffersize - m_out;
 214             memmove(buf, m_buffer + m_out, avail);
 215             buf = (void *)((char *)buf + avail);
 216             total_read += avail;
 217             len -= avail;
 218             m_out += avail;
 219             // Wrap?
 220             if (m_out == m_buffersize)
 221                 m_out = 0;
 222             if (m_out == m_in) {
 223                 m_out = -1; // we have emptied the buffer
 224                 m_in  = 0;
 225             }
 226         }
 227         if (!IsEmpty() || IsShutdown()) {
 228             // regenerate the read token
 229             MPSignalSemaphore(m_readtoken);
 230         }
 231         if (IsShutdown()) {
 232             total_read = -1;
 233             break;
 234         }
 235         // we have read as much as we can or need.  Are we done?
 236         if (flag == kAsync || len == 0 || (flag == kSome && total_read != 0))
 237         {
 238             // yes.
 239             break;
 240         }
 241         // need more data.
 242         assert(IsEmpty());
 243         // loop again
 244     }
 245     assert(total_read < 0 || flag != kWait || total_read == origlen);
 246
 247     return total_read;
 248 }
 249
 250 bool JRingBuffer::WaitForData(long timeout)
 251 {
 252     OSStatus err;
 253     err = MPWaitOnSemaphore(m_readtoken, timeout);
 254     if (err == 0) MPSignalSemaphore(m_readtoken);
 255     return err == 0;
 256 }
 257
 258 void JRingBuffer::MakeEmpty()
 259 {
 260     MPEnterCriticalRegion(m_mutex, kDurationForever);
 261     m_in = 0;
 262     m_out = -1;
 263     MPExitCriticalRegion(m_mutex);
 264 }
 265
 266 void JRingBuffer::Shutdown()
 267 {
 268     MPEnterCriticalRegion(m_mutex, kDurationForever);
 269     m_in = 0;
 270     m_out = -1;
 271     m_shutdown = 1;
 272     MPExitCriticalRegion(m_mutex);
 273     (void)MPSignalSemaphore(m_readtoken);
 274 }
 275