Add async file chunk reading to FileReader

Author: kmichaelfox

Source/Plugins/KWIKFormat/FileSource/KwikFileSource.cpp

@@ -172,7 +172,7 @@ void KWIKFileSource::updateActiveRecord()
     samplePos=0;
     try
     {
-        String path = "/recordings/" + String(availableDataSets[activeRecord]) + "/data";
+        String path = "/recordings/" + String(availableDataSets[activeRecord.get()]) + "/data";
         dataSet = new DataSet(sourceFile->openDataSet(path.toUTF8()));
     }
     catch (FileIException error)
@@ -275,4 +275,4 @@ bool KWIKFileSource::isReady()
 	}
 	else
 		return true;
-}
+}

Source/Processors/FileReader/FileReader.cpp

@@ -30,6 +30,7 @@
 
 FileReader::FileReader()
     : GenericProcessor ("File Reader")
+    , Thread ("filereader_Async_Reader")
     , timestamp             (0)
     , currentSampleRate     (0)
     , currentNumChannels    (0)
@@ -39,6 +40,7 @@ FileReader::FileReader()
     , stopSample            (0)
     , counter               (0)
     , bufferCacheWindow     (0)
+    , m_shouldFillBackBuffer(false)
 {
     setProcessorType (PROCESSOR_TYPE_SOURCE);
 
@@ -63,6 +65,8 @@ FileReader::FileReader()
 
 FileReader::~FileReader()
 {
+    signalThreadShouldExit();
+    notify();
 }
 
 
@@ -180,13 +184,19 @@ bool FileReader::setFile (String fullpath)
 
     static_cast<FileReaderEditor*> (getEditor())->populateRecordings (input);
     setActiveRecording (0);
+    
+    m_samplesPerBuffer.set(float(BUFFER_SIZE) * (getDefaultSampleRate() / 44100.0f));
+    
+    readAndFillBufferCache(bufferA); // pre-fill the front buffer with a blocking read
+    
+    startThread(); // start async file reader thread
 
     return true;
 }
 
 
 void FileReader::setActiveRecording (int index)
