Queue buffers when auto demand is low enough

.github/actions/add_pr_to_smackore_board/action.yml

@@ -24,7 +24,7 @@ runs:
           export STATUS_FIELD_ID=PVTSSF_lADOAYE_z84AWEIBzgOGd1k
           export TARGET_COLUMN_ID=e6b1ee10
 
-          export AUTHOR_ORIGIN=$(curl --request GET --url "https://api.github.com/orgs/membraneframework/members" --header "Authorization: Bearer $GH_TOKEN" -s | python scripts/python/get_author_origin.py $AUTHOR_LOGIN)
+          export AUTHOR_ORIGIN=$(curl --request GET --url "https://api.github.com/orgs/membraneframework/members" --header "Authorization: Bearer $GH_TOKEN" -s | python scripts/python/get_author_origin.py "$AUTHOR_LOGIN")
 
           if [ "$AUTHOR_ORIGIN" == "COMMUNITY" ]
           then

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, only if demand on input pad with `flow_control: :auto` is non-negative. [#654](https://github.com/membraneframework/membrane_core/pull/654)
  * Fix clock selection [#626](https://github.com/membraneframework/membrane_core/pull/626)
 
 ## 1.0.0

lib/membrane/core/element.ex

@@ -159,7 +159,9 @@ defmodule Membrane.Core.Element do
         effective_flow_control: :push,
         handling_action?: 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/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,20 @@ 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 pad_ref in state.awaiting_auto_input_pads do
+        raise "to nie powinno sie zdarzyc dupa 1"
+      end
+
+      if PadModel.get_data!(state, pad_ref, [:auto_flow_queue]) != Qex.new() do
+        raise "to nie powinno sie zdarzyc dupa 2"
+      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

lib/membrane/core/element/demand_controller.ex

@@ -49,7 +49,23 @@ defmodule Membrane.Core.Element.DemandController do
     } = pad_data
 
     if AtomicDemand.get(atomic_demand) > 0 do
-      AutoFlowUtils.auto_adjust_atomic_demand(associated_pads, state)
+      # tutaj powinno mieć miejsce
+      #  - usuniecie pada z mapsetu
+      #  - sflushowanie kolejek, jesli mapset jest pusty
+      # zwroc uwage, czy gdzies w czyms w stylu handle_outgoing_buffers nie wjedzie ci tutaj jakas nieprzyjemna rekurencja
+      # kolejna rzecz: przerwanie rekurencji moze nastąpić, nawet wtedy, gdy kolejki będą miały w sobie bufory
+
+      state = Map.update!(state, :satisfied_auto_output_pads, &MapSet.delete(&1, pad_data.ref))
+
+      # dobra, wyglada git
+
+      state = AutoFlowUtils.pop_auto_flow_queues_while_needed(state)
+
+      if MapSet.size(state.satisfied_auto_output_pads) == 0 do
+        AutoFlowUtils.auto_adjust_atomic_demand(associated_pads, state)
+      else
+        state
+      end
     else
       state
     end
@@ -91,9 +107,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

lib/membrane/core/element/demand_controller/auto_flow_utils.ex

@@ -1,9 +1,16 @@
 defmodule Membrane.Core.Element.DemandController.AutoFlowUtils do
   @moduledoc false
 
+  alias Membrane.Buffer
+  alias Membrane.Event
+  alias Membrane.StreamFormat
+
   alias Membrane.Core.Element.{
     AtomicDemand,
-    State
+    BufferController,
+    EventController,
+    State,
+    StreamFormatController
   }
 
   require Membrane.Core.Child.PadModel, as: PadModel
@@ -59,14 +66,36 @@ defmodule Membrane.Core.Element.DemandController.AutoFlowUtils do
     state
   end
 
-  @spec auto_adjust_atomic_demand(Pad.ref() | [Pad.ref()], State.t()) :: State.t()
-  def auto_adjust_atomic_demand(pad_ref_list, state) when is_list(pad_ref_list) do
-    Enum.reduce(pad_ref_list, state, &auto_adjust_atomic_demand/2)
+  @spec store_buffers_in_queue(Pad.ref(), [Buffer.t()], State.t()) :: State.t()
+  def store_buffers_in_queue(pad_ref, buffers, state) do
+    state = Map.update!(state, :awaiting_auto_input_pads, &MapSet.put(&1, pad_ref))
+    store_in_queue(pad_ref, :buffers, buffers, state)
+  end
+
+  @spec store_event_in_queue(Pad.ref(), Event.t(), State.t()) :: State.t()
+  def store_event_in_queue(pad_ref, event, state) do
+    store_in_queue(pad_ref, :event, event, state)
   end
 
-  def auto_adjust_atomic_demand(pad_ref, state) when Pad.is_pad_ref(pad_ref) do
-    PadModel.get_data!(state, pad_ref)
-    |> do_auto_adjust_atomic_demand(state)
+  @spec store_stream_format_in_queue(Pad.ref(), StreamFormat.t(), State.t()) :: State.t()
+  def store_stream_format_in_queue(pad_ref, stream_format, state) do
+    store_in_queue(pad_ref, :stream_format, stream_format, state)
+  end
+
+  defp store_in_queue(pad_ref, type, item, state) do
+    PadModel.update_data!(state, pad_ref, :auto_flow_queue, &Qex.push(&1, {type, item}))
+  end
+
+  @spec auto_adjust_atomic_demand(Pad.ref() | [Pad.ref()], State.t()) :: State.t()
+  def auto_adjust_atomic_demand(ref_or_ref_list, state)
+      when Pad.is_pad_ref(ref_or_ref_list) or is_list(ref_or_ref_list) do
+    ref_or_ref_list
+    |> Bunch.listify()
+    |> Enum.reduce(state, fn pad_ref, state ->
+      PadModel.get_data!(state, pad_ref)
+      |> do_auto_adjust_atomic_demand(state)
+      |> elem(1)  # todo: usun to :increased / :unchanged, ktore discardujesz w tym elem(1)
+    end)
   end
 
   defp do_auto_adjust_atomic_demand(pad_data, state) when is_input_auto_pad_data(pad_data) do
@@ -83,9 +112,11 @@ defmodule Membrane.Core.Element.DemandController.AutoFlowUtils do
       :ok = AtomicDemand.increase(atomic_demand, diff)
 
       :atomics.put(stalker_metrics.demand, 1, auto_demand_size)
-      PadModel.set_data!(state, ref, :demand, auto_demand_size)
+
+      state = PadModel.set_data!(state, ref, :demand, auto_demand_size)
+      {:increased, state}
     else
-      state
+      {:unchanged, state}
     end
   end
 
@@ -97,14 +128,66 @@ defmodule Membrane.Core.Element.DemandController.AutoFlowUtils do
     state.effective_flow_control == :pull and
       not pad_data.auto_demand_paused? and
       pad_data.demand < pad_data.auto_demand_size / 2 and
-      Enum.all?(pad_data.associated_pads, &atomic_demand_positive?(&1, state))
+      output_auto_demand_positive?(state)
+  end
+
+  @spec pop_auto_flow_queues_while_needed(State.t()) :: State.t()
+  def pop_auto_flow_queues_while_needed(state) do
+    if (state.effective_flow_control == :push or
+          MapSet.size(state.satisfied_auto_output_pads) == 0) and
+         MapSet.size(state.awaiting_auto_input_pads) > 0 do
+      pop_random_auto_flow_queue(state)
+      |> pop_auto_flow_queues_while_needed()
+    else
+      state
+    end
+  end
+
+  defp pop_random_auto_flow_queue(state) do
+    pad_ref = Enum.random(state.awaiting_auto_input_pads)
+
+    PadModel.get_data!(state, pad_ref, :auto_flow_queue)
+    |> Qex.pop()
+    |> case do
+      {{:value, {:buffers, buffers}}, popped_queue} ->
+        state = PadModel.set_data!(state, pad_ref, :auto_flow_queue, popped_queue)
+        state = BufferController.exec_buffer_callback(pad_ref, buffers, state)
+        pop_stream_formats_and_events(pad_ref, state)
+
+      {:empty, _empty_queue} ->
+        Map.update!(state, :awaiting_auto_input_pads, &MapSet.delete(&1, pad_ref))
+    end
   end
 
-  defp atomic_demand_positive?(pad_ref, state) do
-    atomic_demand_value =
-      PadModel.get_data!(state, pad_ref, :atomic_demand)
-      |> AtomicDemand.get()
+  defp pop_stream_formats_and_events(pad_ref, state) do
+    PadModel.get_data!(state, pad_ref, :auto_flow_queue)
+    |> Qex.pop()
+    |> case do
+      {{:value, {type, item}}, popped_queue} when type in [:event, :stream_format] ->
+        state = PadModel.set_data!(state, pad_ref, :auto_flow_queue, popped_queue)
+        state = exec_queue_item_callback(pad_ref, {type, item}, state)
+        pop_stream_formats_and_events(pad_ref, state)
+
+      {{:value, {:buffers, _buffers}}, _popped_queue} ->
+        state
+
+      {:empty, _empty_queue} ->
+        Map.update!(state, :awaiting_auto_input_pads, &MapSet.delete(&1, pad_ref))
+    end
+  end
+
+  defp output_auto_demand_positive?(%State{satisfied_auto_output_pads: pads}),
+    do: MapSet.size(pads) == 0
+
+  defp exec_queue_item_callback(pad_ref, {:buffers, buffers}, state) do
+    BufferController.exec_buffer_callback(pad_ref, buffers, state)
+  end
+
+  defp exec_queue_item_callback(pad_ref, {:event, event}, state) do
+    EventController.exec_handle_event(pad_ref, event, state)
+  end
 
-    atomic_demand_value > 0
+  defp exec_queue_item_callback(pad_ref, {:stream_format, stream_format}, state) do
+    StreamFormatController.exec_handle_stream_format(pad_ref, stream_format, state)
   end
 end

lib/membrane/core/element/effective_flow_controller.ex

@@ -104,8 +104,8 @@ defmodule Membrane.Core.Element.EffectiveFlowController do
 
     state.pads_data
     |> Enum.filter(fn {_ref, %{flow_control: flow_control}} -> flow_control == :auto end)
-    |> Enum.reduce(state, fn
-      {_ref, %{direction: :output} = pad_data}, state ->
+    |> Enum.each(fn
+      {_ref, %{direction: :output} = pad_data} ->
         :ok =
           AtomicDemand.set_sender_status(
             pad_data.atomic_demand,
@@ -120,9 +120,7 @@ defmodule Membrane.Core.Element.EffectiveFlowController do
           [pad_data.other_ref, new_effective_flow_control]
         )
 
-        state
-
-      {pad_ref, %{direction: :input} = pad_data}, state ->
+      {pad_ref, %{direction: :input} = pad_data} ->
         if triggering_pad in [pad_ref, nil] or
              AtomicDemand.get_receiver_status(pad_data.atomic_demand) != :to_be_resolved do
           :ok =
@@ -131,8 +129,13 @@ defmodule Membrane.Core.Element.EffectiveFlowController do
               {:resolved, new_effective_flow_control}
             )
         end
+    end)
 
-        AutoFlowUtils.auto_adjust_atomic_demand(pad_ref, state)
+    state.pads_data
+    |> Enum.flat_map(fn
+      {pad_ref, %{direction: :input, flow_control: :auto}} -> [pad_ref]
+      _other -> []
     end)
+    |> AutoFlowUtils.auto_adjust_atomic_demand(state)
   end
 end

lib/membrane/core/element/event_controller.ex

@@ -18,6 +18,8 @@ defmodule Membrane.Core.Element.EventController do
     State
   }
 
