File: RingBuffer.cpp

package info (click to toggle)
camstream 0.27%2Bdfsg-3
  • links: PTS
  • area: main
  • in suites: lenny
  • size: 12,320 kB
  • ctags: 5,393
  • sloc: cpp: 17,031; sh: 8,154; asm: 455; ansic: 440; makefile: 343
file content (580 lines) | stat: -rw-r--r-- 16,250 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
/*  audiodevs: Abstraction layer for audio hardware & samples
    Copyright (C) 2003-2004 Nemosoft Unv.

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

    For questions, remarks, patches, etc. for this program, the author can be
    reached at camstream@smcc.demon.nl.
*/

/**
  \class CRingBuffer

  \brief A thread-safe circular buffer with multiple readers and multiple writers.

  A ringbuffer is a FIFO buffer: First In, First Out. It gets its name
  because if data is to be written beyond the end of the internal buffer,
  it wraps around to the beginning (provided that there is enough space
  left and we don't overwrite the tail). Thus the buffer is used 'circular',
  in a ring.

  What makes this class special is that it supports multiple writers and
  readers, \b and is thread-safe. So, multiple threads can write to the
  buffer (appending at the front), and multiple threads can read from the
  tail. (Please note that the readers and writers themselves are not
  thread safe! Use separate readers and writers in different threads).

  However, the functionality is not symmetrical: there is only a single head
  pointer, so every writer always appends to the buffer, not overwriting
  anything. But each reader has its own tail pointer and all readers will
  receive the same data from the buffer. Effectively, the data is copied
  to the various reader threads.

  The readers don't have a head pointer; just a tail and the length; the
  length is increased every time data is added by a writer. The tail and
  length are updated at read operations.

  Note: you must use at least Qt 2.3.0, because there was a bug in the
  QMutex class.
*/

#ifndef RINGBUFFER_DEBUG
#undef  RINBGUFFER_DEBUG
//#define RINGBUFFER_DEBUG
#endif

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include <assert.h>
#include <stdlib.h>

#include "MemoryAudioSample.h"
#include "RingBuffer.h"

#define RB_EVENT		0x52420000
#define RB_EVENT_NEW_DATA	(RB_EVENT + 1)
/**
  @param buffer_space The desired buffer length, in bytes

  Constructor; allocates space.
*/

CRingBuffer::CRingBuffer(unsigned int buffer_space)
{
   BufferSpace = buffer_space;
   Buffer = NULL;
   BufferHead = 0;
   MaxLength = 0;
   BytesWritten = 0;
   Overflows = 0;

   if (BufferSpace >= 0)
     Buffer = malloc(BufferSpace);
}


CRingBuffer::~CRingBuffer()
{
   free(Buffer);
   Buffer = NULL;
}


// private

/**
  @param s A new buffer writer

  Adds a new CRingBufferWriter to the pool of writers. A writer has the
  right to add data to the circular buffer. A writer may be attached to
  a ring only once.
*/

void CRingBuffer::AttachWriter(CRingBufferWriter *s)
{
#ifdef RINGBUFFER_DEBUG
   qDebug("CRingBuffer::AttachWriter(%p)", s);
#endif
   Lists.lock();
   Writers.append(s);
   Lists.unlock();
   emit WriterAttached();
}

void CRingBuffer::DetachWriter(CRingBufferWriter *s)
{
#ifdef RINGBUFFER_DEBUG
   qDebug("CRingBuffer::DetachWriter(%p)", s);
#endif
   Lists.lock();
   if (!Writers.remove(s))
     qWarning("CRingBuffer::DetachWriter(%p): failed to find item in list.", s);
   Lists.unlock();
   emit WriterDetached();
}