-{
+{    
     input->setActiveRecord (index);
 
     currentNumChannels  = input->getActiveNumChannels();
@@ -205,7 +215,13 @@ void FileReader::setActiveRecording (int index)
 
     static_cast<FileReaderEditor*> (getEditor())->setTotalTime (samplesToMilliseconds (currentNumSamples));
 
-    readBuffer.malloc (currentNumChannels * BUFFER_SIZE * BUFFER_WINDOW_CACHE_SIZE);
+    bufferA.malloc (currentNumChannels * BUFFER_SIZE * BUFFER_WINDOW_CACHE_SIZE);
+    bufferB.malloc (currentNumChannels * BUFFER_SIZE * BUFFER_WINDOW_CACHE_SIZE);
+    
+    // set the backbuffer so that on the next call to process() we start with bufferA and buffer
+    // cache window id = 0
+    readBuffer = &bufferB;
+    bufferCacheWindow = 0;
 }
 
 
@@ -233,47 +249,20 @@ void FileReader::updateSettings()
 void FileReader::process (AudioSampleBuffer& buffer)
 {
     const int samplesNeededPerBuffer = int (float (buffer.getNumSamples()) * (getDefaultSampleRate() / 44100.0f));
-    const int samplesNeeded = samplesNeededPerBuffer * BUFFER_WINDOW_CACHE_SIZE;
+    m_samplesPerBuffer.set(samplesNeededPerBuffer);
     // FIXME: needs to account for the fact that the ratio might not be an exact
     //        integer value
 
-    int samplesRead = 0;
-    
-    // if window id == 0, we need to read and cache BUFFER_WINDOW_CACHE_SIZE more buffer windows
+    // if cache window id == 0, we need to read and cache BUFFER_WINDOW_CACHE_SIZE more buffer windows
     if (bufferCacheWindow == 0)
     {
-        
-        // should only loop if reached end of file and resuming from start
-        while (samplesRead < samplesNeeded)
-        {
-            int samplesToRead = samplesNeeded - samplesRead;
-            
-            // if reached end of file stream
-            if ( (currentSample + samplesToRead) > stopSample)
-            {
-                samplesToRead = stopSample - currentSample;
-                if (samplesToRead > 0)
-                    input->readData (readBuffer + samplesRead * currentNumChannels, samplesToRead);
-                
-                // reset stream to beginning
-                input->seekTo (startSample);
-                currentSample = startSample;
-            }
-            else // else read the block needed
-            {
-                input->readData (readBuffer + samplesRead * currentNumChannels, samplesToRead);
-                
-                currentSample += samplesToRead;
-            }
-            
-            samplesRead += samplesToRead;
-        }
+        switchBuffer();
     }
 
     for (int i = 0; i < currentNumChannels; ++i)
     {
         // offset readBuffer index by current cache window count * buffer window size * num channels
-        input->processChannelData (readBuffer + (samplesNeededPerBuffer * currentNumChannels * bufferCacheWindow),
+        input->processChannelData (*readBuffer + (samplesNeededPerBuffer * currentNumChannels * bufferCacheWindow),
                                    buffer.getWritePointer (i, 0),
                                    i,
                                    samplesNeededPerBuffer);
@@ -325,3 +314,73 @@ int64 FileReader::millisecondsToSamples (unsigned int ms) const
 {
     return (int64) (currentSampleRate * float (ms) / 1000.f);
 }
+
+void FileReader::switchBuffer()
+{
+    if (readBuffer == &bufferA)
+        readBuffer = &bufferB;
+    else
+        readBuffer = &bufferA;
+    
+    m_shouldFillBackBuffer.set(true);
+    notify();
+}
+
+HeapBlock<int16>* FileReader::getFrontBuffer()
+{
+    return readBuffer;
+}
+
+HeapBlock<int16>* FileReader::getBackBuffer()
+{
+    if (readBuffer == &bufferA) return &bufferB;
+    
+    return &bufferA;
+}
+
+void FileReader::run()
+{
+    while (!threadShouldExit())
+    {
+        if (m_shouldFillBackBuffer.compareAndSetBool(false, true))
+        {
+            readAndFillBufferCache(*getBackBuffer());
+        }
+        
+        wait(30);
+    }
+}
+
+void FileReader::readAndFillBufferCache(HeapBlock<int16> &cacheBuffer)
+{
+    const int samplesNeededPerBuffer = m_samplesPerBuffer.get();
+    const int samplesNeeded = samplesNeededPerBuffer * BUFFER_WINDOW_CACHE_SIZE;
+    
+    int samplesRead = 0;
+    
+    // should only loop if reached end of file and resuming from start
+    while (samplesRead < samplesNeeded)
+    {
+        int samplesToRead = samplesNeeded - samplesRead;
+        
+        // if reached end of file stream
+        if ( (currentSample + samplesToRead) > stopSample)
+        {
+            samplesToRead = stopSample - currentSample;
+            if (samplesToRead > 0)
+                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;
+        }
+        
+        samplesRead += samplesToRead;
+    }
+}

Source/Processors/FileReader/FileReader.h

@@ -32,15 +32,16 @@
 #include "FileSource.h"
 
 #define BUFFER_SIZE 1024
-#define BUFFER_WINDOW_CACHE_SIZE 3
+#define BUFFER_WINDOW_CACHE_SIZE 10
 
 
 /**
   Reads data from a file.
 
   @see GenericProcessor
 */
-class FileReader : public GenericProcessor
+class FileReader : public GenericProcessor,
+    private Thread
 {
 public:
     FileReader();
@@ -92,9 +93,30 @@ class FileReader : public GenericProcessor
 
     ScopedPointer<FileSource> input;
 
-    HeapBlock<int16> readBuffer;
+    HeapBlock<int16> * readBuffer;      // Ptr to the current "front" buffer
+    HeapBlock<int16> bufferA;
+    HeapBlock<int16> bufferB;
 
     HashMap<String, int> supportedExtensions;
+    
+    Atomic<int> m_shouldFillBackBuffer;
+    Atomic<int> m_samplesPerBuffer;
+    
+    /** Swaps the backbuffer to the front and flags the background reader
+        thread to update the new backbuffer */
+    void switchBuffer();
+    
+    HeapBlock<int16>* getFrontBuffer();
+    HeapBlock<int16>* getBackBuffer();
+    
+    /** Executes the background thread task */
+    void run() override;
+    
+    /** Reads a chunk of the file that fills an entire buffer cache.
+     
+        This method will read into the buffer that passed in by the param 
+     */
+    void readAndFillBufferCache(HeapBlock<int16> &cacheBuffer);
 
 
     JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR(FileReader);

Source/Processors/FileReader/FileSource.cpp

@@ -58,7 +58,7 @@ int FileSource::getRecordNumChannels (int index) const
 
 int FileSource::getActiveNumChannels() const
 {
-    return getRecordNumChannels (activeRecord);
+    return getRecordNumChannels (activeRecord.get());
 }
 
 
@@ -70,7 +70,7 @@ float FileSource::getRecordSampleRate (int index) const
 
 float FileSource::getActiveSampleRate() const
 {
-    return getRecordSampleRate (activeRecord);
+    return getRecordSampleRate (activeRecord.get());
 }
 
 
@@ -82,13 +82,13 @@ int64 FileSource::getRecordNumSamples (int index) const
 
 int64 FileSource::getActiveNumSamples() const
 {
-    return getRecordNumSamples (activeRecord);
+    return getRecordNumSamples (activeRecord.get());
 }
 
 
 int FileSource::getActiveRecord() const
 {
-    return activeRecord;
+    return activeRecord.get();
 }
 
 
@@ -100,13 +100,14 @@ RecordedChannelInfo FileSource::getChannelInfo (int recordIndex, int channel) co
 
 RecordedChannelInfo FileSource::getChannelInfo (int channel) const
 {
-    return getChannelInfo (activeRecord, channel);
+    return getChannelInfo (activeRecord.get(), channel);
 }
 
 
 void FileSource::setActiveRecord (int index)
 {
-    activeRecord = index;
+//    activeRecord = index;
+    activeRecord.set(index);
     updateActiveRecord();
 }
 

Source/Processors/FileReader/FileSource.h

@@ -81,7 +81,7 @@ class PLUGIN_API FileSource
 
     bool fileOpened;
     int numRecords;
-    int activeRecord;
+    Atomic<int> activeRecord;       // atomic to protect against threaded data race in FileReader
     String filename;