Playback refactor (#432)

* remove parent message dispatcher and genserver helper

* POC pipeline & elements startup

* termination mvp

* children removal bug fixes

* add linking order test, fix getting clock and bin termination

* remove playback state checks, fix tests

* remove status and add playback

* don't remove children until they're linked

* fix long running sync test

* Add option to name processes to ease debugging with observer

* make dialyzer happy

* fix linking test

* fixed most credo issues

* improve pipeline initialization

* using dynamic filter from test/support

* docs update

* add terminate action

* improve logs in supervisors

* ignore playback action when terminating

* Update lib/membrane/core/bin/action_handler.ex

Co-authored-by: Łukasz Kita <[email protected]>

* Update lib/membrane/core/pipeline/action_handler.ex

Co-authored-by: Łukasz Kita <[email protected]>

* Update lib/membrane/pipeline.ex

Co-authored-by: Łukasz Kita <[email protected]>

* Update lib/membrane/pipeline.ex

Co-authored-by: Łukasz Kita <[email protected]>

* start remote controlled pipeline with start_supervised in tests

* fixes for CR

* add test for element play

* fixes for CR

Co-authored-by: Bartosz Błaszków <[email protected]>

* handle_play -> handle_playing

* play_request? -> playing_requested?

* Play -> Playing

* Apply suggestions from code review

Co-authored-by: Bartosz Błaszków <[email protected]>

* refactor component path

* Improved ChildrenSupervisor, ResourceGuard and Observability (#447)

* supervisors refactor

* resource guard

* add utility supervisor to context

* refactor element tests not to use State.new

* Update lib/membrane/resource_guard.ex

Co-authored-by: Bartosz Błaszków <[email protected]>

* run resource cleanup in a separate process

* refactor observability setup

* improve children supervisor API

* add docs and test for utility supervisor and resource guard

* add timeout to resource

* use a boring function name instead of go_brrr

* unlink cleanup task in resource guard before killing it

* Add list of available callbacks in each component (#449)

* Add an aggregated list of all callbacks that can be implemented by `Filter`, `Endpoint`, `Source` and `Sink` element behaviours.

Co-authored-by: Mateusz Front <[email protected]>

* add docs for Observability.setup

Co-authored-by: Bartosz Błaszków <[email protected]>
Co-authored-by: Łukasz Kita <[email protected]>

* Playback termination (#451)

* handle_terminate_request in parent

* add handle_terminate_request to element

* fix unlinking pads

* add terminate action type to element and bin

* Rewrite espec tests (#452)

* Remove unecessary ESpec tests
* Rewrite some crucial ESpec tests
* Unify the support modules from ExUnit and ESpec

* fix linking test

* fix resource guard test

* Apply suggestions from code review

Co-authored-by: Bartosz Błaszków <[email protected]>
Co-authored-by: Łukasz Kita <[email protected]>

* Update lib/membrane/bin.ex

Co-authored-by: Bartosz Błaszków <[email protected]>

* CallbackContext.Other -> Info

* fix typos <3

* Callback contexts for handle_terminate_request

* add docs for zombies

* use ResourceGuard to report termination via telemetry

* Add a test to check the links between processes in the pipeline_test.exs. Fix the bug in the test case

* add deferred termination tests

* add description for children supervisor

* fix RC in child removal test

* ChildrenSupervisor -> SubprocessSupervisor

Co-authored-by: Łukasz Kita <[email protected]>
Co-authored-by: Bartosz Błaszków <[email protected]>

* Fix spawning children on remote nodes (#456)

* fix spawning children on remote nodes

* remove redundant checks in element and bin

* Fix filter aggregator (#455)

* termporairly fix distribution

* re-enable filter aggregator tests

* make filter aggregator work again

* add context to handle_init

* unify handle_init default impl across components

* filter aggregator test improvements

* remove redundant match

* Fixes for CR

Co-authored-by: Łukasz Kita <[email protected]>

* RemoteControlled.Pipeline await_play -> await_playing

* put back relevant tests

Co-authored-by: Łukasz Kita <[email protected]>
Co-authored-by: Bartosz Błaszków <[email protected]>

Author: 3 people

CHANGELOG.md

@@ -11,7 +11,7 @@
 ## 0.10.0
  * Remove all deprecated stuff [#399](https://github.com/membraneframework/membrane_core/pull/399)
  * Make `Membrane.Pipeline.{prepare, play, stop}` deprecated and add `:playback` action instead
- * Make `Membrane.Pipeline.stop_and_terminate/2` deprecated and add `Membrane.Pipeline.terminate/2` instead
+ * Make `Membrane.Pipeline.stop_and_terminate` deprecated and add `Membrane.Pipeline.terminate/2` instead
  * Add `Membrane.RemoteControlled.Pipeline` - a basic implementation of a `Membrane.Pipeline` that </br>
    can be spawned and controlled by an external process [#366](https://github.com/membraneframework/membrane_core/pull/366)
  * Disallow sending buffers without sending caps first [#341](https://github.com/membraneframework/membrane_core/issues/341)

assets/images/observer_graph.png

@@ -26,7 +26,7 @@ defmodule Membrane.Bin do
 
   @typedoc """
   Defines options that can be passed to `start_link/3` and received
-  in `c:handle_init/1` callback.
+  in `c:handle_init/2` callback.
   """
   @type options_t :: struct | nil
 
@@ -36,23 +36,19 @@ defmodule Membrane.Bin do
   @type name_t :: any()
 
   @doc """
-  Callback invoked on initialization of bin process. It should parse options
-  and initialize bin's internal state. Internally it is invoked inside
-  `c:GenServer.init/1` callback.
-  """
-  @callback handle_init(options :: options_t) :: callback_return_t()
-
-  @doc """
-  Callback invoked when bin is shutting down.
-  Internally called in `c:GenServer.terminate/2` callback.
+  Callback invoked on initialization of bin.
 
-  Useful for any cleanup required.
+  This callback is synchronous: the parent waits until it finishes. Also, any failures
+  that happen in this callback crash the parent as well, regardless of crash groups.
+  For these reasons, it's important to do any long-lasting or complex work in `c:handle_setup/2`,
+  while `handle_init` should be used for things like parsing options, initializing state or
+  spawning children.
   """
-  @callback handle_shutdown(reason, state :: state_t) :: :ok
-            when reason: :normal | :shutdown | {:shutdown, any} | term()
+  @callback handle_init(context :: CallbackContext.Init.t(), options :: options_t) ::
+              callback_return_t()
 
   @doc """
-  Callback that is called when new pad has beed added to bin. Executed
+  Callback that is called when new pad has been added to bin. Executed
   ONLY for dynamic pads.
   """
   @callback handle_pad_added(
@@ -62,7 +58,7 @@ defmodule Membrane.Bin do
             ) :: callback_return_t
 
   @doc """
-  Callback that is called when some pad of the bin has beed removed. Executed
+  Callback that is called when some pad of the bin has been removed. Executed
   ONLY for dynamic pads.
   """
   @callback handle_pad_removed(
@@ -72,54 +68,24 @@ defmodule Membrane.Bin do
             ) :: callback_return_t
 
   @doc """
-  Callback invoked when bin transition from `:stopped` to `:prepared` state has finished,
-  that is all of its children are prepared to enter `:playing` state.
-  """
-  @callback handle_stopped_to_prepared(
-              context :: CallbackContext.PlaybackChange.t(),
-              state :: state_t
-            ) ::
-              callback_return_t
+  Callback invoked on bin startup, right after `c:handle_init/2`.
 
-  @doc """
-  Callback invoked when bin transition from `:playing` to `:prepared` state has finished,
-  that is all of its children are prepared to be stopped.
+  Any long-lasting or complex initialization should happen here.
   """
-  @callback handle_playing_to_prepared(
-              context :: CallbackContext.PlaybackChange.t(),
+  @callback handle_setup(
+              context :: CallbackContext.Setup.t(),
               state :: state_t
-            ) ::
-              callback_return_t
-
-  @doc """
-  Callback invoked when bin is in `:playing` state, i.e. all its children
-  are in this state.
-  """
-  @callback handle_prepared_to_playing(
-              context :: CallbackContext.PlaybackChange.t(),
-              state :: state_t
-            ) ::
-              callback_return_t
+            ) :: callback_return_t
 
   @doc """
-  Callback invoked when bin is in `:playing` state, i.e. all its children
-  are in this state.
+  Callback invoked when bin switches the playback to `:playing`.
   """
-  @callback handle_prepared_to_stopped(
-              context :: CallbackContext.PlaybackChange.t(),
+  @callback handle_playing(
+              context :: CallbackContext.Playing.t(),
               state :: state_t
             ) ::
               callback_return_t
 
-  @doc """
-  Callback invoked when bin is in `:terminating` state, i.e. all its children
-  are in this state.
-  """
-  @callback handle_stopped_to_terminating(
-              context :: CallbackContext.PlaybackChange.t(),
-              state :: state_t
-            ) :: callback_return_t
-
   @doc """
   Callback invoked when a notification comes in from an element.
   """
@@ -143,14 +109,13 @@ defmodule Membrane.Bin do
   Callback invoked when bin receives a message that is not recognized
   as an internal membrane message.
 
-  Useful for receiving data sent from NIFs or other stuff.
+  Can be used for receiving data from non-membrane processes.
   """
   @callback handle_info(
               message :: any,
-              context :: CallbackContext.Other.t(),
+              context :: CallbackContext.Info.t(),
               state :: state_t
-            ) ::
-              callback_return_t
+            ) :: callback_return_t
 
   @doc """
   Callback invoked when a child element starts processing stream via given pad.
@@ -173,11 +138,7 @@ defmodule Membrane.Bin do
             ) :: callback_return_t
 
   @doc """
-  Callback invoked when `Membrane.ParentSpec` is linked and in the same playback
-  state as bin.
-
-  This callback can be started from `c:handle_init/1` callback or as
-  `t:Membrane.Bin.Action.spec_t/0` action.
+  Callback invoked when children of `Membrane.ParentSpec` are started.
   """
   @callback handle_spec_started(
               children :: [Child.name_t()],
@@ -195,22 +156,27 @@ defmodule Membrane.Bin do
               state :: state_t
             ) :: callback_return_t
 
-  @optional_callbacks handle_init: 1,
-                      handle_shutdown: 2,
+  @doc """
+  A callback invoked when the bin is being removed by its parent.
+
+  By default it returns `t:Membrane.Bin.Action.terminate_t/0` with reason `:normal`.
+  """
+  @callback handle_terminate_request(context :: CallbackContext.TerminateRequest.t(), state_t) ::
+              callback_return_t()
+
+  @optional_callbacks handle_init: 2,
                       handle_pad_added: 3,
                       handle_pad_removed: 3,
-                      handle_stopped_to_prepared: 2,
-                      handle_playing_to_prepared: 2,
-                      handle_prepared_to_playing: 2,
-                      handle_prepared_to_stopped: 2,
-                      handle_stopped_to_terminating: 2,
+                      handle_setup: 2,
+                      handle_playing: 2,
                       handle_info: 3,
                       handle_spec_started: 3,
                       handle_element_start_of_stream: 4,
                       handle_element_end_of_stream: 4,
                       handle_child_notification: 4,
                       handle_parent_notification: 3,
-                      handle_tick: 3
+                      handle_tick: 3,
+                      handle_terminate_request: 2
 
   @doc PadsSpecs.def_pad_docs(:input, :bin)
   defmacro def_input_pad(name, spec) do
@@ -314,10 +280,11 @@ defmodule Membrane.Bin do
       def membrane_bin?, do: true
 
       @impl true
-      def handle_init(_options), do: {:ok, %{}}
+      def handle_init(_ctx, %_opt_struct{} = options),
+        do: {:ok, options |> Map.from_struct()}
 
       @impl true
-      def handle_shutdown(_reason, _state), do: :ok
+      def handle_init(_ctx, options), do: {:ok, options}
 
       @impl true
       def handle_pad_added(_pad, _ctx, state), do: {:ok, state}
@@ -326,19 +293,10 @@ defmodule Membrane.Bin do
       def handle_pad_removed(_pad, _ctx, state), do: {:ok, state}
 
       @impl true
-      def handle_stopped_to_prepared(_ctx, state), do: {:ok, state}
-
-      @impl true
-      def handle_prepared_to_playing(_ctx, state), do: {:ok, state}
-
-      @impl true
-      def handle_playing_to_prepared(_ctx, state), do: {:ok, state}
+      def handle_setup(_ctx, state), do: {:ok, state}
 
       @impl true
-      def handle_prepared_to_stopped(_ctx, state), do: {:ok, state}
-
-      @impl true
-      def handle_stopped_to_terminating(_ctx, state), do: {:ok, state}
+      def handle_playing(_ctx, state), do: {:ok, state}
 
       @impl true
       def handle_info(message, _ctx, state), do: {:ok, state}
@@ -358,21 +316,21 @@ defmodule Membrane.Bin do
       @impl true
       def handle_parent_notification(_notification, _ctx, state), do: {:ok, state}
 
-      defoverridable handle_init: 1,
-                     handle_shutdown: 2,
+      @impl true
+      def handle_terminate_request(_ctx, state), do: {{:ok, terminate: :normal}, state}
+
+      defoverridable handle_init: 2,
                      handle_pad_added: 3,
                      handle_pad_removed: 3,
-                     handle_stopped_to_prepared: 2,
-                     handle_playing_to_prepared: 2,
-                     handle_prepared_to_playing: 2,
-                     handle_prepared_to_stopped: 2,
-                     handle_stopped_to_terminating: 2,
+                     handle_setup: 2,
+                     handle_playing: 2,
                      handle_info: 3,
                      handle_spec_started: 3,
                      handle_element_start_of_stream: 4,
                      handle_element_end_of_stream: 4,
                      handle_child_notification: 4,
-                     handle_parent_notification: 3
+                     handle_parent_notification: 3,
+                     handle_terminate_request: 2
     end
   end
 end

lib/membrane/bin.ex

@@ -5,8 +5,7 @@ defmodule Membrane.Bin.Action do
 
   Returning actions is a way of bin interaction with
   other components and parts of framework. Each action may be returned by any
-  callback (except for `c:Membrane.Bin.handle_shutdown/2`, as it
-  does not support returning any actions) unless explicitly stated otherwise.
+  callback unless explicitly stated otherwise.
   """
 
   alias Membrane.{Child, ParentSpec}
@@ -25,8 +24,8 @@ defmodule Membrane.Bin.Action do
   @typedoc """
   Action that instantiates children and links them according to `Membrane.ParentSpec`.
 
-  Children's playback state is changed to the current bin state.
-  `c:Membrane.Parent.handle_spec_started/3` callback is executed once it happens.
+  Children's playback is changed to the current bin playback.
+  `c:Membrane.Parent.handle_spec_started/3` callback is executed once the children are spawned.
   """
   @type spec_t :: {:spec, ParentSpec.t()}
 
@@ -82,6 +81,22 @@ defmodule Membrane.Bin.Action do
   """
   @type stop_timer_t :: {:stop_timer, timer_id :: any}
 
+  @typedoc """
+  Terminates bin with given reason.
+
+  Termination reason follows the OTP semantics:
+  - Use `:normal` for graceful termination. Allowed only when the parent already requested termination,
+  i.e. after `c:Membrane.Bin.handle_terminate_request/2` is called. If the bin has no children, it
+  terminates immediately. Otherwise, it switches to the zombie mode, requests all the children to terminate,
+  waits for them to terminate and then terminates itself. In the zombie mode, no bin callbacks
+  are called and all messages and calls to the bin are ignored (apart from Membrane internal
+  messages)
+  - If the reason is other than `:normal`, the bin terminates immediately. The bin supervisor
+  terminates all the children with the reason `:shutdown`
+  - If the reason is neither `:normal`, `:shutdown` nor `{:shutdown, term}`, an error is logged
+  """
+  @type terminate_t :: {:terminate, reason :: :normal | :shutdown | {:shutdown, term} | term}
+
   @typedoc """
   Type describing actions that can be returned from bin callbacks.
 
@@ -96,4 +111,5 @@ defmodule Membrane.Bin.Action do
           | start_timer_t
           | timer_interval_t
           | stop_timer_t
+          | terminate_t
 end

lib/membrane/bin/action.ex

@@ -1,4 +1,4 @@
-defmodule Membrane.Bin.CallbackContext.Other do
+defmodule Membrane.Bin.CallbackContext.Info do
   @moduledoc """
   Structure representing a context that is passed to the callback when
   bin receives unrecognized message.

lib/membrane/bin/callback_context/other.ex renamed to lib/membrane/bin/callback_context/info.ex

@@ -0,0 +1,6 @@
+defmodule Membrane.Bin.CallbackContext.Init do
+  @moduledoc """
+  Callback context for `c:Membrane.Bin.handle_init/2`.
+  """
+  use Membrane.Core.Bin.CallbackContext
+end

lib/membrane/bin/callback_context/init.ex

@@ -0,0 +1,6 @@
+defmodule Membrane.Bin.CallbackContext.Playing do
+  @moduledoc """
+  Structure representing a context that is passed to the `c:Membrane.Bin.handle_playing/2` callback.
+  """
+  use Membrane.Core.Bin.CallbackContext
+end

lib/membrane/bin/callback_context/playback_change.ex

@@ -0,0 +1,6 @@
+defmodule Membrane.Bin.CallbackContext.Setup do
+  @moduledoc """
+  Structure representing a context that is passed to the `c:Membrane.Bin.handle_setup/2` callback.
+  """
+  use Membrane.Core.Bin.CallbackContext
+end