Optimize DataQueue and RecordThread for per-stream sample number handling

Author: anjaldoshi

Source/Processors/RecordNode/DataQueue.cpp

@@ -67,20 +67,20 @@ void DataQueue::setChannelCount (int nChans)
     m_readSamples.clear();
     m_numChans = nChans;
     m_sampleNumbers.clear();
-    m_lastReadSampleNumbers.clear();
+    m_lastReadSampleNumber = 0;
 
     for (int i = 0; i < nChans; ++i)
     {
         m_fifos.add (new AbstractFifo (m_maxSize));
         m_readSamples.push_back (0);
-        m_sampleNumbers.add (new std::vector<int64>());
+    }
 
-        for (int j = 0; j < m_numBlocks; j++)
-        {
-            m_sampleNumbers.getLast()->push_back (0);
-        }
-        m_lastReadSampleNumbers.push_back (0);
+    // Initialize per-stream sample numbers (one per block)
+    for (int j = 0; j < m_numBlocks; j++)
+    {
+        m_sampleNumbers.push_back (0);
     }
+
     m_buffer.setSize (nChans, m_maxSize);
 }
 
@@ -98,10 +98,12 @@ void DataQueue::resize (int nBlocks)
         m_fifos[i]->setTotalSize (size);
         m_fifos[i]->reset();
         m_readSamples[i] = 0;
-        m_sampleNumbers[i]->resize (nBlocks);
-        m_lastReadSampleNumbers[i] = 0;
     }
 
+    // Resize per-stream sample numbers
+    m_sampleNumbers.resize (nBlocks);
+    m_lastReadSampleNumber = 0;
+
     m_readFTSSamples.clear();
 
     for (int i = 0; i < m_numFTSChans; ++i)
@@ -114,7 +116,7 @@ void DataQueue::resize (int nBlocks)
     m_FTSBuffer.setSize (m_numFTSChans, size);
 }
 
-void DataQueue::fillSampleNumbers (int channel, int index, int size, int64 sampleNumber)
+void DataQueue::fillSampleNumbers (int index, int size, int64 sampleNumber)
 {
     //Search for the next block start.
     int blockMod = index % m_blockSize;
@@ -142,7 +144,7 @@ void DataQueue::fillSampleNumbers (int channel, int index, int size, int64 sampl
         if ((blockStartPos + i) < (index + size))
         {
             latestSampleNumber = startSampleNumber + (i * m_blockSize);
-            m_sampleNumbers[channel]->at (blockIdx) = latestSampleNumber;
+            m_sampleNumbers[blockIdx] = latestSampleNumber;
         }
     }
 }
@@ -202,7 +204,11 @@ float DataQueue::writeChannel (const AudioBuffer<float>& buffer,
                        0,
                        size1);
 
-    fillSampleNumbers (destChannel, index1, size1, sampleNumber);
+    // Only fill sample numbers once per stream (using first channel write)
+    if (destChannel == 0)
+    {
+        fillSampleNumbers (index1, size1, sampleNumber);
+    }
 
     if (size2 > 0)
     {
@@ -213,7 +219,10 @@ float DataQueue::writeChannel (const AudioBuffer<float>& buffer,
                            size1,
                            size2);
 
-        fillSampleNumbers (destChannel, index2, size2, sampleNumber + size1);
+        if (destChannel == 0)
+        {
+            fillSampleNumbers (index2, size2, sampleNumber + size1);
+        }
     }
     m_fifos[destChannel]->finishedWrite (size1 + size2);
 
