Trigger diamon detection between elements WiP

lib/membrane/core/bin.ex

@@ -239,16 +239,6 @@ defmodule Membrane.Core.Bin do
     {:noreply, state}
   end
 
-  # defp do_handle_info(Message.new(:start_diamond_detection), state) do
-  #   :ok = Parent.DiamondDetectionController.start_diamond_detection(state)
-  #   {:noreply, state}
-  # end
-
-  # defp do_handle_info(Message.new(:trigger_diamond_detection), state) do
-  #   :ok = __MODULE__.DiamondDetectionController.trigger_diamond_detection(state)
-  #   {:noreply, state}
-  # end
-
   defp do_handle_info(Message.new(:child_death, [name, reason]), state) do
     case Parent.ChildLifeController.handle_child_death(name, reason, state) do
       {:stop, reason, _state} -> ProcessHelper.notoelo(reason)

lib/membrane/core/element.ex

@@ -306,6 +306,21 @@ defmodule Membrane.Core.Element do
     {:noreply, state}
   end
 
+  defp do_handle_info(Message.new(:start_diamond_detection_trigger, trigger_ref), state) do
+    state = DiamondDetectionController.start_diamond_detection_trigger(trigger_ref, state)
+    {:noreply, state}
+  end
+
+  defp do_handle_info(Message.new(:diamond_detection_trigger, trigger_ref), state) do
+    state = DiamondDetectionController.handle_diamond_detection_trigger(trigger_ref, state)
+    {:noreply, state}
+  end
+
+  defp do_handle_info(Message.new(:delete_diamond_detection_trigger_ref, trigger_ref), state) do
+    state = DiamondDetectionController.delete_diamond_detection_trigger_ref(trigger_ref, 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/diamond_detection_controller.ex

@@ -8,6 +8,8 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
   alias Membrane.Core.Element.State
   alias Membrane.Element.PadData
 
+  # TODO: don't forward diamond detection and triggers in endpoints
+
   @type diamond_detection_path :: [{pid(), Child.name(), Pad.ref()}]
 
   @spec start_diamond_detection(State.t()) :: :ok
@@ -71,7 +73,7 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
   defp forward_diamond_detection_trigger(trigger_ref, state) do
     state.pads_data
     |> Enum.each(fn {_pad_ref, %PadData{} = pad_data} ->
-      if pad_data.direction == :output and pad_data.flow_control != :push do
+      if pad_data.direction == :input and pad_data.flow_control != :push do
         Message.send(pad_data.pid, :diamond_detection_trigger, trigger_ref)
       end
     end)
@@ -88,16 +90,20 @@ defmodule Membrane.Core.Element.DiamondDetectionController do
     uniq_length < length(diamond_detection_path)
   end
 
-  def start_diamond_detection_trigger(spec_ref, state) when map_size(state.pads_data) >= 2 do
-    handle_diamond_detection_trigger(spec_ref, state)
+  def start_diamond_detection_trigger(spec_ref, state) do
+    if map_size(state.pads_data) < 2 or
+         MapSet.member?(state.diamond_detection_trigger_refs, spec_ref) do
+      state
+    else
+      do_handle_diamond_detection_trigger(spec_ref, state)
+    end
   end
 
-  def start_diamond_detection_trigger(_spec_ref, state), do: state
-
   def handle_diamond_detection_trigger(trigger_ref, %State{} = state) do
-    if MapSet.member?(state.diamond_detection_trigger_refs, trigger_ref),
-      do: state,
-      else: do_handle_diamond_detection_trigger(trigger_ref, state)
+    if state.type == :endpoint or
+         MapSet.member?(state.diamond_detection_trigger_refs, trigger_ref),
+       do: state,
+       else: do_handle_diamond_detection_trigger(trigger_ref, state)
   end
 
   defp do_handle_diamond_detection_trigger(trigger_ref, %State{} = state) do

lib/membrane/core/parent/child_life_controller.ex

@@ -11,6 +11,7 @@ defmodule Membrane.Core.Parent.ChildLifeController do
     ChildEntryParser,
     ChildrenModel,
     ClockHandler,
+    DiamondDetectionController,
     Link,
     SpecificationParser
   }
