Detect diamonds

lib/membrane/core/element/diamond_detection_controller.ex

@@ -10,16 +10,17 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
 
   # This algorithm takes the directed graph made by all elements within a single pipeline and
   # finds some diamond-subgraphs where all the edges (links) work in the :pull mode,
-  # it means in :manual flow control or in :auto flow control if the effective flow control
-  # is set to :pull.
+  # that is they're either in the :manual flow control or in the :auto flow control and the
+  # effective flow control is set to :pull.
 
   # These diamonds can be dangerous when used with pull flow control, e.g. let's consider
   # a pipeline that contains:
   #   * MP4 demuxer that has two output pads
   #   * MKV muxer that is linked to both of them
-  # and let's assume that the MP4 container that is consumed by the MP4 demuxer is unbalanced.
+  # and let's assume that the MP4 file that is consumed by the MP4 demuxer is unbalanced
+  # (audio and video streams are not interleaved, for example audio comes first and then video)
   # If the MKV muxer has pads working in pull mode, then demand on one pad will be satisfied,
-  # but on the other won't, because the MP4 container is unbalanced. Then, if MP4 demuxer
+  # but on the other won't, because the source MP4 file is unbalanced. Then, if the MP4 demuxer
   # has pads in auto flow control and its effective flow control is set to :pull, it won't
   # demand on the input, because one of the pads output with :auto flow control doesn't
   # have positive demand, so the whole pipeline will get stuck and won't process more data.
@@ -30,24 +31,24 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
 
   # Let's notice that:
   #   * a new diamond can be created only after linking a new spec
-  #   * if the new spec caused some new diamond to occur, this diamond will contain some of
-  # the links spawned in this spec
+  #   * if the new spec created a new diamond, this diamond will contain some of
+  #     the links spawned in this spec
   # If the diamond contains a link, it must also contain an element whose output pad
   # is part of this link.
 
-  # After the spec status is set to :done, the parent component that returned the spec will trigger
-  # all elements whose output pads have been linked in this spec. The reference of the trigger
-  # is always set to the spec reference.
+  # After the spec status is set to :done, the parent component that returned the spec will
+  # triggerall elements whose output pads have been linked in this spec (`type: :start_trigger`).
+  # The reference of the trigger is always set to the spec reference.
 
-  # If the element is triggered with a specific reference for the first time, it does two
-  # things:
+  # If the element is triggered with a specific reference (`type: :trigger`) for the first time,
+  # it does two things:
   #   * the element forwards the trigger with the same reference via all input pads working
-  #     in the pull mode
+  #     in the pull mode (`type: :trigger`)
   #   * if the element has at least two output pads working in the pull mode, it postpones
-  #     the proper searching that will be spawned from itself. The time between postponing and the
-  #     proper searching is one second. If during this time an element is triggered once
-  #     again with a different reference, it won't cause another postponement of the proper
-  #     searching, this means that at the time there is at most one proper searching
+  #     the proper searching that will be spawned from itself (`type: :start_search`). The time
+  #     between postponing and the proper searching is one second. If during this time an element
+  #     is triggered once again with a different reference, it won't cause another postponement
+  #     of the proper searching, this means that at the time there is at most one proper searching
   #     postponed in the single element
 
   # (2) Proper searching:
@@ -60,6 +61,7 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
   # just started the proper searching, or maybe it was forwarded to it via a link):
   #   * if the element sees the proper searching reference for the first time, then:
   #     - it forwards proper searching via all output pads working in the pull mode
+  #       (`type: :search`)
   #     - when proper searching is forwarded, it remembers the path in the graph through
   #       the elements that it has already passed
   #   * if the element has already seen the reference of proper searching, but there is
@@ -72,23 +74,20 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
   #     and doesn't forward proper searching further.
 
   alias __MODULE__.{DiamondLogger, PathInGraph}
-  alias __MODULE__.PathInGraph.Vertex
   alias Membrane.Core.Element.State
   alias Membrane.Element.PadData
 
   require Membrane.Core.Message, as: Message
   require Membrane.Logger
   require Membrane.Pad, as: Pad
 
