Add writeAllChannels method to DataQueue

Author: jsiegle

Source/Processors/RecordNode/DataQueue.cpp

@@ -158,16 +158,22 @@ float DataQueue::writeSynchronizedTimestamps (double start, double step, int des
         LOGE (__FUNCTION__, " Recording Data Queue Overflow: sz1: ", size1, " sz2: ", size2, " nSamples: ", nSamples);
     }
 
+    // Get direct pointer access for faster writes
+    double* writePtr = m_FTSBuffer.getWritePointer (destChannel);
+
+    // Fill first segment
     for (int i = 0; i < size1; i++)
     {
-        m_FTSBuffer.setSample (destChannel, index1 + i, start + (double) i * step);
+        writePtr[index1 + i] = start + (double) i * step;
     }
 
+    // Fill second segment (wrap-around) if present
     if (size2 > 0)
     {
+        double offset = start + (double) size1 * step;
         for (int i = 0; i < size2; i++)
         {
-            m_FTSBuffer.setSample (destChannel, index2 + i, start + (double) (size1 * step) + double (i * step));
+            writePtr[index2 + i] = offset + (double) i * step;
         }
     }
 
@@ -214,8 +220,69 @@ float DataQueue::writeChannel (const AudioBuffer<float>& buffer,
     return 1.0f - (float) m_fifos[destChannel]->getFreeSpace() / (float) m_fifos[destChannel]->getTotalSize();
 }
 
+float DataQueue::writeAllChannels (const AudioBuffer<float>& buffer,
+                                   const Array<int>& srcChannels,
+                                   int nSamples,
+                                   int64 sampleNumber)
+{
+    if (m_numChans == 0 || nSamples == 0)
+        return 0.0f;
+
+    // Get FIFO indices from first channel - all channels should be in sync
+    int index1, size1, index2, size2;
+    m_fifos[0]->prepareToWrite (nSamples, index1, size1, index2, size2);
+
+    if ((size1 + size2) < nSamples)
+    {
+        LOGE (__FUNCTION__, " Recording Data Queue Overflow: sz1: ", size1, " sz2: ", size2, " nSamples: ", nSamples);
+    }
+
+    float maxUsage = 0.0f;
+
+    // Fill sample numbers once for channel 0
+    fillSampleNumbers (0, index1, size1, sampleNumber);
+    if (size2 > 0)
+    {
+        fillSampleNumbers (0, index2, size2, sampleNumber + size1);
+    }
+
+    // Batch copy and update all channels
+    for (int destChannel = 0; destChannel < m_numChans; ++destChannel)
+    {
+        int srcChannel = srcChannels[destChannel];
+
+        // Copy first segment
+        m_buffer.copyFrom (destChannel, index1, buffer, srcChannel, 0, size1);
+
+        // Copy second segment (wrap-around) if present
+        if (size2 > 0)
+        {
+            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)
+        {
+            int d1, d2, d3, d4;
+            m_fifos[destChannel]->prepareToWrite (nSamples, d1, d2, d3, d4);
+        }
+        m_fifos[destChannel]->finishedWrite (size1 + size2);
+    }
+
+    // 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();
+}
+
 /*
-We could copy the internal circular buffer to an external one, as DataBuffer does. This class
+We could copy the internal circular buffer to an external one, as DataBuffer does.
 is, however, intended for disk writing, which is one of the most CPU-critical systems. Just
 allowing the record subsystem to access the internal buffer is way faster, although it has to be
 done with special care and manually finish the read process.

Source/Processors/RecordNode/DataQueue.h

@@ -69,6 +69,22 @@ class DataQueue
     /** Writes an array of data for one channel */
     float writeChannel (const AudioBuffer<float>& buffer, int srcChannel, int destChannel, int nSamples, int64 sampleNumber);
 
+    /** 
+     * Batch write: writes data for all channels in a single operation.
+     * Much faster than calling writeChannel() repeatedly because it uses
+     * a single FIFO operation for all channels.
+     * 
+     * @param buffer      Source audio buffer
+     * @param srcChannels Array mapping dest channel index to source channel index
+     * @param nSamples    Number of samples to write per channel
+     * @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, 
+                            const Array<int>& srcChannels,
+                            int nSamples, 
+                            int64 sampleNumber);
+
     /** Writes an array of timestamps for one stream */
     float writeSynchronizedTimestamps (double start, double step, int destChannel, int nSamples);
 