+  alias Membrane.Core.Element.DemandController.AutoFlowUtils
+
   require Membrane.Core.Child.PadModel
   require Membrane.Core.Message
   require Membrane.Core.Telemetry
@@ -39,15 +41,22 @@ defmodule Membrane.Core.Element.EventController do
           playback: %State{playback: :playing} <- state do
       Telemetry.report_metric(:event, 1, inspect(pad_ref))
 
-      if not Event.async?(event) and buffers_before_event_present?(data) do
-        PadModel.update_data!(
-          state,
-          pad_ref,
-          :input_queue,
-          &InputQueue.store(&1, :event, event)
-        )
-      else
-        exec_handle_event(pad_ref, event, state)
+      cond do
+        # events goes to the manual flow control input queue
+        not Event.async?(event) and buffers_before_event_present?(data) ->
+          PadModel.update_data!(
+            state,
+            pad_ref,
+            :input_queue,
+            &InputQueue.store(&1, :event, event)
+          )
+
+        # event goes to the auto flow control queue
+        pad_ref in state.awaiting_auto_input_pads ->
+          AutoFlowUtils.store_event_in_queue(pad_ref, event, state)
+
+        true ->
+          exec_handle_event(pad_ref, event, state)
       end
     else
       pad: {:error, :unknown_pad} ->

