flowgraph: improve Oboe flowgraph

There was a problem with disconnecting modules
that could cause high pitched sounds in OboeTester.

This patch simplifies the logic for handling cyclic graphs
and split graphs.

Fixes #902

9 files changed, 42 insertions, 31 deletions

diff --git a/src/common/DataConversionFlowGraph.cpp b/src/common/DataConversionFlowGraph.cpp
index fd6695b3..7ddc0e60 100644
--- a/src/common/DataConversionFlowGraph.cpp
+++ b/src/common/DataConversionFlowGraph.cpp
@@ -209,7 +209,7 @@ int32_t DataConversionFlowGraph::read(void *buffer, int32_t numFrames, int64_t t
     if (mSourceCaller) {
         mSourceCaller->setTimeoutNanos(timeoutNanos);
     }
-    int32_t numRead = mSink->read(mFramePosition, buffer, numFrames);
+    int32_t numRead = mSink->read(buffer, numFrames);
     mFramePosition += numRead;
     return numRead;
 }
@@ -220,7 +220,7 @@ int32_t DataConversionFlowGraph::write(void *inputBuffer, int32_t numFrames) {
     mSource->setData(inputBuffer, numFrames);
     while (true) {
         // Pull and read some data in app format into a small buffer.
-        int32_t framesRead = mSink->read(mFramePosition, mAppBuffer.get(), flowgraph::kDefaultBufferSize);
+        int32_t framesRead = mSink->read(mAppBuffer.get(), flowgraph::kDefaultBufferSize);
         mFramePosition += framesRead;
         if (framesRead <= 0) break;
         // Write to a block adapter, which will call the destination whenever it has enough data.
diff --git a/src/flowgraph/FlowGraphNode.cpp b/src/flowgraph/FlowGraphNode.cpp
index bb6ecc9f..fd267489 100644
--- a/src/flowgraph/FlowGraphNode.cpp
+++ b/src/flowgraph/FlowGraphNode.cpp
@@ -22,23 +22,21 @@
 using namespace flowgraph;
 
 /***************************************************************************/
-int32_t FlowGraphNode::pullData(int64_t framePosition, int32_t numFrames) {
+int32_t FlowGraphNode::pullData(int32_t numFrames, int64_t callCount) {
     int32_t frameCount = numFrames;
     // Prevent recursion and multiple execution of nodes.
-    if (framePosition <= mLastFramePosition && !mBlockRecursion) {
-        mBlockRecursion = true;  // for cyclic graphs
+    if (callCount > mLastCallCount) {
+        mLastCallCount = callCount;
         if (mDataPulledAutomatically) {
             // Pull from all the upstream nodes.
             for (auto &port : mInputPorts) {
                 // TODO fix bug of leaving unused data in some ports if using multiple AudioSource
-                frameCount = port.get().pullData(framePosition, frameCount);
+                frameCount = port.get().pullData(callCount, frameCount);
             }
         }
         if (frameCount > 0) {
             frameCount = onProcess(frameCount);
         }
-        mLastFramePosition += frameCount;
-        mBlockRecursion = false;
         mLastFrameCount = frameCount;
     } else {
         frameCount = mLastFrameCount;
@@ -60,6 +58,7 @@ void FlowGraphNode::pullReset() {
 
 void FlowGraphNode::reset() {
     mLastFrameCount = 0;
+    mLastCallCount = kInitialCallCount;
 }
 
 /***************************************************************************/
@@ -74,9 +73,9 @@ FlowGraphPortFloat::FlowGraphPortFloat(FlowGraphNode &parent,
 }
 
 /***************************************************************************/
-int32_t FlowGraphPortFloatOutput::pullData(int64_t framePosition, int32_t numFrames) {
+int32_t FlowGraphPortFloatOutput::pullData(int64_t callCount, int32_t numFrames) {
     numFrames = std::min(getFramesPerBuffer(), numFrames);
-    return mContainingNode.pullData(framePosition, numFrames);
+    return mContainingNode.pullData(numFrames, callCount);
 }
 
 void FlowGraphPortFloatOutput::pullReset() {
@@ -93,10 +92,10 @@ void FlowGraphPortFloatOutput::disconnect(FlowGraphPortFloatInput *port) {
 }
 
 /***************************************************************************/
-int32_t FlowGraphPortFloatInput::pullData(int64_t framePosition, int32_t numFrames) {
+int32_t FlowGraphPortFloatInput::pullData(int64_t callCount, int32_t numFrames) {
     return (mConnected == nullptr)
             ? std::min(getFramesPerBuffer(), numFrames)
-            : mConnected->pullData(framePosition, numFrames);
+            : mConnected->pullData(callCount, numFrames);
 }
 void FlowGraphPortFloatInput::pullReset() {
     if (mConnected != nullptr) mConnected->pullReset();
@@ -109,3 +108,7 @@ float *FlowGraphPortFloatInput::getBuffer() {
         return mConnected->getBuffer();
     }
 }
+
+int32_t FlowGraphSink::pullData(int32_t numFrames) {
+    return FlowGraphNode::pullData(numFrames, getLastCallCount() + 1);
+}
\ No newline at end of file
diff --git a/src/flowgraph/FlowGraphNode.h b/src/flowgraph/FlowGraphNode.h
index 007131e4..e31cab85 100644
--- a/src/flowgraph/FlowGraphNode.h
+++ b/src/flowgraph/FlowGraphNode.h
@@ -71,15 +71,17 @@ public:
     virtual int32_t onProcess(int32_t numFrames) = 0;
 
     /**
-     * If the framePosition is at or after the last frame position then call onProcess().
+     * If the callCount is at or after the previous callCount then call
+     * pullData on all of the upstreamNodes.
+     * Then call onProcess().
      * This prevents infinite recursion in case of cyclic graphs.
      * It also prevents nodes upstream from a branch from being executed twice.
      *
-     * @param framePosition
+     * @param callCount
      * @param numFrames
      * @return number of frames valid
      */
-    int32_t pullData(int64_t framePosition, int32_t numFrames);
+    int32_t pullData(int32_t numFrames, int64_t callCount);
 
     /**
      * Recursively reset all the nodes in the graph, starting from a Sink.
@@ -118,12 +120,14 @@ public:
         return "FlowGraph";
     }
 
-    int64_t getLastFramePosition() {
-        return mLastFramePosition;
+    int64_t getLastCallCount() {
+        return mLastCallCount;
     }
 
 protected:
-    int64_t  mLastFramePosition = 0;
+
+    static constexpr int64_t  kInitialCallCount = -1;
+    int64_t  mLastCallCount = kInitialCallCount;
 
     std::vector<std::reference_wrapper<FlowGraphPort>> mInputPorts;
 
@@ -394,8 +398,15 @@ public:
         return numFrames;
     }
 
-    virtual int32_t read(int64_t framePosition, void *data, int32_t numFrames) = 0;
+    virtual int32_t read(void *data, int32_t numFrames) = 0;
 
+protected:
+    /**
+     * Pull data through the graph using this nodes last callCount.
+     * @param numFrames
+     * @return
+     */
+    int32_t pullData(int32_t numFrames);
 };
 
 /***************************************************************************/
diff --git a/src/flowgraph/SinkFloat.cpp b/src/flowgraph/SinkFloat.cpp
index f830dafd..20c0ffa0 100644
--- a/src/flowgraph/SinkFloat.cpp
+++ b/src/flowgraph/SinkFloat.cpp
@@ -25,7 +25,7 @@ SinkFloat::SinkFloat(int32_t channelCount)
         : FlowGraphSink(channelCount) {
 }
 
-int32_t SinkFloat::read(int64_t framePosition, void *data, int32_t numFrames) {
+int32_t SinkFloat::read(void *data, int32_t numFrames) {
     // printf("SinkFloat::read(,,%d)\n", numFrames);
     float *floatData = (float *) data;
     int32_t channelCount = input.getSamplesPerFrame();
@@ -33,7 +33,7 @@ int32_t SinkFloat::read(int64_t framePosition, void *data, int32_t numFrames) {
     int32_t framesLeft = numFrames;
     while (framesLeft > 0) {
         // Run the graph and pull data through the input port.
-        int32_t framesPulled = pullData(framePosition, framesLeft);
+        int32_t framesPulled = pullData(framesLeft);
         // printf("SinkFloat::read: framesLeft = %d, framesPulled = %d\n", framesLeft, framesPulled);
         if (framesPulled <= 0) {
             break;
@@ -43,7 +43,6 @@ int32_t SinkFloat::read(int64_t framePosition, void *data, int32_t numFrames) {
         memcpy(floatData, signal, numSamples * sizeof(float));
         floatData += numSamples;
         framesLeft -= framesPulled;
-        framePosition += framesPulled;
     }
     // printf("SinkFloat returning %d\n", numFrames - framesLeft);
     return numFrames - framesLeft;
diff --git a/src/flowgraph/SinkFloat.h b/src/flowgraph/SinkFloat.h
index b6474d18..418f6d7c 100644
--- a/src/flowgraph/SinkFloat.h
+++ b/src/flowgraph/SinkFloat.h
@@ -32,7 +32,7 @@ class SinkFloat : public FlowGraphSink {
 public:
     explicit SinkFloat(int32_t channelCount);
 
-    int32_t read(int64_t framePosition, void *data, int32_t numFrames) override;
+    int32_t read(void *data, int32_t numFrames) override;
 
     const char *getName() override {
         return "SinkFloat";
diff --git a/src/flowgraph/SinkI16.cpp b/src/flowgraph/SinkI16.cpp
index a5904e82..da7fd6b7 100644
--- a/src/flowgraph/SinkI16.cpp
+++ b/src/flowgraph/SinkI16.cpp
@@ -28,14 +28,14 @@ using namespace flowgraph;
 SinkI16::SinkI16(int32_t channelCount)
         : FlowGraphSink(channelCount) {}
 
-int32_t SinkI16::read(int64_t framePosition, void *data, int32_t numFrames) {
+int32_t SinkI16::read(void *data, int32_t numFrames) {
     int16_t *shortData = (int16_t *) data;
     const int32_t channelCount = input.getSamplesPerFrame();
 
     int32_t framesLeft = numFrames;
     while (framesLeft > 0) {
         // Run the graph and pull data through the input port.
-        int32_t framesRead = pullData(framePosition, framesLeft);
+        int32_t framesRead = pullData(framesLeft);
         if (framesRead <= 0) {
             break;
         }
@@ -52,7 +52,6 @@ int32_t SinkI16::read(int64_t framePosition, void *data, int32_t numFrames) {
         }
 #endif
         framesLeft -= framesRead;
-        framePosition += framesRead;
     }
     return numFrames - framesLeft;
 }
diff --git a/src/flowgraph/SinkI16.h b/src/flowgraph/SinkI16.h
index 2cfdfb0c..1e1ce3af 100644
--- a/src/flowgraph/SinkI16.h
+++ b/src/flowgraph/SinkI16.h
@@ -31,7 +31,7 @@ class SinkI16 : public FlowGraphSink {
 public:
     explicit SinkI16(int32_t channelCount);
 
-    int32_t read(int64_t framePosition, void *data, int32_t numFrames) override;
+    int32_t read(void *data, int32_t numFrames) override;
 
     const char *getName() override {
         return "SinkI16";
diff --git a/src/flowgraph/SinkI24.cpp b/src/flowgraph/SinkI24.cpp
index b944b770..a9fb5d22 100644
--- a/src/flowgraph/SinkI24.cpp
+++ b/src/flowgraph/SinkI24.cpp
@@ -30,14 +30,14 @@ using namespace flowgraph;
 SinkI24::SinkI24(int32_t channelCount)
         : FlowGraphSink(channelCount) {}
 
-int32_t SinkI24::read(int64_t framePosition, void *data, int32_t numFrames) {
+int32_t SinkI24::read(void *data, int32_t numFrames) {
     uint8_t *byteData = (uint8_t *) data;
     const int32_t channelCount = input.getSamplesPerFrame();
 
     int32_t framesLeft = numFrames;
     while (framesLeft > 0) {
         // Run the graph and pull data through the input port.
-        int32_t framesRead = pullData(framePosition, framesLeft);
+        int32_t framesRead = pullData(framesLeft);
         if (framesRead <= 0) {
             break;
         }
@@ -61,7 +61,6 @@ int32_t SinkI24::read(int64_t framePosition, void *data, int32_t numFrames) {
         }
 #endif
         framesLeft -= framesRead;
-        framePosition += framesRead;
     }
     return numFrames - framesLeft;
 }
diff --git a/src/flowgraph/SinkI24.h b/src/flowgraph/SinkI24.h
index 7477c8d2..44078a96 100644
--- a/src/flowgraph/SinkI24.h
+++ b/src/flowgraph/SinkI24.h
@@ -32,7 +32,7 @@ class SinkI24 : public FlowGraphSink {
 public:
     explicit SinkI24(int32_t channelCount);
 
-    int32_t read(int64_t framePosition, void *data, int32_t numFrames) override;
+    int32_t read(void *data, int32_t numFrames) override;
 
     const char *getName() override {
         return "SinkI24";