/* -*- c++ -*- */
/*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2006,2009,2010 Free Software Foundation, Inc.
*
* This file is part of GNU Radio.
*
#ifndef _CIRCULAR_BUFFER_H_
#define _CIRCULAR_BUFFER_H_
-#include "mld_threads.h"
+#include <gruel/thread.h>
+#include <iostream>
#include <stdexcept>
+#ifndef DO_DEBUG
#define DO_DEBUG 0
+#endif
#if DO_DEBUG
#define DEBUG(X) do{X} while(0);
#define DEBUG(X) do{} while(0);
#endif
-template <class T> class circular_buffer
+template <class T>
+class circular_buffer
{
private:
// the buffer to use
T* d_buffer;
// the following are in Items (type T)
- UInt32 d_bufLen_I, d_readNdx_I, d_writeNdx_I;
- UInt32 d_n_avail_write_I, d_n_avail_read_I;
+ size_t d_bufLen_I, d_readNdx_I, d_writeNdx_I;
+ size_t d_n_avail_write_I, d_n_avail_read_I;
// stuff to control access to class internals
- mld_mutex_ptr d_internal;
- mld_condition_ptr d_readBlock, d_writeBlock;
+ gruel::mutex* d_internal;
+ gruel::condition_variable* d_readBlock;
+ gruel::condition_variable* d_writeBlock;
// booleans to decide how to control reading, writing, and aborting
bool d_doWriteBlock, d_doFullRead, d_doAbort;
};
public:
- circular_buffer (UInt32 bufLen_I,
+ circular_buffer (size_t bufLen_I,
bool doWriteBlock = true, bool doFullRead = false) {
if (bufLen_I == 0)
throw std::runtime_error ("circular_buffer(): "
d_internal = NULL;
d_readBlock = d_writeBlock = NULL;
reset ();
- DEBUG (fprintf (stderr, "c_b(): buf len (items) = %ld, "
- "doWriteBlock = %s, doFullRead = %s\n", d_bufLen_I,
- (d_doWriteBlock ? "true" : "false"),
- (d_doFullRead ? "true" : "false")));
+ DEBUG (std::cerr << "c_b(): buf len (items) = " << d_bufLen_
+ << ", doWriteBlock = " << (d_doWriteBlock ? "true" : "false")
+ << ", doFullRead = " << (d_doFullRead ? "true" : "false")
+ << std::endl);
};
~circular_buffer () {
delete [] d_buffer;
};
- inline UInt32 n_avail_write_items () {
- d_internal->lock ();
- UInt32 retVal = d_n_avail_write_I;
- d_internal->unlock ();
+ inline size_t n_avail_write_items () {
+ gruel::scoped_lock l (*d_internal);
+ size_t retVal = d_n_avail_write_I;
return (retVal);
};
- inline UInt32 n_avail_read_items () {
- d_internal->lock ();
- UInt32 retVal = d_n_avail_read_I;
- d_internal->unlock ();
+ inline size_t n_avail_read_items () {
+ gruel::scoped_lock l (*d_internal);
+ size_t retVal = d_n_avail_read_I;
return (retVal);
};
- inline UInt32 buffer_length_items () {return (d_bufLen_I);};
+ inline size_t buffer_length_items () {return (d_bufLen_I);};
inline bool do_write_block () {return (d_doWriteBlock);};
inline bool do_full_read () {return (d_doFullRead);};
// create a mutex to handle contention of shared resources;
// any routine needed access to shared resources uses lock()
// before doing anything, then unlock() when finished.
- d_internal = new mld_mutex ();
+ d_internal = new gruel::mutex ();
// link the internal mutex to the read and write conditions;
// when wait() is called, the internal mutex will automatically
- // be unlock()'ed. Upon return (from a signal() to the condition),
+ // be unlock()'ed. Upon return (from a notify_one() to the condition),
// the internal mutex will be lock()'ed.
- d_readBlock = new mld_condition (d_internal);
- d_writeBlock = new mld_condition (d_internal);
+ d_readBlock = new gruel::condition_variable ();
+ d_writeBlock = new gruel::condition_variable ();
};
/*
* buffer length is larger than the instantiated buffer length
*/
- int enqueue (T* buf, UInt32 bufLen_I) {
- DEBUG (fprintf (stderr, "enqueue: buf = %X, bufLen = %ld, #av_wr = %ld, "
- "#av_rd = %ld.\n", (unsigned int)buf, bufLen_I,
- d_n_avail_write_I, d_n_avail_read_I));
+ int enqueue (T* buf, size_t bufLen_I) {
+ DEBUG (std::cerr << "enqueue: buf = " << (void*) buf
+ << ", bufLen = " << bufLen_I
+ << ", #av_wr = " << d_n_avail_write_I
+ << ", #av_rd = " << d_n_avail_read_I << std::endl);
if (bufLen_I > d_bufLen_I) {
- fprintf (stderr, "cannot add buffer longer (%ld"
- ") than instantiated length (%ld"
- ").\n", bufLen_I, d_bufLen_I);
+ std::cerr << "ERROR: cannot add buffer longer ("
+ << bufLen_I << ") than instantiated length ("
+ << d_bufLen_I << ")." << std::endl;
throw std::runtime_error ("circular_buffer::enqueue()");
}
if (!buf)
throw std::runtime_error ("circular_buffer::enqueue(): "
"input buffer is NULL.\n");
- d_internal->lock ();
+ gruel::scoped_lock l (*d_internal);
if (d_doAbort) {
- d_internal->unlock ();
return (2);
}
// set the return value to 1: success; change if needed
if (bufLen_I > d_n_avail_write_I) {
if (d_doWriteBlock) {
while (bufLen_I > d_n_avail_write_I) {
- DEBUG (fprintf (stderr, "enqueue: #len > #a, waiting.\n"));
- // wait will automatically unlock() the internal mutex
- d_writeBlock->wait ();
- // and lock() it here.
+ DEBUG (std::cerr << "enqueue: #len > #a, waiting." << std::endl);
+ // wait; will automatically unlock() the internal mutex via
+ // the scoped lock
+ d_writeBlock->wait (l);
+ // and auto re-lock() it here.
if (d_doAbort) {
- d_internal->unlock ();
- DEBUG (fprintf (stderr, "enqueue: #len > #a, aborting.\n"));
+ DEBUG (std::cerr << "enqueue: #len > #a, aborting." << std::endl);
return (2);
}
- DEBUG (fprintf (stderr, "enqueue: #len > #a, done waiting.\n"));
+ DEBUG (std::cerr << "enqueue: #len > #a, done waiting." << std::endl);
}
} else {
d_n_avail_read_I = d_bufLen_I - bufLen_I;
d_n_avail_write_I = bufLen_I;
- DEBUG (fprintf (stderr, "circular_buffer::enqueue: overflow\n"));
+ DEBUG (std::cerr << "circular_buffer::enqueue: overflow" << std::endl);
retval = -1;
}
}
- UInt32 n_now_I = d_bufLen_I - d_writeNdx_I, n_start_I = 0;
+ size_t n_now_I = d_bufLen_I - d_writeNdx_I, n_start_I = 0;
if (n_now_I > bufLen_I)
n_now_I = bufLen_I;
else if (n_now_I < bufLen_I)
d_writeNdx_I += n_now_I;
d_n_avail_read_I += bufLen_I;
d_n_avail_write_I -= bufLen_I;
- d_readBlock->signal ();
- d_internal->unlock ();
+ d_readBlock->notify_one ();
return (retval);
};
* buffer length is larger than the instantiated buffer length
*/
- int dequeue (T* buf, UInt32* bufLen_I) {
- DEBUG (fprintf (stderr, "dequeue: buf = %X, *bufLen = %ld, #av_wr = %ld, "
- "#av_rd = %ld.\n", (unsigned int)buf, *bufLen_I,
- d_n_avail_write_I, d_n_avail_read_I));
+ int dequeue (T* buf, size_t* bufLen_I) {
+ DEBUG (std::cerr << "dequeue: buf = " << ((void*) buf)
+ << ", *bufLen = " << (*bufLen_I)
+ << ", #av_wr = " << d_n_avail_write_I
+ << ", #av_rd = " << d_n_avail_read_I << std::endl);
if (!bufLen_I)
throw std::runtime_error ("circular_buffer::dequeue(): "
"input bufLen pointer is NULL.\n");
if (!buf)
throw std::runtime_error ("circular_buffer::dequeue(): "
"input buffer pointer is NULL.\n");
- UInt32 l_bufLen_I = *bufLen_I;
+ size_t l_bufLen_I = *bufLen_I;
if (l_bufLen_I == 0)
return (0);
if (l_bufLen_I > d_bufLen_I) {
- fprintf (stderr, "cannot remove buffer longer (%ld"
- ") than instantiated length (%ld"
- ").\n", l_bufLen_I, d_bufLen_I);
+ std::cerr << "ERROR: cannot remove buffer longer ("
+ << l_bufLen_I << ") than instantiated length ("
+ << d_bufLen_I << ")." << std::endl;
throw std::runtime_error ("circular_buffer::dequeue()");
}
- d_internal->lock ();
+ gruel::scoped_lock l (*d_internal);
if (d_doAbort) {
- d_internal->unlock ();
return (2);
}
if (d_doFullRead) {
while (d_n_avail_read_I < l_bufLen_I) {
- DEBUG (fprintf (stderr, "dequeue: #a < #len, waiting.\n"));
- // wait will automatically unlock() the internal mutex
- d_readBlock->wait ();
- // and lock() it here.
+ DEBUG (std::cerr << "dequeue: #a < #len, waiting." << std::endl);
+ // wait; will automatically unlock() the internal mutex via
+ // the scoped lock
+ d_readBlock->wait (l);
+ // and re-lock() it here.
if (d_doAbort) {
- d_internal->unlock ();
- DEBUG (fprintf (stderr, "dequeue: #a < #len, aborting.\n"));
+ DEBUG (std::cerr << "dequeue: #a < #len, aborting." << std::endl);
return (2);
}
- DEBUG (fprintf (stderr, "dequeue: #a < #len, done waiting.\n"));
+ DEBUG (std::cerr << "dequeue: #a < #len, done waiting." << std::endl);
}
} else {
while (d_n_avail_read_I == 0) {
- DEBUG (fprintf (stderr, "dequeue: #a == 0, waiting.\n"));
- // wait will automatically unlock() the internal mutex
- d_readBlock->wait ();
- // and lock() it here.
+ DEBUG (std::cerr << "dequeue: #a == 0, waiting." << std::endl);
+ // wait; will automatically unlock() the internal mutex via
+ // the scoped lock
+ d_readBlock->wait (l);
+ // and re-lock() it here.
if (d_doAbort) {
- d_internal->unlock ();
- DEBUG (fprintf (stderr, "dequeue: #a == 0, aborting.\n"));
+ DEBUG (std::cerr << "dequeue: #a == 0, aborting." << std::endl);
return (2);
}
- DEBUG (fprintf (stderr, "dequeue: #a == 0, done waiting.\n"));
+ DEBUG (std::cerr << "dequeue: #a == 0, done waiting." << std::endl);
}
}
if (l_bufLen_I > d_n_avail_read_I)
l_bufLen_I = d_n_avail_read_I;
- UInt32 n_now_I = d_bufLen_I - d_readNdx_I, n_start_I = 0;
+ size_t n_now_I = d_bufLen_I - d_readNdx_I, n_start_I = 0;
if (n_now_I > l_bufLen_I)
n_now_I = l_bufLen_I;
else if (n_now_I < l_bufLen_I)
*bufLen_I = l_bufLen_I;
d_n_avail_read_I -= l_bufLen_I;
d_n_avail_write_I += l_bufLen_I;
- d_writeBlock->signal ();
- d_internal->unlock ();
+ d_writeBlock->notify_one ();
return (1);
};
void abort () {
- d_internal->lock ();
+ gruel::scoped_lock l (*d_internal);
d_doAbort = true;
- d_writeBlock->signal ();
- d_readBlock->signal ();
- d_internal->unlock ();
+ d_writeBlock->notify_one ();
+ d_readBlock->notify_one ();
};
};
projects / debian / gnuradio / blobdiff