Make demand action accept timestamps & write manual timestamps guide by FelonEkonom · Pull Request #1079 · membraneframework/membrane_core

FelonEkonom · 2026-02-25T16:23:05Z

Closes #1072
Closes #1081

varsill

Just a guide review, I will come back with code review later

varsill · 2026-03-10T11:19:36Z

+element whose output pad with manual flow control is connected to that
+downstream. The callback receives the pad name, the demanded amount, and the
+demand unit. The element is expected to produce and send that amount of data,
+or less if the stream ends.


Add some note that of course the element does not need to fulfill the whole demand all at once, in a single callback's invocation

varsill · 2026-03-10T11:25:45Z

+   flow control — the output pad inherits `:buffers`.
+
+So the output pad can explicitly control the unit it receives demand in, but
+timestamp units are not available on output pads.


I don't understand this "timestamp units are not available on output pads." part 🤔

varsill · 2026-03-10T13:27:02Z

+the demanded value.
+
+Timestamp demand units are only applicable to **input pads with manual flow
+control**. Output pads do not support them. If an input pad uses a timestamp


Let's rephrase it slightly so that to emphasize that the output pad connected to an input pad with timestamps demand unit receives demand in :bytes or :buffers, as specified by output pad's demand unit (defaulting to :buffers)

Isn't it said in the line below? TBH I don't know if this commend was written before, or after I added this line, so let me know it the way how it is written right now looks good to you

varsill · 2026-03-12T09:48:11Z

+end
+```
+
+> #### Do not use redemand in a filter's `handle_demand` {: .warning}


Something is wrong with this {: .warning}

WDYM? It works fine when I run $ mix docs --open

varsill · 2026-03-12T09:49:48Z

+on `:input`, so each demand naturally covers the next one-second slice of the
+stream.
+
+## Filters with manual flow control


I think using :redemand in filters in also quite "canonical" and it's less intuitive than using it in sources - how about providing an example for it as well?

varsill · 2026-03-12T10:02:50Z

+  For count/byte metrics, subtracts `consumed_size` from `demand`.
+  For timestamp metrics, the demand is a duration that does not change as buffers are consumed —


[NIT] I am not sure which should describe how do behaviour implementations work in the behaviour's callback definition

varsill · 2026-03-12T10:04:05Z

-  defp delay_supplying_demand(pad_ref, 0, state) do
-    Membrane.Logger.debug_verbose("Ignoring demand of size of 0 on pad #{inspect(pad_ref)}")
-    state
-  end
-


Why are we removing it?

Because demand 0 is not ignored when unit is timestamp

varsill · 2026-03-12T10:07:12Z

  def decrease_demand_by_outgoing_buffers(pad_ref, buffers, state) do
    pad_data = PadModel.get_data!(state, pad_ref)
-    buffers_size = Buffer.Metric.from_unit(pad_data.demand_unit).buffers_size(buffers)
+    {:ok, buffers_size} = Buffer.Metric.from_unit(pad_data.demand_unit).buffers_size(buffers)


It will cause a nasty match error when buffers_size returns {:error, ...} - shouldn't we propagate the error further or explicitly raise with a meaningfull error message?

varsill · 2026-03-12T10:13:08Z

+
+  When the pad's `demand_unit` is `:timestamp`, `{:timestamp, :pts}`,
+  `{:timestamp, :dts}`, or `{:timestamp, :dts_or_pts}`, the demand size is a
+  `t:Membrane.Time.t/0` duration (in nanoseconds). The queue will deliver


Suggested change

`t:Membrane.Time.t/0` duration (in nanoseconds). The queue will deliver

`t:Membrane.Time.t/0` duration (expressed internally in nanoseconds). The queue will deliver

I will remove this nanoseconds mention

varsill · 2026-03-12T11:29:07Z

As discussed, please put the metrics into separate modules.

varsill · 2026-03-12T11:31:35Z

+
+  defp maybe_assert_timestamps_present!(buffers, outbound_metric, pad_ref) do
+    Enum.each(buffers, fn buffer ->
+      if outbound_metric.nil_timestamp?(buffer) do


I think it would be better to make get_timestamp() a public callback and use it here to compare it with nil - this way we could get rid of nil_timestamp?() callback

varsill · 2026-03-12T11:41:52Z

As discussed - let's:

add is_timestamp_metric guard and use it in InputQueue

unify generate_metric_specific_warning() and nil_timestamp? assertion into a single callback in TimestampMetric behaviour

Co-authored-by: Łukasz Kita <lukasz.kita0@gmail.com>

mat-hek · 2026-03-16T09:26:44Z

+# Manual demands
+
+Elements with pads using manual flow control have two responsibilities:


I'd start with something like

This guide explains how to use manual demands. Make sure you read the introduction to flow control first.

Manual demands are a mechanism for manually controlling the speed of processing data in Membrane pipelines. This mechanism is powerful, but complex - use it only when :auto flow control is not sufficient for your use case.

I'd also link to this guide in https://hexdocs.pm/membrane_core/06_flow_control.html and add examples of when to use and when not to use manual demands either there or here.

mat-hek · 2026-03-16T09:30:04Z

+
+1. If the **output pad** declares `demand_unit: :buffers | :bytes`, that unit
+   is used.
+2. Otherwise, if the linked **input pad** declares `:buffers` or `:bytes` :demand_unit, that


Suggested change

2. Otherwise, if the linked **input pad** declares `:buffers` or `:bytes` :demand_unit, that

2. Otherwise, if the linked **input pad** declares `:buffers` or `:bytes` `:demand_unit`, that

mat-hek · 2026-03-16T09:35:56Z

+demand unit. The element is expected to produce and send that amount of data,
+or less if the stream ends.
+
+The unit in which `demand_size` is expressed is resolved as follows:


Let's first explain each demand unit

mat-hek · 2026-03-16T09:38:43Z

+1. If the **output pad** declares `demand_unit: :buffers | :bytes`, that unit
+   is used.
+2. Otherwise, if the linked **input pad** declares `:buffers` or `:bytes` :demand_unit, that
+   unit is inherited.
+3. Otherwise — when the input pad uses a timestamp demand unit or has auto
+   flow control — the output pad inherits `:buffers`.


mention that the demand unit is automatically converted to what the output expects if it doesn't match what the input uses

mat-hek · 2026-03-16T09:39:29Z

+[`handle_demand/5`](`c:Membrane.Element.WithOutputPads.handle_demand/5`) always
+receives the **total current outstanding demand** from downstream, not a delta.
+There is no need to accumulate demand values across multiple callback invocations 
+— each call tells you the full amount still expected.


Mention that the delta is available in the context

mat-hek · 2026-03-16T09:42:31Z

+end
+```
+
+A source generating random bytes:


