Improved buffer handling

1 files changed, 173 insertions, 57 deletions

diff --git a/ringbuffer.c b/ringbuffer.c
index b52c37e5..3473a9e7 100644
--- a/ringbuffer.c
+++ b/ringbuffer.c
@@ -7,7 +7,7 @@
  * Parts of this file were inspired by the 'ringbuffy.c' from the
  * LinuxDVB driver (see linuxtv.org).
  *
- * $Id: ringbuffer.c 1.20 2004/06/19 12:27:56 kls Exp $
+ * $Id: ringbuffer.c 1.21 2004/10/15 13:49:25 kls Exp $
  */
 
 #include "ringbuffer.h"
@@ -18,11 +18,14 @@
 // --- cRingBuffer -----------------------------------------------------------
 
 #define OVERFLOWREPORTDELTA 5 // seconds between reports
+#define PERCENTAGEDELTA     10
+#define PERCENTAGETHRESHOLD 70
 
 cRingBuffer::cRingBuffer(int Size, bool Statistics)
 {
   size = Size;
   statistics = Statistics;
+  getThreadTid = 0;
   maxFill = 0;
   lastPercent = 0;
   putTimeout = getTimeout = 0;
@@ -36,34 +39,41 @@ cRingBuffer::~cRingBuffer()
      dsyslog("buffer stats: %d (%d%%) used", maxFill, maxFill * 100 / (size - 1));
 }
 
