Update processed specs, when children or link is down

CHANGELOG.md

@@ -15,6 +15,7 @@
  * The flow control of the pad is now set with a single `:flow_control` option instead of `:mode` and `:demand_mode` options.
  * Remove _t suffix from types [#509](https://github.com/membraneframework/membrane_core/pull/509)
  * Implement automatic demands in Membrane Sinks and Endpoints. [#512](https://github.com/membraneframework/membrane_core/pull/512)
+ * Add `handle_child_pad_removed/4` callback in Bins and Pipelines. [#513](https://github.com/membraneframework/membrane_core/pull/513)
 
 ## 0.11.0
  * Separate element_name and pad arguments in handle_element_{start, end}_of_stream signature [#219](https://github.com/membraneframework/membrane_core/issues/219)

lib/membrane/bin.ex

@@ -90,6 +90,20 @@ defmodule Membrane.Bin do
             ) ::
               callback_return
 
+  @doc """
+  Callback invoked when a child removes its pad.
+
+  The callback won't be invoked, when you have initiated the pad removal,
+  eg. when you have returned `t:Membrane.Bin.Action.remove_link()` action
+  which made one of your children's pads be removed.
+  """
+  @callback handle_child_pad_removed(
+              child :: Child.name(),
+              pad :: Pad.ref(),
+              context :: CallbackContext.t(),
+              state :: state
+            ) :: callback_return
+
   @doc """
   Callback invoked when a notification comes in from an element.
   """
@@ -168,8 +182,7 @@ defmodule Membrane.Bin do
   @callback handle_terminate_request(
               context :: CallbackContext.t(),
               state
-            ) ::
-              callback_return()
+            ) :: callback_return
 
   @optional_callbacks handle_init: 2,
                       handle_pad_added: 3,
@@ -183,7 +196,8 @@ defmodule Membrane.Bin do
                       handle_child_notification: 4,
                       handle_parent_notification: 3,
                       handle_tick: 3,
-                      handle_terminate_request: 2
+                      handle_terminate_request: 2,
+                      handle_child_pad_removed: 4
 
   @doc PadsSpecs.def_pad_docs(:input, :bin)
   defmacro def_input_pad(name, spec) do

lib/membrane/core/bin.ex

@@ -183,6 +183,11 @@ defmodule Membrane.Core.Bin do
     {:noreply, state}
   end
 
+  defp do_handle_info(Message.new(:child_pad_removed, [child, pad]), state) do
+    state = Parent.ChildLifeController.handle_child_pad_removed(child, pad, state)
+    {:noreply, state}
+  end
+
   defp do_handle_info(Message.new(:child_notification, [from, notification]), state) do
     state = Parent.LifecycleController.handle_child_notification(from, notification, state)
     {:noreply, state}

lib/membrane/core/bin/pad_controller.ex

@@ -80,6 +80,22 @@ defmodule Membrane.Core.Bin.PadController do
     state
   end
 
+  @spec remove_pad!(Pad.ref(), State.t()) :: State.t()
+  def remove_pad!(pad_ref, state) do
+    cond do
+      Pad.is_dynamic_pad_ref(pad_ref) ->
+        Message.send(state.parent_pid, :child_pad_removed, [state.name, pad_ref])
+        PadModel.delete_data!(state, pad_ref)
+
+      Pad.is_static_pad_ref(pad_ref) and state.terminating? ->
+        state
+
+      Pad.is_static_pad_ref(pad_ref) ->
+        raise Membrane.PadError,
+              "Tried to unlink bin static pad #{inspect(pad_ref)}. Static pads cannot be unlinked unless bin is terminating"
+    end
+  end
+
   @spec handle_linking_timeout(Pad.ref(), State.t()) :: :ok | no_return()
   def handle_linking_timeout(pad_ref, state) do
     case PadModel.get_data(state, pad_ref) do
@@ -193,58 +209,80 @@ defmodule Membrane.Core.Bin.PadController do
           Core.Bin.State.t()
         ) :: {Core.Element.PadController.link_call_reply(), Core.Bin.State.t()}
   def handle_link(direction, endpoint, other_endpoint, params, state) do
-    pad_data = PadModel.get_data!(state, endpoint.pad_ref)
-
     Membrane.Logger.debug("Handle link #{inspect(endpoint, pretty: true)}")
 
-    %{spec_ref: spec_ref, endpoint: child_endpoint, name: pad_name} = pad_data
-
-    pad_props =
-      Map.merge(endpoint.pad_props, child_endpoint.pad_props, fn key,
-                                                                 external_value,
-                                                                 internal_value ->
-        if key in [
-             :target_queue_size,
-             :min_demand_factor,
-             :auto_demand_size,
-             :toilet_capacity,
-             :throttling_factor
-           ] do
-          external_value || internal_value
-        else
-          internal_value
-        end
-      end)
-
-    child_endpoint = %{child_endpoint | pad_props: pad_props}
-
-    if params.initiator == :sibling do
-      :ok =
-        Child.PadController.validate_pad_mode!(
-          {endpoint.pad_ref, pad_data},
-          {other_endpoint.pad_ref, params.other_info}
+    with {:ok, pad_data} <- PadModel.get_data(state, endpoint.pad_ref) do
+      %{spec_ref: spec_ref, endpoint: child_endpoint, name: pad_name} = pad_data
+
+      pad_props =
+        Map.merge(endpoint.pad_props, child_endpoint.pad_props, fn key,
+                                                                   external_value,
+                                                                   internal_value ->
+          if key in [
+               :target_queue_size,
+               :min_demand_factor,
+               :auto_demand_size,
+               :toilet_capacity,
+               :throttling_factor
+             ] do
+            external_value || internal_value
+          else
+            internal_value
+          end
+        end)
+
+      child_endpoint = %{child_endpoint | pad_props: pad_props}
+
+      if params.initiator == :sibling do
+        :ok =
+          Child.PadController.validate_pad_mode!(
+            {endpoint.pad_ref, pad_data},
+            {other_endpoint.pad_ref, params.other_info}
+          )
+      end
+
+      params =
+        Map.update!(
+          params,
+          :stream_format_validation_params,
+          &[{state.module, pad_name} | &1]
         )
-    end
 
-    params =
-      Map.update!(
-        params,
-        :stream_format_validation_params,
-        &[{state.module, pad_name} | &1]
-      )
+      handle_link_response =
+        Message.call(child_endpoint.pid, :handle_link, [
+          direction,
+          child_endpoint,
+          other_endpoint,
+          params
+        ])
+
+      case handle_link_response do
+        {:error, {:call_failure, reason}} ->
+          Membrane.Logger.debug("""
+          Tried to link pad #{inspect(endpoint.pad_ref)}, but child #{inspect(endpoint.child)},
+          that was supposed to be linked to this pad, is not alive.
+          """)
+
+          {{:error, {:child_dead, reason}}, state}
+
+        {:error, _reason} = error ->
+          {error, state}
+
+        reply ->
+          state = PadModel.set_data!(state, endpoint.pad_ref, :linked?, true)
+          state = PadModel.set_data!(state, endpoint.pad_ref, :endpoint, child_endpoint)
+          state = ChildLifeController.proceed_spec_startup(spec_ref, state)
+          {reply, state}
+      end
+    else
+      {:error, :unknown_pad} ->
+        Membrane.Logger.debug("""
+        Tried to link pad #{inspect(endpoint.pad_ref)}, but this pad is not existing
+        in the moment of receiving :handle_link. It could be removed or never existed.
+        """)
 
-    reply =
-      Message.call!(child_endpoint.pid, :handle_link, [
-        direction,
-        child_endpoint,
-        other_endpoint,
-        params
-      ])
-
-    state = PadModel.set_data!(state, endpoint.pad_ref, :linked?, true)
-    state = PadModel.set_data!(state, endpoint.pad_ref, :endpoint, child_endpoint)
-    state = ChildLifeController.proceed_spec_startup(spec_ref, state)
-    {reply, state}
+        {{:error, {:unknown_pad, state.name, endpoint.pad_ref}}, state}
+    end
   end
 
   @doc """

lib/membrane/core/element/pad_controller.ex

@@ -120,7 +120,18 @@ defmodule Membrane.Core.Element.PadController do
         {:ok, state}
 
       {:error, {:call_failure, reason}} ->
+        Membrane.Logger.debug("""
+        Tried to link pad #{inspect(endpoint.pad_ref)}, but neighbour #{inspect(other_endpoint.child)}
+        is not alive.
+        """)
+
         {{:error, {:neighbor_dead, reason}}, state}
+
+      {:error, {:unknown_pad, _name, _pad_ref}} = error ->
+        {error, state}
+
+      {:error, {:child_dead, reason}} ->
+        {{:error, {:neighbor_child_dead, reason}}, state}
     end
   end