/**
  \brief Append data to buffer-front
  \param data Pointer to data
  \param len Number of byte to put into buffer
  \param must_fit When there's not enough space in the buffer, do not attempt to write data into it
  \return Number of bytes put in buffer; may be smaller than \ref len
*/
int CRingBuffer::AddToBuffer(void *data, int len, bool must_fit)
{
   int plus;
   CRingBufferReader *reader;

   /* Although the lock in this function is rather long, and includes a
      (lenghty) memcpy() operation, there's not much we can do about it.
      There's all sorts of problems with trying to keep the memcpy() outside
      the lock, mainly consistency (a second writer-thread could increase
      the reader buffer lengths, and thus the reader-'heads', to a point
      beyond where the mempcy() of the first writer thread is, should that
      thread have run out of its timeslice).

      So, effectively all writes are serialized, but it doesn't matter
      because usually there are more readers than writers, and the total
      time of all the memcpy()s combined is needed anyway.
    */

   Head.lock();
   if (len + MaxLength > BufferSpace) { // Dang, not enough space left
//qWarning("CRingBuffer::AddToBuffer: buffer overflow (len = %d, space = %d)", len, BufferSpace - MaxLength);
     if (must_fit)
       len = 0;  // Don't bail out yet; fall through to the reader update part below
     else
       len = BufferSpace - MaxLength;
     Overflows++;
   }

   if (len > 0) { // We have space left.
     /* copy data, split if necessary */
     if (BufferHead + len > BufferSpace) {
       /* hrmpf. Copy data in 2 strokes */
       plus = BufferSpace - BufferHead;
       memcpy((char *)Buffer + BufferHead, data, plus);
       memcpy(Buffer, (char *)data + plus, len - plus);
     }
     else // 'ANSI C++ forbids using pointer of type 'void *' in arithmetic'... How inconvenient.
       memcpy((char *)Buffer + BufferHead, data, len);

     /* advance head */
     BufferHead = (BufferHead + len) % BufferSpace;
     BytesWritten += len;
   }

   /* Data is now in the buffer; adjust lengths of our readers, finding
      MaxLength at the same time.
      NB: we always process our readers here, even if our buffer does get full;
      this prevents dead-lock. If we had returned above when len == 0 we would
      never get here, and notice that data got read by our readers. Second,
      any Readers waiting for data will get notified.
    */

   Lists.lock(); // prevent list manipulation while we're busy
   MaxLength = 0;
   reader = Readers.first();
   while (reader != NULL) {
      reader->Lock.lock();
      reader->MyBufferLength += len;
      if (reader->MyBufferLength > MaxLength)
        MaxLength = reader->MyBufferLength;
      if (reader->LowWaterMark > 0 && reader->MyBufferLength > reader->LowWaterMark)
        reader->DataReady.wakeAll(); // In case our thread is waiting for data
      reader->Lock.unlock();
// FIXME: post event when low water mark is *crossed*
//      QThread::postEvent(reader, new QEvent((QEvent::Type)RB_EVENT_NEW_DATA));
      reader = Readers.next();
   }
   Lists.unlock();
   Head.unlock();

   return len;
}

/**
  \brief Return number of bytes available in buffer
*/

int CRingBuffer::SpaceLeft()
{
   int n;
   Head.lock();
   n = BufferSpace - MaxLength;
   Head.unlock();
   return n;
}

/**
  \brief Return number of bytes used in buffer

  Updates internal counters.
*/
int CRingBuffer::SpaceUsed()
{
   CRingBufferReader *reader;
   int n;

   Head.lock();
   Lists.lock(); // prevent list manipulation while we're busy
   MaxLength = 0;
   reader = Readers.first();
   while (reader != NULL) {
      reader->Lock.lock();
      if (reader->MyBufferLength > MaxLength)
        MaxLength = reader->MyBufferLength;
      reader->Lock.unlock();
      reader = Readers.next();
   }
   Lists.unlock();

   n = MaxLength;
   Head.unlock();
   return n;
}

/**
  Adds a reader to our pool; also initializes the Tail pointer.
*/
void CRingBuffer::AttachReader(CRingBufferReader *r)
{
#ifdef RINGBUFFER_DEBUG
   qDebug("CRingBuffer::AttachReader(%p)", r);
#endif
   /* We have to be careful we get a valid BufferHead pointer. If we are
      attaching while we we are in the middle of AddToBuffer, all kinds of
      weird situations might happen.
    */
   Head.lock(); // make sure BufferHead doesn't change
   r->BufferTail = BufferHead;
   Lists.lock();
   Readers.append(r);
   connect(this, SIGNAL(BufferFlushed()), r, SIGNAL(BufferFlushed()));
   Lists.unlock();
   Head.unlock();
   emit ReaderAttached();
}

