Auto push full impl

lib/membrane/core/element.ex

@@ -222,14 +222,14 @@ defmodule Membrane.Core.Element do
   end
 
   defp do_handle_info(
-         Message.new(:other_effective_flow_control_resolved, [
+         Message.new(:sender_effective_flow_control_resolved, [
            input_pad_ref,
            effective_flow_control
          ]),
          state
        ) do
     state =
-      EffectiveFlowController.handle_other_effective_flow_control(
+      EffectiveFlowController.handle_sender_effective_flow_control(
         input_pad_ref,
         effective_flow_control,
         state

lib/membrane/core/element/buffer_controller.ex

@@ -57,11 +57,10 @@ defmodule Membrane.Core.Element.BufferController do
   @spec do_handle_buffer(Pad.ref(), PadModel.pad_data(), [Buffer.t()] | Buffer.t(), State.t()) ::
           State.t()
   defp do_handle_buffer(pad_ref, %{flow_control: :auto} = data, buffers, state) do
-    %{lacking_buffer_size: lacking_buffer_size, demand_unit: demand_unit} = data
+    %{demand: demand, demand_unit: demand_unit} = data
     buf_size = Buffer.Metric.from_unit(demand_unit).buffers_size(buffers)
 
-    state =
-      PadModel.set_data!(state, pad_ref, :lacking_buffer_size, lacking_buffer_size - buf_size)
+    state = PadModel.set_data!(state, pad_ref, :demand, demand - buf_size)
 
     state = AutoFlowUtils.auto_adjust_demand_counter(pad_ref, state)
     exec_buffer_callback(pad_ref, buffers, state)

lib/membrane/core/element/demand_controller.ex

@@ -90,7 +90,12 @@ defmodule Membrane.Core.Element.DemandController do
   def decrease_demand_by_outgoing_buffers(pad_ref, buffers, state) do
     pad_data = PadModel.get_data!(state, pad_ref)
 
-    demand_unit = pad_data.demand_unit || pad_data.other_demand_unit || :buffers
+    demand_unit =
+      case pad_data.flow_control do
+        :push -> pad_data.other_demand_unit || :buffers
+        _pull_or_auto -> pad_data.demand_unit
+      end
+
     buffers_size = Buffer.Metric.from_unit(demand_unit).buffers_size(buffers)
 
     demand_snapshot = pad_data.demand_snapshot - buffers_size

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

@@ -26,13 +26,13 @@ defmodule Membrane.Core.Element.DemandController.AutoFlowUtils do
 
   defp do_auto_adjust_demand_counter(pad_data, state) when is_input_auto_pad_data(pad_data) do
     if increase_demand_counter?(pad_data, state) do
-      diff = pad_data.auto_demand_size - pad_data.lacking_buffer_size
+      diff = pad_data.auto_demand_size - pad_data.demand
       :ok = DemandCounter.increase(pad_data.demand_counter, diff)
 
       PadModel.set_data!(
         state,
         pad_data.ref,
-        :lacking_buffer_size,
+        :demand,
         pad_data.auto_demand_size
       )
     else
@@ -46,7 +46,7 @@ defmodule Membrane.Core.Element.DemandController.AutoFlowUtils do
 
   defp increase_demand_counter?(pad_data, state) do
     state.effective_flow_control == :pull and
-      pad_data.lacking_buffer_size < pad_data.auto_demand_size / 2 and
+      pad_data.demand < pad_data.auto_demand_size / 2 and
       Enum.all?(pad_data.associated_pads, &demand_counter_positive?(&1, state))
   end
 

lib/membrane/core/element/demand_counter.ex

@@ -1,10 +1,11 @@
 defmodule Membrane.Core.Element.DemandCounter do
   @moduledoc false
+  alias Membrane.Core.Element.DemandCounter.DistributedFlowStatus
   alias Membrane.Core.Element.EffectiveFlowController
 
   alias __MODULE__.{
     DistributedAtomic,
-    DistributedFlowMode
+    DistributedFlowStatus
   }
 
   require Membrane.Core.Message, as: Message
@@ -17,9 +18,9 @@ defmodule Membrane.Core.Element.DemandCounter do
 
   @opaque t :: %__MODULE__{
             counter: DistributedAtomic.t(),
-            receiver_mode: DistributedFlowMode.t(),
+            receiver_status: DistributedFlowStatus.t(),
             receiver_process: Process.dest(),
-            sender_mode: DistributedFlowMode.t(),
+            sender_status: DistributedFlowStatus.t(),
             sender_process: Process.dest(),
             sender_pad_ref: Pad.ref(),
             toilet_capacity: neg_integer(),
@@ -29,13 +30,13 @@ defmodule Membrane.Core.Element.DemandCounter do
             receiver_demand_unit: Membrane.Buffer.Metric.unit()
           }
 
-  @type flow_mode :: DistributedFlowMode.flow_mode_value()
+  @type flow_mode :: DistributedFlowStatus.value()
 
   @enforce_keys [
     :counter,
-    :receiver_mode,
+    :receiver_status,
     :receiver_process,
-    :sender_mode,
+    :sender_status,
     :sender_process,
     :sender_pad_ref,
     :throttling_factor,
@@ -46,7 +47,7 @@ defmodule Membrane.Core.Element.DemandCounter do
   defstruct @enforce_keys ++ [buffered_decrementation: 0, toilet_overflowed?: false]
 
   @spec new(
-          receiver_mode :: EffectiveFlowController.effective_flow_control(),
+          receiver_effective_flow_control :: EffectiveFlowController.effective_flow_control(),
           receiver_process :: Process.dest(),
           receiver_demand_unit :: Membrane.Buffer.Metric.unit(),
           sender_process :: Process.dest(),
@@ -55,7 +56,7 @@ defmodule Membrane.Core.Element.DemandCounter do
           throttling_factor :: pos_integer() | nil
         ) :: t
   def new(
-        receiver_mode,
+        receiver_effective_flow_control,
         receiver_process,
         receiver_demand_unit,
         sender_process,
@@ -72,11 +73,13 @@ defmodule Membrane.Core.Element.DemandCounter do
         true -> @distributed_default_throttling_factor
       end
 
+    receiver_status = DistributedFlowStatus.new({:resolved, receiver_effective_flow_control})
+
     %__MODULE__{
       counter: counter,
-      receiver_mode: DistributedFlowMode.new(receiver_mode),
+      receiver_status: receiver_status,
       receiver_process: receiver_process,
-      sender_mode: DistributedFlowMode.new(:to_be_resolved),
+      sender_status: DistributedFlowStatus.new(:to_be_resolved),
       sender_process: sender_process,
       sender_pad_ref: sender_pad_ref,
       toilet_capacity: toilet_capacity || default_toilet_capacity(receiver_demand_unit),
@@ -85,30 +88,30 @@ defmodule Membrane.Core.Element.DemandCounter do
     }
   end
 
-  @spec set_sender_mode(t, EffectiveFlowController.effective_flow_control()) :: :ok
-  def set_sender_mode(%__MODULE__{} = demand_counter, mode) do
-    DistributedFlowMode.put(
-      demand_counter.sender_mode,
+  @spec set_sender_status(t, DistributedFlowStatus.value()) :: :ok
+  def set_sender_status(%__MODULE__{} = demand_counter, mode) do
+    DistributedFlowStatus.put(
+      demand_counter.sender_status,
       mode
     )
   end
 
-  @spec get_sender_mode(t) :: flow_mode()
-  def get_sender_mode(%__MODULE__{} = demand_counter) do
-    DistributedFlowMode.get(demand_counter.sender_mode)
+  @spec get_sender_status(t) :: DistributedFlowStatus.value()
+  def get_sender_status(%__MODULE__{} = demand_counter) do
+    DistributedFlowStatus.get(demand_counter.sender_status)
   end
 
-  @spec set_receiver_mode(t, flow_mode()) :: :ok
-  def set_receiver_mode(%__MODULE__{} = demand_counter, mode) do
-    DistributedFlowMode.put(
-      demand_counter.receiver_mode,
+  @spec set_receiver_status(t, DistributedFlowStatus.value()) :: :ok
+  def set_receiver_status(%__MODULE__{} = demand_counter, mode) do
+    DistributedFlowStatus.put(
+      demand_counter.receiver_status,
       mode
     )
   end
 
-  @spec get_receiver_mode(t) :: flow_mode()
-  def get_receiver_mode(%__MODULE__{} = demand_counter) do
-    DistributedFlowMode.get(demand_counter.receiver_mode)
+  @spec get_receiver_status(t) :: DistributedFlowStatus.value()
+  def get_receiver_status(%__MODULE__{} = demand_counter) do
+    DistributedFlowStatus.get(demand_counter.receiver_status)
   end
 
   @spec increase(t, non_neg_integer()) :: :ok
@@ -153,8 +156,8 @@ defmodule Membrane.Core.Element.DemandCounter do
     demand_counter = %{demand_counter | buffered_decrementation: 0}
 
     if not demand_counter.toilet_overflowed? and
-         get_receiver_mode(demand_counter) == :pull and
-         get_sender_mode(demand_counter) == :push and
+         get_receiver_status(demand_counter) == {:resolved, :pull} and
+         get_sender_status(demand_counter) == {:resolved, :push} and
          -1 * counter_value > demand_counter.toilet_capacity do
       overflow(demand_counter, counter_value)
     else

lib/membrane/core/element/demand_counter/distributed_atomic.ex

@@ -19,7 +19,7 @@ defmodule Membrane.Core.Element.DemandCounter.DistributedAtomic do
   @spec new(integer() | nil) :: t
   def new(initial_value \\ nil) do
     atomic_ref = :atomics.new(1, [])
-    {:ok, worker} = Worker.start_link()
+    {:ok, worker} = Worker.start_link(self())
 
     distributed_atomic = %__MODULE__{
       atomic_ref: atomic_ref,

lib/membrane/core/element/demand_counter/distributed_atomic/worker.ex

@@ -8,12 +8,13 @@ defmodule Membrane.Core.Element.DemandCounter.DistributedAtomic.Worker do
 
   @type t :: pid()
 
-  @spec start_link() :: {:ok, t}
-  def start_link(), do: GenServer.start_link(__MODULE__, nil)
+  @spec start_link(pid()) :: {:ok, t}
+  def start_link(owner_pid), do: GenServer.start_link(__MODULE__, owner_pid)
 
   @impl true
-  def init(_arg) do
-    {:ok, nil, :hibernate}
+  def init(owner_pid) do
+    ref = Process.monitor(owner_pid)
+    {:ok, %{ref: ref}, :hibernate}
   end
 
   @impl true
@@ -39,4 +40,9 @@ defmodule Membrane.Core.Element.DemandCounter.DistributedAtomic.Worker do
     :atomics.put(atomic_ref, 1, value)
     {:noreply, nil}
   end
+
+  @impl true
+  def handle_info({:DOWN, ref, _process, _pid, _reason}, %{ref: ref} = state) do
+    {:stop, :normal, state}
+  end
 end

lib/membrane/core/element/demand_counter/distributed_flow_status.ex

@@ -0,0 +1,37 @@
+defmodule Membrane.Core.Element.DemandCounter.DistributedFlowStatus do
+  @moduledoc false
+
+  alias Membrane.Core.Element.DemandCounter.DistributedAtomic
+  alias Membrane.Core.Element.EffectiveFlowController
+
+  @type t :: DistributedAtomic.t()
+  @type value :: {:resolved, EffectiveFlowController.effective_flow_control()} | :to_be_resolved
+
+  @spec new(value) :: t
+  def new(initial_value) do
+    initial_value
+    |> flow_status_to_int()
+    |> DistributedAtomic.new()
+  end
+
+  @spec get(t) :: value()
+  def get(distributed_atomic) do
+    distributed_atomic
+    |> DistributedAtomic.get()
+    |> int_to_flow_status()
+  end
+
+  @spec put(t, value()) :: :ok
+  def put(distributed_atomic, value) do
+    value = flow_status_to_int(value)
+    DistributedAtomic.put(distributed_atomic, value)
+  end
+
+  defp int_to_flow_status(0), do: :to_be_resolved
+  defp int_to_flow_status(1), do: {:resolved, :push}
+  defp int_to_flow_status(2), do: {:resolved, :pull}
+
+  defp flow_status_to_int(:to_be_resolved), do: 0
+  defp flow_status_to_int({:resolved, :push}), do: 1
+  defp flow_status_to_int({:resolved, :pull}), do: 2
+end