Merge branch 'harp-timestamps-and-recording-optimization' of into lfp-viewer-filter-car`

Author: anjaldoshi

Source/Processors/FileReader/BinaryFileSource/BinaryFileSource.cpp

@@ -306,6 +306,7 @@ void BinaryFileSource::fillRecordInfo()
 
 void BinaryFileSource::processEventData (EventInfo& eventInfo, int64 start, int64 stop)
 {
+    const ScopedLock sl (m_readLock);
     int64 local_start = start % getActiveNumSamples();
     int64 local_stop = stop % getActiveNumSamples();
 
@@ -333,6 +334,7 @@ void BinaryFileSource::processEventData (EventInfo& eventInfo, int64 start, int6
 
 void BinaryFileSource::updateActiveRecord (int index)
 {
+    const ScopedLock sl (m_readLock);
     m_dataFile.reset();
     m_dataFile = std::make_unique<MemoryMappedFile> (m_dataFileArray[index], MemoryMappedFile::readOnly);
     m_samplePos = 0;
@@ -348,11 +350,16 @@ void BinaryFileSource::updateActiveRecord (int index)
 
 void BinaryFileSource::seekTo (int64 sample)
 {
+    const ScopedLock sl (m_readLock);
     m_samplePos = sample % getActiveNumSamples();
 }
 
 int BinaryFileSource::readData (float* buffer, int nSamples)
 {
+    const ScopedLock sl (m_readLock);
+
+    if (m_dataFile == nullptr || buffer == nullptr || nSamples <= 0 || numActiveChannels <= 0)
+        return 0;
     int64 samplesToRead;
 
     if (m_samplePos + nSamples > getActiveNumSamples())
@@ -366,6 +373,9 @@ int BinaryFileSource::readData (float* buffer, int nSamples)
 
     int16* data = static_cast<int16*> (m_dataFile->getData()) + (m_samplePos * numActiveChannels);
 
+    if (samplesToRead <= 0)
+        return 0;
+
     for (int i = 0; i < samplesToRead * numActiveChannels; i++)
     {
         *(buffer + i) = *(data + i) * bitVolts[i % numActiveChannels];

Source/Processors/FileReader/BinaryFileSource/BinaryFileSource.h

@@ -66,6 +66,7 @@ class BinaryFileSource : public FileSource
     void processEventData (EventInfo& info, int64 fromSampleNumber, int64 toSampleNumber) override;
 
 private:
+    CriticalSection m_readLock;
     int numActiveChannels;
     Array<float> bitVolts;
 

Source/Processors/FileReader/FileReader.cpp

@@ -46,6 +46,7 @@ FileReader::FileReader() : GenericProcessor ("File Reader"),
                            startSample (0),
                            stopSample (0),
                            bufferCacheWindow (0),
+                           needsBufferReset (false),
                            m_shouldFillBackBuffer (false),
                            m_bufferSize (1024),
                            m_sysSampleRate (44100),
@@ -381,6 +382,8 @@ void FileReader::setActiveStream (int index, bool reset)
         channelInfo.add (input->getChannelInfo (index, i));
 
     input->seekTo (startSample);
+    firstProcess = false;
+    needsBufferReset.set (false);
 
     updateSettings();
     CoreServices::updateSignalChain (this);
@@ -421,6 +424,15 @@ bool FileReader::startAcquisition()
 
     checkAudioDevice();
 
+    {
+        const ScopedLock sl(bufferLock);
+        readAndFillBufferCache (bufferA);
+        readBuffer = &bufferA;
+        bufferCacheWindow = 0;
+        needsBufferReset.set (false);
+        m_shouldFillBackBuffer.set (true);
+    }
+
     /* Start asynchronous file reading thread */
     startThread();
 
@@ -449,8 +461,10 @@ int64 FileReader::getCurrentSample()
 
 void FileReader::setCurrentSample(int64 sampleNumber)
 {
+    const bool wasThreadRunning = isThreadRunning();
     // Stop background thread before modifying shared state
-    stopThread(100);
+    if (wasThreadRunning)
+        stopThread(100);
 
     const ScopedLock sl(bufferLock);
 
@@ -460,22 +474,34 @@ void FileReader::setCurrentSample(int64 sampleNumber)
     // Reset file position
     input->seekTo(sampleNumber);
 
-    // Get the current back buffer without switching
-    HeapBlock<float>* backBuffer = getBackBuffer();
-
-    // Fill only the back buffer first
-    readAndFillBufferCache(*backBuffer);
+    if (wasThreadRunning)
+    {
+        // Get the current back buffer without switching
+        HeapBlock<float>* backBuffer = getBackBuffer();
+        readAndFillBufferCache(*backBuffer);
+    }
 
-    // Signal that we want to switch buffers on next process() call
-    bufferCacheWindow.set(BUFFER_WINDOW_CACHE_SIZE - 1); // Force buffer switch on next process
-    needsBufferReset.set(true);
+    if (wasThreadRunning)
+    {
+        // Signal that we want to switch buffers on next process() call
+        bufferCacheWindow.set(BUFFER_WINDOW_CACHE_SIZE - 1); // Force buffer switch on next process
+        needsBufferReset.set(true);
 
-    // The process() thread will handle the buffer switch and trigger
-    // the background thread to fill the new back buffer
-    m_shouldFillBackBuffer.set(false);
+        // The process() thread will handle the buffer switch and trigger
+        // the background thread to fill the new back buffer
+        m_shouldFillBackBuffer.set(false);
+    }
+    else
+    {
+        bufferCacheWindow.set(0);
+        needsBufferReset.set(false);
+        readBuffer = &bufferA;
+        m_shouldFillBackBuffer.set(false);
+    }
 
     // Restart background thread
-    startThread();
+    if (wasThreadRunning)
+        startThread();
 }
 
 void FileReader::setPlaybackStart (int64 startSample)
@@ -626,14 +652,9 @@ void FileReader::updateSettings()
     input->seekTo (startSample);
     currentSample = startSample;
 
-    /* Pre-fills the front buffer with a blocking read */
-    readAndFillBufferCache (bufferA);
-
-    readBuffer = &bufferB;
+    readBuffer = &bufferA;
     bufferCacheWindow = 0;
     m_shouldFillBackBuffer.set (false);
-    if (firstProcess)
-        switchBuffer();
 
     LOGD ("File Reader finished updating custom settings.");
 }
@@ -660,11 +681,9 @@ void FileReader::checkAudioDevice()
         input->seekTo (startSample);
         currentSample = startSample;
 
-        /* Pre-fills the front buffer with a blocking read */
-        readAndFillBufferCache (bufferA);
-
-        readBuffer = &bufferB;
+        readBuffer = &bufferA;
         bufferCacheWindow = 0;
+        needsBufferReset.set (false);
         m_shouldFillBackBuffer.set (false);
     }
 }
@@ -846,38 +865,44 @@ void FileReader::readAndFillBufferCache (HeapBlock<float>& cacheBuffer)
 {
     const int samplesNeededPerBuffer = m_samplesPerBuffer.get();
     const int samplesNeeded = samplesNeededPerBuffer * BUFFER_WINDOW_CACHE_SIZE;
+    if (samplesNeeded <= 0 || stopSample <= startSample)
+        return;
 
     int samplesRead = 0;
 
     // should only loop if reached end of file and resuming from start
     while (samplesRead < samplesNeeded)
     {
         int samplesToRead = samplesNeeded - samplesRead;
+        int samplesJustRead = 0;
+
+        const bool wrapsAtEnd = (currentSample + samplesToRead) > stopSample;
 
         // if reached end of file stream
-        if ((currentSample + samplesToRead) > stopSample)
+        if (wrapsAtEnd)
         {
             samplesToRead = int (stopSample - currentSample);
             if (samplesToRead > 0)
-                input->readData (cacheBuffer + samplesRead * currentNumChannels, samplesToRead);
+                samplesJustRead = input->readData (cacheBuffer + samplesRead * currentNumChannels, samplesToRead);
 
             // reset stream to beginning
             input->seekTo (startSample);
             currentSample = startSample;
         }
         else // else read the block needed
         {
-            input->readData (cacheBuffer + samplesRead * currentNumChannels, samplesToRead);
-
-            currentSample += samplesToRead;
+            samplesJustRead = input->readData (cacheBuffer + samplesRead * currentNumChannels, samplesToRead);
+            currentSample += samplesJustRead;
         }
 
-        samplesRead += samplesToRead;
+        samplesRead += samplesJustRead;
 
         //LOGD("CURRENT SAMPLE: ", currentSample, " samplesRead: ", samplesRead, " samplesNeeded: ", samplesNeeded);
-
         if (samplesRead < 0)
             return;
+
+        if (samplesJustRead <= 0 && ! wrapsAtEnd)
+            return;
     }
 }
 

Source/Processors/RecordNode/BinaryFormat/BinaryRecording.cpp

@@ -97,8 +97,6 @@ void BinaryRecording::openFiles (File rootFolder, int experimentNumber, int reco
 
         if (streamId != lastStreamId)
         {
-            wroteFirstSampleNumber[streamId] = false;
-
             firstChannels.add (channelInfo);
             streamIndex++;
 
@@ -678,12 +676,6 @@ void BinaryRecording::writeContinuousData (int writeChannel,
 
         uint32 streamId = getContinuousChannel (realChannel)->getStreamId();
 
-        if (! wroteFirstSampleNumber[streamId])
-        {
-            firstSampleNumber[streamId] = baseSampleNumber;
-            wroteFirstSampleNumber[streamId] = true;
-        }
-
         /* Generate sequential sample numbers using SIMD-optimized fill */
         SIMDConverter::fillSequentialInt64 (reinterpret_cast<int64_t*> (m_sampleNumberBuffer.getData()), baseSampleNumber, size);
 
@@ -792,6 +784,7 @@ void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
                                                 int numSamples,
                                                 int fileIndex)
 {
+    ignoreUnused (fileIndex);
     if (numSamples == 0 || numChannels == 0)
         return;
 
@@ -826,8 +819,12 @@ void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
         m_batchIntBufferPtrs.resize (numChannels);
     }
 
+    // Resolve the file index from recorded channel mapping.
+    // This is robust even when some source streams have zero recorded channels.
+    const int resolvedFileIndex = m_fileIndexes[writeChannels[0]];
+
     // Get file and validate
-    if (fileIndex < 0 || fileIndex >= m_continuousFiles.size() || ! m_continuousFiles[fileIndex])
+    if (resolvedFileIndex < 0 || resolvedFileIndex >= m_continuousFiles.size() || ! m_continuousFiles[resolvedFileIndex])
         return;
 
     // Setup scale factors and output buffer pointers for each channel
@@ -849,12 +846,12 @@ void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
     // Get starting sample position (all channels in a stream have same position)
     uint64 startPos = m_samplesWritten[writeChannels[0]];
 
-    //LOGD("BinaryRecording::writeContinuousDataBatch: Writing ", numSamples, " samples for ", numChannels, " channels at position ", startPos, " to file index ", fileIndex);
+    //LOGD("BinaryRecording::writeContinuousDataBatch: Writing ", numSamples, " samples for ", numChannels, " channels at position ", startPos, " to file index ", resolvedFileIndex);
 
     // Try batch interleaving if we have all channels for this file
     // The file's channel count is determined by the stream's channel count
     // If we have a partial batch, fall back to per-channel writes
-    bool useBatchWrite = m_continuousFiles[fileIndex] != nullptr && m_continuousFiles[fileIndex]->writeChannelBatch (startPos, m_batchIntBufferPtrs.data(), numChannels, numSamples);
+    bool useBatchWrite = m_continuousFiles[resolvedFileIndex] != nullptr && m_continuousFiles[resolvedFileIndex]->writeChannelBatch (startPos, m_batchIntBufferPtrs.data(), numChannels, numSamples);
 
     if (! useBatchWrite)
     {
@@ -863,7 +860,7 @@ void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
         {
             int writeChannel = writeChannels[i];
             int channelIdx = m_channelIndexes[writeChannel];
-            m_continuousFiles[fileIndex]->writeChannel (
+            m_continuousFiles[resolvedFileIndex]->writeChannel (
                 startPos,
                 channelIdx,
                 m_batchIntBufferPtrs[i],
@@ -886,20 +883,14 @@ void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
             int64 baseSampleNumber = getLatestSampleNumber (writeChannels[i]);
             uint32 streamId = getContinuousChannel (realChannels[i])->getStreamId();
 
-            if (! wroteFirstSampleNumber[streamId])
-            {
-                firstSampleNumber[streamId] = baseSampleNumber;
-                wroteFirstSampleNumber[streamId] = true;
-            }
-
             /* Generate sequential sample numbers using SIMD-optimized fill */
             SIMDConverter::fillSequentialInt64 (reinterpret_cast<int64_t*> (m_sampleNumberBuffer.getData()), baseSampleNumber, numSamples);
 
-            m_dataTimestampFiles[fileIndex]->writeData (m_sampleNumberBuffer, numSamples * sizeof (int64));
-            m_dataTimestampFiles[fileIndex]->increaseRecordCount (numSamples);
+            m_dataTimestampFiles[resolvedFileIndex]->writeData (m_sampleNumberBuffer, numSamples * sizeof (int64));
+            m_dataTimestampFiles[resolvedFileIndex]->increaseRecordCount (numSamples);
 
-            m_dataSyncTimestampFiles[fileIndex]->writeData (timestampBuffer, numSamples * sizeof (double));
-            m_dataSyncTimestampFiles[fileIndex]->increaseRecordCount (numSamples);
+            m_dataSyncTimestampFiles[resolvedFileIndex]->writeData (timestampBuffer, numSamples * sizeof (double));
+            m_dataSyncTimestampFiles[resolvedFileIndex]->increaseRecordCount (numSamples);
             break;
         }
     }
@@ -913,13 +904,7 @@ void BinaryRecording::writeTimestampSyncText (uint64 streamId, int64 sampleNumbe
     String syncString = text + ": " + String (sampleNumber);
     LOGD (syncString);
 
-    int64 fsn = firstSampleNumber[streamId];
-
-    if (streamId > 0)
-        jassert (fsn == sampleNumber);
-
     m_syncTextFile->writeText (syncString + "\r\n", false, false, nullptr);
-    // Note: flush removed - file will be flushed on close or by OS buffering
 }
 
 RecordEngineManager* BinaryRecording::getEngineManager()

Source/Processors/RecordNode/BinaryFormat/BinaryRecording.h

@@ -141,9 +141,6 @@ class BinaryRecording : public RecordEngine
     int m_experimentNum;
     Array<int64> m_samplesWritten;
 
-    std::map<uint64, int64> firstSampleNumber;
-    std::map<uint64, bool> wroteFirstSampleNumber;
-
     const int samplesPerBlock { 8192 };  // Larger blocks reduce memory block operations
 };
 #endif

Source/Processors/RecordNode/BinaryFormat/SIMDConverter.h

@@ -31,6 +31,7 @@
 
 #include <cstdint>
 #include <string>
+#include "../../../TestableExport.h"
 
 /**
  * SIMD-optimized float-to-int16 conversion utilities.
@@ -47,7 +48,7 @@
  * (FloatVectorOperations::copyWithMultiply + AudioDataConverters::convertFloatToInt16LE)
  * into a single pass, eliminating the intermediate buffer and improving cache utilization.
  */
-class SIMDConverter
+class TESTABLE SIMDConverter
 {
 public:
     /**