Some inspects

benchmark/metric/in_progress_memory.ex

@@ -11,7 +11,7 @@ defmodule Benchmark.Metric.InProgressMemory do
 
     if cumulative_memory > cumulative_memory_ref * (1 + @tolerance_factor),
       do:
-        raise(
+        IO.warn(
           "The memory performance has got worse! For test case: #{inspect(test_case, pretty: true)}
         the cumulative memory used to be: #{cumulative_memory_ref} MB and now it is: #{cumulative_memory} MB"
         )

benchmark/metric/message_queues_length.ex

@@ -4,6 +4,8 @@ defmodule Benchmark.Metric.MessageQueuesLength do
   @tolerance_factor 0.5
   @sampling_period 100
 
+  require Membrane.Logger
+
   @impl true
   def assert(queues_lengths, queues_lengths_ref, test_case) do
     cumulative_queues_length = integrate(queues_lengths)
@@ -12,7 +14,7 @@ defmodule Benchmark.Metric.MessageQueuesLength do
     if cumulative_queues_length >
          cumulative_queues_length_ref * (1 + @tolerance_factor),
        do:
-         raise(
+         IO.warn(
            "The cumulative queues length has got worse! For test case: #{inspect(test_case, pretty: true)}
           the cumulative queues length to be: #{cumulative_queues_length_ref} and now it is: #{cumulative_queues_length}"
          )

benchmark/metric/time.ex

@@ -5,12 +5,12 @@ defmodule Benchmark.Metric.Time do
 
   @impl true
   def assert(time, time_ref, test_case) do
-    if time > time_ref * (1 + @tolerance_factor),
-      do:
-        raise(
+    if time > time_ref * (1 + @tolerance_factor) do
+        IO.warn(
           "The time performance has got worse! For test case: #{inspect(test_case, pretty: true)} the test
         used to take: #{time_ref} ms and now it takes: #{time} ms"
         )
+    end
 
     :ok
   end

benchmark/run.exs

@@ -61,43 +61,43 @@ defmodule Benchmark.Run do
   require Membrane.Pad
 
   @test_cases [
-    linear: [
-      reductions: 1_000,
-      max_random: 1,
-      number_of_filters: 10,
-      number_of_buffers: 500_000,
-      buffer_size: 1
-    ],
-    linear: [
-      reductions: 1_000,
-      max_random: 1,
-      number_of_filters: 100,
-      number_of_buffers: 50_000,
-      buffer_size: 1
-    ],
-    linear: [
-      reductions: 1_000,
-      max_random: 5,
-      number_of_filters: 10,
-      number_of_buffers: 50_000,
-      buffer_size: 1
-    ],
+    # linear: [
+    #   reductions: 1_000,
+    #   max_random: 1,
+    #   number_of_filters: 10,
+    #   number_of_buffers: 500_000,
+    #   buffer_size: 1
+    # ],
+    # linear: [
+    #   reductions: 1_000,
+    #   max_random: 1,
+    #   number_of_filters: 100,
+    #   number_of_buffers: 50_000,
+    #   buffer_size: 1
+    # ],
+    # linear: [
+    #   reductions: 1_000,
+    #   max_random: 5,
+    #   number_of_filters: 10,
+    #   number_of_buffers: 50_000,
+    #   buffer_size: 1
+    # ],
     with_branches: [
       struct: [{1, 3}, {3, 2}, {2, 1}],
       reductions: 100,
       number_of_buffers: 50_000,
       buffer_size: 1,
       max_random: 1
-    ],
-    with_branches: [
-      struct: [{1, 2}, {1, 2}, {2, 1}, {2, 1}],
-      reductions: 100,
-      number_of_buffers: 500_000,
-      buffer_size: 1,
-      max_random: 10
+    # ],
+    # with_branches: [
+    #   struct: [{1, 2}, {1, 2}, {2, 1}, {2, 1}],
+    #   reductions: 100,
+    #   number_of_buffers: 500_000,
+    #   buffer_size: 1,
+    #   max_random: 10
     ]
   ]
-  @how_many_tries 5
+  @how_many_tries 3
   # [ms]
   @test_timeout 300_000
   # the greater the factor is, the more unevenly distributed by the dispatcher will the buffers be

benchmark/run.sh

@@ -0,0 +1,11 @@
+echo "FEATURE BRANCH"
+git checkout queue-buffers-when-auto-demand-is-low-v2
+MIX_ENV=benchmark mix do deps.get, deps.compile --force --all, run benchmark/run.exs feature_branch_results
+
+echo "MASTER BRANCH"
+git checkout master
+MIX_ENV=benchmark mix do deps.get, deps.compile --force --all, run benchmark/run.exs master_results
+
+MIX_ENV=benchmark mix run benchmark/compare.exs feature_branch_results master_results
+
+git checkout queue-buffers-when-auto-demand-is-low-v2

lib/membrane/core/element/buffer_controller.ex

@@ -77,6 +77,7 @@ defmodule Membrane.Core.Element.BufferController do
         raise "cannot execute handle_buffer callback for an awaiting input pad"
       end
 
+      state = Map.update!(state, :unqueued_buffers, & &1 + 1)
       state = exec_buffer_callback(pad_ref, buffers, state)
       AutoFlowUtils.auto_adjust_atomic_demand(pad_ref, state)
     end

lib/membrane/core/element/demand_controller/auto_flow_utils.ex

@@ -244,6 +244,7 @@ defmodule Membrane.Core.Element.DemandController.AutoFlowUtils do
     |> case do
       {{:value, {:buffer, buffer}}, popped_queue} ->
         state = PadModel.set_data!(state, pad_ref, :auto_flow_queue, popped_queue)
+        state = Map.update!(state, :queued_buffers, & &1 + 1)
         state = BufferController.exec_buffer_callback(pad_ref, [buffer], state)
         pop_stream_formats_and_events(pad_ref, state)
 

lib/membrane/core/element/event_controller.ex

@@ -131,6 +131,8 @@ defmodule Membrane.Core.Element.EventController do
           state
         )
 
+      IO.inspect({state.name, state.queued_buffers, state.unqueued_buffers}, label: "STATS")
+
       Message.send(
         state.parent_pid,
         :stream_management_event,

lib/membrane/core/element/state.ex

@@ -85,6 +85,8 @@ defmodule Membrane.Core.Element.State do
     :playback_queue,
     :pads_data,
     :satisfied_auto_output_pads,
-    :awaiting_auto_input_pads
+    :awaiting_auto_input_pads,
+    queued_buffers: 0,
+    unqueued_buffers: 0
   ]
 end