Implement suggestions from CR & delete unnecessary state field

Author: FelonEkonom

lib/membrane/bin/pad_data.ex

@@ -27,7 +27,9 @@ defmodule Membrane.Bin.PadData do
           link_id: private_field,
           endpoint: private_field,
           linked?: private_field,
-          response_received?: private_field
+          response_received?: private_field,
+          linking_timeout_id: private_field,
+          linked_in_spec?: private_field
         }
 
   @enforce_keys [
@@ -40,7 +42,9 @@ defmodule Membrane.Bin.PadData do
     :endpoint,
     :linked?,
     :response_received?,
-    :spec_ref
+    :spec_ref,
+    :linking_timeout_id,
+    :linked_in_spec?
   ]
 
   defstruct @enforce_keys

lib/membrane/core/bin.ex

@@ -209,8 +209,8 @@ defmodule Membrane.Core.Bin do
     {:noreply, state}
   end
 
-  defp do_handle_info(Message.new(:linking_timeout, pad_ref), state) do
-    state = PadController.handle_linking_timeout(pad_ref, state)
+  defp do_handle_info(Message.new(:linking_timeout, [pad_ref, linking_timeout_id]), state) do
+    :ok = PadController.handle_linking_timeout(pad_ref, linking_timeout_id, state)
     {:noreply, state}
   end
 

lib/membrane/core/bin/pad_controller.ex

@@ -8,11 +8,10 @@ defmodule Membrane.Core.Bin.PadController do
   alias Membrane.{Core, LinkError, Pad}
   alias Membrane.Core.Bin.{ActionHandler, CallbackContext, State}
   alias Membrane.Core.{CallbackHandler, Child, Message}
-  alias Membrane.Core.Child.PadModel
   alias Membrane.Core.Element.StreamFormatController
   alias Membrane.Core.Parent.{ChildLifeController, Link, SpecificationParser}
 
-  require Membrane.Core.Child.PadModel
+  require Membrane.Core.Child.PadModel, as: PadModel
   require Membrane.Core.Message
   require Membrane.Logger
   require Membrane.Pad
@@ -50,8 +49,7 @@ defmodule Membrane.Core.Bin.PadController do
     state =
       case PadModel.get_data(state, pad_ref) do
         {:error, :unknown_pad} ->
-          init_pad_data(pad_ref, pad_info, state)
-          |> Map.update!(:pad_refs, &[pad_ref | &1])
+          init_pad_data(pad_ref, state)
 
         # This case is for pads that were instantiated before the external link request,
         # that is in the internal link request (see `handle_internal_link_request/4`).
@@ -69,9 +67,17 @@ defmodule Membrane.Core.Bin.PadController do
           state
       end
 
-    state = PadModel.update_data!(state, pad_ref, &%{&1 | link_id: link_id, options: pad_options})
+    linking_timeout_id = make_ref()
 
-    _ref = Process.send_after(self(), Message.new(:linking_timeout, pad_ref), 5000)
+    state =
+      PadModel.update_data!(
+        state,
+        pad_ref,
+        &%{&1 | link_id: link_id, linking_timeout_id: linking_timeout_id, options: pad_options}
+      )
+
+    message = Message.new(:linking_timeout, [pad_ref, linking_timeout_id])
+    _ref = Process.send_after(self(), message, 5000)
 
     maybe_handle_pad_added(pad_ref, state)
   end
@@ -96,20 +102,16 @@ defmodule Membrane.Core.Bin.PadController do
     end
   end
 
-  @spec handle_linking_timeout(Pad.ref(), State.t()) :: State.t() | no_return()
-  def handle_linking_timeout(pad_ref, state) do
-    case Map.fetch(state.linking_timeout_counters, pad_ref) do
-      {:ok, 1} ->
-        Map.update!(state, :linking_timeout_counters, &Map.delete(&1, pad_ref))
-
-      {:ok, counter} when counter > 1 ->
-        put_in(state.linking_timeout_counters[pad_ref], counter - 1)
-
-      _else ->
-        raise Membrane.LinkError, """
-        Bin pad #{inspect(pad_ref)} wasn't linked internally within timeout. Pad data: #{PadModel.get_data(state, pad_ref) |> inspect(pretty: true)}
-        """
+  @spec handle_linking_timeout(Pad.ref(), reference(), State.t()) :: :ok | no_return()
+  def handle_linking_timeout(pad_ref, linking_timeout_id, state) do
+    with {:ok, pad_data} <- PadModel.get_data(state, pad_ref),
+         %{linking_timeout_id: ^linking_timeout_id, linked_in_spec?: false} <- pad_data do
+      raise Membrane.LinkError, """
+      Bin pad #{inspect(pad_ref)} wasn't linked internally within timeout. Pad data: #{PadModel.get_data(state, pad_ref) |> inspect(pretty: true)}
+      """
     end