void CRingBuffer::DetachReader(CRingBufferReader *r)
{
#ifdef RINGBUFFER_DEBUG
   qDebug("CRingBuffer::DetachReader(%p)", r);
#endif
   Lists.lock();
   if (!Readers.remove(r))
     qWarning("CRingBuffer::DetachReader(%p): failed to find item in list.", r);
   Lists.unlock();
   emit ReaderDetached();
}


/** \brief Flush contents of buffer
*/
void CRingBuffer::Flush()
{
   CRingBufferReader *reader = 0;

   Head.lock(); // make sure BufferHead doesn't change
   Lists.lock();
   MaxLength = 0;
   reader = Readers.first();
   while (reader != NULL) {
      reader->Lock.lock();
      reader->BufferTail = BufferHead; // reset poiinters
      reader->MyBufferLength = 0;
      reader->Lock.unlock();
      reader = Readers.next();
   }
   Lists.unlock();
   Head.unlock();

   emit BufferFlushed();
}

// public



unsigned int CRingBuffer::GetBufferLength()
{
   return BufferSpace;
}




/****************************************************************************/

/**
  \brief Constructor for writer; attaches to ring automatically

  There is no default constructor.
*/
CRingBufferWriter::CRingBufferWriter(CRingBuffer *ring)
{
qDebug(">> CRingBufferWriter::CRingBufferWriter(CRingBuffer *)");
   assert(ring != NULL);
   pRing = ring;

   pRing->AttachWriter(this);
qDebug("<< CRingBufferWriter::CRingBufferWriter(CRingBuffer *)");
}

/**
  \brief Destructor; detaches from ring automatically
*/
CRingBufferWriter::~CRingBufferWriter()
{
   pRing->DetachWriter(this);
}


/**
  \brief Put data in buffer
  \param data A pointer to the data
  \param len Number of bytes to write
  \param must_fit Data must fit in buffer
  \return Number of bytes written

  This function tries to put new data in the circular buffer. It returns
  the number of bytes that actually got written, which may be < len. If
  must_fit is TRUE then there must be enough space available in the buffer
  to write the whole data block. When there's not enough room, 0 will
  be returned.

  This function has an interesting side effect: when there are no readers,
  the buffer is never filled!
*/
int CRingBufferWriter::WriteToBuffer(void *data, int len, bool must_fit) const
{
#ifdef RINGBUFFER_DEBUG
   qDebug("CRingBufferWriter::WriteToBuffer(%d) this = %p", len, this);
#endif
   return pRing->AddToBuffer(data, len, must_fit);
}

int CRingBufferWriter::SpaceLeft() const
{
   return pRing->SpaceLeft();
}

int CRingBufferWriter::SpaceUsed() const
{
   return pRing->SpaceUsed();
}

void CRingBufferWriter::Flush() const
{
   pRing->Flush();
}


/*****************/

/**
  \brief Contructor

  Constructs a \ref CRingBuffer reader, and adds it to the pool of readers.
  Initially, our 'buffer' will be empty; we will only see data that is
  written to the ringbuffer after we are attached.
*/

CRingBufferReader::CRingBufferReader(CRingBuffer *ring)
{
qDebug(">> CRingBufferReader::CRingBufferReader(CRingBuffer *)");
   pRing = ring;
   MyBufferLength = 0;
   LowWaterMark = HighWaterMark = 0;
   pRing->AttachReader(this);
qDebug("<< CRingBufferReader::CRingBufferReader(CRingBuffer *)");
}


/**
  \brief Destructor

  The destructor will automatically detach this object from the ringbuffer.
*/
CRingBufferReader::~CRingBufferReader()
{
   pRing->DetachReader(this);
}

// private


// public

int CRingBufferReader::SpaceUsed() const
{
   return MyBufferLength;
}

