WaveRecorder: fix hang in StreamingThread (#106)

The read() could hang forever waiting for data when the
WaveRecorder was stopped.
Now it terminates the thread.

Also fix bug in AudioFifo when partially full. It would not read any data!

Fixes #105

5 files changed, 118 insertions, 29 deletions

diff --git a/src/main/java/com/jsyn/io/AudioFifo.java b/src/main/java/com/jsyn/io/AudioFifo.java
index 0c563e4..b7321c6 100644
--- a/src/main/java/com/jsyn/io/AudioFifo.java
+++ b/src/main/java/com/jsyn/io/AudioFifo.java
@@ -37,6 +37,7 @@ public class AudioFifo implements AudioInputStream, AudioOutputStream {
     private int sizeMask;
     private boolean writeWaitEnabled = true;
     private boolean readWaitEnabled = true;
+    private volatile boolean mOpen = true;
     final Lock lock = new ReentrantLock();
     final Condition notFull  = lock.newCondition();
     final Condition notEmpty = lock.newCondition();
@@ -69,7 +70,12 @@ public class AudioFifo implements AudioInputStream, AudioOutputStream {
 
     @Override
     public void close() {
-        // TODO Maybe we should tell any thread that is waiting that the FIFO is closed.
+        // Tell any thread that is waiting that the FIFO is closed.
+        mOpen = false;
+        lock.lock();
+        notEmpty.signal();
+        notFull.signal();
+        lock.unlock();
     }
 
     @Override
@@ -78,20 +84,22 @@ public class AudioFifo implements AudioInputStream, AudioOutputStream {
         if (readWaitEnabled) {
             lock.lock();
             try {
-              while (available() < 1) {
+              while (mOpen && available() < 1) {
                 try {
                     notEmpty.await();
                 } catch (InterruptedException e) {
                     return Double.NaN;
                 }
               }
-              value = readOneInternal();
+              if (mOpen) {
+                  value = readOneInternal();
+              }
             } finally {
               lock.unlock();
             }
 
         } else {
-            if (readIndex != writeIndex) {
+            if (mOpen && readIndex != writeIndex) {
                 value = readOneInternal();
             }
         }
@@ -116,7 +124,7 @@ public class AudioFifo implements AudioInputStream, AudioOutputStream {
         if (writeWaitEnabled) {
             lock.lock();
             try {
-                while (available() == buffer.length)
+                while (mOpen && available() == buffer.length)
                 {
                     try {
                         notFull.await();
@@ -124,7 +132,9 @@ public class AudioFifo implements AudioInputStream, AudioOutputStream {
                         return; // Silently fail
                     }
                 }
-                writeOneInternal(value);
+                if (mOpen) {
+                    writeOneInternal(value);
+                }
             } finally {
                 lock.unlock();
             }
@@ -154,20 +164,20 @@ public class AudioFifo implements AudioInputStream, AudioOutputStream {
 
     @Override
     public int read(double[] buffer, int start, int count) {
-        if (readWaitEnabled) {
-            for (int i = 0; i < count; i++) {
-                buffer[i + start] = read();
-            }
-        } else {
-            if (available() < count) {
-                count = available();
-            } else {
-                for (int i = 0; i < count; i++) {
-                    buffer[i + start] = read();
-                }
-            }
+        if (!mOpen) {
+            return 0;
+        }
+        if (!readWaitEnabled) {
+            count = Math.min(available(), count);
+        }
+        int numRead = 0;
+        for (int i = 0; mOpen && i < count; i++) {
+            double value = read();
+            if (value == Double.NaN) break;
+            buffer[i + start] = value;
+            numRead++;
         }
-        return count;
+        return numRead;
     }
 
     @Override
diff --git a/src/main/java/com/jsyn/util/StreamingThread.java b/src/main/java/com/jsyn/util/StreamingThread.java
index 682f476..7377698 100644
--- a/src/main/java/com/jsyn/util/StreamingThread.java
+++ b/src/main/java/com/jsyn/util/StreamingThread.java
@@ -4,9 +4,9 @@
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
- * 
+ *
  *     http://www.apache.org/licenses/LICENSE-2.0
- * 
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -23,7 +23,7 @@ import com.jsyn.io.AudioOutputStream;
 
 /**
  * Read from an AudioInputStream and write to an AudioOutputStream as a background thread.
- * 
+ *
  * @author Phil Burk (C) 2011 Mobileer Inc
  */
 public class StreamingThread extends Thread {
@@ -47,17 +47,18 @@ public class StreamingThread extends Thread {
         try {
             transportModel.firePositionChanged(framePosition);
             transportModel.fireStateChanged(TransportModel.STATE_RUNNING);
-            int framesToRead = geteFramesToRead(buffer);
+            int framesToRead = getFramesToRead(buffer);
             while (go && (framesToRead > 0)) {
                 int samplesToRead = framesToRead * samplesPerFrame;
                 while (samplesToRead > 0) {
                     int samplesRead = inputStream.read(buffer, 0, samplesToRead);
                     outputStream.write(buffer, 0, samplesRead);
                     samplesToRead -= samplesRead;
+                    if (samplesRead < samplesToRead) break; // stream closed
                 }
                 framePosition += framesToRead;
                 transportModel.firePositionChanged(framePosition);
-                framesToRead = geteFramesToRead(buffer);
+                framesToRead = getFramesToRead(buffer);
             }
             transportModel.fireStateChanged(TransportModel.STATE_STOPPED);
         } catch (IOException e) {
@@ -65,7 +66,7 @@ public class StreamingThread extends Thread {
         }
     }
 
-    private int geteFramesToRead(double[] buffer) {
+    private int getFramesToRead(double[] buffer) {
         if (maxFrames > 0) {
             long numToRead = maxFrames - framePosition;
             if (numToRead < 0) {
@@ -85,7 +86,7 @@ public class StreamingThread extends Thread {
 
     /**
      * Only call this before the thread has started.
-     * 
+     *
      * @param framesPerBuffer
      */
     public void setFramesPerBuffer(int framesPerBuffer) {
diff --git a/src/main/java/com/jsyn/util/WaveRecorder.java b/src/main/java/com/jsyn/util/WaveRecorder.java
index 8008d1d..d189f06 100644
--- a/src/main/java/com/jsyn/util/WaveRecorder.java
+++ b/src/main/java/com/jsyn/util/WaveRecorder.java
@@ -24,7 +24,8 @@ import com.jsyn.Synthesizer;
 import com.jsyn.ports.UnitInputPort;
 
 /**
- * Connect a unit generator to the input. Then start() recording. The signal will be written to a
+ * Connect a unit generator to the input. Then start() recording.
+ * The signal will be written to a
  * WAV format file that can be read by other programs.
  *
  * @author Phil Burk (C) 2011 Mobileer Inc
@@ -60,7 +61,8 @@ public class WaveRecorder {
      * @param bitsPerSample 16 or 24
      * @throws FileNotFoundException
      */
-    public WaveRecorder(Synthesizer synth, File outputFile, int samplesPerFrame, int bitsPerSample)
+    public WaveRecorder(Synthesizer synth, File outputFile,
+            int samplesPerFrame, int bitsPerSample)
             throws FileNotFoundException {
         this.synth = synth;
         reader = new AudioStreamReader(synth, samplesPerFrame);
@@ -86,10 +88,12 @@ public class WaveRecorder {
 
     public void stop() {
         if (thread != null) {
+            reader.close();
             thread.requestStop();
             try {
                 thread.join(500);
             } catch (InterruptedException ignored) {
+                System.out.println("join() " + ignored);
             }
             thread = null;
         }
diff --git a/src/test/java/com/jsyn/engine/TestFifo.java b/src/test/java/com/jsyn/engine/TestFifo.java
index 1c681fb..3c57053 100644
--- a/src/test/java/com/jsyn/engine/TestFifo.java
+++ b/src/test/java/com/jsyn/engine/TestFifo.java
@@ -34,7 +34,7 @@ public class TestFifo {
         fifo.allocate(8);
         assertEquals(0, fifo.available(), "start empty");
 
-        assertEquals(Double.NaN, fifo.read(), "read back Nan when emopty");
+        assertEquals(Double.NaN, fifo.read(), "read back Nan when empty");
 
         fifo.write(1.0);
         assertEquals(1, fifo.available(), "added one value");
@@ -52,6 +52,29 @@ public class TestFifo {
         watchdog.interrupt();
     }
 
+    @Test
+    public void testReadAvailable() {
+        Thread watchdog = startWatchdog(600);
+
+        AudioFifo fifo = new AudioFifo();
+        fifo.setReadWaitEnabled(false);
+        fifo.allocate(8);
+
+        final double x1 = 0.234;
+        fifo.write(x1);
+        final double x2 = 0.750;
+        fifo.write(x2);
+        assertEquals(2, fifo.available(), "added two values");
+
+        double[] buffer = new double[6];
+        int numRead = fifo.read(buffer, 2, buffer.length - 2);
+        assertEquals(2, numRead, "should have read two values");
+        assertEquals(x1, buffer[2], "read x1");
+        assertEquals(x2, buffer[3], "read x2");
+
+        watchdog.interrupt();
+    }
+
     /**
      * Wrap around several times to test masking.
      */
diff --git a/src/test/java/com/jsyn/util/DebugWaveRecorderFinish.java b/src/test/java/com/jsyn/util/DebugWaveRecorderFinish.java
new file mode 100644
index 0000000..c30be88
--- /dev/null
+++ b/src/test/java/com/jsyn/util/DebugWaveRecorderFinish.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2022 Phil Burk, Mobileer Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.jsyn.util;
+
+import com.jsyn.util.WaveRecorder;
+
+import com.jsyn.JSyn;
+import com.jsyn.Synthesizer;
+import com.jsyn.instruments.DualOscillatorSynthVoice;
+import com.softsynth.shared.time.TimeStamp;
+
+/**
+ * Debug the thread not finishing.
+ * Contributed by Zoran Stojanović on GitHub.
+ */
+public class DebugWaveRecorderFinish {
+    public static void main(String[] args) throws Exception {
+        System.out.println("Started!");
+        Synthesizer synth = JSyn.createSynthesizer();
+        synth.setRealTime(false);
+        DualOscillatorSynthVoice voice = new DualOscillatorSynthVoice();
+        synth.add(voice);
+        WaveRecorder recorder = new WaveRecorder(synth, new java.io.File("test.wav"), 1);
+        voice.getOutput().connect(0, recorder.getInput(), 0);
+        synth.start();
+        recorder.start();
+
+        voice.noteOn(440, 1, new TimeStamp(0));
+        voice.noteOff(new TimeStamp(3));
+        synth.sleepUntil(4.0);
+
+        recorder.stop();
+        recorder.close();
+        synth.stop();
+        System.out.println("Finished!");
+    }
+}