Add minimum block size for writing in RecordThread

Author: anjaldoshi

Source/Processors/RecordNode/DataQueue.cpp

@@ -300,9 +300,25 @@ const SynchronizedTimestampBuffer& DataQueue::getTimestampBufferReference() cons
     return m_FTSBuffer;
 }
 
+int DataQueue::getNumSamplesReady() const
+{
+    if (m_numChans == 0)
+        return 0;
+
+    int minReady = INT_MAX;
+    for (int chan = 0; chan < m_numChans; ++chan)
+    {
+        int ready = m_fifos.getUnchecked (chan)->getNumReady();
+        if (ready < minReady)
+            minReady = ready;
+    }
+    return (minReady == INT_MAX) ? 0 : minReady;
+}
+
 bool DataQueue::startRead (std::vector<CircularBufferIndexes>& dataBufferIdxs,
                            std::vector<CircularBufferIndexes>& timestampBufferIdxs,
                            int64& sampleNumber,
+                           int nMin,
                            int nMax)
 {
     //This should never happen, but it never hurts to be on the safe side.
@@ -324,8 +340,8 @@ bool DataQueue::startRead (std::vector<CircularBufferIndexes>& dataBufferIdxs,
     // Apply nMax limit to the minimum
     int samplesToRead = ((minSamplesAvailable > nMax) && (nMax > 0)) ? nMax : minSamplesAvailable;
 
-    // If no samples available on any channel, nothing to read
-    if (samplesToRead == 0)
+    // If not enough samples available (below nMin threshold), skip this read
+    if (samplesToRead < nMin)
     {
         // Initialize all indexes to zero
         for (int chan = 0; chan < m_numChans; ++chan)

Source/Processors/RecordNode/DataQueue.h

@@ -80,18 +80,29 @@ class DataQueue
      * @param sampleNumber Starting sample number for this block
      * @return Maximum FIFO usage across all channels (0.0 to 1.0)
      */
-    float writeAllChannels (const AudioBuffer<float>& buffer, 
+    float writeAllChannels (const AudioBuffer<float>& buffer,
                             const Array<int>& srcChannels,
-                            int nSamples, 
+                            int nSamples,
                             int64 sampleNumber);
 
     /** Writes an array of timestamps for one stream */
     float writeSynchronizedTimestamps (double start, double step, int destChannel, int nSamples);
 
-    /** Start reading data for all channels in this stream */
+    /** Returns the number of samples available to read (minimum across all channels) */
+    int getNumSamplesReady() const;
+
+    /** Start reading data for all channels in this stream 
+     *  @param dataBufferIdxs Output: indices for reading continuous data
+     *  @param timestampBufferIdxs Output: indices for reading timestamps
+     *  @param sampleNumber Output: sample number for the start of the read
+     *  @param nMin Minimum samples required - returns false if fewer available
+     *  @param nMax Maximum samples to read
+     *  @return true if read was started successfully, false otherwise
+     */
     bool startRead (std::vector<CircularBufferIndexes>& dataBufferIdxs,
                     std::vector<CircularBufferIndexes>& timestampBufferIdxs,
                     int64& sampleNumber,
+                    int nMin,
                     int nMax);
 
     /** Called when data read is finished */
@@ -120,8 +131,8 @@ class DataQueue
 
     std::vector<int> m_readSamples;
     std::vector<int> m_readFTSSamples;
-    std::vector<int64> m_sampleNumbers;       // Per-stream sample numbers (one per block)
-    int64 m_lastReadSampleNumber;             // Last sample number read for the stream
+    std::vector<int64> m_sampleNumbers; // Per-stream sample numbers (one per block)
+    int64 m_lastReadSampleNumber; // Last sample number read for the stream
 
     int m_numChans;
     int m_numFTSChans;

Source/Processors/RecordNode/RecordThread.cpp

@@ -31,9 +31,18 @@ RecordThread::RecordThread (RecordNode* parentNode, RecordEngine* engine) : Thre
                                                                             m_engine (engine),
                                                                             recordNode (parentNode),
                                                                             m_receivedFirstBlock (false),
-                                                                            m_cleanExit (true)
-//samplesWritten(0)
+                                                                            m_cleanExit (true),
+                                                                            m_minWriteSamples (BLOCK_DEFAULT_MIN_WRITE_SAMPLES),
+                                                                            m_maxWriteSamples (BLOCK_DEFAULT_MAX_WRITE_SAMPLES)
 {
+    // Ensure max >= min to avoid deadlock
+    if (m_maxWriteSamples < m_minWriteSamples)
+    {
+        LOGC ("WARNING: MAX_WRITE_SAMPLES (", m_maxWriteSamples, ") < MIN_WRITE_SAMPLES (", m_minWriteSamples, "), setting MAX = MIN");
+        m_maxWriteSamples = m_minWriteSamples;
+    }
+
+    LOGD ("RecordThread initialized with MIN_WRITE_SAMPLES=", m_minWriteSamples, " MAX_WRITE_SAMPLES=", m_maxWriteSamples);
 }
 
 RecordThread::~RecordThread()
@@ -137,15 +146,15 @@ void RecordThread::run()
         {
             int64 streamSampleNum = queue->getSampleNumberForBlock (0);
             m_perStreamSampleNumbers[streamIdx] = streamSampleNum;
-            
+
             // Count channels in this stream and set sample numbers for all of them
             int numChannelsInStream = 0;
             for (int ch = 0; ch < m_numChannels; ch++)
             {
                 if (m_timestampBufferChannelArray[ch] == streamIdx)
                     numChannelsInStream++;
             }
-            
+
             for (int i = 0; i < numChannelsInStream; i++)
             {
                 if (globalChan < m_numChannels)
@@ -158,7 +167,7 @@ void RecordThread::run()
 
     //3-Normal loop
     while (! threadShouldExit())
-        writeData (BLOCK_MAX_WRITE_SAMPLES, BLOCK_MAX_WRITE_EVENTS, BLOCK_MAX_WRITE_SPIKES);
+        writeData (m_minWriteSamples, m_maxWriteSamples, BLOCK_MAX_WRITE_EVENTS, BLOCK_MAX_WRITE_SPIKES);
 
     //LOGD(__FUNCTION__, " Exiting record thread");
     //4-Before closing the thread, try to write the remaining samples
@@ -167,8 +176,8 @@ void RecordThread::run()
 
     if (! closeEarly)
     {
-        // flush the buffers
-        writeData (BLOCK_MAX_WRITE_SAMPLES, BLOCK_MAX_WRITE_EVENTS, BLOCK_MAX_WRITE_SPIKES, true);
+        // flush the buffers - use 0 for minSamples to write all remaining data
+        writeData (0, m_maxWriteSamples, BLOCK_MAX_WRITE_EVENTS, BLOCK_MAX_WRITE_SPIKES, true);
 
         //5-Close files
         m_engine->closeFiles();
@@ -179,14 +188,20 @@ void RecordThread::run()
     //LOGC("RecordThread received ", spikesReceived, " spikes and wrote ", spikesWritten, ".");
 }
 
-void RecordThread::writeData (int maxSamples,
+void RecordThread::writeData (int minSamples,
+                              int maxSamples,
                               int maxEvents,
                               int maxSpikes,
                               bool lastBlock)
 {
     int numStreams = m_dataQueues->size();
     int globalChannelOffset = 0;
 
+    // Use sample-based thresholds directly - they naturally scale with channel count:
+    // High channels = large byte writes (efficient), Low channels = small byte writes (fine)
+    int effectiveMinSamples = lastBlock ? 0 : minSamples;
+    int effectiveMaxSamples = maxSamples;
+
     // Process each stream independently - this avoids race conditions
     // between streams with different sample rates
     for (int streamIdx = 0; streamIdx < numStreams; streamIdx++)
@@ -207,12 +222,12 @@ void RecordThread::writeData (int maxSamples,
         }
 
         m_perStreamDataIdxs[streamIdx].resize (numChannelsInStream);
-        m_perStreamTimestampIdxs[streamIdx].resize (1);  // One timestamp stream per queue
+        m_perStreamTimestampIdxs[streamIdx].resize (1); // One timestamp stream per queue
 
         // Get single sample number for this stream
         int64 streamSampleNumber = 0;
 
-        if (queue->startRead (m_perStreamDataIdxs[streamIdx], m_perStreamTimestampIdxs[streamIdx], streamSampleNumber, maxSamples))
+        if (queue->startRead (m_perStreamDataIdxs[streamIdx], m_perStreamTimestampIdxs[streamIdx], streamSampleNumber, effectiveMinSamples, effectiveMaxSamples))
         {
             // Update global sample numbers using the stream's sample number
             m_perStreamSampleNumbers[streamIdx] = streamSampleNumber;

Source/Processors/RecordNode/RecordThread.h

@@ -29,8 +29,17 @@
 #include "DataQueue.h"
 #include "EventQueue.h"
 #include <atomic>
+#include <chrono>
+#include <cstdlib>
 
-#define BLOCK_MAX_WRITE_SAMPLES 4096
+// Block write size limits (in samples) - tune these for performance
+// Sample-based thresholds naturally scale with channel count:
+//   - High channel count = large writes (efficient for high data rates)
+//   - 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_MAX_WRITE_EVENTS 50000
 #define BLOCK_MAX_WRITE_SPIKES 50000
 
@@ -79,8 +88,15 @@ class RecordThread : public Thread
     RecordNode* recordNode;
 
 private:
-    /** Writes continuous data from all per-stream queues */
-    void writeData (int maxSamples,
+    /** Writes continuous data from all per-stream queues 
+     *  @param minSamples Minimum samples required before writing (0 = write any available)
+     *  @param maxSamples Maximum samples per write batch
+     *  @param maxEvents Maximum events to write
+     *  @param maxSpikes Maximum spikes to write
+     *  @param lastBlock If true, write all remaining data regardless of minSamples
+     */
+    void writeData (int minSamples,
+                    int maxSamples,
                     int maxEvents,
                     int maxSpikes,
                     bool lastBlock = false);
@@ -89,16 +105,16 @@ class RecordThread : public Thread
     Array<int> m_channelArray;
     Array<int> m_timestampBufferChannelArray;
 
-    OwnedArray<DataQueue>* m_dataQueues;  // Array of per-stream queues
+    OwnedArray<DataQueue>* m_dataQueues; // Array of per-stream queues
     EventMsgQueue* m_eventQueue;
     SpikeMsgQueue* m_spikeQueue;
 
     std::atomic<bool> m_receivedFirstBlock;
     std::atomic<bool> m_cleanExit;
 
-    Array<int64> sampleNumbers;  // Global sample numbers (indexed by global channel)
-    std::vector<int64> m_perStreamSampleNumbers;  // Per-stream sample numbers
-    
+    Array<int64> sampleNumbers; // Global sample numbers (indexed by global channel)
+    std::vector<int64> m_perStreamSampleNumbers; // Per-stream sample numbers
+
     // Per-stream buffer index arrays for independent queue reading
     std::vector<std::vector<CircularBufferIndexes>> m_perStreamDataIdxs;
     std::vector<std::vector<CircularBufferIndexes>> m_perStreamTimestampIdxs;
@@ -107,15 +123,19 @@ class RecordThread : public Thread
     int spikesWritten;
 
     // Batch write support - pre-allocated buffers for grouping channels by stream
-    std::vector<int> m_batchWriteChannels;       // Write channel indices for current batch
-    std::vector<int> m_batchRealChannels;        // Real channel indices for current batch
-    std::vector<const float*> m_batchDataPtrs;   // Data buffer pointers for current batch
+    std::vector<int> m_batchWriteChannels; // Write channel indices for current batch
+    std::vector<int> m_batchRealChannels; // Real channel indices for current batch
+    std::vector<const float*> m_batchDataPtrs; // Data buffer pointers for current batch
 
     File m_rootFolder;
     int m_experimentNumber;
     int m_recordingNumber;
     int m_numChannels;
 
+    // Block size limits (in samples)
+    int m_minWriteSamples; // Minimum samples before writing
+    int m_maxWriteSamples; // Maximum samples per write batch
+
     JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (RecordThread);
 };