git.gag.com Git - debian/gnuradio/blob - gr-audio-osx/src/circular_buffer.h

   1 /* -*- c++ -*- */
   2 /*
   3  * Copyright 2006.2009 Free Software Foundation, Inc.
   4  *
   5  * This file is part of GNU Radio.
   6  *
   7  * GNU Radio is free software; you can redistribute it and/or modify
   8  * it under the terms of the GNU General Public License as published by
   9  * the Free Software Foundation; either version 3, or (at your option)
  10  * any later version.
  11  *
  12  * GNU Radio is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  * GNU General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU General Public License
  18  * along with GNU Radio; see the file COPYING.  If not, write to
  19  * the Free Software Foundation, Inc., 51 Franklin Street,
  20  * Boston, MA 02110-1301, USA.
  21  */
  22
  23 #ifndef _CIRCULAR_BUFFER_H_
  24 #define _CIRCULAR_BUFFER_H_
  25
  26 #include "mld_threads.h"
  27 #include <stdexcept>
  28
  29 #ifndef DO_DEBUG
  30 #define DO_DEBUG 0
  31 #endif
  32
  33 #if DO_DEBUG
  34 #define DEBUG(X) do{X} while(0);
  35 #else
  36 #define DEBUG(X) do{} while(0);
  37 #endif
  38
  39 template <class T> class circular_buffer
  40 {
  41 private:
  42 // the buffer to use
  43   T* d_buffer;
  44
  45 // the following are in Items (type T)
  46   size_t d_bufLen_I, d_readNdx_I, d_writeNdx_I;
  47   size_t d_n_avail_write_I, d_n_avail_read_I;
  48
  49 // stuff to control access to class internals
  50   mld_mutex_ptr d_internal;
  51   mld_condition_ptr d_readBlock, d_writeBlock;
  52
  53 // booleans to decide how to control reading, writing, and aborting
  54   bool d_doWriteBlock, d_doFullRead, d_doAbort;
  55
  56   void delete_mutex_cond () {
  57     if (d_internal) {
  58       delete d_internal;
  59       d_internal = NULL;
  60     }
  61     if (d_readBlock) {
  62       delete d_readBlock;
  63       d_readBlock = NULL;
  64     }
  65     if (d_writeBlock) {
  66       delete d_writeBlock;
  67       d_writeBlock = NULL;
  68     }
  69   };
  70
  71 public:
  72   circular_buffer (size_t bufLen_I,
  73                    bool doWriteBlock = true, bool doFullRead = false) {
  74     if (bufLen_I == 0)
  75       throw std::runtime_error ("circular_buffer(): "
  76                                 "Number of items to buffer must be > 0.\n");
  77     d_bufLen_I = bufLen_I;
  78     d_buffer = (T*) new T[d_bufLen_I];
  79     d_doWriteBlock = doWriteBlock;
  80     d_doFullRead = doFullRead;
  81     d_internal = NULL;
  82     d_readBlock = d_writeBlock = NULL;
  83     reset ();
  84     DEBUG (fprintf (stderr, "c_b(): buf len (items) = %ld, "
  85                     "doWriteBlock = %s, doFullRead = %s\n", d_bufLen_I,
  86                     (d_doWriteBlock ? "true" : "false"),
  87                     (d_doFullRead ? "true" : "false")););
  88   };
  89
  90   ~circular_buffer () {
  91     delete_mutex_cond ();
  92     delete [] d_buffer;
  93   };
  94
  95   inline size_t n_avail_write_items () {
  96     d_internal->lock ();
  97     size_t retVal = d_n_avail_write_I;
  98     d_internal->unlock ();
  99     return (retVal);
 100   };
 101
 102   inline size_t n_avail_read_items () {
 103     d_internal->lock ();
 104     size_t retVal = d_n_avail_read_I;
 105     d_internal->unlock ();
 106     return (retVal);
 107   };
 108
 109   inline size_t buffer_length_items () {return (d_bufLen_I);};
 110   inline bool do_write_block () {return (d_doWriteBlock);};
 111   inline bool do_full_read () {return (d_doFullRead);};
 112
 113   void reset () {
 114     d_doAbort = false;
 115     bzero (d_buffer, d_bufLen_I * sizeof (T));
 116     d_readNdx_I = d_writeNdx_I = d_n_avail_read_I = 0;
 117     d_n_avail_write_I = d_bufLen_I;
 118     delete_mutex_cond ();
 119     // create a mutex to handle contention of shared resources;
 120     // any routine needed access to shared resources uses lock()
 121     // before doing anything, then unlock() when finished.
 122     d_internal = new mld_mutex ();
 123     // link the internal mutex to the read and write conditions;
 124     // when wait() is called, the internal mutex will automatically
 125     // be unlock()'ed.  Upon return (from a signal() to the condition),
 126     // the internal mutex will be lock()'ed.
 127     d_readBlock = new mld_condition (d_internal);
 128     d_writeBlock = new mld_condition (d_internal);
 129   };
 130
 131 /*
 132  * enqueue: add the given buffer of item-length to the queue,
 133  *     first-in-first-out (FIFO).
 134  *
 135  * inputs:
 136  *     buf: a pointer to the buffer holding the data
 137  *
 138  *     bufLen_I: the buffer length in items (of the instantiated type)
 139  *
 140  * returns:
 141  *    -1: on overflow (write is not blocking, and data is being
 142  *                     written faster than it is being read)
 143  *     0: if nothing to do (0 length buffer)
 144  *     1: if success
 145  *     2: in the process of aborting, do doing nothing
 146  *
 147  * will throw runtime errors if inputs are improper:
 148  *     buffer pointer is NULL
 149  *     buffer length is larger than the instantiated buffer length
 150  */
 151
 152   int enqueue (T* buf, size_t bufLen_I) {
 153     DEBUG (fprintf (stderr, "enqueue: buf = %X, bufLen = %ld, #av_wr = %ld, "
 154                     "#av_rd = %ld.\n", (unsigned int)buf, bufLen_I,
 155                     d_n_avail_write_I, d_n_avail_read_I););
 156     if (bufLen_I > d_bufLen_I) {
 157       fprintf (stderr, "cannot add buffer longer (%ld"
 158                ") than instantiated length (%ld"
 159                ").\n", bufLen_I, d_bufLen_I);
 160       throw std::runtime_error ("circular_buffer::enqueue()");
 161     }
 162
 163     if (bufLen_I == 0)
 164       return (0);
 165     if (!buf)
 166       throw std::runtime_error ("circular_buffer::enqueue(): "
 167                                 "input buffer is NULL.\n");
 168     d_internal->lock ();
 169     if (d_doAbort) {
 170       d_internal->unlock ();
 171       return (2);
 172     }
 173     // set the return value to 1: success; change if needed
 174     int retval = 1;
 175     if (bufLen_I > d_n_avail_write_I) {
 176       if (d_doWriteBlock) {
 177         while (bufLen_I > d_n_avail_write_I) {
 178           DEBUG (fprintf (stderr, "enqueue: #len > #a, waiting.\n"););
 179           // wait will automatically unlock() the internal mutex
 180           d_writeBlock->wait ();
 181           // and lock() it here.
 182           if (d_doAbort) {
 183             d_internal->unlock ();
 184             DEBUG (fprintf (stderr, "enqueue: #len > #a, aborting.\n"););
 185             return (2);
 186           }
 187           DEBUG (fprintf (stderr, "enqueue: #len > #a, done waiting.\n"););
 188         }
 189       } else {
 190         d_n_avail_read_I = d_bufLen_I - bufLen_I;
 191         d_n_avail_write_I = bufLen_I;
 192         DEBUG (fprintf (stderr, "circular_buffer::enqueue: overflow\n"););
 193         retval = -1;
 194       }
 195     }
 196     size_t n_now_I = d_bufLen_I - d_writeNdx_I, n_start_I = 0;
 197     if (n_now_I > bufLen_I)
 198       n_now_I = bufLen_I;
 199     else if (n_now_I < bufLen_I)
 200       n_start_I = bufLen_I - n_now_I;
 201     bcopy (buf, &(d_buffer[d_writeNdx_I]), n_now_I * sizeof (T));
 202     if (n_start_I) {
 203       bcopy (&(buf[n_now_I]), d_buffer, n_start_I * sizeof (T));
 204       d_writeNdx_I = n_start_I;
 205     } else
 206       d_writeNdx_I += n_now_I;
 207     d_n_avail_read_I += bufLen_I;
 208     d_n_avail_write_I -= bufLen_I;
 209     d_readBlock->signal ();
 210     d_internal->unlock ();
 211     return (retval);
 212   };
 213
 214 /*
 215  * dequeue: removes from the queue the number of items requested, or
 216  *     available, into the given buffer on a FIFO basis.
 217  *
 218  * inputs:
 219  *     buf: a pointer to the buffer into which to copy the data
 220  *
 221  *     bufLen_I: pointer to the number of items to remove in items
 222  *         (of the instantiated type)
 223  *
 224  * returns:
 225  *     0: if nothing to do (0 length buffer)
 226  *     1: if success
 227  *     2: in the process of aborting, do doing nothing
 228  *
 229  * will throw runtime errors if inputs are improper:
 230  *     buffer pointer is NULL
 231  *     buffer length pointer is NULL
 232  *     buffer length is larger than the instantiated buffer length
 233  */
 234
 235   int dequeue (T* buf, size_t* bufLen_I) {
 236     DEBUG (fprintf (stderr, "dequeue: buf = %X, *bufLen = %ld, #av_wr = %ld, "
 237                     "#av_rd = %ld.\n", (unsigned int)buf, *bufLen_I,
 238                     d_n_avail_write_I, d_n_avail_read_I););
 239     if (!bufLen_I)
 240       throw std::runtime_error ("circular_buffer::dequeue(): "
 241                                 "input bufLen pointer is NULL.\n");
 242     if (!buf)
 243       throw std::runtime_error ("circular_buffer::dequeue(): "
 244                                 "input buffer pointer is NULL.\n");
 245     size_t l_bufLen_I = *bufLen_I;
 246     if (l_bufLen_I == 0)
 247       return (0);
 248     if (l_bufLen_I > d_bufLen_I) {
 249       fprintf (stderr, "cannot remove buffer longer (%ld"
 250                ") than instantiated length (%ld"
 251                ").\n", l_bufLen_I, d_bufLen_I);
 252       throw std::runtime_error ("circular_buffer::dequeue()");
 253     }
 254
 255     d_internal->lock ();
 256     if (d_doAbort) {
 257       d_internal->unlock ();
 258       return (2);
 259     }
 260     if (d_doFullRead) {
 261       while (d_n_avail_read_I < l_bufLen_I) {
 262         DEBUG (fprintf (stderr, "dequeue: #a < #len, waiting.\n"););
 263         // wait will automatically unlock() the internal mutex
 264         d_readBlock->wait ();
 265         // and lock() it here.
 266         if (d_doAbort) {
 267           d_internal->unlock ();
 268           DEBUG (fprintf (stderr, "dequeue: #a < #len, aborting.\n"););
 269           return (2);
 270         }
 271         DEBUG (fprintf (stderr, "dequeue: #a < #len, done waiting.\n"););
 272      }
 273     } else {
 274       while (d_n_avail_read_I == 0) {
 275         DEBUG (fprintf (stderr, "dequeue: #a == 0, waiting.\n"););
 276         // wait will automatically unlock() the internal mutex
 277         d_readBlock->wait ();
 278         // and lock() it here.
 279         if (d_doAbort) {
 280           d_internal->unlock ();
 281           DEBUG (fprintf (stderr, "dequeue: #a == 0, aborting.\n"););
 282           return (2);
 283         }
 284         DEBUG (fprintf (stderr, "dequeue: #a == 0, done waiting.\n"););
 285       }
 286     }
 287     if (l_bufLen_I > d_n_avail_read_I)
 288       l_bufLen_I = d_n_avail_read_I;
 289     size_t n_now_I = d_bufLen_I - d_readNdx_I, n_start_I = 0;
 290     if (n_now_I > l_bufLen_I)
 291       n_now_I = l_bufLen_I;
 292     else if (n_now_I < l_bufLen_I)
 293       n_start_I = l_bufLen_I - n_now_I;
 294     bcopy (&(d_buffer[d_readNdx_I]), buf, n_now_I * sizeof (T));
 295     if (n_start_I) {
 296       bcopy (d_buffer, &(buf[n_now_I]), n_start_I * sizeof (T));
 297       d_readNdx_I = n_start_I;
 298     } else
 299       d_readNdx_I += n_now_I;
 300     *bufLen_I = l_bufLen_I;
 301     d_n_avail_read_I -= l_bufLen_I;
 302     d_n_avail_write_I += l_bufLen_I;
 303     d_writeBlock->signal ();
 304     d_internal->unlock ();
 305     return (1);
 306   };
 307
 308   void abort () {
 309     d_internal->lock ();
 310     d_doAbort = true;
 311     d_writeBlock->signal ();
 312     d_readBlock->signal ();
 313     d_internal->unlock ();
 314   };
 315 };
 316
 317 #endif /* _CIRCULAR_BUFFER_H_ */