Merge branch 'master' into resource-guard-test

Author: mat-hek

.github/actions/add_pr_to_smackore_board/action.yml

@@ -1,5 +1,5 @@
 name: 'Add PR to Smackore board, if author is from community'
-description: 'Adds PR to "New issues by community" column in Smackore project board, if PR author is from outside Membrane Team.'
+description: '(disabled due to github-side bug) Adds PR to "New issues by community" column in Smackore project board, if PR author is from outside Membrane Team.'
 inputs:
   GITHUB_TOKEN:
     description: 'GitHub token'
@@ -19,21 +19,23 @@ runs:
         repository: membraneframework/membrane_core
     - name: Maybe add PR to board and set ticket status
       run: | 
-          export PROJECT_NUMBER=19
-          export PROJECT_ID=PVT_kwDOAYE_z84AWEIB
-          export STATUS_FIELD_ID=PVTSSF_lADOAYE_z84AWEIBzgOGd1k
-          export TARGET_COLUMN_ID=e6b1ee10
+          # currently this causes github action crash, more info here: https://github.com/membraneframework/membrane_core/issues/749
+
+          # export PROJECT_NUMBER=19
+          # export PROJECT_ID=PVT_kwDOAYE_z84AWEIB
+          # export STATUS_FIELD_ID=PVTSSF_lADOAYE_z84AWEIBzgOGd1k
+          # export TARGET_COLUMN_ID=e6b1ee10
           
-          export AUTHOR_ORIGIN=$(gh api -H "Accept: application/vnd.github+json" -H "X-GitHub-Api-Version: 2022-11-28" /orgs/membraneframework/teams/membraneteam/members | python scripts/python/get_author_origin.py $AUTHOR_LOGIN)
+          # export AUTHOR_ORIGIN=$(gh api -H "Accept: application/vnd.github+json" -H "X-GitHub-Api-Version: 2022-11-28" /orgs/membraneframework/teams/membraneteam/members | python scripts/python/get_author_origin.py $AUTHOR_LOGIN)
 
-          if [ "$AUTHOR_ORIGIN" == "COMMUNITY" ]
-          then
-            gh pr edit "$PR_URL" --add-project Smackore
-            sleep 10
+          # if [ "$AUTHOR_ORIGIN" == "COMMUNITY" ]
+          # then
+          #   gh pr edit "$PR_URL" --add-project Smackore
+          #   sleep 10
 
-            export TICKET_ID=$(gh project item-list $PROJECT_NUMBER --owner membraneframework --format json --limit 10000000 | python scripts/python/get_ticket_id.py "$PR_URL")
-            gh project item-edit --id $TICKET_ID --field-id $STATUS_FIELD_ID --project-id $PROJECT_ID --single-select-option-id $TARGET_COLUMN_ID
-          fi
+          #   export TICKET_ID=$(gh project item-list $PROJECT_NUMBER --owner membraneframework --format json --limit 10000000 | python scripts/python/get_ticket_id.py "$PR_URL")
+          #   gh project item-edit --id $TICKET_ID --field-id $STATUS_FIELD_ID --project-id $PROJECT_ID --single-select-option-id $TARGET_COLUMN_ID
+          # fi
 
       env:
         GH_TOKEN: ${{ inputs.GITHUB_TOKEN }}

CHANGELOG.md

@@ -1,11 +1,16 @@
 # Changelog
 