lib/membrane/core/element/pad_controller.ex

@@ -330,10 +330,15 @@ defmodule Membrane.Core.Element.PadController do
 
     state = update_associated_pads(pad_data, state)
 
-    if pad_data.direction == :input and pad_data.flow_control == :auto do
-      AutoFlowUtils.auto_adjust_atomic_demand(endpoint.pad_ref, state)
-    else
-      state
+    case pad_data do
+      %{direction: :output, flow_control: :auto} ->
+        Map.update!(state, :satisfied_auto_output_pads, &MapSet.put(&1, pad_data.ref))
+
+      %{direction: :input, flow_control: :auto} ->
+        AutoFlowUtils.auto_adjust_atomic_demand(endpoint.pad_ref, state)
+
+      _pad_data ->
+        state
     end
   end
 
@@ -484,6 +489,9 @@ defmodule Membrane.Core.Element.PadController do
             PadModel.update_data!(state, pad, :associated_pads, &List.delete(&1, pad_data.ref))
           end)
           |> PadModel.set_data!(pad_ref, :associated_pads, [])
+          |> Map.update!(:satisfied_auto_output_pads, &MapSet.delete(&1, pad_ref))
+          |> Map.update!(:awaiting_auto_input_pads, &MapSet.delete(&1, pad_ref))
+          |> AutoFlowUtils.pop_auto_flow_queues_while_needed()
 
         if pad_data.direction == :output,
           do: AutoFlowUtils.auto_adjust_atomic_demand(pad_data.associated_pads, state),