-  @component_path_prefix "__membrane_component_path_64_byte_prefix________________________"
-
   @type diamond_detection_message() :: %{
           :type =>
-            :start
+            :start_search
+            | :search
+            | :delete_search_ref
             | :start_trigger
-            | :diamond_detection
             | :trigger
-            | :delete_ref
             | :delete_trigger_ref,
           optional(:ref) => reference(),
           optional(:path) => PathInGraph.t(),
@@ -98,50 +97,51 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
   @spec handle_diamond_detection_message(diamond_detection_message(), State.t()) :: State.t()
   def handle_diamond_detection_message(%{type: type} = message, state) do
     case type do
-      :start ->
-        start_diamond_detection(state)
+      :start_search ->
+        :ok = start_search(state)
+        state
 
-      :start_trigger ->
-        start_diamond_detection_trigger(message.ref, state)
+      :search ->
+        handle_and_forward_search(message.pad_ref, message.ref, message.path, state)
 
-      :diamond_detection ->
-        continue_diamond_detection(message.pad_ref, message.ref, message.path, state)
+      :delete_search_ref ->
+        delete_search_ref(message.ref, state)
 
-      :trigger ->
-        handle_diamond_detection_trigger(message.ref, state)
+      :start_trigger ->
+        start_trigger(message.ref, state)
 
-      :delete_ref ->
-        delete_diamond_detection_ref(message.ref, state)
+      :trigger ->
+        handle_and_forward_trigger(message.ref, state)
 
       :delete_trigger_ref ->
-        delete_diamond_detection_trigger_ref(message.ref, state)
+        delete_trigger_ref(message.ref, state)
     end
   end
 
-  @spec start_diamond_detection(State.t()) :: State.t()
-  defp start_diamond_detection(state) do
-    {component_path, state} = get_component_path(state)
+  @spec start_search(State.t()) :: :ok
+  defp start_search(state) do
+    component_path = Membrane.ComponentPath.get_formatted()
 
     diamond_detection_path = [
       %PathInGraph.Vertex{pid: self(), component_path: component_path}
     ]
 
     :ok =
       make_ref()
-      |> forward_diamond_detection(diamond_detection_path, state)
+      |> forward_search(diamond_detection_path, state)
 
-    state
+    :ok
   end
 
-  @spec continue_diamond_detection(Pad.ref(), reference(), PathInGraph.t(), State.t()) ::
+  @spec handle_and_forward_search(Pad.ref(), reference(), PathInGraph.t(), State.t()) ::
           State.t()
-  defp continue_diamond_detection(
+  defp handle_and_forward_search(
          input_pad_ref,
          diamond_detection_ref,
          diamond_detecton_path,
          state
        ) do
-    {component_path, state} = get_component_path(state)
+    component_path = Membrane.ComponentPath.get_formatted()
 
     new_path_vertex = %PathInGraph.Vertex{
       pid: self(),
@@ -152,16 +152,16 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
     diamond_detecton_path = [new_path_vertex | diamond_detecton_path]
 
     cond do
-      not is_map_key(state.diamond_detection.ref_to_path, diamond_detection_ref) ->
-        :ok = forward_diamond_detection(diamond_detection_ref, diamond_detecton_path, state)
+      not is_map_key(state.diamond_detection_state.ref_to_path, diamond_detection_ref) ->
+        :ok = forward_search(diamond_detection_ref, diamond_detecton_path, state)
 
         :ok =
-          %{type: :delete_ref, ref: diamond_detection_ref}
+          %{type: :delete_search_ref, ref: diamond_detection_ref}
           |> send_after_to_self()
 
         state
         |> put_in(
-          [:diamond_detection, :ref_to_path, diamond_detection_ref],
+          [:diamond_detection_state, :ref_to_path, diamond_detection_ref],
           diamond_detecton_path
         )
 
@@ -170,35 +170,30 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
 
       have_common_prefix?(
         diamond_detecton_path,
-        state.diamond_detection.ref_to_path[diamond_detection_ref]
+        state.diamond_detection_state.ref_to_path[diamond_detection_ref]
       ) ->
         state
 
       true ->
-        old_diamond_detection_path =
-          state.diamond_detection.ref_to_path[diamond_detection_ref]
-          |> remove_component_path_prefix()
-
         :ok =
-          diamond_detecton_path
-          |> remove_component_path_prefix()
-          |> DiamondLogger.log_diamond(old_diamond_detection_path)
+          state.diamond_detection_state.ref_to_path[diamond_detection_ref]
+          |> DiamondLogger.log_diamond(diamond_detecton_path)
 
         state
     end
   end
 
-  @spec delete_diamond_detection_ref(reference(), State.t()) :: State.t()
-  defp delete_diamond_detection_ref(diamond_detection_ref, state) do
+  @spec delete_search_ref(reference(), State.t()) :: State.t()
+  defp delete_search_ref(diamond_detection_ref, state) do
     {_path, %State{} = state} =
       state
-      |> pop_in([:diamond_detection, :ref_to_path, diamond_detection_ref])
+      |> pop_in([:diamond_detection_state, :ref_to_path, diamond_detection_ref])
 
     state
   end
 
-  @spec forward_diamond_detection(reference(), PathInGraph.t(), State.t()) :: :ok
-  defp forward_diamond_detection(diamond_detection_ref, diamond_detection_path, state) do
+  @spec forward_search(reference(), PathInGraph.t(), State.t()) :: :ok
+  defp forward_search(diamond_detection_ref, diamond_detection_path, state) do
     auto_pull_mode? = state.effective_flow_control == :pull
     [current_entry | diamond_detection_path_tail] = diamond_detection_path
 
@@ -209,7 +204,7 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
         diamond_detection_path = [current_entry | diamond_detection_path_tail]
 
         message = %{
-          type: :diamond_detection,
+          type: :search,
           pad_ref: pad_data.other_ref,
           ref: diamond_detection_ref,
           path: diamond_detection_path
@@ -241,29 +236,29 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
     uniq_length < length(diamond_detection_path)
   end
 
-  @spec start_diamond_detection_trigger(reference(), State.t()) :: State.t()
-  defp start_diamond_detection_trigger(spec_ref, state) do
+  @spec start_trigger(reference(), State.t()) :: State.t()
+  defp start_trigger(spec_ref, state) do
     if map_size(state.pads_data) < 2 or
-         MapSet.member?(state.diamond_detection.trigger_refs, spec_ref) do
+         MapSet.member?(state.diamond_detection_state.trigger_refs, spec_ref) do
       state
     else
-      do_handle_diamond_detection_trigger(spec_ref, state)
+      do_handle_and_forward_trigger(spec_ref, state)
     end
   end
 
-  @spec handle_diamond_detection_trigger(reference(), State.t()) :: State.t()
-  defp handle_diamond_detection_trigger(trigger_ref, %State{} = state) do
+  @spec handle_and_forward_trigger(reference(), State.t()) :: State.t()
+  defp handle_and_forward_trigger(trigger_ref, %State{} = state) do
     if state.type == :endpoint or
-         MapSet.member?(state.diamond_detection.trigger_refs, trigger_ref),
+         MapSet.member?(state.diamond_detection_state.trigger_refs, trigger_ref),
        do: state,
-       else: do_handle_diamond_detection_trigger(trigger_ref, state)
+       else: do_handle_and_forward_trigger(trigger_ref, state)
   end
 
-  defp do_handle_diamond_detection_trigger(trigger_ref, %State{} = state) do
+  defp do_handle_and_forward_trigger(trigger_ref, %State{} = state) do
     state =
       state
       |> update_in(
-        [:diamond_detection, :trigger_refs],
+        [:diamond_detection_state, :trigger_refs],
         &MapSet.put(&1, trigger_ref)
       )
 
@@ -278,22 +273,23 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
       else: state
   end
 
-  defp postpone_diamond_detection(%State{} = state) when state.diamond_detection.postponed? do
+  defp postpone_diamond_detection(%State{} = state)
+       when state.diamond_detection_state.postponed? do
     state
   end
 
   defp postpone_diamond_detection(%State{} = state) do
-    :ok = %{type: :start} |> send_after_to_self(1)
+    :ok = %{type: :start_search} |> send_after_to_self(1)
 
     state
-    |> put_in([:diamond_detection, :postponed?], true)
+    |> put_in([:diamond_detection_state, :postponed?], true)
   end
 
-  @spec delete_diamond_detection_trigger_ref(reference(), State.t()) :: State.t()
-  defp delete_diamond_detection_trigger_ref(trigger_ref, state) do
+  @spec delete_trigger_ref(reference(), State.t()) :: State.t()
+  defp delete_trigger_ref(trigger_ref, state) do
     state
     |> update_in(
-      [:diamond_detection, :trigger_refs],
+      [:diamond_detection_state, :trigger_refs],
       &MapSet.delete(&1, trigger_ref)
     )
   end
@@ -312,36 +308,5 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
     :ok
   end
 
-  defp get_component_path(state) do
-    case state.diamond_detection.serialized_component_path do
-      nil ->
-        # adding @component_path_prefix to component path causes that component path
-        # always has more than 64 bytes, so it won't be copied during sending a message
-
-        component_path =
-          [@component_path_prefix | Membrane.ComponentPath.get()]
-          |> Enum.join()
-
-        state =
-          state
-          |> put_in(
-            [:diamond_detection, :serialized_component_path],
-            component_path
-          )
-
-        {component_path, state}
-
-      component_path ->
-        {component_path, state}
-    end
-  end
-
   defp have_common_prefix?(path_a, path_b), do: List.last(path_a) == List.last(path_b)
-
-  defp remove_component_path_prefix(path_in_graph) do
-    path_in_graph
-    |> Enum.map(fn %Vertex{component_path: @component_path_prefix <> component_path} = vertex ->
-      %{vertex | component_path: component_path}
-    end)
-  end
 end

lib/membrane/core/element/diamond_detection_controller/diamond_detection_state.ex

@@ -0,0 +1,17 @@
+defmodule Membrane.Core.Element.DiamondDetectionController.DiamondDatectionState do
+  @moduledoc false
+
+  use Bunch.Access
+
+  alias Membrane.Core.Element.DiamondDetectionController.PathInGraph
+
+  defstruct ref_to_path: %{},
+            trigger_refs: MapSet.new(),
+            postponed?: false
+
+  @type t :: %__MODULE__{
+          ref_to_path: %{optional(reference()) => PathInGraph.t()},
+          trigger_refs: MapSet.t(reference()),
+          postponed?: boolean()
+        }
+end