Implement live queueing strategy and callbacks system

lib/membrane/video_compositor/compositor_core_format.ex

@@ -1,7 +1,6 @@
 defmodule Membrane.VideoCompositor.CompositorCoreFormat do
-  @moduledoc """
-  Describes CoreVC input format.
-  """
+  @moduledoc false
+  # Describes CoreVC input format.
 
   alias Membrane.{Pad, RawVideo}
 

lib/membrane/video_compositor/core.ex

@@ -1,7 +1,6 @@
 defmodule Membrane.VideoCompositor.Core do
-  @moduledoc """
-  The element responsible for composing frames.
-  """
+  @moduledoc false
+  # The element responsible for composing frames.
 
   use Membrane.Filter
 
@@ -10,9 +9,8 @@ defmodule Membrane.VideoCompositor.Core do
   alias Membrane.VideoCompositor.{CompositorCoreFormat, Scene, SceneChangeEvent, WgpuAdapter}
 
   defmodule State do
-    @moduledoc """
-    The internal state of the compositor
-    """
+    @moduledoc false
+    # The internal state of the compositor
 
     @enforce_keys [:wgpu_state, :output_stream_format]
     defstruct @enforce_keys ++
@@ -97,7 +95,8 @@ defmodule Membrane.VideoCompositor.Core do
           wgpu_state: wgpu_state,
           scene: scene,
           input_stream_format: stream_format,
-          update_videos?: update_videos?
+          update_videos?: update_videos?,
+          output_stream_format: output_stream_format
         }
       ) do
     if update_videos? do
@@ -110,11 +109,14 @@ defmodule Membrane.VideoCompositor.Core do
     end
 
     {:ok, rendered_frame} =
-      payload
-      |> Map.to_list()
-      |> Enum.filter(fn {pad, _frame} -> Map.has_key?(scene.video_configs, pad) end)
-      |> Enum.map(fn {pad, frame} -> {Map.get(pads_to_ids, pad), frame, pts} end)
-      |> then(fn pads_frames -> send_pads_frames(wgpu_state, pads_frames) end)
+      if payload == %{} do
+        {:ok, get_blank_frame(output_stream_format)}
+      else
+        payload
+        |> Map.to_list()
+        |> Enum.map(fn {pad, frame} -> {Map.fetch!(pads_to_ids, pad), frame, pts} end)
+        |> then(fn pads_frames -> send_pads_frames(wgpu_state, pads_frames) end)
+      end
 
     output_buffer = %Buffer{pts: pts, dts: pts, payload: rendered_frame}
 
@@ -154,4 +156,10 @@ defmodule Membrane.VideoCompositor.Core do
         raise "Core should render frame only on last buffer"
     end
   end
+
+  @spec get_blank_frame(RawVideo.t()) :: binary()
+  defp get_blank_frame(%RawVideo{width: width, height: height}) do
+    pixels = div(width * height * 3, 2)
+    <<0::size(pixels)>>
+  end
 end

lib/membrane/video_compositor/handler.ex

@@ -11,6 +11,11 @@ defmodule Membrane.VideoCompositor.Handler do
   alias Membrane.VideoCompositor.Handler.InputProperties
   alias Membrane.VideoCompositor.Scene
 
+  @typedoc """
+  Module implementing `#{inspect(__MODULE__)}` behaviour.
+  """
+  @type t :: module()
+
   @typedoc """
   Type of user-managed inner state of the handler.
   """

lib/membrane/video_compositor/queue.ex

@@ -1,12 +1,13 @@
 defmodule Membrane.VideoCompositor.Queue do
-  @moduledoc """
-  Defines input pads and compositor core contracts, that each
-  implementation of a queue should meet.
-  """
+  @moduledoc false
+
+  # Defines input pads and compositor core contracts, that each
+  # implementation of a queue should meet.
 
-  alias Membrane.{Buffer, Pad, RawVideo}
+  alias Membrane.{Buffer, Pad}
   alias Membrane.VideoCompositor
