Auto push full impl

lib/membrane/bin.ex

@@ -18,7 +18,6 @@ defmodule Membrane.Bin do
   alias Membrane.Core.OptionsSpecs
 
   require Membrane.Core.Message
-  require Membrane.Logger
 
   @type state :: any()
 

lib/membrane/children_spec.ex

@@ -538,11 +538,11 @@ defmodule Membrane.ChildrenSpec do
   """
   @spec via_in(builder(), Pad.name() | Pad.ref(),
           options: pad_options(),
-          toilet_capacity: number | nil,
-          target_queue_size: number | nil,
-          min_demand_factor: number | nil,
-          auto_demand_size: number | nil,
-          throttling_factor: number | nil
+          toilet_capacity: non_neg_integer() | nil,
+          target_queue_size: non_neg_integer() | nil,
+          min_demand_factor: non_neg_integer() | nil,
+          auto_demand_size: non_neg_integer() | nil,
+          throttling_factor: non_neg_integer() | nil
         ) :: builder() | no_return
   def via_in(builder, pad, props \\ [])
 
@@ -567,7 +567,7 @@ defmodule Membrane.ChildrenSpec do
         min_demand_factor: [default: nil],
         auto_demand_size: [default: nil],
         toilet_capacity: [default: nil],
-        throttling_factor: [default: 1]
+        throttling_factor: [default: nil]
       )
       |> case do
         {:ok, props} ->

lib/membrane/core/bin/pad_controller.ex

@@ -200,12 +200,10 @@ defmodule Membrane.Core.Bin.PadController do
           SpecificationParser.raw_endpoint(),
           SpecificationParser.raw_endpoint(),
           %{
-            initiator: :parent,
             stream_format_validation_params:
               StreamFormatController.stream_format_validation_params()
           }
           | %{
-              initiator: :sibling,
               other_info: PadModel.pad_info() | nil,
               link_metadata: map,
               stream_format_validation_params:
@@ -238,7 +236,7 @@ defmodule Membrane.Core.Bin.PadController do
 
       child_endpoint = %{child_endpoint | pad_props: pad_props}
 
-      if params.initiator == :sibling do
+      if direction == :input do
         :ok =
           Child.PadController.validate_pad_mode!(
             {endpoint.pad_ref, pad_data},
@@ -297,11 +295,10 @@ defmodule Membrane.Core.Bin.PadController do
   def handle_unlink(pad_ref, state) do
     with {:ok, %{availability: :on_request}} <- PadModel.get_data(state, pad_ref) do
       state = maybe_handle_pad_removed(pad_ref, state)
-      endpoint = PadModel.get_data!(state, pad_ref, :endpoint)
       {pad_data, state} = PadModel.pop_data!(state, pad_ref)
 
-      if endpoint do
-        Message.send(endpoint.pid, :handle_unlink, endpoint.pad_ref)
+      if pad_data.endpoint do
+        Message.send(pad_data.endpoint.pid, :handle_unlink, pad_data.endpoint.pad_ref)
         ChildLifeController.proceed_spec_startup(pad_data.spec_ref, state)
       else
         Membrane.Logger.debug("""

lib/membrane/core/child/pad_controller.ex

@@ -5,7 +5,6 @@ defmodule Membrane.Core.Child.PadController do
   alias Membrane.{LinkError, Pad}
 
   require Membrane.Core.Child.PadModel
-  require Membrane.Logger
 
   @type state :: Membrane.Core.Bin.State.t() | Membrane.Core.Element.State.t()
 

lib/membrane/core/child/pad_model.ex

@@ -6,6 +6,7 @@ defmodule Membrane.Core.Child.PadModel do
   use Bunch
 
   alias Membrane.Core.Child
