Implement batch writing (10-15% recording speedup)

Author: jsiegle

Source/Processors/RecordNode/BinaryFormat/BinaryRecording.cpp

@@ -786,6 +786,109 @@ void BinaryRecording::writeSpike (int electrodeIndex, const Spike* spike)
     increaseEventCounts (rec);
 }
 
+void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
+                                                 const int* realChannels,
+                                                 const float* const* dataBuffers,
+                                                 const double* timestampBuffer,
+                                                 int numChannels,
+                                                 int numSamples,
+                                                 int fileIndex)
+{
+    if (numSamples == 0 || numChannels == 0)
+        return;
+
+    // Ensure we have enough batch buffer space
+    if (numSamples > m_batchBufferSamples || numChannels > m_batchBufferChannels)
+    {
+        int newSamples = jmax (numSamples, m_batchBufferSamples);
+        int newChannels = jmax (numChannels, m_batchBufferChannels);
+        
+        LOGD ("BinaryRecording::writeContinuousDataBatch: Resizing batch buffer to ", 
+              newChannels, " channels x ", newSamples, " samples");
+        
+        m_batchIntBuffer.malloc (newSamples * newChannels);
+        m_batchBufferSamples = newSamples;
+        m_batchBufferChannels = newChannels;
+    }
+
+    // Ensure sample number buffer is large enough
+    if (numSamples > m_bufferSize)
+    {
+        m_sampleNumberBuffer.malloc (numSamples);
+        m_bufferSize = numSamples;
+    }
+
+    // Resize batch pointer arrays if needed
+    if (m_batchScaleFactors.size() < static_cast<size_t> (numChannels))
+    {
+        m_batchScaleFactors.resize (numChannels);
+        m_batchIntBufferPtrs.resize (numChannels);
+    }
+
+    // Get file and validate
+    if (fileIndex < 0 || fileIndex >= m_continuousFiles.size() || ! m_continuousFiles[fileIndex])
+        return;
+
+    // Setup scale factors and output buffer pointers for each channel
+    // Each channel gets a contiguous region of m_batchIntBuffer
+    for (int i = 0; i < numChannels; i++)
+    {
+        m_batchScaleFactors[i] = 1.0f / getContinuousChannel (realChannels[i])->getBitVolts();
+        m_batchIntBufferPtrs[i] = m_batchIntBuffer.getData() + i * numSamples;
+    }
+
+    // Batch convert all channels using SIMD
+    SIMDConverter::convertFloatToInt16Batch (
+        dataBuffers,
+        m_batchIntBufferPtrs.data(),
+        m_batchScaleFactors.data(),
+        numChannels,
+        numSamples);
+
+    // Get starting sample position (all channels in a stream have same position)
+    uint64 startPos = m_samplesWritten[writeChannels[0]];
+
+    // Write all channels at once using batch interleaving
+    m_continuousFiles[fileIndex]->writeChannelBatch (
+        startPos,
+        m_batchIntBufferPtrs.data(),
+        numChannels,
+        numSamples);
+
+    // Update samples written for all channels
+    for (int i = 0; i < numChannels; i++)
+    {
+        m_samplesWritten.set (writeChannels[i], m_samplesWritten[writeChannels[i]] + numSamples);
+    }
+
+    // Write timestamps (only once per stream, using the first channel)
+    // Find which channel in this batch is the first channel for this file (channel index 0 within stream)
+    for (int i = 0; i < numChannels; i++)
+    {
+        if (m_channelIndexes[writeChannels[i]] == 0)
+        {
+            int64 baseSampleNumber = getLatestSampleNumber (writeChannels[i]);
+            uint32 streamId = getContinuousChannel (realChannels[i])->getStreamId();
+
+            if (! wroteFirstSampleNumber[streamId])
+            {
+                firstSampleNumber[streamId] = baseSampleNumber;
+                wroteFirstSampleNumber[streamId] = true;
+            }
+
+            for (int s = 0; s < numSamples; s++)
+                m_sampleNumberBuffer[s] = baseSampleNumber + s;
+
+            m_dataTimestampFiles[fileIndex]->writeData (m_sampleNumberBuffer, numSamples * sizeof (int64));
+            m_dataTimestampFiles[fileIndex]->increaseRecordCount (numSamples);
+
+            m_dataSyncTimestampFiles[fileIndex]->writeData (timestampBuffer, numSamples * sizeof (double));
+            m_dataSyncTimestampFiles[fileIndex]->increaseRecordCount (numSamples);
+            break;
+        }
+    }
+}
+
 void BinaryRecording::writeTimestampSyncText (uint64 streamId, int64 sampleNumber, float sourceSampleRate, String text)
 {
     if (! m_syncTextFile)

Source/Processors/RecordNode/BinaryFormat/BinaryRecording.h

@@ -62,6 +62,18 @@ class BinaryRecording : public RecordEngine
                               const double* timestampBuffer,
                               int size);
 