@@ -239,11 +248,11 @@ float DataQueue::writeAllChannels (const AudioBuffer<float>& buffer,
 
     float maxUsage = 0.0f;
 
-    // Fill sample numbers once for channel 0
-    fillSampleNumbers (0, index1, size1, sampleNumber);
+    // Fill sample numbers once for the stream (not per-channel)
+    fillSampleNumbers (index1, size1, sampleNumber);
     if (size2 > 0)
     {
-        fillSampleNumbers (0, index2, size2, sampleNumber + size1);
+        fillSampleNumbers (index2, size2, sampleNumber + size1);
     }
 
     // Batch copy and update all channels
@@ -260,14 +269,6 @@ float DataQueue::writeAllChannels (const AudioBuffer<float>& buffer,
             m_buffer.copyFrom (destChannel, index2, buffer, srcChannel, size1, size2);
         }
 
-        // Copy sample numbers from channel 0 (faster than calling fillSampleNumbers for each)
-        if (destChannel > 0)
-        {
-            std::memcpy (m_sampleNumbers[destChannel]->data(),
-                         m_sampleNumbers[0]->data(),
-                         m_numBlocks * sizeof (int64));
-        }
-
         // Update FIFO state - need to call prepareToWrite for channels > 0
         if (destChannel > 0)
         {
@@ -278,7 +279,8 @@ float DataQueue::writeAllChannels (const AudioBuffer<float>& buffer,
     }
 
     // Return usage from last channel (all should be the same)
-    return 1.0f - (float) m_fifos[m_numChans - 1]->getFreeSpace() / (float) m_fifos[m_numChans - 1]->getTotalSize();
+    const float usage = 1.0f - (float) m_fifos[m_numChans - 1]->getFreeSpace() / (float) m_fifos[m_numChans - 1]->getTotalSize();
+    return usage;
 }
 
 /*
@@ -300,7 +302,7 @@ const SynchronizedTimestampBuffer& DataQueue::getTimestampBufferReference() cons
 
 bool DataQueue::startRead (std::vector<CircularBufferIndexes>& dataBufferIdxs,
                            std::vector<CircularBufferIndexes>& timestampBufferIdxs,
-                           Array<int64>& sampleNumbers,
+                           int64& sampleNumber,
                            int nMax)
 {
     //This should never happen, but it never hurts to be on the safe side.
@@ -344,37 +346,39 @@ bool DataQueue::startRead (std::vector<CircularBufferIndexes>& dataBufferIdxs,
             idx.size2 = 0;
             m_readFTSSamples[chan] = 0;
         }
+        sampleNumber = m_lastReadSampleNumber;
         m_readInProgress = false;
         return false;
     }
 
-    // Second pass: read the same number of samples from all channels
-    for (int chan = 0; chan < m_numChans; ++chan)
-    {
-        CircularBufferIndexes& idx = dataBufferIdxs[chan];
+    // Get sample number for the stream (using first channel's indices)
+    CircularBufferIndexes& firstIdx = dataBufferIdxs[0];
+    m_fifos.getUnchecked (0)->prepareToRead (samplesToRead, firstIdx.index1, firstIdx.size1, firstIdx.index2, firstIdx.size2);
+    m_readSamples[0] = firstIdx.size1 + firstIdx.size2;
 
-        m_fifos.getUnchecked (chan)->prepareToRead (samplesToRead, idx.index1, idx.size1, idx.index2, idx.size2);
-        m_readSamples[chan] = idx.size1 + idx.size2;
+    int blockMod = firstIdx.index1 % m_blockSize;
+    int blockDiff = (blockMod == 0) ? 0 : (m_blockSize - blockMod);
 
-        int blockMod = idx.index1 % m_blockSize;
-        int blockDiff = (blockMod == 0) ? 0 : (m_blockSize - blockMod);
+    // If the next sample number block is within the data we're reading, include the translated sample number
+    if (blockDiff < (firstIdx.size1 + firstIdx.size2))
+    {
+        int blockIdx = ((firstIdx.index1 + blockDiff) / m_blockSize) % m_numBlocks;
+        sampleNumber = m_sampleNumbers[blockIdx] - blockDiff;
+    }
+    else
+    {
+        // If not, use the last sent sample number
+        sampleNumber = m_lastReadSampleNumber;
+    }
 
-        //If the next sample number block is within the data we're reading, include the translated sample number in the output
-        int64 sampleNum;
+    m_lastReadSampleNumber = sampleNumber + firstIdx.size1 + firstIdx.size2;
 
-        if (blockDiff < (idx.size1 + idx.size2))
-        {
-            int blockIdx = ((idx.index1 + blockDiff) / m_blockSize) % m_numBlocks;
-            sampleNum = m_sampleNumbers[chan]->at (blockIdx) - blockDiff;
-        }
-        //If not, copy the last sent again
-        else
-        {
-            sampleNum = m_lastReadSampleNumbers[chan];
-        }
-
-        sampleNumbers.set (chan, sampleNum); // expensive operation?
-        m_lastReadSampleNumbers[chan] = sampleNum + idx.size1 + idx.size2;
+    // Read remaining channels with same parameters
+    for (int chan = 1; chan < m_numChans; ++chan)
+    {
+        CircularBufferIndexes& idx = dataBufferIdxs[chan];
+        m_fifos.getUnchecked (chan)->prepareToRead (samplesToRead, idx.index1, idx.size1, idx.index2, idx.size2);
+        m_readSamples[chan] = idx.size1 + idx.size2;
     }
 
     // Also find minimum for timestamp streams and read consistently
@@ -418,11 +422,7 @@ void DataQueue::stopRead()
     m_readInProgress = false;
 }
 
-void DataQueue::getSampleNumbersForBlock (int idx, Array<int64>& sampleNumbers) const
+int64 DataQueue::getSampleNumberForBlock (int idx) const
 {
-    sampleNumbers.clear();
-    for (int chan = 0; chan < m_numChans; ++chan)
-    {
-        sampleNumbers.add ((*m_sampleNumbers[chan])[idx]);
-    }
+    return m_sampleNumbers[idx];
 }

Source/Processors/RecordNode/DataQueue.h

@@ -52,7 +52,7 @@ class DataQueue
     ~DataQueue();
 
     /// -----------  NOT THREAD SAFE  -------------- //
-    /** Sets the number of continuous channel buffers needed */
+    /** Sets the number of continuous channel buffers needed (all channels belong to one stream) */
     void setChannelCount (int nChans);
 
     /** Sets the number of timestamp buffers needed */
@@ -61,8 +61,8 @@ class DataQueue
     /** Changes the number of blocks in the queue */
     void resize (int nBlocks);
 
-    /** Returns an array of sample numbers for a given block*/
-    void getSampleNumbersForBlock (int idx, Array<int64>& sampleNumbers) const;
+    /** Returns the sample number for a given block (same for all channels in stream) */
+    int64 getSampleNumberForBlock (int idx) const;
 
     /// -----------  THREAD SAFE  -------------- //
 
@@ -88,10 +88,10 @@ class DataQueue
     /** Writes an array of timestamps for one stream */
     float writeSynchronizedTimestamps (double start, double step, int destChannel, int nSamples);
 
-    /** Start reading data for one channel */
+    /** Start reading data for all channels in this stream */
     bool startRead (std::vector<CircularBufferIndexes>& dataBufferIdxs,
                     std::vector<CircularBufferIndexes>& timestampBufferIdxs,
-                    Array<int64>& sampleNumbers,
+                    int64& sampleNumber,
                     int nMax);
 
     /** Called when data read is finished */
@@ -107,8 +107,8 @@ class DataQueue
     int getBlockSize();
 
 private:
-    /** Fills the sample number buffer for a given channel */
-    void fillSampleNumbers (int channel, int index, int size, int64 sampleNumber);
+    /** Fills the sample number buffer for the stream */
+    void fillSampleNumbers (int index, int size, int64 sampleNumber);
 
     int lastIdx;
 
@@ -120,8 +120,8 @@ class DataQueue
 
     std::vector<int> m_readSamples;
     std::vector<int> m_readFTSSamples;
-    OwnedArray<std::vector<int64>> m_sampleNumbers;
-    std::vector<int64> m_lastReadSampleNumbers;
+    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/RecordNode.cpp

@@ -929,7 +929,7 @@ void RecordNode::startRecording()
     }
 
     recordThread->setFileComponents (rootFolder, experimentNumber, recordingNumber);