+  alias Membrane.Core.Element.EffectiveFlowController
   alias Membrane.{Pad, UnknownPadError}
 
   @type bin_pad_data :: %Membrane.Bin.PadData{
@@ -24,11 +25,13 @@ defmodule Membrane.Core.Child.PadModel do
   @type element_pad_data :: %Membrane.Element.PadData{
           availability: Pad.availability(),
           stream_format: Membrane.StreamFormat.t() | nil,
-          demand: integer() | nil,
+          demand_snapshot: integer() | nil,
+          manual_demand_size: integer(),
           start_of_stream?: boolean(),
           end_of_stream?: boolean(),
           direction: Pad.direction(),
           flow_control: Pad.flow_control(),
+          other_effective_flow_control: EffectiveFlowController.effective_flow_control() | nil,
           name: Pad.name(),
           ref: Pad.ref(),
           demand_unit: Membrane.Buffer.Metric.unit() | nil,
@@ -38,7 +41,6 @@ defmodule Membrane.Core.Child.PadModel do
           sticky_messages: [Membrane.Event.t()],
           input_queue: Membrane.Core.Element.InputQueue.t() | nil,
           options: %{optional(atom) => any},
-          toilet: Membrane.Core.Element.Toilet.t() | nil,
           auto_demand_size: pos_integer() | nil,
           associated_pads: [Pad.ref()] | nil,
           sticky_events: [Membrane.Event.t()]
@@ -52,6 +54,7 @@ defmodule Membrane.Core.Child.PadModel do
           required(:availability) => Pad.availability(),
           required(:direction) => Pad.direction(),
           required(:name) => Pad.name(),
+          required(:accepted_formats_str) => String.t(),
           optional(:flow_control) => Pad.flow_control(),
           optional(:demand_unit) => Membrane.Buffer.Metric.unit(),
           optional(:other_demand_unit) => Membrane.Buffer.Metric.unit()

lib/membrane/core/element.ex

@@ -18,13 +18,15 @@ defmodule Membrane.Core.Element do
   use Bunch
   use GenServer
 
+  alias Membrane.Core.Element.DemandHandler
   alias Membrane.{Clock, Core, ResourceGuard, Sync}
 
   alias Membrane.Core.{SubprocessSupervisor, TimerController}
 
   alias Membrane.Core.Element.{
     BufferController,
     DemandController,
+    EffectiveFlowController,
     EventController,
     LifecycleController,
     PadController,
@@ -171,9 +173,13 @@ defmodule Membrane.Core.Element do
 
   @compile {:inline, do_handle_info: 2}
 
-  defp do_handle_info(Message.new(:demand, size, _opts) = msg, state) do
-    pad_ref = Message.for_pad(msg)
-    state = DemandController.handle_demand(pad_ref, size, state)
+  defp do_handle_info(Message.new(:demand_counter_increased, pad_ref), state) do
+    state = DemandController.snapshot_demand_counter(pad_ref, state)
+    {:noreply, state}
+  end
+
+  defp do_handle_info(Message.new(:resume_handle_demand_loop), state) do
+    state = DemandHandler.handle_delayed_demands(state)
     {:noreply, state}
   end
 
@@ -215,6 +221,23 @@ defmodule Membrane.Core.Element do
     {:noreply, state}
   end
 
+  defp do_handle_info(
+         Message.new(:sender_effective_flow_control_resolved, [
+           input_pad_ref,
+           effective_flow_control
+         ]),
+         state
+       ) do
+    state =
+      EffectiveFlowController.handle_sender_effective_flow_control(
+        input_pad_ref,
+        effective_flow_control,
+        state
+      )
+
+    {:noreply, state}
+  end
+
   defp do_handle_info(Message.new(:terminate), state) do
     state = LifecycleController.handle_terminate_request(state)
     {:noreply, state}

lib/membrane/core/element/action_handler.ex

@@ -20,7 +20,15 @@ defmodule Membrane.Core.Element.ActionHandler do
   }
 
   alias Membrane.Core.Child.PadModel
-  alias Membrane.Core.Element.{DemandHandler, PadController, State, StreamFormatController}
+
+  alias Membrane.Core.Element.{
+    DemandController,
+    DemandHandler,
+    PadController,
+    State,
+    StreamFormatController
+  }
+
   alias Membrane.Core.{Events, Message, Telemetry, TimerController}
   alias Membrane.Element.Action
 
@@ -307,11 +315,11 @@ defmodule Membrane.Core.Element.ActionHandler do
          }
          when stream_format != nil <- pad_data do
       state =
-        DemandHandler.handle_outgoing_buffers(pad_ref, pad_data, buffers, state)
+        DemandController.decrease_demand_by_outgoing_buffers(pad_ref, buffers, state)
         |> PadModel.set_data!(pad_ref, :start_of_stream?, true)
 
       Message.send(pid, :buffer, buffers, for_pad: other_ref)
-      state
+      DemandController.snapshot_demand_counter(pad_ref, state)
     else
       %{direction: :input} ->
         raise PadDirectionError, action: :buffer, direction: :input, pad: pad_ref
@@ -354,7 +362,8 @@ defmodule Membrane.Core.Element.ActionHandler do
         StreamFormatController.validate_stream_format!(
           :output,
           stream_format_validation_params,
-          stream_format
+          stream_format,
+          state
         )
 
       state = PadModel.set_data!(state, pad_ref, :stream_format, stream_format)

lib/membrane/core/element/buffer_controller.ex

@@ -12,14 +12,14 @@ defmodule Membrane.Core.Element.BufferController do
   alias Membrane.Core.Element.{
     ActionHandler,
     CallbackContext,
-    DemandController,
     DemandHandler,
     EventController,
     InputQueue,
     PlaybackQueue,
     State
   }
 
+  alias Membrane.Core.Element.DemandController.AutoFlowUtils
   alias Membrane.Core.Telemetry
 
   require Membrane.Core.Child.PadModel
@@ -58,15 +58,17 @@ defmodule Membrane.Core.Element.BufferController do
           State.t()
   defp do_handle_buffer(pad_ref, %{flow_control: :auto} = data, buffers, state) do
     %{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, :demand, demand - buf_size)
-    state = DemandController.send_auto_demand_if_needed(pad_ref, state)
+
+    state = AutoFlowUtils.auto_adjust_demand_counter(pad_ref, state)
     exec_buffer_callback(pad_ref, buffers, state)
   end
 
   defp do_handle_buffer(pad_ref, %{flow_control: :manual} = data, buffers, state) do
     %{input_queue: old_input_queue} = data
+
     input_queue = InputQueue.store(old_input_queue, buffers)
     state = PadModel.set_data!(state, pad_ref, :input_queue, input_queue)
 
@@ -77,7 +79,7 @@ defmodule Membrane.Core.Element.BufferController do
     end
   end
 
-  defp do_handle_buffer(pad_ref, _data, buffers, state) do
+  defp do_handle_buffer(pad_ref, %{flow_control: :push}, buffers, state) do
     exec_buffer_callback(pad_ref, buffers, state)
   end
 
@@ -92,20 +94,18 @@ defmodule Membrane.Core.Element.BufferController do
   def exec_buffer_callback(pad_ref, buffers, %State{type: :filter} = state) do
     Telemetry.report_metric("buffer", 1, inspect(pad_ref))
 
-    CallbackHandler.exec_and_handle_callback(
-      :handle_buffers_batch,
-      ActionHandler,
-      %{context: &CallbackContext.from_state/1},
-      [pad_ref, buffers],
-      state
-    )
+    do_exec_buffer_callback(pad_ref, buffers, state)
   end
 
   def exec_buffer_callback(pad_ref, buffers, %State{type: type} = state)
       when type in [:sink, :endpoint] do
     Telemetry.report_metric(:buffer, length(List.wrap(buffers)))
     Telemetry.report_bitrate(buffers)
 
+    do_exec_buffer_callback(pad_ref, buffers, state)
+  end
+
+  defp do_exec_buffer_callback(pad_ref, buffers, state) do
     CallbackHandler.exec_and_handle_callback(
       :handle_buffers_batch,
       ActionHandler,