-  alias Membrane.VideoCompositor.{CompositorCoreFormat, SceneChangeEvent}
+  alias Membrane.VideoCompositor.{CompositorCoreFormat, QueueingStrategy, SceneChangeEvent}
+  alias Membrane.VideoCompositor.Queue.Live, as: LiveQueue
   alias Membrane.VideoCompositor.Queue.Offline, as: OfflineQueue
 
   @typedoc """
@@ -42,12 +43,13 @@ defmodule Membrane.VideoCompositor.Queue do
   @type queue_bin :: OfflineQueue.t()
 
   @spec get_queue(VideoCompositor.init_options()) :: OfflineQueue.t()
-  def get_queue(%VideoCompositor{
-        queuing_strategy: queuing_strategy,
-        output_stream_format: %RawVideo{framerate: framerate}
-      }) do
-    case queuing_strategy do
-      :offline -> %OfflineQueue{output_framerate: framerate}
+  def get_queue(options = %VideoCompositor{}) do
+    case options.queuing_strategy do
+      QueueingStrategy.Offline ->
+        %OfflineQueue{vc_init_options: options}
+
+      %QueueingStrategy.Live{latency: latency} ->
+        %LiveQueue{vc_init_options: options, latency: latency}
     end
   end
 end

lib/membrane/video_compositor/queue/live.ex

@@ -0,0 +1,231 @@
+defmodule Membrane.VideoCompositor.Queue.Live do
+  @moduledoc false
+  # Module responsible for frames / events enqueueing accordingly to live composing strategy
+
+  use Membrane.Filter
+
+  alias Membrane.RawVideo
+  alias Membrane.VideoCompositor
+  alias Membrane.VideoCompositor.CompositorCoreFormat
+  alias Membrane.VideoCompositor.Queue.Live.State, as: LiveState
+  alias Membrane.VideoCompositor.Queue.State
+  alias Membrane.VideoCompositor.Queue.State.{HandlerState, PadState}
+  alias Membrane.VideoCompositor.QueueingStrategy.Live
+
+  @type latency :: Membrane.Time.non_neg_t() | :wait_for_start_event
+
+  @type start_timer_message :: :start_timer | {:start_timer, delay :: Membrane.Time.non_neg_t()}
+
+  def_options vc_init_options: [
+                spec: VideoCompositor.init_options()
+              ],
+              latency: [
+                spec: latency()
+              ]
+
+  def_input_pad :input,
+    accepted_format: %RawVideo{pixel_format: :I420},
+    availability: :on_request,
+    demand_mode: :auto,
+    options: [
+      timestamp_offset: [
+        spec: Membrane.Time.non_neg_t(),
+        description: "Input stream PTS offset in nanoseconds. Must be non-negative.",
+        default: 0
+      ],
+      metadata: [
+        spec: VideoCompositor.init_metadata()
+      ]
+    ]
+
+  def_output_pad :output,
+    accepted_format: %CompositorCoreFormat{},
+    availability: :always,
+    demand_mode: :auto
+
+  @impl true
+  def handle_init(_ctx, %{
+        vc_init_options:
+          vc_init_options = %VideoCompositor{
+            output_stream_format: %RawVideo{framerate: framerate},
+            queuing_strategy: %Live{latency: latency}
+          }
+      }) do
+    {[],
+     %State{
+       output_framerate: framerate,
+       custom_strategy_state: %LiveState{
+         latency: latency
+       },
+       handler: HandlerState.new(vc_init_options)
+     }}
+  end
+
+  @impl true
+  def handle_pad_added(pad, context, state) do
+    state = Bunch.Struct.put_in(state, [:pads_states, pad], PadState.new(context.options))
+
+    {[], state}
+  end
+
+  @impl true
+  def handle_start_of_stream(_pad, _ctx, state = %State{}) do
+    if state.custom_strategy_state.timer_started? do
+      {[], state}
+    else
+      state = Bunch.Struct.put_in(state, [:custom_strategy_state, :timer_started?], true)
+
+      case state.custom_strategy_state.latency do
+        :wait_for_start_event ->
+          {[], state}
+
+        latency_time ->
+          {[start_timer: {:initializer, latency_time}], state}
+      end
+    end
+  end
+
+  @impl true
+  def handle_stream_format(pad, stream_format, _ctx, state) do
+    state = State.put_event(state, {{:stream_format, stream_format}, pad})
+    {[], state}
+  end
+
+  @impl true
+  def handle_end_of_stream(pad, _ctx, state) do
+    state = State.put_event(state, {:end_of_stream, pad})
+    {[], state}
+  end
+
+  @impl true
+  def handle_process(pad, buffer, _ctx, state) do
+    state = State.put_event(state, {{:frame, buffer.pts, buffer.payload}, pad})
+    {[], state}
+  end
+
+  @impl true
+  def handle_tick(:initializer, _ctx, state) do
+    {[stop_timer: :initializer, start_timer: {:buffer_scheduler, get_tick_ratio(state)}], state}
+  end
+
+  @impl true
+  def handle_tick(
+        :buffer_scheduler,
+        _ctx,
+        initial_state = %State{next_buffer_pts: buffer_pts}
+      ) do
+    state = drop_eos_pads(initial_state)
+
+    indexes =
+      state.pads_states
+      |> Enum.map(fn {pad, %PadState{events_queue: events_queue}} ->
+        {pad, nearest_frame_index(events_queue, buffer_pts)}
+      end)
+      |> Enum.reject(fn {_pad, index} -> index == :no_frame end)
+      |> Enum.into(%{})
+
+    {pads_frames, new_state} = State.pop_events(state, indexes, true)
+
+    actions = State.get_actions(new_state, initial_state, pads_frames, buffer_pts)
+
+    if all_pads_eos?(new_state) do
+      {actions ++ [stop_timer: :buffer_scheduler, end_of_stream: :output], new_state}
+    else
+      {actions, new_state}
+    end
+  end
+
+  @impl true
+  def handle_parent_notification(:start_timer, _ctx, state) do
+    check_timer_started(state)
+    {[start_timer: {:buffer_scheduler, get_tick_ratio(state)}], state}
+  end
+
+  @impl true
+  def handle_parent_notification({:start_timer, delay}, _ctx, state) do
+    check_timer_started(state)
+    {[start_timer: {:initializer, delay}], state}
+  end
+
+  @impl true
+  def handle_parent_notification(msg, _ctx, state) do
+    state = State.put_event(state, {:message, msg})
+    {[], state}
+  end
+
+  @spec nearest_frame_index([PadState.pad_event()], Membrane.Time.non_neg_t()) ::
+          non_neg_integer() | :no_frame
+  defp nearest_frame_index(events_queue, tick_pts) do
+    events_queue
+    |> Enum.with_index()
+    |> Enum.reduce_while(
+      {:no_frame, :no_frame},
+      fn {event, index}, {best_diff, best_diff_index} ->
+        case event do
+          {:frame, frame_pts, _frame_data}
+          when best_diff == :no_frame or best_diff > abs(frame_pts - tick_pts) ->
+            {:cont, {abs(frame_pts - tick_pts), index}}
+
+          {:frame, _frame_pts, _frame_data} ->
+            {:halt, {best_diff, best_diff_index}}
+
+          _else ->
+            {:cont, {best_diff, best_diff_index}}
+        end
+      end
+    )
+    |> then(fn {_best_diff, best_diff_index} -> best_diff_index end)
+  end
+
+  defp get_tick_ratio(%State{output_framerate: {output_fps_num, output_fps_den}}) do
+    %Ratio{numerator: output_fps_den * Membrane.Time.second(), denominator: output_fps_num}
+  end
+
+  defp check_timer_started(state) do
+    if state.custom_strategy_state.timer_started? do
+      raise "Failed to start timer. Timer already started."
+    end
+  end
+
+  @spec all_pads_eos?(State.t()) :: boolean()
+  defp all_pads_eos?(%State{pads_states: pads_states, next_buffer_pts: buffer_pts}) do
+    pads_states
+    |> Map.values()
+    |> Enum.all?(fn %PadState{events_queue: events_queue} ->
+      eos_before_pts?(events_queue, buffer_pts)
+    end)
+  end
+
+  @spec drop_eos_pads(State.t()) :: State.t()
+  defp drop_eos_pads(
+         state = %State{
+           pads_states: pads_states,
+           output_format: %CompositorCoreFormat{pad_formats: pad_formats},
+           next_buffer_pts: buffer_pts
+         }
+       ) do
+    eos_pads =
+      pads_states
+      |> Enum.filter(fn {_pad, %PadState{events_queue: events_queue}} ->
+        eos_before_pts?(events_queue, buffer_pts)
+      end)
+      |> Enum.map(fn {pad, _pad_state} -> pad end)
+
+    %State{
+      state
+      | pads_states: Map.drop(pads_states, eos_pads),
+        output_format: %CompositorCoreFormat{pad_formats: Map.drop(pad_formats, eos_pads)}
+    }
+  end
+
+  @spec eos_before_pts?(list(PadState.pad_event()), Membrane.Time.non_neg_t()) :: boolean()
+  defp eos_before_pts?(events_queue, buffer_pts) do
+    Enum.reduce_while(events_queue, false, fn event, _eos_before_pts? ->
+      case event do
+        {:frame, pts, _data} when pts > buffer_pts -> {:halt, false}
+        :end_of_stream -> {:halt, true}
+        _else -> {:cont, false}
+      end
+    end)
+  end
+end

lib/membrane/video_compositor/queue/live/state.ex

@@ -0,0 +1,13 @@
+defmodule Membrane.VideoCompositor.Queue.Live.State do
+  @moduledoc false
+
+  alias Membrane.VideoCompositor.Queue.Live
+
+  @enforce_keys [:latency]
+  defstruct @enforce_keys ++ [timer_started?: false]
+
+  @type t :: %__MODULE__{
+          latency: Live.latency(),
+          timer_started?: boolean()
+        }
+end