Queue buffers when auto demand is low enough

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## 1.0.1
  * Specify the order in which state fields will be printed in the error logs. [#614](https://github.com/membraneframework/membrane_core/pull/614)
+ * Handle buffers from input pads having `flow_control: :auto` only if demand on all output pads having `flow_control: :auto` is positive. [#693](https://github.com/membraneframework/membrane_core/pull/693)
  * Fix clock selection [#626](https://github.com/membraneframework/membrane_core/pull/626)
  * Log messages in the default handle_info implementation [#680](https://github.com/membraneframework/membrane_core/pull/680)
  * Fix typespecs in Membrane.UtilitySupervisor [#681](https://github.com/membraneframework/membrane_core/pull/681)

lib/membrane/core/child/pad_model.ex

@@ -42,7 +42,6 @@ defmodule Membrane.Core.Child.PadModel do
           input_queue: Membrane.Core.Element.InputQueue.t() | nil,
           options: %{optional(atom) => any},
           auto_demand_size: pos_integer() | nil,
-          associated_pads: [Pad.ref()] | nil,
           sticky_events: [Membrane.Event.t()],
           stalker_metrics: %{atom => :atomics.atomics_ref()}
         }

lib/membrane/core/element.ex

@@ -158,8 +158,11 @@ defmodule Membrane.Core.Element do
         setup_incomplete?: false,
         effective_flow_control: :push,
         handling_action?: false,
+        popping_queue?: false,
         pads_to_snapshot: MapSet.new(),
-        stalker: options.stalker
+        stalker: options.stalker,
+        satisfied_auto_output_pads: MapSet.new(),
+        awaiting_auto_input_pads: MapSet.new()
       }
       |> PadSpecHandler.init_pads()
 

lib/membrane/core/element/action_handler.ex

@@ -24,11 +24,11 @@ defmodule Membrane.Core.Element.ActionHandler do
   alias Membrane.Core.Element.{
     DemandController,
     DemandHandler,
-    PadController,
     State,
     StreamFormatController
   }
 
+  alias Membrane.Core.Element.DemandController.AutoFlowUtils
   alias Membrane.Core.{Events, TimerController}
   alias Membrane.Element.Action
 
@@ -466,8 +466,9 @@ defmodule Membrane.Core.Element.ActionHandler do
   @spec handle_outgoing_event(Pad.ref(), Event.t(), State.t()) :: State.t()
   defp handle_outgoing_event(pad_ref, %Events.EndOfStream{}, state) do
     with %{direction: :output, end_of_stream?: false} <- PadModel.get_data!(state, pad_ref) do
-      state = PadController.remove_pad_associations(pad_ref, state)
-      PadModel.set_data!(state, pad_ref, :end_of_stream?, true)
+      Map.update!(state, :satisfied_auto_output_pads, &MapSet.delete(&1, pad_ref))
+      |> PadModel.set_data!(pad_ref, :end_of_stream?, true)
+      |> AutoFlowUtils.pop_queues_and_bump_demand()
     else
       %{direction: :input} ->
         raise PadDirectionError, action: "end of stream", direction: :input, pad: pad_ref

lib/membrane/core/element/atomic_demand.ex

@@ -139,14 +139,14 @@ defmodule Membrane.Core.Element.AtomicDemand do
     :ok
   end
 
-  @spec decrease(t, non_neg_integer()) :: t
+  @spec decrease(t, non_neg_integer()) :: {{:decreased, integer()}, t} | {:unchanged, t}
   def decrease(%__MODULE__{} = atomic_demand, value) do
     atomic_demand = Map.update!(atomic_demand, :buffered_decrementation, &(&1 + value))
 
     if atomic_demand.buffered_decrementation >= atomic_demand.throttling_factor do
       flush_buffered_decrementation(atomic_demand)
     else
-      atomic_demand
+      {:unchanged, atomic_demand}
     end
   end
 
@@ -164,14 +164,17 @@ defmodule Membrane.Core.Element.AtomicDemand do
 
     atomic_demand = %{atomic_demand | buffered_decrementation: 0}
 
-    if not atomic_demand.toilet_overflowed? and
-         get_receiver_status(atomic_demand) == {:resolved, :pull} and
-         get_sender_status(atomic_demand) == {:resolved, :push} and
-         -1 * atomic_demand_value > atomic_demand.toilet_capacity do
-      overflow(atomic_demand, atomic_demand_value)
-    else
-      atomic_demand
-    end
+    atomic_demand =
+      if not atomic_demand.toilet_overflowed? and
+           get_receiver_status(atomic_demand) == {:resolved, :pull} and
+           get_sender_status(atomic_demand) == {:resolved, :push} and
+           -1 * atomic_demand_value > atomic_demand.toilet_capacity do
+        overflow(atomic_demand, atomic_demand_value)
+      else
+        atomic_demand
+      end
+
+    {{:decreased, atomic_demand_value}, atomic_demand}
   end
 
   defp overflow(atomic_demand, atomic_demand_value) do

lib/membrane/core/element/buffer_controller.ex

@@ -69,8 +69,17 @@ defmodule Membrane.Core.Element.BufferController do
     state = PadModel.set_data!(state, pad_ref, :demand, demand - buf_size)
     :atomics.put(stalker_metrics.demand, 1, demand - buf_size)
 
-    state = AutoFlowUtils.auto_adjust_atomic_demand(pad_ref, state)
-    exec_buffer_callback(pad_ref, buffers, state)
+    if state.effective_flow_control == :pull and MapSet.size(state.satisfied_auto_output_pads) > 0 do
+      AutoFlowUtils.store_buffers_in_queue(pad_ref, buffers, state)
+    else
+      if MapSet.member?(state.awaiting_auto_input_pads, pad_ref) or
+           PadModel.get_data!(state, pad_ref, [:auto_flow_queue]) != Qex.new() do
+        raise "cannot execute handle_buffer callback for an awaiting input pad"
+      end
+
+      state = exec_buffer_callback(pad_ref, buffers, state)
+      AutoFlowUtils.auto_adjust_atomic_demand(pad_ref, state)
+    end
   end
 
   defp do_handle_buffer(pad_ref, %{flow_control: :manual} = data, buffers, state) do
@@ -93,11 +102,7 @@ defmodule Membrane.Core.Element.BufferController do
   @doc """
   Executes `handle_buffer` callback.
   """
-  @spec exec_buffer_callback(
-          Pad.ref(),
-          [Buffer.t()] | Buffer.t(),
-          State.t()
-        ) :: State.t()
+  @spec exec_buffer_callback(Pad.ref(), [Buffer.t()], State.t()) :: State.t()
   def exec_buffer_callback(pad_ref, buffers, %State{type: :filter} = state) do
     Telemetry.report_metric("buffer", 1, inspect(pad_ref))
 

lib/membrane/core/element/demand_controller.ex

@@ -23,13 +23,20 @@ defmodule Membrane.Core.Element.DemandController do
 
   @spec snapshot_atomic_demand(Pad.ref(), State.t()) :: State.t()
   def snapshot_atomic_demand(pad_ref, state) do
-    with {:ok, pad_data} <- PadModel.get_data(state, pad_ref),
+    with {:ok, pad_data} when not pad_data.end_of_stream? <- PadModel.get_data(state, pad_ref),
          %State{playback: :playing} <- state do
       if pad_data.direction == :input,
         do: raise("cannot snapshot atomic counter in input pad")
 
       do_snapshot_atomic_demand(pad_data, state)
     else
+      {:ok, %{end_of_stream?: true}} ->
+        Membrane.Logger.debug_verbose(
+          "Skipping snapshot of pad #{inspect(pad_ref)}, because it has flag :end_of_stream? set to true"
+        )
+
+        state
+
       {:error, :unknown_pad} ->
         # We've got a :atomic_demand_increased message on already unlinked pad
         state
@@ -43,13 +50,10 @@ defmodule Membrane.Core.Element.DemandController do
          %{flow_control: :auto} = pad_data,
          %{effective_flow_control: :pull} = state
        ) do
-    %{
-      atomic_demand: atomic_demand,
-      associated_pads: associated_pads
-    } = pad_data
-
-    if AtomicDemand.get(atomic_demand) > 0 do
-      AutoFlowUtils.auto_adjust_atomic_demand(associated_pads, state)
+    if AtomicDemand.get(pad_data.atomic_demand) > 0 do
+      state
+      |> Map.update!(:satisfied_auto_output_pads, &MapSet.delete(&1, pad_data.ref))
+      |> AutoFlowUtils.pop_queues_and_bump_demand()
     else
       state
     end
@@ -91,9 +95,15 @@ defmodule Membrane.Core.Element.DemandController do
     buffers_size = Buffer.Metric.from_unit(pad_data.demand_unit).buffers_size(buffers)
 
     demand = pad_data.demand - buffers_size
-    atomic_demand = AtomicDemand.decrease(pad_data.atomic_demand, buffers_size)
+    {decrease_result, atomic_demand} = AtomicDemand.decrease(pad_data.atomic_demand, buffers_size)
 
-    PadModel.set_data!(state, pad_ref, %{
+    with {:decreased, new_value} when new_value <= 0 <- decrease_result,
+         %{flow_control: :auto} <- pad_data do
+      Map.update!(state, :satisfied_auto_output_pads, &MapSet.put(&1, pad_ref))
+    else
+      _other -> state
+    end
+    |> PadModel.set_data!(pad_ref, %{
       pad_data
       | demand: demand,
         atomic_demand: atomic_demand