I think the sentence 'a source generating random bytes/buffers' is not precise - each element generates buffers. I'd say 'source handling demand when the unit is :buffers'

mat-hek · 2026-03-16T10:09:01Z

+> Returning [`:redemand`](`t:Membrane.Element.Action.redemand/0`) from
+> [`handle_demand/5`](`c:Membrane.Element.WithOutputPads.handle_demand/5`) in a
+> filter is illegal. [`handle_demand/5`](`c:Membrane.Element.WithOutputPads.handle_demand/5`)


Is it? I don't see it mentioned anywhere in the docs, nor prohibited in the code. I think it rather doesn't make sense, but it's not illegal

AFAIK it is illegal, or at least it was made illegal by recent PRs

mat-hek · 2026-03-16T10:13:47Z

+> [`handle_buffer/4`](`c:Membrane.Element.WithInputPads.handle_buffer/4`) when
+> the data actually arrives.
+
+## Auto flow control


I'd move this to the beginning or to https://hexdocs.pm/membrane_core/06_flow_control.html and add an example when auto flow control is not sufficient

Hm, I added it here deliberately, to as a reference for somebody how will read this guide, to remind that auto flow control exists

This reverts commit 8afb3a1.

FelonEkonom · 2026-03-18T14:33:58Z

TODO: deprecate old buffer metric modules and introduce one big private module to rule them all instead

mat-hek · 2026-03-19T14:12:18Z

As discussed, let's remove the behaviours

mat-hek · 2026-04-20T10:37:04Z

+              [%Buffer{}] | [],
+              non_neg_integer | Membrane.Time.t(),
+              first_consumed_buffer :: Buffer.t() | nil,
+              last_consumed_buffer :: Buffer.t() | nil
+            ) :: {[%Buffer{}] | [], [%Buffer{}] | []}


Suggested change

[%Buffer{}] | [],

non_neg_integer | Membrane.Time.t(),

first_consumed_buffer :: Buffer.t() | nil,

last_consumed_buffer :: Buffer.t() | nil

) :: {[%Buffer{}] | [], [%Buffer{}] | []}

[%Buffer{}],

non_neg_integer | Membrane.Time.t(),

first_consumed_buffer :: Buffer.t() | nil,

last_consumed_buffer :: Buffer.t() | nil

) :: {[%Buffer{}], [%Buffer{}]}

mat-hek · 2026-04-20T10:44:07Z