+    /** 
+     * Writes continuous data for all channels of a stream in a single batch operation.
+     * Uses SIMD-optimized conversion and cache-efficient interleaving.
+     */
+    void writeContinuousDataBatch (const int* writeChannels,
+                                   const int* realChannels,
+                                   const float* const* dataBuffers,
+                                   const double* timestampBuffer,
+                                   int numChannels,
+                                   int numSamples,
+                                   int fileIndex) override;
+
     /** Writes an event to disk */
     void writeEvent (int eventIndex, const EventPacket& packet);
 
@@ -102,6 +114,14 @@ class BinaryRecording : public RecordEngine
     int m_bufferSize;
     int m_syncTimestampBufferSize;
 
+    // Batch conversion buffers
+    static const int MAX_BATCH_CHANNELS = 512;  // Maximum channels per batch
+    std::vector<float> m_batchScaleFactors;
+    std::vector<int16*> m_batchIntBufferPtrs;
+    HeapBlock<int16> m_batchIntBuffer;  // Separate buffer for batch writes
+    int m_batchBufferSamples { 0 };      // Samples per channel in batch buffer
+    int m_batchBufferChannels { 0 };     // Number of channels in batch buffer
+
     Array<unsigned int> m_channelIndexes;
     Array<unsigned int> m_fileIndexes;
 

Source/Processors/RecordNode/BinaryFormat/SequentialBlockFile.cpp

@@ -129,6 +129,108 @@ bool SequentialBlockFile::writeChannel (uint64 startPos, int channel, int16* dat
     return true;
 }
 
+bool SequentialBlockFile::writeChannelBatch (uint64 startPos, int16* const* channelData, int numChannels, int nSamples)
+{
+    if (! m_file)
+    {
+        printf ("[RN]SequentialBlockFile::writeChannelBatch returned false: (!m_file)\n");
+        return false;
+    }
+
+    if (numChannels != m_nChannels)
+    {
+        printf ("[RN]SequentialBlockFile::writeChannelBatch: channel count mismatch (%d vs %d)\n", 
+                numChannels, m_nChannels);
+        return false;
+    }
+
+    int bIndex = m_memBlocks.size() - 1;
+    if ((bIndex < 0) || (m_memBlocks[bIndex]->getOffset() + m_samplesPerBlock) < (startPos + nSamples))
+        allocateBlocks (startPos, nSamples);
+
+    for (bIndex = m_memBlocks.size() - 1; bIndex >= 0; bIndex--)
+    {
+        if (m_memBlocks[bIndex]->getOffset() <= startPos)
+            break;
+    }
+    if (bIndex < 0)
+    {
+        return false;
+    }
+
+    int writtenSamples = 0;
+    uint64 startIdx = startPos - m_memBlocks[bIndex]->getOffset();
+    int dataIdx = 0;
+    int lastBlockIdx = m_memBlocks.size() - 1;
+
+    // Process in blocks for better cache utilization
+    // Block size chosen to fit in L1 cache (typically 32-64KB)
+    const int cacheBlockSamples = 64;  // Process 64 samples at a time
+
+    // Cache blocking parameters - chosen to fit in L1 cache (~32KB)
+    // For a tile of TILE_SAMPLES x TILE_CHANNELS:
+    // - Input: TILE_CHANNELS pointers (8 bytes each) + TILE_SAMPLES * TILE_CHANNELS * 2 bytes data
+    // - Output: TILE_SAMPLES * nChannels * 2 bytes (but we only write TILE_CHANNELS at a time)
+    // With 256 samples x 64 channels: 256 * 64 * 2 = 32KB input data per tile
+    const int TILE_SAMPLES = 256;
+    const int TILE_CHANNELS = 64;
+
+    while (writtenSamples < nSamples)
+    {
+        int16* blockPtr = m_memBlocks[bIndex]->getData();
+        int samplesToWrite = jmin ((nSamples - writtenSamples), (m_samplesPerBlock - int (startIdx)));
+
+        uint64 baseMemPos = startIdx * m_nChannels;
+
+        // Process in tiles to optimize cache usage
+        // For each tile of samples:
+        for (int sampleTileStart = 0; sampleTileStart < samplesToWrite; sampleTileStart += TILE_SAMPLES)
+        {
+            int sampleTileEnd = jmin (sampleTileStart + TILE_SAMPLES, samplesToWrite);
+
+            // For each tile of channels:
+            for (int channelTileStart = 0; channelTileStart < m_nChannels; channelTileStart += TILE_CHANNELS)
+            {
+                int channelTileEnd = jmin (channelTileStart + TILE_CHANNELS, m_nChannels);
+
+                // Process this tile - iterate samples in outer loop to optimize output writes
+                for (int s = sampleTileStart; s < sampleTileEnd; s++)
+                {
+                    uint64 memPos = baseMemPos + s * m_nChannels + channelTileStart;
+                    int srcIdx = dataIdx + s;
+
+                    // Write channels in this tile for this sample
+                    for (int ch = channelTileStart; ch < channelTileEnd; ch++)
+                    {
+                        blockPtr[memPos + (ch - channelTileStart)] = channelData[ch][srcIdx];
+                    }
+                }
+            }
+        }
+
+        writtenSamples += samplesToWrite;
+        dataIdx += samplesToWrite;
+
+        // Update the last block fill index
+        size_t samplePos = startIdx + samplesToWrite;
+        if (bIndex == lastBlockIdx && samplePos > m_lastBlockFill)
+        {
+            m_lastBlockFill = samplePos;
+        }
+
+        startIdx = 0;
+        bIndex++;
+    }
+
+    // Update current block for all channels
+    for (int ch = 0; ch < m_nChannels; ch++)
+    {
+        m_currentBlock.set (ch, bIndex - 1);
+    }
+
+    return true;
+}
+
 void SequentialBlockFile::allocateBlocks (uint64 startIndex, int numSamples)
 {
     //First deallocate full blocks

Source/Processors/RecordNode/BinaryFormat/SequentialBlockFile.h

@@ -66,6 +66,19 @@ class PLUGIN_API SequentialBlockFile
     /** Writes nSamples of data for a particular channel */
     bool writeChannel (uint64 startPos, int channel, int16* data, int nSamples);
 
+    /** 
+     * Writes data for all channels at once with optimized interleaving.
+     * This is more efficient than calling writeChannel() for each channel separately
+     * because it performs interleaving in a cache-friendly manner.
+     * 
+     * @param startPos     Starting sample position in the file
+     * @param channelData  Array of pointers to int16 data for each channel
+     * @param numChannels  Number of channels (must match m_nChannels)
+     * @param nSamples     Number of samples per channel
+     * @return true if write was successful
+     */
+    bool writeChannelBatch (uint64 startPos, int16* const* channelData, int numChannels, int nSamples);
+
 private:
     std::shared_ptr<FileOutputStream> m_file;
     const int m_nChannels;

Source/Processors/RecordNode/RecordEngine.cpp

@@ -131,6 +131,22 @@ int RecordEngine::getNumRecordedSpikeChannels() const
     return recordNode->getTotalSpikeChannels();
 }
 
