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

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