Implement dynamic write threshold adjustment based on buffer usage

Author: anjaldoshi

Source/Processors/RecordNode/DataQueue.cpp

@@ -22,6 +22,7 @@
 */
 
 #include "DataQueue.h"
+#include <climits>
 
 DataQueue::DataQueue (int blockSize, int nBlocks) : m_buffer (0, blockSize * nBlocks),
                                                     m_numChans (0),
@@ -36,7 +37,7 @@ DataQueue::~DataQueue()
 {
 }
 
-int DataQueue::getBlockSize()
+int DataQueue::getBlockSize() const
 {
     return m_blockSize;
 }

Source/Processors/RecordNode/DataQueue.h

@@ -115,7 +115,7 @@ class DataQueue
     const SynchronizedTimestampBuffer& getTimestampBufferReference() const;
 
     /** Returns the current block size*/
-    int getBlockSize();
+    int getBlockSize() const;
 
 private:
     /** Fills the sample number buffer for the stream */

Source/Processors/RecordNode/RecordThread.cpp

@@ -49,6 +49,54 @@ RecordThread::~RecordThread()
 {
 }
 
+void RecordThread::adjustWriteThresholdsBasedOnBufferUsage (float bufferUsage)
+{
+    // Dynamic adjustment strategy:
+    // - Low usage (0-40%): Use larger batches for efficiency
+    // - Medium usage (40-70%): Use moderate batches  
+    // - High usage (70-85%): Use smaller batches, write more frequently
+    // - Critical usage (85%+): Minimize delay, write as soon as possible
+    
+    int oldMin = m_minWriteSamples;
+    int oldMax = m_maxWriteSamples;
+    
+    if (bufferUsage < 0.4f)
+    {
+        // Low buffer usage - optimize for disk efficiency with larger batches
+        m_minWriteSamples = BLOCK_DEFAULT_MIN_WRITE_SAMPLES;
+        m_maxWriteSamples = BLOCK_DEFAULT_MAX_WRITE_SAMPLES;
+    }
+    else if (bufferUsage < 0.7f)
+    {
+        // Medium buffer usage - balanced approach
+        m_minWriteSamples = BLOCK_DEFAULT_MIN_WRITE_SAMPLES / 2;  // 256
+        m_maxWriteSamples = BLOCK_DEFAULT_MAX_WRITE_SAMPLES / 2;  // 2048
+    }
+    else if (bufferUsage < 0.85f)
+    {
+        // High buffer usage - prioritize emptying buffer
+        m_minWriteSamples = BLOCK_DEFAULT_MIN_WRITE_SAMPLES / 4;  // 128
+        m_maxWriteSamples = BLOCK_DEFAULT_MAX_WRITE_SAMPLES / 4;  // 1024
+    }
+    else
+    {
+        // Critical buffer usage - write immediately with any available data
+        m_minWriteSamples = 64;  // Very low threshold
+        m_maxWriteSamples = 512; // Smaller batches to reduce latency
+    }
+    
+    // Ensure max >= min to avoid deadlock
+    if (m_maxWriteSamples < m_minWriteSamples)
+        m_maxWriteSamples = m_minWriteSamples;
+    
+    // Log threshold changes to help monitor adaptive behavior
+    if (oldMin != m_minWriteSamples || oldMax != m_maxWriteSamples)
+    {
+        LOGC ("Buffer usage: ", int (bufferUsage * 100), "% - Adjusted write thresholds: MIN=", 
+              m_minWriteSamples, " MAX=", m_maxWriteSamples);
+    }
+}
+
 void RecordThread::setEngine (RecordEngine* engine)
 {
     m_engine = engine;
@@ -170,7 +218,28 @@ void RecordThread::run()
 
     //4 - Normal loop
     while (! threadShouldExit())
+    {
+        // Calculate maximum buffer usage across all streams to adjust write strategy
+        float maxBufferUsage = 0.0f;
+        for (int streamIdx = 0; streamIdx < numStreams; streamIdx++)
+        {
+            DataQueue* queue = (*m_dataQueues)[streamIdx];
+            if (queue != nullptr)
+            {
+                // Calculate buffer usage based on samples ready vs total capacity
+                int samplesReady = queue->getNumSamplesReady();
+                int totalCapacity = queue->getBlockSize() * DATA_BUFFER_NBLOCKS;
+                float usage = (float) samplesReady / (float) totalCapacity;
+                if (usage > maxBufferUsage)
+                    maxBufferUsage = usage;
+            }
+        }
+        
+        // Dynamically adjust write thresholds based on buffer fullness
+        adjustWriteThresholdsBasedOnBufferUsage (maxBufferUsage);
+        
         writeData (m_minWriteSamples, m_maxWriteSamples, BLOCK_MAX_WRITE_EVENTS, BLOCK_MAX_WRITE_SPIKES);
+    }
 
     //LOGD(__FUNCTION__, " Exiting record thread");
     //5 - Before closing the thread, try to write the remaining samples

Source/Processors/RecordNode/RecordThread.h

@@ -38,8 +38,8 @@
 //   - Low channel count = small writes (fine for low data rates, keeps latency low)
 //
 
-#define BLOCK_DEFAULT_MIN_WRITE_SAMPLES 8192
-#define BLOCK_DEFAULT_MAX_WRITE_SAMPLES 16384
+#define BLOCK_DEFAULT_MIN_WRITE_SAMPLES 512
+#define BLOCK_DEFAULT_MAX_WRITE_SAMPLES 4096
 #define BLOCK_MAX_WRITE_EVENTS 50000
 #define BLOCK_MAX_WRITE_SPIKES 50000
 
@@ -139,6 +139,11 @@ class RecordThread : public Thread
     int m_minWriteSamples; // Minimum samples before writing
     int m_maxWriteSamples; // Maximum samples per write batch
 
+    /** Dynamically adjusts write thresholds based on buffer fullness 
+     *  @param bufferUsage Current buffer usage (0.0 to 1.0)
+     */
+    void adjustWriteThresholdsBasedOnBufferUsage (float bufferUsage);
+
     JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (RecordThread);
 };