+## 1.1.0-rc0
+ * Deprecate `handle_spec_started/3` callback in Bins and Pipelines. [#708](https://github.com/membraneframework/membrane_core/pull/708)
+ * Handle buffers from input pads having `flow_control: :auto` only if demand on all output pads having `flow_control: :auto` is positive. [#693](https://github.com/membraneframework/membrane_core/pull/693)
+
 ## 1.0.1
  * Specify the order in which state fields will be printed in the error logs. [#614](https://github.com/membraneframework/membrane_core/pull/614)
  * Fix clock selection [#626](https://github.com/membraneframework/membrane_core/pull/626)
  * Log messages in the default handle_info implementation [#680](https://github.com/membraneframework/membrane_core/pull/680)
  * Fix typespecs in Membrane.UtilitySupervisor [#681](https://github.com/membraneframework/membrane_core/pull/681)
  * Improve callback return types and group actions types [#702](https://github.com/membraneframework/membrane_core/pull/702) 
+ * Add `crash_reason` to `handle_crash_group_down/3` callback context in bins and pipelines. [#720](https://github.com/membraneframework/membrane_core/pull/720)
 
 ## 1.0.0
  * Introduce `:remove_link` action in pipelines and bins.

README.md

@@ -53,10 +53,6 @@ Membrane.Pipeline.start_link(MyPipeline, mp3_url)
 ```
 
 This is an [Elixir](elixir-lang.org) snippet, that streams an mp3 via HTTP and plays it on your speaker. Here's how to run it:
-- Install [libmad](https://github.com/markjeee/libmad) and [portaudio](https://github.com/PortAudio/portaudio). Membrane uses these libs to decode the mp3 and to access your speaker, respectively. You can use these commands:
-  - On Mac OS: `brew install libmad portaudio pkg-config`
-  - On Debian: `apt install libmad0-dev portaudio19-dev`
-
 - Option 1: Click the button below:
 
   [![Run in Livebook](https://livebook.dev/badge/v1/blue.svg)](https://livebook.dev/run?url=https%3A%2F%2Fgithub.com%2Fmembraneframework%2Fmembrane_core%2Fblob%2Fmaster%2Fexample.livemd)
@@ -96,7 +92,7 @@ Apart from plugins, Membrane has stream formats, which live in `membrane_X_forma
 The API for creating pipelines (and custom elements too) is provided by [membrane_core](https://github.com/membraneframework/membrane_core). To install it, add the following line to your `deps` in `mix.exs` and run `mix deps.get`
 
 ```elixir
-{:membrane_core, "~> 1.0.0"}
+{:membrane_core, "~> 1.0"}
 ```
 
 **Standalone libraries**

lib/membrane/bin.ex

@@ -167,9 +167,11 @@ defmodule Membrane.Bin do
             ) :: callback_return
 
   @doc """
+  This callback is deprecated since v1.1.0-rc0
+
   Callback invoked when children of `Membrane.ChildrenSpec` are started.
 
-  By default, it does nothing.
+  It is invoked, only if pipeline module contains its definition. Otherwise, nothing happens.
   """
   @callback handle_spec_started(
               children :: [Child.name()],
@@ -309,6 +311,7 @@ defmodule Membrane.Bin do
       alias unquote(__MODULE__)
       @behaviour unquote(__MODULE__)
       @before_compile unquote(__MODULE__)
+      @after_compile {Membrane.Core.Parent, :check_deprecated_callbacks}
 
       unquote(bring_spec)
       unquote(bring_pad)
@@ -354,9 +357,6 @@ defmodule Membrane.Bin do
         {[], state}
       end
 
-      @impl true
-      def handle_spec_started(new_children, _ctx, state), do: {[], state}
-
       @impl true
       def handle_element_start_of_stream(_element, _pad, _ctx, state), do: {[], state}
 
@@ -381,7 +381,6 @@ defmodule Membrane.Bin do
                      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,

lib/membrane/bin/callback_context.ex

@@ -12,8 +12,8 @@ defmodule Membrane.Bin.CallbackContext do
   Field `:start_of_stream_received?` is present only in
   `c:Membrane.Bin.handle_element_end_of_stream/4`.
 
-  Fields `:members` and `:crash_initiator` are present only in
-  `c:Membrane.Pipeline.handle_crash_group_down/3`.
+  Fields `:members`, `:crash_initiator` and `crash_reason` and  are present only in
+  `c:Membrane.Bin.handle_crash_group_down/3`.
   """
   @type t :: %{
           :clock => Membrane.Clock.t(),
@@ -27,6 +27,7 @@ defmodule Membrane.Bin.CallbackContext do
           optional(:pad_options) => map(),
           optional(:members) => [Membrane.Child.name()],
           optional(:crash_initiator) => Membrane.Child.name(),
+          optional(:crash_reason) => :normal | :shutdown | {:shutdown, term()} | term(),
           optional(:start_of_stream_received?) => boolean()
         }
 end

lib/membrane/core/bin/callback_context.ex

@@ -3,7 +3,11 @@ defmodule Membrane.Core.Bin.CallbackContext do
 
   @type optional_fields ::
           [pad_options: map()]
-          | [members: [Membrane.Child.name()], crash_initiator: Membrane.Child.name()]
+          | [
+              members: [Membrane.Child.name()],
+              crash_initiator: Membrane.Child.name(),
+              crash_reason: :normal | :shutdown | {:shutdown, term()} | term()
+            ]
           | [start_of_stream_received?: boolean()]
 
   @spec from_state(Membrane.Core.Bin.State.t(), optional_fields()) ::

lib/membrane/core/callback_handler.ex

@@ -8,6 +8,7 @@ defmodule Membrane.Core.CallbackHandler do
   use Bunch
 
   alias Membrane.CallbackError
+  alias Membrane.Core.Component
 
   require Membrane.Logger
 
@@ -191,6 +192,13 @@ defmodule Membrane.Core.CallbackHandler do
     was_handling_action? = state.handling_action?
     state = %{state | handling_action?: true}
 
+    # Updating :supplying_demand? flag value here is a temporal fix.
+    # Setting it to `true` while handling actions causes postponing calls
+    # of handle_redemand/2 and supply_demand/2 until a moment, when all
+    # actions returned from the callback are handled
+    was_supplying_demand? = Map.get(state, :supplying_demand?, false)
+    state = if Component.is_element?(state), do: %{state | supplying_demand?: true}, else: state
+
     state =
       Enum.reduce(actions, state, fn action, state ->
         try do
@@ -210,6 +218,11 @@ defmodule Membrane.Core.CallbackHandler do
         do: state,
         else: %{state | handling_action?: false}
 
+    state =
+      if Component.is_element?(state) and not was_supplying_demand?,
+        do: %{state | supplying_demand?: false},
+        else: state
+
     handler_module.handle_end_of_actions(state)
   end
 end

lib/membrane/core/child/pad_model.ex

@@ -42,7 +42,6 @@ defmodule Membrane.Core.Child.PadModel do
           input_queue: Membrane.Core.Element.InputQueue.t() | nil,
           options: %{optional(atom) => any},
           auto_demand_size: pos_integer() | nil,
-          associated_pads: [Pad.ref()] | nil,
           sticky_events: [Membrane.Event.t()],
           stalker_metrics: %{atom => :atomics.atomics_ref()}
         }

lib/membrane/core/element.ex

@@ -158,8 +158,12 @@ defmodule Membrane.Core.Element do
         setup_incomplete?: false,
         effective_flow_control: :push,
         handling_action?: false,
+        popping_auto_flow_queue?: false,
         pads_to_snapshot: MapSet.new(),
-        stalker: options.stalker
+        stalker: options.stalker,
+        satisfied_auto_output_pads: MapSet.new(),
+        awaiting_auto_input_pads: MapSet.new(),
+        auto_input_pads: []
       }
       |> PadSpecHandler.init_pads()
 
@@ -256,20 +260,7 @@ defmodule Membrane.Core.Element do
   end
 
   defp do_handle_info(Message.new(:timer_tick, timer_id), state) do
-    # Guarding the `TimerController.handle_tick/2` invocation is
-    # required since there might be a case in which `handle_tick`
-    # callback's implementation returns demand action.
-    # In this scenario, without this guard, there would a possibility that
-    # the `handle_buffer` would be called immediately, returning
-    # some action that would affect the timer and the original state
-    # of the timer, set with actions returned from `handle_tick`,
-    # would be overwritten with that action.
-    #
-    # For more information see: https://github.com/membraneframework/membrane_core/issues/670
-    state = %{state | supplying_demand?: true}
     state = TimerController.handle_tick(timer_id, state)
-    state = %{state | supplying_demand?: false}
-    state = Membrane.Core.Element.DemandHandler.handle_delayed_demands(state)
     {:noreply, state}
   end
 

lib/membrane/core/element/action_handler.ex

@@ -24,11 +24,11 @@ defmodule Membrane.Core.Element.ActionHandler do
   alias Membrane.Core.Element.{
     DemandController,
     DemandHandler,
-    PadController,
     State,
     StreamFormatController
   }
 
+  alias Membrane.Core.Element.DemandController.AutoFlowUtils
   alias Membrane.Core.{Events, TimerController}
   alias Membrane.Element.Action
 
@@ -38,22 +38,47 @@ defmodule Membrane.Core.Element.ActionHandler do
   require Membrane.Logger
 
   @impl CallbackHandler
-  def transform_actions(actions, callback, _handler_params, state) do
+  def transform_actions(actions, _callback, _handler_params, state) do
     actions = join_buffers(actions)
-    ensure_nothing_after_redemand(actions, callback, state)
     {actions, state}
   end
 
   defguardp is_demand_size(size) when is_integer(size) or is_function(size)
 
   @impl CallbackHandler
-  def handle_end_of_actions(state) when not state.handling_action? do
-    Enum.reduce(state.pads_to_snapshot, state, &DemandController.snapshot_atomic_demand/2)
-    |> Map.put(:pads_to_snapshot, MapSet.new())
+  def handle_end_of_actions(state) do
+    # Fixed order of handling demand of manual and auto pads would lead to
+    # favoring manual pads over auto pads (or vice versa), especially after
+    # introducting auto flow queues.
+    manual_demands_first? = Enum.random([1, 2]) == 1
+
+    state =
+      if manual_demands_first?,
+        do: maybe_handle_delayed_demands(state),
+        else: state
+
+    state = maybe_handle_pads_to_snapshot(state)
+
+    state =
+      if manual_demands_first?,
+        do: state,
+        else: maybe_handle_delayed_demands(state)
+
+    state
   end
 
-  @impl CallbackHandler
-  def handle_end_of_actions(state), do: state
+  defp maybe_handle_delayed_demands(state) do
+    with %{supplying_demand?: false} <- state do
+      DemandHandler.handle_delayed_demands(state)
+    end
+  end
+
+  defp maybe_handle_pads_to_snapshot(state) do
+    with %{handling_action?: false} <- state do
+      Enum.reduce(state.pads_to_snapshot, state, &DemandController.snapshot_atomic_demand/2)
+      |> Map.put(:pads_to_snapshot, MapSet.new())
+    end
+  end
 
   @impl CallbackHandler
   def handle_action({action, _}, :handle_init, _params, _state)
@@ -176,7 +201,11 @@ defmodule Membrane.Core.Element.ActionHandler do
         _other -> :output
       end
 
-    pads = state |> PadModel.filter_data(%{direction: dir}) |> Map.keys()
+    pads =
+      Enum.flat_map(state.pads_data, fn
+        {pad_ref, %{direction: ^dir}} -> [pad_ref]
+        _pad_entry -> []
+      end)
 
     Enum.reduce(pads, state, fn pad, state ->
       action =
@@ -280,30 +309,6 @@ defmodule Membrane.Core.Element.ActionHandler do
     )
   end
 
-  defp ensure_nothing_after_redemand(actions, callback, state) do
-    {redemands, actions_after_redemands} =
-      actions
-      |> Enum.drop_while(fn
-        {:redemand, _args} -> false
-        _other_action -> true
-      end)
-      |> Enum.split_while(fn
-        {:redemand, _args} -> true
-        _other_action -> false
-      end)
-
-    case {redemands, actions_after_redemands} do
-      {_redemands, []} ->
-        :ok
-
-      {[redemand | _redemands], _actions_after_redemands} ->
-        raise ActionError,
-          reason: :actions_after_redemand,
-          action: redemand,
-          callback: {state.module, callback}
-    end
-  end
-
   @spec send_buffer(Pad.ref(), [Buffer.t()] | Buffer.t(), State.t()) :: State.t()
   defp send_buffer(_pad_ref, [], state) do
     state
@@ -466,8 +471,10 @@ defmodule Membrane.Core.Element.ActionHandler do
   @spec handle_outgoing_event(Pad.ref(), Event.t(), State.t()) :: State.t()
   defp handle_outgoing_event(pad_ref, %Events.EndOfStream{}, state) do
     with %{direction: :output, end_of_stream?: false} <- PadModel.get_data!(state, pad_ref) do
-      state = PadController.remove_pad_associations(pad_ref, state)
-      PadModel.set_data!(state, pad_ref, :end_of_stream?, true)
+      DemandHandler.remove_pad_from_delayed_demands(pad_ref, state)
+      |> Map.update!(:satisfied_auto_output_pads, &MapSet.delete(&1, pad_ref))
+      |> PadModel.set_data!(pad_ref, :end_of_stream?, true)
+      |> AutoFlowUtils.pop_queues_and_bump_demand()
     else
       %{direction: :input} ->
         raise PadDirectionError, action: "end of stream", direction: :input, pad: pad_ref

lib/membrane/core/element/atomic_demand.ex

@@ -139,14 +139,14 @@ defmodule Membrane.Core.Element.AtomicDemand do
     :ok
   end
 
-  @spec decrease(t, non_neg_integer()) :: t
+  @spec decrease(t, non_neg_integer()) :: {{:decreased, integer()}, t} | {:unchanged, t}
   def decrease(%__MODULE__{} = atomic_demand, value) do
     atomic_demand = Map.update!(atomic_demand, :buffered_decrementation, &(&1 + value))
 
     if atomic_demand.buffered_decrementation >= atomic_demand.throttling_factor do
       flush_buffered_decrementation(atomic_demand)
     else
-      atomic_demand
+      {:unchanged, atomic_demand}
     end
   end
 
@@ -164,14 +164,17 @@ defmodule Membrane.Core.Element.AtomicDemand do
 
     atomic_demand = %{atomic_demand | buffered_decrementation: 0}
 
-    if not atomic_demand.toilet_overflowed? and
-         get_receiver_status(atomic_demand) == {:resolved, :pull} and
-         get_sender_status(atomic_demand) == {:resolved, :push} and
-         -1 * atomic_demand_value > atomic_demand.toilet_capacity do
-      overflow(atomic_demand, atomic_demand_value)
-    else
-      atomic_demand
-    end
+    atomic_demand =
+      if not atomic_demand.toilet_overflowed? and
+           get_receiver_status(atomic_demand) == {:resolved, :pull} and
+           get_sender_status(atomic_demand) == {:resolved, :push} and
+           -1 * atomic_demand_value > atomic_demand.toilet_capacity do
+        overflow(atomic_demand, atomic_demand_value)
+      else
+        atomic_demand
+      end
+
+    {{:decreased, atomic_demand_value}, atomic_demand}
   end
 
   defp overflow(atomic_demand, atomic_demand_value) do