Source/Processors/RecordNode/RecordNode.cpp

@@ -791,6 +791,7 @@ bool RecordNode::stopAcquisition()
 // called by GenericProcessor::setRecording() and CoreServices::setRecordingStatus()
 void RecordNode::startRecording()
 {
+    
     Array<int> chanProcessorMap;
     Array<int> chanOrderinProc;
     OwnedArray<RecordProcessorInfo> procInfo;
@@ -867,6 +868,9 @@ void RecordNode::startRecording()
     int bufferSize = ads.bufferSize;
 
     dataQueues.clear();
+    streamSourceChannels.clear();
+    int globalChannelIdx = 0;
+
     for (int i = 0; i < dataStreams.size(); i++)
     {
         // Only create queue if this stream has recorded channels
@@ -876,12 +880,24 @@ void RecordNode::startRecording()
             queue->setChannelCount (recordedChannelsPerStream[i]);
             queue->setTimestampStreamCount (1);  // Each queue has one timestamp stream
             dataQueues.add (queue);
+
+            // Pre-compute source channel array for this stream
+            Array<int>* srcChannels = new Array<int>();
+            srcChannels->ensureStorageAllocated (recordedChannelsPerStream[i]);
+            for (int j = 0; j < recordedChannelsPerStream[i]; j++)
+            {
+                srcChannels->add (channelMap[globalChannelIdx + j]);
+            }
+            streamSourceChannels.add (srcChannels);
         }
         else
         {
             // Add nullptr placeholder to maintain stream index alignment
             dataQueues.add (nullptr);
+            streamSourceChannels.add (nullptr);
         }
+
+        globalChannelIdx += recordedChannelsPerStream[i];
     }
 
     recordThread->setQueuePointers (&dataQueues, eventQueue.get(), spikeQueue.get());
@@ -1099,7 +1115,10 @@ void RecordNode::process (AudioBuffer<float>& buffer)
         {
             streamIndex++;
 
-            int recordChanCount = (*stream)["channels"].getArray()->size();
+            // Use pre-computed array size to avoid JSON lookup on every call
+            int recordChanCount = (streamSourceChannels[streamIndex] != nullptr)
+                                      ? streamSourceChannels[streamIndex]->size()
+                                      : 0;
 
             if (recordChanCount == 0)
                 continue;
@@ -1111,8 +1130,6 @@ void RecordNode::process (AudioBuffer<float>& buffer)
 
             int64 sampleNumber = getFirstSampleNumberForBlock (streamId);
 
-            float totalFifoUsage = 0.0f;
-
             double first, second;
 
             // Skip if no queue for this stream (no recorded channels)
@@ -1142,22 +1159,21 @@ void RecordNode::process (AudioBuffer<float>& buffer)
                     numSamples);
             }
 
-            for (int i = 0; i < recordChanCount; i++)
+            if (numSamples > 0 && recordChanCount > 0 && streamSourceChannels[streamIndex] != nullptr)
             {
-                channelIndex++;
-
-                if (numSamples > 0)
-                {
-                    // Write to per-stream queue with local channel index (i)
-                    totalFifoUsage += dataQueues[streamIndex]->writeChannel (buffer,
-                                                               channelMap[channelIndex],
-                                                               i,  // local channel index within stream
-                                                               numSamples,
-                                                               sampleNumber);
-                }
+                // Batch write all channels at once using pre-computed source channels
+                fifoUsage[streamId] = dataQueues[streamIndex]->writeAllChannels (
+                    buffer,
+                    *streamSourceChannels[streamIndex],
+                    numSamples,
+                    sampleNumber);
+            }
+            else
+            {
+                fifoUsage[streamId] = 0.0f;
             }
 
-            fifoUsage[streamId] = totalFifoUsage / recordChanCount;
+            channelIndex += recordChanCount;
 
             if (fifoUsage[streamId] > 0.9)
                 fifoAlmostFull = true;

Source/Processors/RecordNode/RecordNode.h

@@ -293,6 +293,7 @@ class TESTABLE RecordNode : public GenericProcessor,
     int recordingNumber;
 
     OwnedArray<DataQueue> dataQueues;  // One DataQueue per stream for race-condition-free reading
+    OwnedArray<Array<int>> streamSourceChannels;  // Pre-computed source channel indices per stream
     std::unique_ptr<EventMsgQueue> eventQueue;
     std::unique_ptr<SpikeMsgQueue> spikeQueue;