@@ -163,23 +164,10 @@ defmodule Membrane.Core.Parent.ChildLifeController do
         awaiting_responses: MapSet.new()
       })
 
-    # |> trigger_diamond_detection()
-
     state = StartupUtils.exec_handle_spec_started(all_children_names, state)
     proceed_spec_startup(spec_ref, state)
   end
 
-  # defp trigger_diamond_detection(state) do
-  #   cond do
-  #     Component.bin?(state) ->
-  #       :ok = Bin.DiamondDetectionController.trigger_diamond_detection(state)
-  #       state
-
-  #     Component.pipeline?(state) ->
-  #       Pipeline.DiamondDetectionController.trigger_diamond_detection(state)
-  #   end
-  # end
-
   defp assert_all_static_pads_linked_in_spec!(children_definitions, links) do
     pads_to_link =
       links
@@ -317,8 +305,6 @@ defmodule Membrane.Core.Parent.ChildLifeController do
     withl spec_data: {:ok, spec_data} <- Map.fetch(state.pending_specs, spec_ref),
           do: {spec_data, state} = do_proceed_spec_startup(spec_ref, spec_data, state),
           status: :ready <- spec_data.status do
-
-
       cleanup_spec_startup(spec_ref, state)
     else
       spec_data: :error -> state
@@ -440,7 +426,7 @@ defmodule Membrane.Core.Parent.ChildLifeController do
     end
   end
 
-  defp do_proceed_spec_startup(_spec_ref, %{status: :ready} = spec_data, state) do
+  defp do_proceed_spec_startup(spec_ref, %{status: :ready} = spec_data, state) do
     state =
       Enum.reduce(spec_data.children_names, state, fn child, state ->
         %{pid: pid, terminating?: terminating?} = state.children[child]
@@ -454,6 +440,13 @@ defmodule Membrane.Core.Parent.ChildLifeController do
         put_in(state.children[child].ready?, true)
       end)
 
+    spec_data.links_ids
+    |> Enum.map(&state.links[&1].from.name)
+    |> Enum.uniq()
+    |> Enum.each(fn child_name ->
+      DiamondDetectionController.start_diamond_detection_trigger(child_name, spec_ref, state)
+    end)
+
     state =
       with %{playback: :playing} <- state do
         handle_children_playing(spec_data.children_names, state)

lib/membrane/core/parent/diamond_detection_controller.ex

@@ -1,15 +1,17 @@
-# defmodule Membrane.Core.Parent.DiamondDetectionController do
-#   @moduledoc false
+defmodule Membrane.Core.Parent.DiamondDetectionController do
+  @moduledoc false
 
-#   require Membrane.Core.Message, as: Message
+  require Membrane.Core.Message, as: Message
 
-#   alias Membrane.Core.Parent
+  alias Membrane.Child
+  alias Membrane.Core.Parent
 
-#   @spec start_diamond_detection(Parent.state()) :: :ok
-#   def start_diamond_detection(state) do
-#     state.children
-#     |> Enum.each(fn {_child, %{pid: child_pid}} ->
-#         Message.send(child_pid, :start_diamond_detection)
-#     end)
-#   end
-# end
+  @spec start_diamond_detection_trigger(Child.name(), reference(), Parent.state()) :: :ok
+  def start_diamond_detection_trigger(child_name, trigger_ref, state) do
+    with %{component_type: :element, pid: pid} <- state.children[child_name] do
+      Message.send(pid, :start_diamond_detection_trigger, trigger_ref)
+    end
+
+    :ok
+  end
+end

lib/membrane/core/pipeline.ex

@@ -130,16 +130,6 @@ defmodule Membrane.Core.Pipeline do
     {:noreply, state}
   end
 
-  # defp do_handle_info(Message.new(:start_diamond_detection), state) do
-  #   state = __MODULE__.DiamondDetectionController.start_diamond_detection(state)
-  #   {:noreply, state}
-  # end
-
-  # defp do_handle_info(Message.new(:trigger_diamond_detection), state) do
-  #   state = __MODULE__.DiamondDetectionController.trigger_diamond_detection(state)
-  #   {:noreply, state}
-  # end
-
   defp do_handle_info(Message.new(:initialized, child), state) do
     state = ChildLifeController.handle_child_initialized(child, state)
     {:noreply, state}