-void cRingBuffer::WaitForPut(void)
+void cRingBuffer::UpdatePercentage(int Fill)
 {
-  if (putTimeout) {
-     putMutex.Lock();
-     readyForPut.TimedWait(putMutex, putTimeout);
-     putMutex.Unlock();
+  if (Fill > maxFill)
+     maxFill = Fill;
+  int percent = Fill * 100 / (Size() - 1) / PERCENTAGEDELTA * PERCENTAGEDELTA;
+  if (percent != lastPercent) {
+     if (percent >= PERCENTAGETHRESHOLD && percent > lastPercent || percent < PERCENTAGETHRESHOLD && lastPercent >= PERCENTAGETHRESHOLD) {
+        dsyslog("buffer usage: %d%% (tid=%ld)", percent, getThreadTid);
+        lastPercent = percent;
+        }
      }
 }
 
+void cRingBuffer::WaitForPut(void)
+{
+  if (putTimeout)
+     readyForPut.Wait(putTimeout);
+}
+
 void cRingBuffer::WaitForGet(void)
 {
-  if (getTimeout) {
-     getMutex.Lock();
-     readyForGet.TimedWait(getMutex, getTimeout);
-     getMutex.Unlock();
-     }
+  if (getTimeout)
+     readyForGet.Wait(getTimeout);
 }
 
 void cRingBuffer::EnablePut(void)
 {
-  if (putTimeout)
-     readyForPut.Broadcast();
+  if (putTimeout && Free() > Size() / 3)
+     readyForPut.Signal();
 }
 
 void cRingBuffer::EnableGet(void)
 {
-  if (getTimeout)
-     readyForGet.Broadcast();
+  if (getTimeout && Available() > Size() / 3)
+     readyForGet.Signal();
 }
 
 void cRingBuffer::SetTimeouts(int PutTimeout, int GetTimeout)
@@ -85,70 +95,168 @@ void cRingBuffer::ReportOverflow(int Bytes)
 
 // --- cRingBufferLinear -----------------------------------------------------
 
-cRingBufferLinear::cRingBufferLinear(int Size, int Margin, bool Statistics)
+#ifdef DEBUGRINGBUFFERS
+#define MAXRBLS 30
+#define DEBUGRBLWIDTH 45
+
+cRingBufferLinear *cRingBufferLinear::RBLS[MAXRBLS] = { NULL };
+
+void cRingBufferLinear::AddDebugRBL(cRingBufferLinear *RBL)
+{
+  for (int i = 0; i < MAXRBLS; i++) {
+      if (!RBLS[i]) {
+         RBLS[i] = RBL;
+         break;
+         }
+      }
+}
+
+void cRingBufferLinear::DelDebugRBL(cRingBufferLinear *RBL)
+{
+  for (int i = 0; i < MAXRBLS; i++) {
+      if (RBLS[i] == RBL) {
+         RBLS[i] = NULL;
+         break;
+         }
+      }
+}
+
+void cRingBufferLinear::PrintDebugRBL(void)
+{
+  bool printed = false;
+  for (int i = 0; i < MAXRBLS; i++) {
+      cRingBufferLinear *p = RBLS[i];
+      if (p) {
+         printed = true;
+         int lh = p->lastHead;
+         int lt = p->lastTail;
+         int h = lh * DEBUGRBLWIDTH / p->Size();
+         int t = lt * DEBUGRBLWIDTH / p->Size();
+         char buf[DEBUGRBLWIDTH + 10];
+         memset(buf, '-', DEBUGRBLWIDTH);
+         if (lt <= lh)
+            memset(buf + t, '*', max(h - t, 1));
+         else {
+            memset(buf, '*', h);
+            memset(buf + t, '*', DEBUGRBLWIDTH - t);
+            }
+         buf[t] = '<';
+         buf[h] = '>';
+         buf[DEBUGRBLWIDTH] = 0;
+         printf("%2d %s %8d %8d %s\n", i, buf, p->lastPut, p->lastGet, p->description);
+         }
+      }
+  if (printed)
+     printf("\n");
+  }
+#endif
+
+cRingBufferLinear::cRingBufferLinear(int Size, int Margin, bool Statistics, const char *Description)
 :cRingBuffer(Size, Statistics)
 {
-  margin = Margin;
+  description = Description ? strdup(Description) : NULL;
+  tail = head = margin = Margin;
   buffer = NULL;
-  getThreadTid = 0;
   if (Size > 1) { // 'Size - 1' must not be 0!
-     buffer = MALLOC(uchar, Size);
-     if (!buffer)
-        esyslog("ERROR: can't allocate ring buffer (size=%d)", Size);
-     Clear();
+     if (Margin <= Size / 2) {
+        buffer = MALLOC(uchar, Size);
+        if (!buffer)
+           esyslog("ERROR: can't allocate ring buffer (size=%d)", Size);
+        Clear();
+        }
+     else
+        esyslog("ERROR: illegal margin for ring buffer (%d > %d)", Margin, Size / 2);
      }
   else
      esyslog("ERROR: illegal size for ring buffer (%d)", Size);
+#ifdef DEBUGRINGBUFFERS
+  lastHead = head;
+  lastTail = tail;
+  lastPut = lastGet = -1;
+  AddDebugRBL(this);
+#endif
 }
 
 cRingBufferLinear::~cRingBufferLinear()
 {
+#ifdef DEBUGRINGBUFFERS
+  DelDebugRBL(this);
+#endif
   free(buffer);
+  free(description);
 }
 
 int cRingBufferLinear::Available(void)
 {
-  Lock();
   int diff = head - tail;
-  Unlock();
   return (diff >= 0) ? diff : Size() + diff - margin;
 }
 
 void cRingBufferLinear::Clear(void)
 {
-  Lock();
-  head = tail = margin;
-  lastGet = -1;
-  Unlock();
+  tail = head;
+#ifdef DEBUGRINGBUFFERS
+  lastHead = head;
+  lastTail = tail;
+  lastPut = lastGet = -1;
+#endif
+  maxFill = 0;
   EnablePut();
+}
+
+int cRingBufferLinear::Read(int FileHandle, int Max)
+{
+  int Tail = tail;
+  int diff = Tail - head;
+  int free = (diff > 0) ? diff - 1 : Size() - head;
+  if (Tail <= margin)
+     free--;
+  int Count = 0;
+  if (free > 0) {
+     if (0 < Max && Max < free)
+        free = Max;
+     Count = safe_read(FileHandle, buffer + head, free);
+     if (Count > 0) {
+        int Head = head + Count;
+        if (Head >= Size())
+           Head = margin;
+        head = Head;
+        if (statistics) {
+           int fill = head - Tail;
+           if (fill < 0)
+              fill = Size() + fill;
+           else if (fill >= Size())
+              fill = Size() - 1;
+           UpdatePercentage(fill);
+           }
+        }
+     }
+#ifdef DEBUGRINGBUFFERS
+  lastHead = head;
+  lastPut = Count;
+#endif
   EnableGet();
+  if (free == 0)
+     WaitForPut();
+  return Count;
 }
 
 int cRingBufferLinear::Put(const uchar *Data, int Count)
 {
   if (Count > 0) {
-     Lock();
+     int Tail = tail;
      int rest = Size() - head;
-     int diff = tail - head;
-     int free = ((tail < margin) ? rest : (diff > 0) ? diff : Size() + diff - margin) - 1;
+     int diff = Tail - head;
+     int free = ((Tail < margin) ? rest : (diff > 0) ? diff : Size() + diff - margin) - 1;
      if (statistics) {
         int fill = Size() - free - 1 + Count;
         if (fill >= Size())
            fill = Size() - 1;
-        if (fill > maxFill)
-           maxFill = fill;
-        int percent = maxFill * 100 / (Size() - 1) / 5 * 5;
-        if (abs(lastPercent - percent) >= 5) {
-           if (percent > 75)
-              dsyslog("buffer usage: %d%% (tid=%ld)", percent, getThreadTid);
-           lastPercent = percent;
-           }
+        UpdatePercentage(fill);
         }
      if (free > 0) {
         if (free < Count)
            Count = free;
-        if (Count > maxFill)
-           maxFill = Count;
         if (Count >= rest) {
            memcpy(buffer + head, Data, rest);
            if (Count - rest)
@@ -162,7 +270,10 @@ int cRingBufferLinear::Put(const uchar *Data, int Count)
         }
      else
         Count = 0;
-     Unlock();
+#ifdef DEBUGRINGBUFFERS
+     lastHead = head;
+     lastPut = Count;
+#endif
      EnableGet();
      if (Count == 0)
         WaitForPut();
@@ -173,25 +284,24 @@ int cRingBufferLinear::Put(const uchar *Data, int Count)
 uchar *cRingBufferLinear::Get(int &Count)
 {
   uchar *p = NULL;
-  Lock();
+  int Head = head;
   if (getThreadTid <= 0)
      getThreadTid = pthread_self();
   int rest = Size() - tail;
-  if (rest < margin && head < tail) {
+  if (rest < margin && Head < tail) {
      int t = margin - rest;
      memcpy(buffer + t, buffer + tail, rest);
      tail = t;
-     rest = head - tail;
+     rest = Head - tail;
      }
-  int diff = head - tail;
+  int diff = Head - tail;
   int cont = (diff >= 0) ? diff : Size() + diff - margin;
   if (cont > rest)
      cont = rest;
   if (cont >= margin) {
      p = buffer + tail;
-     Count = lastGet = cont;
+     Count = gotten = cont;
      }
-  Unlock();
   if (!p)
      WaitForGet();
   return p;
@@ -199,17 +309,23 @@ uchar *cRingBufferLinear::Get(int &Count)
 
 void cRingBufferLinear::Del(int Count)
 {
-  if (Count > 0 && Count <= lastGet) {
-     Lock();
-     tail += Count;
-     lastGet -= Count;
-     if (tail >= Size())
-        tail = margin;
-     Unlock();
+  if (Count > gotten) {
+     esyslog("ERROR: invalid Count in cRingBufferLinear::Del: %d (limited to %d)", Count, gotten);
+     Count = gotten;
+     }
+  if (Count > 0) {
+     int Tail = tail;
+     Tail += Count;
+     gotten -= Count;
+     if (Tail >= Size())
+        Tail = margin;
+     tail = Tail;
      EnablePut();
      }
-  else
-     esyslog("ERROR: invalid Count in cRingBufferLinear::Del: %d", Count);
+#ifdef DEBUGRINGBUFFERS
+  lastTail = tail;
+  lastGet = Count;
+#endif
 }
 
 // --- cFrame ----------------------------------------------------------------