+void RecordEngine::writeContinuousDataBatch (const int* writeChannels,
+                                             const int* realChannels,
+                                             const float* const* dataBuffers,
+                                             const double* timestampBuffer,
+                                             int numChannels,
+                                             int numSamples,
+                                             int fileIndex)
+{
+    // Default implementation: fall back to per-channel writes
+    // Subclasses can override for optimized batch processing
+    for (int i = 0; i < numChannels; i++)
+    {
+        writeContinuousData (writeChannels[i], realChannels[i], dataBuffers[i], timestampBuffer, numSamples);
+    }
+}
+
 void RecordEngine::configureEngine()
 {
     if (! manager)

Source/Processors/RecordNode/RecordEngine.h

@@ -98,6 +98,26 @@ class PLUGIN_API RecordEngine
                                       const double* timestampBuffer,
                                       int size) = 0;
 
+    /** 
+     * Write continuous data for all channels of a stream in a single batch operation.
+     * This is more efficient than calling writeContinuousData() for each channel.
+     * 
+     * @param writeChannels   Array of write channel indices (indices among all recorded channels)
+     * @param realChannels    Array of real channel indices (indices within processor)
+     * @param dataBuffers     Array of pointers to float data buffers (one per channel)
+     * @param timestampBuffer Pointer to synchronized timestamp buffer (shared across channels)
+     * @param numChannels     Number of channels in this batch
+     * @param numSamples      Number of samples per channel
+     * @param fileIndex       Index of the file/stream to write to
+     */
+    virtual void writeContinuousDataBatch (const int* writeChannels,
+                                           const int* realChannels,
+                                           const float* const* dataBuffers,
+                                           const double* timestampBuffer,
+                                           int numChannels,
+                                           int numSamples,
+                                           int fileIndex);
+
     /** Write a single event to disk (TTL or TEXT) */
     virtual void writeEvent (int eventChannel, const EventPacket& event) = 0;