-    recordThread->startThread();
+    recordThread->startThread(Thread::Priority::highest);
     isRecording = true;
 
     if (settingsNeeded)

Source/Processors/RecordNode/RecordThread.cpp

@@ -103,6 +103,7 @@ void RecordThread::run()
     int numStreams = recordNode->getNumDataStreams();
     m_perStreamDataIdxs.resize (numStreams);
     m_perStreamTimestampIdxs.resize (numStreams);
+    m_perStreamSampleNumbers.resize (numStreams, 0);
 
     bool closeEarly = true;
 
@@ -131,12 +132,21 @@ void RecordThread::run()
         DataQueue* queue = (*m_dataQueues)[streamIdx];
         if (queue != nullptr)
         {
-            Array<int64> streamSampleNums;
-            queue->getSampleNumbersForBlock (0, streamSampleNums);
-            for (int i = 0; i < streamSampleNums.size(); i++)
+            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)
-                    initSampleNumbers.add (streamSampleNums[i]);
+                    initSampleNumbers.add (streamSampleNum);
                 globalChan++;
             }
         }
@@ -196,21 +206,18 @@ void RecordThread::writeData (int maxSamples,
         m_perStreamDataIdxs[streamIdx].resize (numChannelsInStream);
         m_perStreamTimestampIdxs[streamIdx].resize (1);  // One timestamp stream per queue
 
-        // Create local sample numbers array for this stream
-        Array<int64> streamSampleNumbers;
-        for (int i = 0; i < numChannelsInStream; i++)
-            streamSampleNumbers.add (0);
+        // Get single sample number for this stream
+        int64 streamSampleNumber = 0;
 
-        if (queue->startRead (m_perStreamDataIdxs[streamIdx], m_perStreamTimestampIdxs[streamIdx], streamSampleNumbers, maxSamples))
+        if (queue->startRead (m_perStreamDataIdxs[streamIdx], m_perStreamTimestampIdxs[streamIdx], streamSampleNumber, maxSamples))
         {
-            // Update global sample numbers
-            int localChan = 0;
+            // Update global sample numbers using the stream's sample number
+            m_perStreamSampleNumbers[streamIdx] = streamSampleNumber;
             for (int globalChan = 0; globalChan < m_numChannels; globalChan++)
             {
                 if (m_timestampBufferChannelArray[globalChan] == streamIdx)
                 {
-                    sampleNumbers.set (globalChan, streamSampleNumbers[localChan]);
-                    localChan++;
+                    sampleNumbers.set (globalChan, streamSampleNumber);
                 }
             }
             m_engine->updateLatestSampleNumbers (sampleNumbers);
@@ -271,14 +278,14 @@ void RecordThread::writeData (int maxSamples,
                     int numSamples2 = m_perStreamDataIdxs[streamIdx][0].size2;
                     int bufferIndex2 = m_perStreamDataIdxs[streamIdx][0].index2;
 
-                    // Update sample numbers
-                    localChan = 0;
+                    // Update sample numbers using per-stream value
+                    int64 updatedStreamSampleNumber = m_perStreamSampleNumbers[streamIdx] + numSamples;
+                    m_perStreamSampleNumbers[streamIdx] = updatedStreamSampleNumber;
                     for (int globalChan = 0; globalChan < m_numChannels; globalChan++)
                     {
                         if (m_timestampBufferChannelArray[globalChan] == streamIdx)
                         {
-                            sampleNumbers.set (globalChan, sampleNumbers[globalChan] + numSamples);
-                            localChan++;
+                            sampleNumbers.set (globalChan, updatedStreamSampleNumber);
                         }
                     }
                     m_engine->updateLatestSampleNumbers (sampleNumbers, m_batchWriteChannels[0]);

Source/Processors/RecordNode/RecordThread.h

@@ -97,6 +97,7 @@ class RecordThread : public Thread
     std::atomic<bool> m_cleanExit;
 
     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;