Create encoder

c_src/membrane_vpx_plugin/vpx_decoder.c

@@ -74,7 +74,7 @@ UNIFEX_TERM decode_frame(UnifexEnv *env, UnifexPayload *frame, State *state) {
   vpx_image_t *img = NULL;
   PixelFormat pixel_format = PIXEL_FORMAT_I420;
   unsigned int frames_cnt = 0, allocated_frames = 1;
-  UnifexPayload **output_frames = unifex_alloc(allocated_frames * sizeof(*output_frames));
+  UnifexPayload **output_frames = unifex_alloc(allocated_frames * sizeof(UnifexPayload*));
 
   if (vpx_codec_decode(&state->codec_context, frame->data, frame->size, NULL, 0)) {
     return result_error(

c_src/membrane_vpx_plugin/vpx_decoder.spec.exs

@@ -12,4 +12,4 @@ spec decode_frame(payload, state) ::
        {:ok :: label, frames :: [payload], pixel_format :: pixel_format}
        | {:error :: label, reason :: atom}
 
-dirty :cpu, create: 1, decode_frame: 2
+dirty :cpu, [:create, :decode_frame]

c_src/membrane_vpx_plugin/vpx_encoder.c

@@ -90,9 +90,9 @@ UNIFEX_TERM encode(UnifexEnv *env, vpx_image_t *img, vpx_codec_pts_t pts, State
   const vpx_codec_cx_pkt_t *packet = NULL;
 
   unsigned int frames_cnt = 0, allocated_frames = 1;
-  UnifexPayload **encoded_frames = unifex_alloc(allocated_frames * sizeof(*encoded_frames));
+  UnifexPayload **encoded_frames = unifex_alloc(allocated_frames * sizeof(UnifexPayload*));
   vpx_codec_pts_t *encoded_frames_timestamps =
-      unifex_alloc(allocated_frames * sizeof(*encoded_frames_timestamps));
+      unifex_alloc(allocated_frames * sizeof(vpx_codec_pts_t));
 
   do {
     // Reasoning for the do-while and while loops comes from the description of vpx_codec_encode:
@@ -155,4 +155,4 @@ UNIFEX_TERM encode_frame(
   return encode(env, &state->img, pts, state);
 }
 
-UNIFEX_TERM flush(UnifexEnv *env, State *state) { return encode(env, NULL, 0, state); }
+UNIFEX_TERM flush(UnifexEnv *env, State *state) { return encode(env, NULL, 0, state); }

c_src/membrane_vpx_plugin/vpx_encoder.spec.exs

@@ -23,4 +23,4 @@ spec flush(state) ::
        {:ok :: label, frames :: [payload], timestamps :: [int64]}
        | {:error :: label, reason :: atom}
 
-dirty :cpu, create: 5, encode_frame: 3, flush: 1
+dirty :cpu, [:create, :encode_frame, :flush]

lib/membrane_vpx/encoder/vp8_encoder.ex

@@ -7,13 +7,15 @@ defmodule Membrane.VP8.Encoder do
   alias Membrane.{VP8, VPx}
 
   def_options encoding_deadline: [
-                spec: non_neg_integer(),
-                default: 1,
+                spec: Membrane.Time.t() | :auto,
+                default: :auto,
                 description: """
-                Determines how long should it take the encoder to encode a frame (in microseconds).
+                Determines how long should it take the encoder to encode a frame.
                 The longer the encoding takes the better the quality will be. If set to 0 the
                 encoder will take as long as it needs to produce the best frame possible. Note that
                 this is a soft limit, there is no guarantee that the encoding process will never exceed it.
+                If set to `:auto` the deadline will be calculated based on the framerate provided by
+                incoming stream format. If it's `nil` a fixed deadline of 10ms will be set.
                 """
               ]
 

lib/membrane_vpx/encoder/vp9_encoder.ex

@@ -7,13 +7,15 @@ defmodule Membrane.VP9.Encoder do
   alias Membrane.{VP9, VPx}
 
   def_options encoding_deadline: [
-                spec: non_neg_integer(),
-                default: 1,
+                spec: Membrane.Time.t() | :auto,
+                default: :auto,
                 description: """
-                Determines how long should it take the encoder to encode a frame (in microseconds).
+                Determines how long should it take the encoder to encode a frame.
                 The longer the encoding takes the better the quality will be. If set to 0 the
                 encoder will take as long as it needs to produce the best frame possible. Note that
                 this is a soft limit, there is no guarantee that the encoding process will never exceed it.
+                If set to `:auto` the deadline will be calculated based on the framerate provided by
+                incoming stream format. If it's `nil` a fixed deadline of 10ms will be set.
                 """
               ]
 

lib/membrane_vpx/encoder/vpx_encoder.ex

@@ -5,6 +5,8 @@ defmodule Membrane.VPx.Encoder do
   alias Membrane.Element.CallbackContext
   alias Membrane.VPx.Encoder.Native
 
+  @default_encoding_deadline Membrane.Time.milliseconds(10)
+
   defmodule State do
     @moduledoc false
 
@@ -42,25 +44,23 @@ defmodule Membrane.VPx.Encoder do
 
   @spec handle_stream_format(:input, RawVideo.t(), CallbackContext.t(), State.t()) ::
           callback_return()
-  def handle_stream_format(
-        :input,
-        raw_video_format,
-        _ctx,
-        %State{codec_module: codec_module} = state
-      ) do
+  def handle_stream_format(:input, stream_format, ctx, state) do
     %RawVideo{
       width: width,
       height: height,
-      framerate: framerate,
-      pixel_format: pixel_format
-    } = raw_video_format
+      framerate: framerate
+    } = stream_format
 
     output_stream_format =
-      struct(codec_module, width: width, height: height, framerate: framerate)
+      struct(state.codec_module, width: width, height: height, framerate: framerate)
 
-    native = Native.create!(state.codec, width, height, pixel_format, state.encoding_deadline)
+    {flushed_buffers, encoder_ref} =
+      maybe_recreate_encoder(ctx.pads.input.stream_format, stream_format, state)
 
-    {[stream_format: {:output, output_stream_format}], %{state | encoder_ref: native}}
+    {
+      [buffer: {:output, flushed_buffers}, stream_format: {:output, output_stream_format}],
+      %{state | encoder_ref: encoder_ref}
+    }
   end
 
   @spec handle_buffer(:input, Membrane.Buffer.t(), CallbackContext.t(), State.t()) ::
@@ -77,12 +77,48 @@ defmodule Membrane.VPx.Encoder do
 
   @spec handle_end_of_stream(:input, CallbackContext.t(), State.t()) :: callback_return()
   def handle_end_of_stream(:input, _ctx, state) do
-    {:ok, encoded_frames, timestamps} = Native.flush(state.encoder_ref)
+    buffers = flush(state.encoder_ref)
+    {[buffer: {:output, buffers}, end_of_stream: :output], state}
+  end
 
-    buffers =
-      Enum.zip(encoded_frames, timestamps)
-      |> Enum.map(fn {frame, frame_pts} -> %Buffer{payload: frame, pts: frame_pts} end)
+  @spec maybe_recreate_encoder(
+          previous_stream_format :: RawVideo.t(),
+          new_stream_format :: RawVideo.t(),
+          State.t()
+        ) :: {flushed_buffers :: [Buffer.t()], encoder_ref :: reference()}
+  defp maybe_recreate_encoder(unchanged_stream_format, unchanged_stream_format, state) do
+    {[], state.encoder_ref}
+  end
 
-    {[buffer: {:output, buffers}, end_of_stream: :output], state}
+  defp maybe_recreate_encoder(_previous_stream_format, new_stream_format, state) do
+    %RawVideo{
+      width: width,
+      height: height,
+      framerate: framerate,
+      pixel_format: pixel_format
+    } = new_stream_format
+
+    encoding_deadline =
+      case {state.encoding_deadline, framerate} do
+        {:auto, nil} -> @default_encoding_deadline |> Membrane.Time.as_microseconds(:round)
+        {:auto, {num, denom}} -> div(denom * 1_000_000, num)
+        {fixed_deadline, _framerate} -> fixed_deadline |> Membrane.Time.as_microseconds(:round)
+      end
+
+    new_encoder_ref =
+      Native.create!(state.codec, width, height, pixel_format, encoding_deadline)
+
+    case state.encoder_ref do
+      nil -> {[], new_encoder_ref}
+      old_encoder_ref -> {flush(old_encoder_ref), new_encoder_ref}
+    end
+  end
+
+  @spec flush(reference()) :: [Membrane.Buffer.t()]
+  defp flush(encoder_ref) do
+    {:ok, encoded_frames, timestamps} = Native.flush(encoder_ref)
+
+    Enum.zip(encoded_frames, timestamps)
+    |> Enum.map(fn {frame, frame_pts} -> %Buffer{payload: frame, pts: frame_pts} end)
   end
 end

mix.exs

@@ -40,7 +40,7 @@ defmodule Membrane.VPx.Plugin.Mixfile do
     [
       {:membrane_core, "~> 1.0"},
       {:unifex, "~> 1.2"},
-      {:membrane_raw_video_format, "~> 0.4.0", override: true},
+      {:membrane_raw_video_format, "~> 0.4.0"},
       {:membrane_vp8_format, "~> 0.5.0"},
       {:membrane_vp9_format, "~> 0.5.0"},
       {:membrane_precompiled_dependency_provider, "~> 0.1.0"},

mix.lock

@@ -20,11 +20,11 @@
   "makeup_elixir": {:hex, :makeup_elixir, "0.16.2", "627e84b8e8bf22e60a2579dad15067c755531fea049ae26ef1020cad58fe9578", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "41193978704763f6bbe6cc2758b84909e62984c7752b3784bd3c218bb341706b"},
   "makeup_erlang": {:hex, :makeup_erlang, "1.0.0", "6f0eff9c9c489f26b69b61440bf1b238d95badae49adac77973cbacae87e3c2e", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "ea7a9307de9d1548d2a72d299058d1fd2339e3d398560a0e46c27dab4891e4d2"},
   "membrane_core": {:hex, :membrane_core, "1.1.0", "c3bbaa5af7c26a7c3748e573efe343c2104801e3463b9e491a607e82860334a4", [:mix], [{:bunch, "~> 1.6", [hex: :bunch, repo: "hexpm", optional: false]}, {:qex, "~> 0.3", [hex: :qex, repo: "hexpm", optional: false]}, {:ratio, "~> 3.0 or ~> 4.0", [hex: :ratio, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "b3209d7f7e86d736cb7caffbba16b075c571cebb9439ab939ed6119c50fb59a5"},
-  "membrane_file_plugin": {:hex, :membrane_file_plugin, "0.17.0", "e855a848e84eaed537b41fd4436712038fc5518059eadc8609c83cd2d819653a", [:mix], [{:logger_backends, "~> 1.0", [hex: :logger_backends, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}], "hexpm", "9c3653ca9f13bb409b36257d6094798d4625c739ab7a4035c12308622eb16e0b"},
+  "membrane_file_plugin": {:hex, :membrane_file_plugin, "0.17.1", "055a904823506e806e1e1a43643de2dfbe9baf3c1fe2f6f055d2e9b3710767dd", [:mix], [{:logger_backends, "~> 1.0", [hex: :logger_backends, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}], "hexpm", "0b209e17a7bafb8e281fe5d15b3760d9c6f8b3af628ed4589267ba01b7774d8f"},
   "membrane_ivf_plugin": {:git, "https://github.com/membraneframework/membrane_ivf_plugin.git", "e112040f22fe87dbe6142ee85c551abf202b426f", [branch: "fix-plugin"]},
   "membrane_precompiled_dependency_provider": {:hex, :membrane_precompiled_dependency_provider, "0.1.2", "8af73b7dc15ba55c9f5fbfc0453d4a8edfb007ade54b56c37d626be0d1189aba", [:mix], [{:bundlex, "~> 1.4", [hex: :bundlex, repo: "hexpm", optional: false]}], "hexpm", "7fe3e07361510445a29bee95336adde667c4162b76b7f4c8af3aeb3415292023"},
   "membrane_raw_video_format": {:hex, :membrane_raw_video_format, "0.4.1", "d7344499c2d80f236a7ef962b5490c651341a501052ee43dec56cf0319fa3936", [:mix], [], "hexpm", "9920b7d445b5357608a364fec5685acdfce85334c647f745045237a0d296c442"},
-  "membrane_raw_video_parser_plugin": {:hex, :membrane_raw_video_parser_plugin, "0.12.1", "fc0ac1f995411c3e3ccd93ac7ff8fe30930f8ff76d404b2f2a585d7efed6636f", [:mix], [{:bunch, "~> 1.3", [hex: :bunch, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_file_plugin, "~> 0.17.0", [hex: :membrane_file_plugin, repo: "hexpm", optional: false]}, {:membrane_raw_video_format, "~> 0.3.0", [hex: :membrane_raw_video_format, repo: "hexpm", optional: false]}], "hexpm", "bdc7859c9d576f59dd221cfa2a29940b4c58637b321279c23cb7c9e413436b65"},
+  "membrane_raw_video_parser_plugin": {:hex, :membrane_raw_video_parser_plugin, "0.12.2", "7a1f11e122dfc1481654fd5a9ac43db80f7851ad569662cfca2e8a818403101c", [:mix], [{:bunch, "~> 1.3", [hex: :bunch, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_file_plugin, "~> 0.17.0", [hex: :membrane_file_plugin, repo: "hexpm", optional: false]}, {:membrane_raw_video_format, "~> 0.4.0", [hex: :membrane_raw_video_format, repo: "hexpm", optional: false]}], "hexpm", "c9254cc52c96ba0b575a65e4ab41f9218cef91ee5953cf6c1180835a21873907"},
   "membrane_vp8_format": {:hex, :membrane_vp8_format, "0.5.0", "a589c20bb9d97ddc9b717684d00cefc84e2500ce63a0c33c4b9618d9b2f9b2ea", [:mix], [], "hexpm", "d29e0dae4bebc6838e82e031c181fe626d168c687e4bc617c1d0772bdeed19d5"},
   "membrane_vp9_format": {:hex, :membrane_vp9_format, "0.5.0", "c6a4f2cbfc39dba5d80ad8287162c52b5cf6488676bd64435c1ac957bd16e66f", [:mix], [], "hexpm", "68752d8cbe7270ec222fc84a7d1553499f0d8ff86ef9d9e89f8955d49e20278e"},
   "mime": {:hex, :mime, "2.0.5", "dc34c8efd439abe6ae0343edbb8556f4d63f178594894720607772a041b04b02", [:mix], [], "hexpm", "da0d64a365c45bc9935cc5c8a7fc5e49a0e0f9932a761c55d6c52b142780a05c"},

test/membrane_vpx_plugin/vpx_decoder_test.exs

@@ -47,5 +47,7 @@ defmodule Membrane.VPx.DecoderTest do
     assert_end_of_stream(pid, :sink, :input, 2000)
 
     assert File.read!(ref_path) == File.read!(output_path)
+
+    Membrane.Testing.Pipeline.terminate(pid)
   end
 end

test/membrane_vpx_plugin/vpx_encoder_test.exs

@@ -55,5 +55,7 @@ defmodule Membrane.VPx.EncoderTest do
     assert_end_of_stream(pid, :sink, :input, 10_000)
 
     assert File.read!(ref_path) == File.read!(output_path)
+
+    Membrane.Testing.Pipeline.terminate(pid)
   end
 end