WIP termination

Author: mat-hek

lib/membrane/component_path.ex

@@ -45,15 +45,15 @@ defmodule Membrane.ComponentPath do
   Returns formatted string of given path's names joined with separator.
   """
   @spec format(path_t(), String.t()) :: String.t()
-  def format(path, separator \\ "/") do
+  def format(path, separator \\ "") do
     path |> Enum.join(separator)
   end
 
   @doc """
   Works the same as `format/2` but uses currently stored path
   """
   @spec get_formatted(String.t()) :: String.t()
-  def get_formatted(separator \\ "/") do
+  def get_formatted(separator \\ "") do
     Process.get(@key, []) |> format(separator)
   end
 

lib/membrane/core/bin.ex

@@ -27,7 +27,8 @@ defmodule Membrane.Core.Bin do
           parent: pid,
           user_options: Membrane.Bin.options_t(),
           parent_clock: Membrane.Clock.t(),
-          log_metadata: Keyword.t()
+          log_metadata: Keyword.t(),
+          sync: :membrane_no_sync
         }
 
   @doc """
@@ -62,11 +63,22 @@ defmodule Membrane.Core.Bin do
       bin options: #{inspect(user_options)}
       """)
 
+      start_fun =
+        &GenServer.start_link(Membrane.Core.Bin, Map.put(options, :children_supervisor, &1))
+
       # rpc if necessary
       if node do
-        :rpc.call(node, GenServer, method, [Membrane.Core.Bin, options])
+        :rpc.call(node, Membrane.Core.Parent.Supervisor, :go_brrr, [
+          method,
+          start_fun,
+          options.setup_logger
+        ])
       else
-        apply(GenServer, method, [Membrane.Core.Bin, options])
+        Membrane.Core.Parent.Supervisor.go_brrr(
+          method,
+          start_fun,
+          options.setup_logger
+        )
       end
     else
       raise """
@@ -76,25 +88,13 @@ defmodule Membrane.Core.Bin do
     end
   end
 
-  @doc """
-  Changes bin's playback state to `:stopped` and terminates its process
-  """
-  @spec stop_and_terminate(bin :: pid) :: :ok
-  def stop_and_terminate(bin) do
-    Message.send(bin, :terminate)
-    :ok
-  end
-
   @impl GenServer
   def init(options) do
-    %{parent: parent, name: name, module: module, log_metadata: log_metadata} = options
-
-    Process.monitor(parent)
-
-    name_str = if String.valid?(name), do: name, else: inspect(name)
-    :ok = Membrane.Logger.set_prefix(name_str <> " bin")
-    :ok = Logger.metadata(log_metadata)
-    :ok = ComponentPath.set_and_append(log_metadata[:parent_path] || [], name_str <> " bin")
+    %{name: name, module: module} = options
+    self_pid = self()
+    setup_logger = fn -> options.setup_logger.(self_pid) end
+    log_metadata = setup_logger.()
+    Message.send(options.children_supervisor, :setup_logger, setup_logger)
 
     Telemetry.report_init(:bin)
 
@@ -116,7 +116,8 @@ defmodule Membrane.Core.Bin do
           stream_sync: Sync.no_sync(),
           latency: 0
         },
-        children_log_metadata: log_metadata
+        children_log_metadata: log_metadata,
+        children_supervisor: options.children_supervisor
       }
       |> Child.PadSpecHandler.init_pads()
 
@@ -129,7 +130,13 @@ defmodule Membrane.Core.Bin do
         state
       )
 
-    {:ok, state}
+    {:ok, state, {:continue, :init}}
+  end
+
+  @impl GenServer
+  def handle_continue(:init, state) do
+    state = Parent.LifecycleController.handle_setup(state)
+    {:noreply, state}
   end
 
   @impl GenServer
@@ -153,21 +160,6 @@ defmodule Membrane.Core.Bin do
     {:noreply, state}
   end
 
-  @impl GenServer
-  def handle_info(
-        Message.new(:playback_state_changed, [pid, new_playback_state]),
-        state
-      ) do
-    state = Parent.ChildLifeController.child_playback_changed(pid, new_playback_state, state)
-    {:noreply, state}
-  end
-
-  @impl GenServer
-  def handle_info(Message.new(:change_playback_state, new_state), state) do
-    state = Parent.LifecycleController.change_playback_state(new_state, state)
-    {:noreply, state}
-  end
-
   @impl GenServer
   def handle_info(Message.new(:stream_management_event, [element_name, pad_ref, event]), state) do
     state =
@@ -206,25 +198,32 @@ defmodule Membrane.Core.Bin do
   end
 
   @impl GenServer