/**
  \brief Get data from tail of buffer
  \param data Pointer where to store data
  \param len Number of requested bytes
  \return Number of bytes put in data pointer; may be less than \e len

  This function reads data from the tail of the buffer. It updates our
  tailpointer (CRingBuffer will check the tailpointers during a write,
  to see how much data is still in use).
*/
int CRingBufferReader::ReadFromBufferTail(void *data, unsigned int len)
{
   unsigned int buflen;

#ifdef RINGBUFFER_DEBUG
   qDebug("CRingBufferReader::ReadFromBufferTail(%d)", len);
#endif
   buflen = MyBufferLength;
   if (len > buflen)
     len = buflen;
   if (len > 0) {
     // Copy data in 1 or 2 strokes
     if (BufferTail + len > pRing->BufferSpace) {
       buflen = pRing->BufferSpace - BufferTail;
       memcpy(data, (char *)pRing->Buffer + BufferTail, buflen);
       memcpy((char *)data + buflen, pRing->Buffer, len - buflen);
     }
     else
       memcpy(data, (char *)pRing->Buffer + BufferTail, len);
   }
   Lock.lock(); // keep this lock minimal
   MyBufferLength -= len;
   BufferTail = (BufferTail + len) % pRing->BufferSpace;
   Lock.unlock();
   return len;
}



/**
  \brief Get data from head of buffer
  \param data Pointer where to store data
  \param len Number of requested bytes
  \param time In case buffer is empty, wait for data a maximum of \b time milliseconds
  \param clear If TRUE, will reset our lengthpointer (pretend we read all)
  \return Number of bytes put in data pointer; may be less than \e len

  This functions allows you to peek at the data at the head of the buffer.
  It will not update the lengthpointer, unless you set \e clear to TRUE;
  then it will reset our reader-buffer (this does not affect other readers).
*/
int CRingBufferReader::ReadFromBufferHead(void *data, unsigned int len, bool clear, unsigned long time)
{
   unsigned int buflen, head;

#ifdef RINGBUFFER_DEBUG
   qDebug("CRingBufferReader::ReadFromBufferHead(%d, %c, %lu)", len, clear ? 'T' : 'F', time);
#endif
   // If we have a LowWaterMark and we don't have enough data, wait a bit.
   if (LowWaterMark > 0 && MyBufferLength < LowWaterMark) {
//qDebug("ReadFromBufferHead: Waiting for data for %lu ms...", time);
     if (!DataReady.wait(time))
       qDebug("CRingBufferReader::ReadFromBufferHead() Timed out after %lu ms.", time);
   }
   Lock.lock();
   buflen = MyBufferLength;
   if (len > buflen)
     len = buflen;
   head = (BufferTail + buflen - len) % pRing->BufferSpace;
   Lock.unlock();

   if (len > 0) {
     // Copy data in 1 or 2 strokes.
     if (head + len > pRing->BufferSpace) {
       buflen = pRing->BufferSpace - head; // restant
       memcpy(data, (char *)pRing->Buffer + head, buflen);
       memcpy((char *)data + buflen, pRing->Buffer, len - buflen);
     }
     else
       memcpy(data, (char *)pRing->Buffer + head, len);
   }
   if (clear) {
     Lock.lock();
     MyBufferLength = 0;
     BufferTail = pRing->BufferHead;
     Lock.unlock();
   }
   return len;
}


bool CRingBufferReader::event(QEvent *e)
{
   if (e->type() == RB_EVENT_NEW_DATA) {
qDebug("CRingBufferReader::event(dataArrived)");
     emit DataArrived(0);
     return true;
   }
   return false;
}

/**
  \brief Set reader low water mark
  \param mark The new water mark; -1 disables the mark.

  A 'low water mark' is used in a buffer to indicate a minimum fill
  before action is/should be taken; it is often used in combination
  with a high water mark. It prevents things like the 'silly buffer syndrome'
  or nefarious polling on the buffer.

  Basicly, the low and high water marks specify the \e preferred minimum and
  maximum fill of the buffer; you expect the buffer to have at least low mark,
  but not more than high mark bytes of data in it; however, the buffer
  may still drain completely or fill up. But usually action is taken when
  the low or high water mark is crossed.

*/

void CRingBufferReader::SetLowWaterMark(unsigned int mark)
{
   if (mark > pRing->BufferSpace)
     return;
   LowWaterMark = mark;
}

unsigned int CRingBufferReader::GetLowWaterMark() const
{
   return LowWaterMark;
}

void CRingBufferReader::SetHighWaterMark(unsigned int mark)
{
   if (mark > pRing->BufferSpace)
     return;
   HighWaterMark = mark;
}

unsigned int CRingBufferReader::GetHighWaterMark() const
{
   return HighWaterMark;
}

// signals

/**
   \function void CRingBufferReader::DataArrived(int bytes)
   \param bytes Number of newly arrived bytes in buffer


   void BufferFlushed();

*/