src/ptlib/unix/maccoreaudio/circular_buffer.inl

   1 /*
   2  * circularbuffer.inl
   3  *
   4  * Copyright (c) 2004 Network for Educational Technology ETH
   5  *
   6  * Written by Hannes Friederich, Andreas Fenkart.
   7  * Based on work of Shawn Pai-Hsiang Hsiao
   8  *
   9  * The contents of this file are subject to the Mozilla Public License
  10  * Version 1.0 (the "License"); you may not use this file except in
  11  * compliance with the License. You may obtain a copy of the License at
  12  * http://www.mozilla.org/MPL/
  13  *
  14  * Software distributed under the License is distributed on an "AS IS"
  15  * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
  16  * the License for the specific language governing rights and limitations
  17  * under the License.
  18  * maccoreaudio.h
  19  *
  20  */
  21
  22
  23 #ifndef CIRCULAR_BUFFER_H
  24 #define CIRCULAR_BUFFER_H
  25
  26
  27 //#define PTRACE_CIRC(level, str) PTRACE(level, str)
  28 #define PTRACE_CIRC(level, str)
  29
  30 /**
  31  * Simple circular buffer for one producer and consumer with a head and a
  32  * tail that chase each other. The capacity is 1 byte bigger than necessary
  33  * due to a sentinel, to tell apart full from empty by the following equations:
  34  *
  35  * full  := head_next == tail
  36  * empty := head      == tail
  37  *
  38  * Inspired by CircularBuffer from beaudio.
  39  * We need a lock when updating the tail index, due to the overwrite mechanism
  40  * in case of buffer overflow. The head index needs no locking because
  41  * overwrite does not make sense in case of buffer underrun.
  42  *
  43  * Keep in mind that this buffer does no know about frames or packets, it's up
  44  * to you to make sure you get the right number of bytes. In doubt use size()
  45  * to compute how many frames/packets it is save to drain/fill.
  46  *
  47  */
  48
  49 class CircularBuffer
  50 {
  51  public:
  52
  53    explicit CircularBuffer(PINDEX len)
  54    :   capacity(len + 1),  /* plus sentinel */ head(0), tail(0)
  55    {
  56       buffer = (char *)malloc(capacity*sizeof(char));
  57       if(!buffer) {
  58          PTRACE(1, "Could not allocate circular buffer");
  59          init = false;
  60       } else {
  61          init = true;
  62       }
  63
  64       /*
  65        * Mutex to block write thread when buffer is full
  66        */
  67       int rval;
  68       rval = pthread_mutex_init(&mutex, NULL);
  69       if (rval) {
  70         PTRACE(1, __func__ << " can not init mutex");
  71         init = false;
  72         //return FALSE;
  73       }
  74       rval = pthread_cond_init(&cond, NULL);
  75       if (rval) {
  76         PTRACE(1, __func__ << " can not init cond");
  77         init = false;
  78         //return FALSE;
  79       }
  80    }
  81
  82    ~CircularBuffer()
  83    {
  84       if(buffer){
  85          std::free(buffer);
  86          buffer = NULL;
  87       }
  88
  89       pthread_mutex_destroy(&mutex);
  90       pthread_cond_destroy(&cond);
  91    }
  92
  93
  94    BOOL Full()
  95    {
  96       /* head + 1 == tail */
  97       return head_next() == tail;
  98    }
  99
 100    BOOL Empty()
 101    {
 102       return head == tail;
 103    }
 104
 105    PINDEX size(){
 106       /* sentinel is outside of occupied area */
 107       return (head < tail ) ? head + capacity - tail : head - tail;
 108    }
 109
 110    PINDEX free(){
 111       return (capacity - size() -1 /*sentinel */);
 112    }
 113
 114    /**
 115     * Fill inserts data into the circular buffer.
 116     * Remind that lock and overwrite are mutually exclusive. If you set lock,
 117     * overwrite will be ignored
 118     */
 119    PINDEX Fill(const char* inbuf, PINDEX len, Boolean lock = true,
 120                      Boolean overwrite = false);
 121
 122
 123    /** See also Fill */
 124    PINDEX Drain(char* outbuf, PINDEX len, Boolean lock = true);
 125
 126  private:
 127    PINDEX head_next()
 128    {
 129       //return (head + 1 == capacity) ? 0 : (head + 1);
 130       return (head + 1 % capacity);
 131    }
 132
 133    void increment_index(PINDEX &index, PINDEX inc)
 134    {
 135       index += inc;
 136       index %= capacity;
 137    }
 138
 139    void increment_head(PINDEX inc)
 140    {
 141       increment_index(head, inc);
 142    }
 143
 144    void increment_tail(PINDEX inc)
 145    {
 146       increment_index(tail, inc);
 147    }
 148
 149
 150
 151  private:
 152    char* buffer;
 153    const PINDEX capacity;
 154    PINDEX head, tail;
 155    Boolean init;
 156
 157    pthread_mutex_t mutex;
 158    pthread_cond_t cond;
 159 };
 160
 161
 162 int CircularBuffer::Fill(const char *inbuf, PINDEX len,
 163          Boolean lock, Boolean overwrite)
 164 {
 165    int done = 0, todo = 0;
 166
 167    PTRACE_CIRC(1, "Fill " << len << " bytes, contains " << size());
 168
 169    if(inbuf== NULL){
 170       PTRACE(1, __func__ << "Input buffer is empty");
 171       return 0;
 172    }
 173
 174    while(done != len && !Full()) {
 175       if(head >= tail) {
 176          // head unwrapped, fill from head till end of buffer
 177          if(tail == 0) /* buffer[capacity] == sentinel */
 178             todo = MIN(capacity -1 /*sentinel*/ - head, len - done);
 179          else
 180             todo = MIN(capacity - head, len - done);
 181       } else {
 182          // fill from head till tail
 183          todo = MIN(tail -1 /*sentinel*/ - head, len - done);
 184       }
 185       memcpy(buffer + head, inbuf + done, todo);
 186       done += todo;
 187       increment_head(todo);
 188       PTRACE_CIRC(1, "copied " << done << " from input buffer"
 189          << " available " << size());
 190    }
 191
 192    // What to do if buffer is full and more bytes
 193    // need to be copied ?
 194    if(Full() && done != len && (overwrite || lock)) {
 195       PTRACE_CIRC(1, __func__ << "Circular buffer is full, while Fill "
 196             << len);
 197       if(lock) {
 198          pthread_mutex_lock(&mutex);
 199          PTRACE_CIRC(1, "Fill going to sleep");
 200          pthread_cond_wait(&cond, &mutex);
 201          // pthread_cond_timedwait
 202          PTRACE_CIRC(1, "Fill woke up");
 203          pthread_mutex_unlock(&mutex);
 204       } else if(overwrite){
 205          pthread_mutex_lock(&mutex);
 206          if(Full()){
 207             tail += len - done; // also shifts sentinel
 208             tail %= capacity; // wrap around
 209          }
 210          pthread_mutex_unlock(&mutex);
 211       }
 212       // try again
 213       done += Fill(inbuf + done, len - done, lock, overwrite);
 214    }
 215
 216    // wake up read thread if necessary
 217    if(!Empty())
 218       pthread_cond_signal(&cond);
 219
 220
 221    PTRACE_CIRC(1, __func__ << " " << len << " bytes, stored " << done
 222          << " available " << size());
 223    return done;
 224 }
 225
 226
 227 PINDEX CircularBuffer::Drain(char *outbuf, PINDEX len, Boolean lock) {
 228    PINDEX done = 0, todo = 0;
 229
 230    PTRACE_CIRC(6, __func__ << " " << len << " bytes, available " << size() );
 231
 232    if(outbuf == NULL){
 233       PTRACE(1, __func__ << " Out buffer is NULL");
 234       return PINDEX(0);
 235    }
 236
 237    /* protect agains buffer corruption when write thread
 238     * is overwriting */
 239    pthread_mutex_lock(&mutex);
 240    PTRACE(6, "aquired lock");
 241
 242    while(done != len && !Empty()) {
 243       if(head >= tail) {
 244          // head unwrapped
 245          todo = MIN(head - tail, len - done);
 246       } else {
 247          // head wrapped
 248          todo = MIN(capacity - tail, len - done);
 249       }
 250       memcpy(outbuf + done, buffer + tail, todo);
 251       done += todo;
 252       increment_tail(todo);
 253       PTRACE_CIRC(1, __func__ << " for " << len " bytes, copied " << done
 254             << " to output buffer,  available in buffer " << size());
 255    }
 256
 257
 258    if(done != len && (lock)) /* && Empty() */ {
 259       PTRACE(3, "Buffer underrun for Drain " << len << " bytes");
 260    }
 261
 262    // what to do if not as many bytes are available then
 263    // requested ?
 264    if(done != len && (lock)) /* && Empty() */ {
 265       if (lock){
 266          pthread_cond_wait(&cond, &mutex);
 267          PTRACE_CIRC(2, "Read thread woke up, calling Drain");
 268          pthread_mutex_unlock(&mutex); // race with write thread...
 269          done += Drain(outbuf + done, len - done, lock);
 270          PTRACE_CIRC(2, "Returned from recursive");
 271       }
 272    }
 273
 274    pthread_mutex_unlock(&mutex);
 275
 276    if(!Full())
 277       pthread_cond_signal(&cond);
 278
 279    PTRACE_CIRC(2, "End Drain " << len << " bytes, fetched " << done
 280          << " buffer fill " << size() );
 281    return done;
 282
 283 }
 284 #endif