-  def handle_info({:membrane_clock_ratio, clock, ratio}, state) do
-    state = TimerController.handle_clock_update(clock, ratio, state)
+  def handle_info(Message.new(:child_death, [name, reason]), state) do
+    state = Parent.ChildLifeController.handle_child_death(name, reason, state)
     {:noreply, state}
   end
 
   @impl GenServer
-  def handle_info({:DOWN, _ref, :process, pid, reason} = message, state) do
-    cond do
-      is_child_pid?(pid, state) ->
-        state = Parent.ChildLifeController.handle_child_death(pid, reason, state)
-        {:noreply, state}
-
-      is_parent_pid?(pid, state) ->
-        {:stop, {:shutdown, :parent_crash}, state}
-
-      true ->
-        state = Parent.LifecycleController.handle_info(message, state)
-        {:noreply, state}
-    end
+  def handle_info(Message.new(:play), state) do
+    state = Parent.LifecycleController.handle_play(state)
+    {:noreply, state}
+  end
+
+  @impl GenServer
+  def handle_info(Message.new(:initialized, child), state) do
+    state = Parent.ChildLifeController.handle_child_initialized(child, state)
+    {:noreply, state}
+  end
+
+  @impl GenServer
+  def handle_info(Message.new(_type, _args, _opts) = message, _state) do
+    raise Membrane.BinError, "Received invalid message #{inspect(message)}"
+  end
+
+  @impl GenServer
+  def handle_info({:membrane_clock_ratio, clock, ratio}, state) do
+    state = TimerController.handle_clock_update(clock, ratio, state)
+    {:noreply, state}
   end
 
   @impl GenServer
@@ -251,14 +250,6 @@ defmodule Membrane.Core.Bin do
   @impl GenServer
   def terminate(reason, state) do
     Telemetry.report_terminate(:bin)
-    :ok = state.module.handle_shutdown(reason, state.internal_state)
-  end
-
-  defp is_parent_pid?(pid, state) do
-    state.parent_pid == pid
-  end
-
-  defp is_child_pid?(pid, state) do
-    Enum.any?(state.children, fn {_name, entry} -> entry.pid == pid end)
+    Parent.LifecycleController.handle_terminate(reason, state)
   end
 end

lib/membrane/core/bin/action_handler.ex

@@ -22,7 +22,7 @@ defmodule Membrane.Core.Bin.ActionHandler do
 
   @impl CallbackHandler
   def handle_action({:remove_child, children}, _cb, _params, state) do
-    Parent.ChildLifeController.handle_remove_child(children, state)
+    Parent.ChildLifeController.handle_remove_children(children, state)
   end
 
   @impl CallbackHandler

lib/membrane/core/bin/state.ex

@@ -43,7 +43,7 @@ defmodule Membrane.Core.Bin.State do
           pending_specs: LinkHandler.pending_specs_t()
         }
 
-  @enforce_keys [:module, :synchronization]
+  @enforce_keys [:module, :synchronization, :children_supervisor]
   defstruct @enforce_keys ++
               [
                 internal_state: nil,
@@ -57,6 +57,9 @@ defmodule Membrane.Core.Bin.State do
                 crash_groups: %{},
                 children_log_metadata: [],
                 links: [],
-                pending_specs: %{}
+                pending_specs: %{},
+                status: :initializing,
+                play_request?: false,
+                terminating?: false
               ]
 end

lib/membrane/core/element.ex

@@ -30,7 +30,7 @@ defmodule Membrane.Core.Element do
     State
   }
 
-  alias Membrane.Core.{PlaybackHandler, TimerController}
+  alias Membrane.Core.TimerController
 
   require Membrane.Core.Message, as: Message
   require Membrane.Core.Telemetry, as: Telemetry
@@ -89,35 +89,11 @@ defmodule Membrane.Core.Element do
     end
   end
 
-  @doc """
-  Stops given element process.
-
-  It will wait for reply for amount of time passed as second argument
-  (in milliseconds).
-
-  Will trigger calling `c:Membrane.Element.Base.handle_shutdown/2`
-  callback.
-  """
-  @spec shutdown(pid, timeout) :: :ok
-  def shutdown(server, timeout \\ 5000) do
-    GenServer.stop(server, :normal, timeout)
-    :ok
-  end
-
   @impl GenServer
   def init(options) do
