Ticket #10019: 0002-ringbuffer-Adapt-readahead-for-low-bit-rate-high-lat.patch

mythtv/libs/libmythtv/ringbuffer.cpp

@@ -407 +407 @@
     // WARNING: readahead_frames can greatly overestimate or underestimate
     //          the number of frames available in the read ahead buffer
     //          when rh_frames is less than the keyframe distance.
+    if (fps == 0.)
+        return false;
     double bytes_per_frame = kbits_per_sec * (1000.0/8.0) / fps;
+    if (bytes_per_frame == 0.)
+        return false;
     double readahead_frames = sz / bytes_per_frame;
 
     bool near_end = ((vvf + readahead_frames) < 10.0) || (sz < rbs*1.5);
@@ -446 +450 @@
     return ret;
 }
 
+inline int RingBuffer::ReadBufUsed() const
+{
+    return (bufferSize - 1) - ReadBufFree();
+}
+
+inline bool RingBuffer::ReadsAllowed() const
+{
+    return ateof || setswitchtonext || commserror ||
+        // Ensure some hysteresis around fill_min
+        ReadBufUsed() >= (readsallowed ? 1 : fill_min);
+}
+
 /** \fn RingBuffer::ResetReadAhead(long long)
  *  \brief Restart the read-ahead thread at the 'newinternal' position.
  *
@@ -888 +904 @@
             }
         }
 
-        int used = bufferSize - ReadBufFree();
-
         bool reads_were_allowed = readsallowed;
 
-        if ((0 == read_return) || (numfailures > 5) ||
-            (readsallowed != (used >= fill_min || ateof ||
-                              setswitchtonext || commserror)))
+        if (0 == read_return || numfailures > 5 || ReadsAllowed() != readsallowed)
         {
             // If readpos changes while the lock is released
             // we should not handle the 0 read_return now.
@@ -905 +917 @@
 
             commserror |= (numfailures > 5);
 
-            readsallowed = used >= fill_min || ateof ||
-                setswitchtonext || commserror;
+            bool bReadsAllowed = ReadsAllowed();
+            if (readsallowed != bReadsAllowed)
+            {
+                readsallowed = bReadsAllowed;
+                LOG(VB_FILE, LOG_INFO, LOC + (bReadsAllowed ?
+                    QString("Reads allowed: %1 bytes available").arg(ReadBufUsed()) :
+                    QString("Rebuffering %1..%2").arg(ReadBufUsed()).arg(fill_min)) );
+            }
 
             if (0 == read_return && old_readpos == readpos)
             {
@@ -929 +947 @@
 
             rwlock.unlock();
             rwlock.lockForRead();
-            used = bufferSize - ReadBufFree();
         }
 
         LOG(VB_FILE, LOG_DEBUG, LOC + "@ end of read ahead loop");
 
         if (!readsallowed || commserror || ateof || setswitchtonext ||
-            (wanttoread <= used && wanttoread > 0))
+            (wanttoread <= ReadBufUsed() && wanttoread > 0))
         {
             // To give other threads a good chance to handle these
             // conditions, even if they are only requesting a read lock
@@ -943 +960 @@
             generalWait.wakeAll();
             rwlock.unlock();
             usleep(5 * 1000);
-            rwlock.lockForRead();            
+            rwlock.lockForRead();
         }
         else
         {
             // yield if we have nothing to do...
             if (!request_pause && reads_were_allowed &&
-                (used >= fill_threshold || ateof || setswitchtonext))
+                (ReadBufUsed() >= fill_threshold || ateof || setswitchtonext || ignoreliveeof))
             {
                 generalWait.wait(&rwlock, 50);
             }
@@ -959 +976 @@
                 generalWait.wakeAll();
                 rwlock.unlock();
                 usleep(5 * 1000);
-                rwlock.lockForRead();            
+                rwlock.lockForRead();
             }
         }
     }
@@ -1015 +1032 @@
     while (!readsallowed && !stopreads &&
            !request_pause && !commserror && readaheadrunning)
     {
-        generalWait.wait(&rwlock, 1000);
-        if (!readsallowed && t.elapsed() > 1000)
+        // The timeout should allow for congestion of internet streamed media
+        if (t.elapsed() >= 30000)
         {
-            LOG(VB_GENERAL, LOG_WARNING, LOC +
-                "Taking too long to be allowed to read..");
-
-            if (t.elapsed() > 10000)
-            {
-                LOG(VB_GENERAL, LOG_ERR, LOC + "Took more than 10 seconds to "
-                                               "be allowed to read, aborting.");
-                return false;
-            }
+            LOG(VB_GENERAL, LOG_ERR, LOC +
+                QString("Waited %1 seconds to be allowed to read, aborting.")
+                .arg(t.elapsed()/1000) );
+            return false;
         }
+
+        generalWait.wait(&rwlock, 250);
     }
 
-    return readsallowed;
+    if (t.elapsed() >= 500)
+    {
+        LOG(VB_GENERAL, LOG_WARNING, LOC +
+            QString("Waited %1 mS to be allowed to read (avail=%2 fill_min=%3)..")
+            .arg(t.elapsed()).arg(ReadBufAvail()).arg(fill_min) );
+    }
+    return true;
 }
 
 bool RingBuffer::WaitForAvail(int count)
@@ -1054 +1074 @@
         generalWait.wakeAll();
     }
 
-    MythTimer t;
-    t.start();
-    while ((avail < count) && !stopreads &&
-           !request_pause && !commserror && readaheadrunning)
+    MythTimer t; t.start();
+    wanttoread = count;
+    while (avail < count && !stopreads && !request_pause &&
+            !commserror && readaheadrunning)
     {
-        wanttoread = count;
-        generalWait.wait(&rwlock, 250);
-        avail = ReadBufAvail();
-
-        if (ateof && avail < count)
-            count = avail;
-
-        if (avail < count)
+        uint elapsed = t.elapsed();
+        if (elapsed >= 10000)
         {
             int elapsed = t.elapsed();
             if (elapsed > 500 && low_buffers && avail >= fill_min)
@@ -1093 +1107 @@
                 return false;
             }
         }
+        else if (elapsed >= 100 && avail)
+        {
+            LOG(VB_GENERAL, LOG_INFO, LOC +
+                QString("Waited %1 mS for %2 bytes (wanted %3)")
+                .arg(elapsed).arg(avail).arg(count) );
+            count = avail;
+            generalWait.wakeAll();
+            break;
+        }
+
+        generalWait.wait(&rwlock, 100);
+        avail = ReadBufAvail();
+        if (ateof && avail < count)
+            count = avail;
     }
 
     wanttoread = 0;
@@ -1154 +1182 @@
         if (new_pos != old_pos)
         {
             LOG(VB_GENERAL, LOG_ERR, LOC +
-                QString("Peek() Failed to return from new "
+                QString("Seek() Failed to return from new "
                         "position %1 to old position %2, now "
                         "at position %3")
                     .arg(old_pos - ret).arg(old_pos).arg(new_pos));
@@ -1174 +1202 @@
  */
 int RingBuffer::ReadPriv(void *buf, int count, bool peek)
 {
-    QString loc_desc = QString("ReadPriv(..%1, %2)")
+    const QString loc_desc = QString("ReadPriv(..%1, %2)")
         .arg(count).arg(peek?"peek":"normal");
-    LOG(VB_FILE, LOG_DEBUG, LOC + loc_desc +
-        QString(" @%1 -- begin").arg(rbrpos));
 
     rwlock.lockForRead();
+
+    LOG(VB_FILE, LOG_DEBUG, LOC + loc_desc +
+        QString(" @%1 avail=%2 -- begin").arg(rbrpos).arg(ReadBufAvail()));
+
     if (writemode)
     {
         LOG(VB_GENERAL, LOG_ERR, LOC + loc_desc +
@@ -1210 +1240 @@
         if (request_pause || stopreads ||
             !readaheadrunning || (ignorereadpos >= 0))
         {
+            LOG(VB_FILE, LOG_DEBUG, LOC + loc_desc + " -- direct read");
             int ret = ReadDirect(buf, count, peek);
             LOG(VB_FILE, LOG_DEBUG, LOC + loc_desc +
                 QString(": ReadDirect checksum %1")
@@ -1224 +1255 @@
     if (!WaitForReadsAllowed())
     {
         LOG(VB_FILE, LOG_NOTICE, LOC + loc_desc + ": !WaitForReadsAllowed()");
-        rwlock.unlock();
-        stopreads = true; // this needs to be outside the lock
-        rwlock.lockForWrite();
-        wanttoread = 0;
+        // NB don't set stopreads or else the next ReadPriv will call ReadDirect
+        // which, if there's any readahead, will cause data to be returned out
+        // of sequence
         rwlock.unlock();
         return 0;
     }

mythtv/libs/libmythtv/ringbuffer.h

@@ -167 +167 @@
 
     int ReadBufFree(void) const;
     int ReadBufAvail(void) const;
+    int ReadBufUsed() const;
+    bool ReadsAllowed() const;
 
     void ResetReadAhead(long long newinternal);
     void KillReadAheadThread(void);