Make CircBuf are lock-free, threadsafe buffer. (We don't actually

take advantage of this yet.)
This commit is contained in:
Glenn Maynard
2004-03-16 03:06:10 +00:00
parent ab50b4c426
commit eb86c76b9f
2 changed files with 188 additions and 39 deletions
+3 -3
View File
@@ -203,7 +203,7 @@ void RageSound::Update(float delta)
/* Return the number of bytes available in the input buffer. */ /* Return the number of bytes available in the input buffer. */
int RageSound::Bytes_Available() const int RageSound::Bytes_Available() const
{ {
return databuf.size(); return databuf.num_readable();
} }
@@ -277,7 +277,7 @@ int RageSound::FillBuf(int bytes)
while(bytes > 0) while(bytes > 0)
{ {
if(read_block_size > databuf.capacity() - databuf.size()) if(read_block_size > databuf.num_writable())
break; /* full */ break; /* full */
char inbuf[10240]; char inbuf[10240];
@@ -347,7 +347,7 @@ int RageSound::GetData(char *buffer, int size)
} else { } else {
/* Feed data out of our streaming buffer. */ /* Feed data out of our streaming buffer. */
ASSERT(Sample); ASSERT(Sample);
got = min(int(databuf.size()), size); got = min(int(databuf.num_readable()), size);
if(buffer) if(buffer)
databuf.read(buffer, got); databuf.read(buffer, got);
} }
+181 -32
View File
@@ -1,67 +1,216 @@
#ifndef RAGE_UTIL_CIRCULAR_BUFFER #ifndef RAGE_UTIL_CIRCULAR_BUFFER
#define RAGE_UTIL_CIRCULAR_BUFFER #define RAGE_UTIL_CIRCULAR_BUFFER
/* Lock-free circular buffer. This should be threadsafe if one thread is reading
* and another is writing. */
template<class T> template<class T>
class CircBuf class CircBuf
{ {
basic_string<T> buf; T *buf;
unsigned cnt, start; /* read_pos is the position data is read from; write_pos is the position
* data is written to. If read_pos == write_pos, the buffer is empty.
*
* There will always be at least one position empty, as a completely full
* buffer (read_pos == write_pos) is indistinguishable from an empty buffer.
*
* Invariants: read_pos < size, write_pos < size. */
unsigned size;
/* These are volatile to prevent reads and writes to them from being optimized. */
volatile unsigned read_pos, write_pos;
public: public:
CircBuf() { clear(); } CircBuf()
unsigned size() const { return cnt; } {
unsigned capacity() const { return buf.size(); } buf = NULL;
clear();
}
~CircBuf()
{
delete[] buf;
}
/* Return the number of elements available to read. */
unsigned num_readable() const
{
const int rpos = read_pos;
const int wpos = write_pos;
if( rpos < wpos )
/* The buffer looks like "eeeeDDDDeeee" (e = empty, D = data). */
return wpos - rpos;
else if( rpos > wpos )
/* The buffer looks like "DDeeeeeeeeDD" (e = empty, D = data). */
return size - (rpos - wpos);
else // if( rpos == wpos )
/* The buffer looks like "eeeeeeeeeeee" (e = empty, D = data). */
return 0;
}
/* Return the number of elements writable. Note that there must always
* be one */
unsigned num_writable() const
{
const int rpos = read_pos;
const int wpos = write_pos;
int ret;
if( rpos < wpos )
/* The buffer looks like "eeeeDDDDeeee" (e = empty, D = data). */
ret = size - (wpos - rpos);
else if( rpos > wpos )
/* The buffer looks like "DDeeeeeeeeDD" (e = empty, D = data). */
ret = rpos - wpos;
else // if( rpos == wpos )
/* The buffer looks like "eeeeeeeeeeee" (e = empty, D = data). */
ret = size;
/* Subtract one, to account for the element that we never fill. */
return ret - 1;
}
unsigned capacity() const { return size; }
void reserve( unsigned n ) void reserve( unsigned n )
{ {
/* Reserve an extra byte. We'll never fill more than n bytes; the extra
* byte is to guarantee that read_pos != write_pos when the buffer is full,
* since that would be ambiguous with an empty buffer. */
clear(); clear();
buf.erase(); delete[] buf;
buf.insert( buf.end(), n, 0 ); buf = new T[n+1];
size = n+1;
} }
void clear() void clear()
{ {
cnt = start = 0; read_pos = write_pos = 0;
} }
void write( const T *buffer, unsigned buffer_size ) /* Indicate that n elements have been written. */
void advance_write_pointer( int n )
{ {
ASSERT( size() + buffer_size <= capacity() ); /* overflow */ write_pos = (write_pos + n) % size;
}
while( buffer_size ) /* Indicate that n elements have been read. */
void advance_read_pointer( int n )
{ {
unsigned write_pos = start + size(); read_pos = (read_pos + n) % size;
if( write_pos >= buf.size() ) }
write_pos -= buf.size();
const int cpy = int(min(buffer_size, buf.size() - write_pos)); void get_write_pointers( T *pPointers[2], unsigned pSizes[2] )
buf.replace( write_pos, cpy, buffer, cpy ); {
const int rpos = read_pos;
const int wpos = write_pos;
cnt += cpy; if( rpos <= wpos )
{
/* The buffer looks like "eeeeDDDDeeee" or "eeeeeeeeeeee" (e = empty, D = data). */
pPointers[0] = buf+wpos;
pPointers[1] = buf;
buffer += cpy; pSizes[0] = size - wpos;
buffer_size -= cpy; pSizes[1] = rpos;
}
else if( rpos > wpos )
{
/* The buffer looks like "DDeeeeeeeeDD" (e = empty, D = data). */
pPointers[0] = buf+wpos;
pPointers[1] = NULL;
pSizes[0] = rpos - wpos;
pSizes[1] = 0;
}
/* Subtract one, to account for the element that we never fill. */
if( pSizes[1] )
--pSizes[1];
else
--pSizes[0];
}
void get_read_pointers( T *pPointers[2], unsigned pSizes[2] )
{
const int rpos = read_pos;
const int wpos = write_pos;
if( rpos < wpos )
{
/* The buffer looks like "eeeeDDDDeeee" (e = empty, D = data). */
pPointers[0] = buf+rpos;
pPointers[1] = NULL;
pSizes[0] = wpos - rpos;
pSizes[1] = 0;
}
else if( rpos > wpos )
{
/* The buffer looks like "DDeeeeeeeeDD" (e = empty, D = data). */
pPointers[0] = buf+rpos;
pPointers[1] = buf;
pSizes[0] = size - rpos;
pSizes[1] = wpos;
}
else
{
/* The buffer looks like "eeeeeeeeeeee" (e = empty, D = data). */
pPointers[0] = NULL;
pPointers[1] = NULL;
pSizes[0] = 0;
pSizes[1] = 0;
} }
} }
void read( T *buffer, unsigned buffer_size ) /* Write buffer_size elements from buffer, and advance the write pointer. If
* the data will not fit entirely, the write pointer will be unchanged
* and false will be returned. */
bool write( const T *buffer, unsigned buffer_size )
{ {
ASSERT( size() >= buffer_size ); /* underflow */ T *p[2];
unsigned sizes[2];
get_write_pointers( p, sizes );
while( buffer_size ) if( buffer_size > sizes[0] + sizes[1] )
{ return false;
const unsigned total = static_cast<unsigned>(min(buf.size() - start, size()));
const unsigned cpy = min( buffer_size, total );
buf.copy( buffer, cpy, start );
start += cpy; const int from_first = min( buffer_size, sizes[0] );
if( start == buf.size() ) memcpy( p[0], buffer, from_first*sizeof(T) );
start = 0; if( buffer_size > sizes[0] )
cnt -= cpy; memcpy( p[1], buffer+from_first, max(buffer_size-sizes[0], 0u)*sizeof(T) );
buffer += cpy; advance_write_pointer( buffer_size );
buffer_size -= cpy;
return true;
} }
/* Read buffer_size elements from buffer, and advance the read pointer. If
* the buffer can not be filled completely, the read pointer will be unchanged
* and false will be returned. */
bool read( T *buffer, unsigned buffer_size )
{
T *p[2];
unsigned sizes[2];
get_read_pointers( p, sizes );
if( buffer_size > sizes[0] + sizes[1] )
return false;
const int from_first = min( buffer_size, sizes[0] );
memcpy( buffer, p[0], from_first*sizeof(T) );
if( buffer_size > sizes[0] )
memcpy( buffer+from_first, p[1], max(buffer_size-sizes[0], 0u)*sizeof(T) );
/* Set the data that we just read to 0xFF. This way, if we're passing pointesr
* through, we can tell if we accidentally get a stale pointer. */
memset( p[0], 0xFF, from_first*sizeof(T) );
if( buffer_size > sizes[0] )
memset( p[1], 0xFF, max(buffer_size-sizes[0], 0u)*sizeof(T) );
advance_read_pointer( buffer_size );
return true;
} }
}; };