-    %{parent: parent, name: name, log_metadata: log_metadata} = options
-
-    Process.monitor(parent)
-    name_str = if String.valid?(name), do: name, else: inspect(name)
-    :ok = Membrane.Logger.set_prefix(name_str)
-    :ok = Logger.metadata(log_metadata)
-    :ok = ComponentPath.set_and_append(log_metadata[:parent_path] || [], name_str)
-
+    options.setup_logger.(self())
     Telemetry.report_init(:element)
-
     state = Map.take(options, [:module, :name, :parent_clock, :sync, :parent]) |> State.new()
-
     state = LifecycleController.handle_init(options.user_options, state)
     {:ok, state, {:continue, :init}}
   end
@@ -131,7 +107,7 @@ defmodule Membrane.Core.Element do
   @impl GenServer
   def terminate(reason, state) do
     Telemetry.report_terminate(:element)
-    LifecycleController.handle_shutdown(reason, state)
+    LifecycleController.handle_terminate(reason, state)
   end
 
   @impl GenServer
@@ -161,12 +137,6 @@ defmodule Membrane.Core.Element do
           "Received invalid message #{inspect(message)} from #{inspect(pid)}"
   end
 
-  @impl GenServer
-  def handle_info({:DOWN, _ref, :process, parent_pid, reason}, %{parent_pid: parent_pid} = state) do
-    :ok = LifecycleController.handle_pipeline_down(reason, state)
-    {:stop, {:shutdown, :parent_crash}, state}
-  end
-
   @impl GenServer
   def handle_info(message, state) do
     Telemetry.report_metric(
@@ -179,17 +149,6 @@ defmodule Membrane.Core.Element do
 
   @compile {:inline, do_handle_info: 2}
 
-  defp do_handle_info(Message.new(:change_playback_state, new_playback_state), state) do
-    {:ok, state} =
-      PlaybackHandler.change_playback_state(
-        new_playback_state,
-        Element.LifecycleController,
-        state
-      )
-
-    {:noreply, state}
-  end
-
   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)
@@ -229,20 +188,25 @@ defmodule Membrane.Core.Element do
     {:noreply, state}
   end
 
-  defp do_handle_info({:membrane_clock_ratio, clock, ratio}, state) do
-    state = TimerController.handle_clock_update(clock, ratio, state)
-    {:noreply, state}
-  end
-
   defp do_handle_info(Message.new(:parent_notification, notification), state) do
     state = Core.Child.LifecycleController.handle_parent_notification(notification, state)
     {:noreply, state}
   end
 
+  defp do_handle_info(Message.new(:terminate), state) do
+    state = LifecycleController.handle_terminate_request(state)
+    {:stop, :normal, state}
+  end
+
   defp do_handle_info(Message.new(_type, _args, _opts) = message, _state) do
     raise Membrane.ElementError, "Received invalid message #{inspect(message)}"
   end
 
+  defp do_handle_info({:membrane_clock_ratio, clock, ratio}, state) do
+    state = TimerController.handle_clock_update(clock, ratio, state)
+    {:noreply, state}
+  end
+
   defp do_handle_info(message, state) do
     state = LifecycleController.handle_info(message, state)
     {:noreply, state}

lib/membrane/core/element/action_handler.ex

@@ -10,8 +10,8 @@ defmodule Membrane.Core.Element.ActionHandler do
 
   alias Membrane.{ActionError, Buffer, Caps, ElementError, Event, Pad, PadDirectionError}
   alias Membrane.Core.Child.PadModel
-  alias Membrane.Core.Element.{DemandHandler, LifecycleController, PadController, State}
-  alias Membrane.Core.{Events, Message, PlaybackHandler, TimerController}
+  alias Membrane.Core.Element.{DemandHandler, PadController, State}
+  alias Membrane.Core.{Events, Message, TimerController}
   alias Membrane.Core.Telemetry
   alias Membrane.Element.Action
 
@@ -72,24 +72,6 @@ defmodule Membrane.Core.Element.ActionHandler do
     )
   end
 
-  @impl CallbackHandler
-  def handle_action({:playback_change, :suspend}, cb, _params, state)
-      when cb in [
-             :handle_stopped_to_prepared,
-             :handle_playing_to_prepared,
-             :handle_prepared_to_playing,
-             :handle_prepared_to_stopped
-           ] do
-    {:ok, state} = PlaybackHandler.suspend_playback_change(state)
-    state
-  end
-
-  @impl CallbackHandler
-  def handle_action({:playback_change, :resume}, _cb, _params, state) do
-    {:ok, state} = PlaybackHandler.continue_playback_change(LifecycleController, state)
-    state
-  end
-
   @impl CallbackHandler
   def handle_action({:buffer, _args} = action, cb, _params, %State{
         playback: %{state: playback_state}