+    [first | rest] = buffers
+
+    _last =
+      Enum.reduce(rest, first, fn


Iterating over all buffers multiple times may introduce some overhead. We do it here and in assert_non_nil_timestamps and split_timestamp_buffers. I'd do that once if possible.

mat-hek · 2026-04-20T13:38:29Z

Is it a copy of Membrane.Core.Element.ManualFlowController.BufferMetric?

Implement :demand in time WiP

59f6fa6

FelonEkonom changed the title ~~Demand in timestamps~~ Make demand action accept timestamps Feb 25, 2026

FelonEkonom added 18 commits February 26, 2026 17:08

WiP

f1cb61b

Fix typo

25967f0

Fix bug in input queue

a334c5c

Demands in time PoC

c3d62a1

Handle offsets

0305ac4

Handle demand: 0 action WiP

171de3a

Handle demand: 0 action, warn on unmonotonic timestamps

e828fbd

Refactor

94082ac

Fix & improve tests

5875168

Improve tests

4f30b98

Improve docs WiP

ad9798b

wip guide

181eaa5

wip guide

07ccb03

Improve the guide

284f559

Improve the guide

fd014c9

Improve warn log

0c038b6

Update changelog

3721382

Fix credo

b5b96dd

FelonEkonom self-assigned this Mar 9, 2026

Merge remote-tracking branch 'origin/master' into time-on-demands

38df1ea

FelonEkonom changed the title ~~Make demand action accept timestamps~~ Make demand action accept timestamps, write manual timestamps guide Mar 9, 2026

FelonEkonom marked this pull request as ready for review March 9, 2026 15:10

FelonEkonom requested a review from mat-hek as a code owner March 9, 2026 15:10

Refactor demands test

dd3cca4

FelonEkonom marked this pull request as draft March 9, 2026 15:12

FelonEkonom added 3 commits March 9, 2026 17:55

Apply Klaudiusz CR comments

48a52b4

Add vibe coding marks

c943b49

Merge remote-tracking branch 'origin/master' into time-on-demands

313cfc9

FelonEkonom marked this pull request as ready for review March 9, 2026 17:16

FelonEkonom changed the title ~~Make demand action accept timestamps, write manual timestamps guide~~ Make demand action accept timestamps & write manual timestamps guide Mar 9, 2026

Format code snippets from guide

4960c7c

FelonEkonom requested a review from varsill March 10, 2026 09:44

Remove rendundant function in timestamp metrics

06b0c97

varsill requested changes Mar 12, 2026

View reviewed changes

Apply suggestions from code review

8f8123f

Co-authored-by: Łukasz Kita <lukasz.kita0@gmail.com>

mat-hek requested changes Mar 16, 2026

View reviewed changes

FelonEkonom added 4 commits March 17, 2026 17:37

CR wip

2a49446

Refactor metrics WiP

555e022

Refactor timestamp metrics

8afb3a1

Revert "Refactor timestamp metrics"

e44b4f6

This reverts commit 8afb3a1.

mat-hek reviewed Mar 19, 2026

View reviewed changes

Comment thread lib/membrane/buffer/metric/timestamp.ex Outdated

Copy link
Copy Markdown

Member

mat-hek Mar 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As discussed, let's remove the behaviours

FelonEkonom added 12 commits April 13, 2026 16:15

Merge remote-tracking branch 'origin/master' into time-on-demands

ec6d37b

Refactor WiP

09ac2d5

Refactor wip

2b3e361

Minor refactor improvements, credo fix

81eaed9

Restore and deprecate old metric impelmentations

185768a

Fix docs warnings

22d0199

Rename Core Metric into BufferMetric

8aff242

Add redemand in filter example

5fcebec

Fix test

85b5906

mix format

2060df1

Refactor guide, apply CR suggestions

7789b6f

Apply CR suggestion

db67f58

FelonEkonom requested review from mat-hek and varsill April 15, 2026 10:46

mat-hek reviewed Apr 20, 2026

View reviewed changes

		For count/byte metrics, subtracts `consumed_size` from `demand`.
		For timestamp metrics, the demand is a duration that does not change as buffers are consumed —

	`t:Membrane.Time.t/0` duration (in nanoseconds). The queue will deliver
	`t:Membrane.Time.t/0` duration (expressed internally in nanoseconds). The queue will deliver

		# Manual demands

		Elements with pads using manual flow control have two responsibilities:

	2. Otherwise, if the linked input pad declares `:buffers` or `:bytes` :demand_unit, that
	2. Otherwise, if the linked input pad declares `:buffers` or `:bytes` `:demand_unit`, that

Conversation

FelonEkonom commented Feb 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

varsill left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

FelonEkonom commented Mar 18, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

FelonEkonom commented Feb 25, 2026 •

edited

Loading