+
+    :ok
   end
 
   @doc """
@@ -137,7 +139,7 @@ defmodule Membrane.Core.Bin.PadController do
 
         # Static pads can be linked internally before the external link request
         pad_info.availability == :always ->
-          init_pad_data(pad_ref, pad_info, state)
+          init_pad_data(pad_ref, state)
 
         true ->
           raise LinkError, "Dynamic pads must be firstly linked externally, then internally"
@@ -282,7 +284,6 @@ defmodule Membrane.Core.Bin.PadController do
     with {:ok, %{availability: :on_request}} <- PadModel.get_data(state, pad_ref) do
       {pad_data, state} =
         maybe_handle_pad_removed(pad_ref, state)
-        |> Map.update!(:pad_refs, &List.delete(&1, pad_ref))
         |> PadModel.pop_data!(pad_ref)
 
       if pad_data.endpoint do
@@ -349,9 +350,15 @@ defmodule Membrane.Core.Bin.PadController do
     end
   end
 
-  defp init_pad_data(pad_ref, pad_info, state) do
+  @spec init_pad_data(Pad.ref(), State.t()) :: State.t()
+  def init_pad_data(pad_ref, state) do
+    if PadModel.assert_instance(state, pad_ref) == :ok do
+      raise "Cannot init pad data for pad #{inspect(pad_ref)}, because it already exists"
+    end
+
     pad_data =
-      pad_info
+      state.pads_info
+      |> Map.get(Pad.name_by_ref(pad_ref))
       |> Map.delete(:accepted_formats_str)
       |> Map.merge(%{
         ref: pad_ref,
@@ -360,10 +367,12 @@ defmodule Membrane.Core.Bin.PadController do
         linked?: false,
         response_received?: false,
         spec_ref: nil,
-        options: nil
+        options: nil,
+        linking_timeout_id: nil,
+        linked_in_spec?: false
       })
       |> then(&struct!(Membrane.Bin.PadData, &1))
 
-    put_in(state, [:pads_data, pad_ref], pad_data)
+    put_in(state.pads_data[pad_ref], pad_data)
   end
 end

lib/membrane/core/bin/state.ex

@@ -20,7 +20,6 @@ defmodule Membrane.Core.Bin.State do
           children: ChildrenModel.children(),
           subprocess_supervisor: pid(),
           name: Membrane.Bin.name() | nil,
-          pad_refs: [Pad.ref()],
           pads_info: PadModel.pads_info() | nil,
           pads_data: PadModel.pads_data() | nil,
           parent_pid: pid,
@@ -45,7 +44,6 @@ defmodule Membrane.Core.Bin.State do
           terminating?: boolean(),
           resource_guard: Membrane.ResourceGuard.t(),
           setup_incomplete?: boolean(),
-          linking_timeout_counters: %{optional(Pad.ref()) => integer()},
           stalker: Membrane.Core.Stalker.t()
         }
 
@@ -63,7 +61,6 @@ defmodule Membrane.Core.Bin.State do
             parent_pid: nil,
             playback: :stopped,
             internal_state: nil,
-            pad_refs: [],
             pads_info: nil,
             children: %{},
             links: %{},
@@ -77,6 +74,5 @@ defmodule Membrane.Core.Bin.State do
             resource_guard: nil,
             subprocess_supervisor: nil,
             children_log_metadata: [],
-            linking_timeout_counters: %{},
             pads_data: nil
 end

lib/membrane/core/child/pad_spec_handler.ex

@@ -21,8 +21,7 @@ defmodule Membrane.Core.Child.PadSpecHandler do
       | pads_info:
           get_pads(state)
           |> Map.new(),
-        pads_data: %{},
-        pad_refs: []
+        pads_data: %{}
     }
   end
 

lib/membrane/core/element/pad_controller.ex

@@ -229,9 +229,7 @@ defmodule Membrane.Core.Element.PadController do
       state = generate_eos_if_needed(pad_ref, state)
       state = maybe_handle_pad_removed(pad_ref, state)
 
-      {pad_data, state} =
-        Map.update!(state, :pad_refs, &List.delete(&1, pad_ref))
-        |> PadModel.pop_data!(pad_ref)
+      {pad_data, state} = PadModel.pop_data!(state, pad_ref)
 
       with %{direction: :input, flow_control: :auto, other_effective_flow_control: :pull} <-
              pad_data do
@@ -321,9 +319,7 @@ defmodule Membrane.Core.Element.PadController do
       |> then(&struct!(Membrane.Element.PadData, &1))
 
     state =
-      state
-      |> put_in([:pads_data, endpoint.pad_ref], pad_data)
-      |> Map.update!(:pad_refs, &[endpoint.pad_ref | &1])
+      put_in(state, [:pads_data, endpoint.pad_ref], pad_data)
 
     :ok =
       AtomicDemand.set_sender_status(

lib/membrane/core/element/state.ex

@@ -19,7 +19,6 @@ defmodule Membrane.Core.Element.State do
           type: Element.type(),
           name: Element.name(),
           internal_state: Element.state() | nil,
-          pad_refs: [Pad.ref()] | nil,
           pads_info: PadModel.pads_info() | nil,
           pads_data: PadModel.pads_data() | nil,
           parent_pid: pid,
@@ -65,7 +64,6 @@ defmodule Membrane.Core.Element.State do
             playback: :stopped,
             type: nil,
             internal_state: nil,
-            pad_refs: [],
             pads_info: %{},
             synchronization: nil,
             delayed_demands: MapSet.new(),

lib/membrane/core/parent/child_life_controller.ex

@@ -5,6 +5,7 @@ defmodule Membrane.Core.Parent.ChildLifeController do
   alias __MODULE__.{CrashGroupUtils, LinkUtils, StartupUtils}
   alias Membrane.{Child, ChildrenSpec}
   alias Membrane.Core.{Bin, CallbackHandler, Component, Parent, Pipeline}
+  alias Membrane.Core.Bin.PadController
 
   alias Membrane.Core.Parent.{
     ChildEntryParser,
@@ -17,6 +18,7 @@ defmodule Membrane.Core.Parent.ChildLifeController do
   alias Membrane.Pad
   alias Membrane.ParentError
 
+  require Membrane.Core.Child.PadModel, as: PadModel
   require Membrane.Core.Component
   require Membrane.Core.Message, as: Message
   require Membrane.Logger
@@ -312,17 +314,25 @@ defmodule Membrane.Core.Parent.ChildLifeController do
   defp do_proceed_spec_startup(spec_ref, %{status: :created} = spec_data, state) do
     state =
       with %Bin.State{} <- state do
-        linking_timeout_counters =
+        bin_pads_linked_in_spec =
           spec_data.links_ids
           |> Enum.map(&Map.fetch!(state.links, &1))
           |> Enum.flat_map(&[&1.from, &1.to])
-          |> Enum.filter(&(&1.child == {Membrane.Bin, :itself}))
-          |> Enum.reduce(
-            state.linking_timeout_counters,
-            &Map.update(&2, &1.pad_ref, 1, fn i -> i + 1 end)
-          )
-
-        %{state | linking_timeout_counters: linking_timeout_counters}
+          |> Enum.flat_map(fn
+            %{child: {Membrane.Bin, :itself}, pad_ref: pad_ref} -> [pad_ref]
+            _other -> []
+          end)
+
+        bin_pads_linked_in_spec
+        |> Enum.reduce(state, fn pad_ref, state ->
+          with {:error, :unknown_pad} <- PadModel.assert_instance(state, pad_ref) do
+            PadController.init_pad_data(pad_ref, state)
+            # |> Map.update!(:pad_refs, &[pad_ref | &1])
+          else
+            :ok -> state
+          end
+          |> PadModel.set_data!(pad_ref, :linked_in_spec?, true)
+        end)
       end
 
     do_proceed_spec_startup(spec_ref, %{spec_data | status: :initializing}, state)

test/membrane/core/element/pad_controller_test.exs

@@ -65,8 +65,8 @@ defmodule Membrane.Core.Element.PadControllerTest do
                  state
                )
 
-      assert Map.drop(new_state, [:pads_data, :pad_refs]) ==
-               Map.drop(state, [:pads_data, :pad_refs])
+      assert Map.delete(new_state, :pads_data) ==
+               Map.delete(state, :pads_data)
 
       assert PadModel.assert_instance(new_state, :input) == :ok
     end