Implement basic functionalities #1

FelonEkonom · 2024-03-19T17:00:55Z

Continuation of membraneframework/membrane_core#756
Relates to membraneframework/membrane_core#734

mat-hek · 2024-03-28T14:37:51Z

lib/membrane/timestamp_queue.ex

+  def new(options \\ []) do
+    metric =
+      options
+      |> Keyword.get(:pause_demand_boundary_unit, :buffers)


as we spoke, the most appropriate unit seems to be time

I will implement having boundary unit expressed in time in the another PR

mat-hek · 2024-03-28T14:55:46Z

lib/membrane/timestamp_queue.ex

+  Makes the queue not return any buffer in `pop_batch/3`, until a buffer or end of stream arrival
+  from `pad_ref`.
+  """
+  @spec wait_on_pad(t(), Pad.ref()) :: t()
+  def wait_on_pad(%__MODULE__{} = timestamp_queue, pad_ref) do


Suggested change

Makes the queue not return any buffer in `pop_batch/3`, until a buffer or end of stream arrival

from `pad_ref`.

"""

@spec wait_on_pad(t(), Pad.ref()) :: t()

def wait_on_pad(%__MODULE__{} = timestamp_queue, pad_ref) do

Registers an input pad in the queue without pushing anything on that pad.

Once a pad is registered, the `pop_batch/3` function won't return buffers

until a `buffer` or `end_of_stream` is available on the registered pad.

Pushing a buffer on an unregistered pad automatically registers it.

"""

@spec register_pad(t(), Pad.ref()) :: t()

def register_pad(%__MODULE__{} = timestamp_queue, pad_ref) do

After calling this function, stream formats and events still can occur in the returned batch

Calling push_stream_format or push_event doesn't cause waiting on the pad

I updated the suggestion

mat-hek · 2024-03-29T09:43:57Z

lib/membrane/timestamp_queue.ex

+  @doc """
+  Creates and returnes new #{inspect(__MODULE__)}.
+
+  Accepts `t:options()`.
+  """


this says nothing :P

Suggested change

@doc """

Creates and returnes new #{inspect(__MODULE__)}.

Accepts `t:options()`.

"""

mat-hek · 2024-03-29T09:48:05Z

lib/membrane/timestamp_queue.ex

+  A queue, that accepts buffers, stream formats and events from various pads and sorts them based on
+  their timestamps.
+  """
+  @type t :: %__MODULE__{


this should probably be opaque

lib/membrane/timestamp_queue.ex

mat-hek · 2024-03-29T09:54:46Z

lib/membrane/timestamp_queue.ex

+      Buffer #{inspect(buffer, pretty: true)} from pad #{inspect(pad_ref)} has timestamp equal \
+      #{inspect(buffer_timestamp)}, but previous buffer pushed on queue from this pad had timestamp \
+      equal #{inspect(max_timestamp)}. Buffers from a single pad must have non-decreasing timestamps.


It would be nice to change the word timestamp to PTS or DTS, whichever we rely on

I can refactor error message, but simply changing timestamp to PTS or DTS will make this log untrue, since having buffers with monotonic DTS and non-monotonic PTS is totally ok

having buffers with monotonic DTS and non-monotonic PTS is totally ok

If the queue relies on PTS and they are not monotonic, the queue can't handle this case no matter what DTS are, so I think the log would be true. Such a situation would probably mean a bug anyway - I believe PTS and DTS should either be always set or never set.

If buffers from a specific pad contain DTS, queue will look only at DTS and won't look at PTS at all

mat-hek · 2024-03-29T10:14:33Z

lib/membrane/timestamp_queue.ex

+  Pops as many buffers from the queue, as it is possible.
+
+  Returns suggested actions list, list of popped buffers and the updated queue.
+
+  If the amount of buffers associated with any pad in the queue falls below the
+  `pause_demand_boundary`, the suggested actions list contains `t:Action.resume_auto_demand()`
+  actions, otherwise it is an empty list.
+
+  If the queue cannot return any buffer, empty list is returned. Note, that queue doesn't have to be
+  empty to be unable to return a buffer - sometimes queue must keep up to 1 buffer for each pad,
+  to be able to work correctly.
+
+  To be able to maintain proper order of buffers from varius pads, queue won't return next buffer,
+  if the next buffer that should be returned:
+   - is the only buffer from a certain pad
+   - and that certain pad has not received the end of stream yet


Here's my try, correct me if I'm wrong:

Suggested change

Pops as many buffers from the queue, as it is possible.

Returns suggested actions list, list of popped buffers and the updated queue.

If the amount of buffers associated with any pad in the queue falls below the

`pause_demand_boundary`, the suggested actions list contains `t:Action.resume_auto_demand()`

actions, otherwise it is an empty list.

If the queue cannot return any buffer, empty list is returned. Note, that queue doesn't have to be

empty to be unable to return a buffer - sometimes queue must keep up to 1 buffer for each pad,

to be able to work correctly.

To be able to maintain proper order of buffers from varius pads, queue won't return next buffer,

if the next buffer that should be returned:

- is the only buffer from a certain pad

- and that certain pad has not received the end of stream yet

Pops items from the queue while they are available.

An item that is not a buffer is always considered available. A buffer is

available when the following conditions are met:

- There is another buffer or `end_of_stream` enqueued on the pad

- On each other pad there is either `end_of_stream` or a buffer with a lower timestamp.

The returned value is a suggested actions list, a list of popped buffers and the updated queue.

If the amount of buffers associated with any pad in the queue falls below the

`pause_demand_boundary`, the suggested actions list contains `t:Action.resume_auto_demand()`

actions, otherwise it is an empty list.

lib/membrane/timestamp_queue.ex

mat-hek · 2024-04-04T13:13:05Z

lib/membrane/timestamp_queue.ex

+    timestamp_field = if pad_queue.use_pts?, do: "pts", else: "dts"
    max_timestamp = pad_queue.max_timestamp_on_qex

    if is_integer(max_timestamp) and max_timestamp > buffer_timestamp do
      raise """


Suggested change

timestamp_field = if pad_queue.use_pts?, do: "pts", else: "dts"

max_timestamp = pad_queue.max_timestamp_on_qex

if is_integer(max_timestamp) and max_timestamp > buffer_timestamp do

raise """

max_timestamp = pad_queue.max_timestamp_on_qex

if is_integer(max_timestamp) and max_timestamp > buffer_timestamp do

timestamp_field = if pad_queue.use_pts?, do: "pts", else: "dts"

raise """

FelonEkonom self-assigned this Mar 19, 2024

FelonEkonom mentioned this pull request Mar 19, 2024

Timestamp queue POC membraneframework/membrane_core#756

Closed

Copy content from membraneframework/membrane_core#756

f7febd7

FelonEkonom force-pushed the basic-functionalities branch from b19c76f to f7febd7 Compare March 19, 2024 17:06

FelonEkonom mentioned this pull request Mar 25, 2024

Helper queue for multi-input elements membraneframework/membrane_core#734

Closed

Refactor docs and test names

58c1df0

FelonEkonom force-pushed the basic-functionalities branch from d7ef045 to 58c1df0 Compare March 26, 2024 10:19

FelonEkonom requested a review from mat-hek March 26, 2024 10:28

FelonEkonom added the enhancement New feature or request label Mar 26, 2024

Eradicate leftovers from membrane_template_plugin

b469dbf

FelonEkonom force-pushed the basic-functionalities branch from 5b507ea to b469dbf Compare March 26, 2024 10:38

Set codeowner

6d27d43

mat-hek requested changes Mar 29, 2024

View reviewed changes

FelonEkonom added 2 commits April 4, 2024 13:47

Refactor docs and error messages

520d432

mix format

1ac231e

FelonEkonom requested a review from mat-hek April 4, 2024 11:48

mat-hek reviewed Apr 4, 2024

View reviewed changes

FelonEkonom force-pushed the basic-functionalities branch from 000fa9f to 1ac231e Compare April 5, 2024 12:42

Implement suggestion from CR

d6e0587

FelonEkonom requested a review from mat-hek April 5, 2024 12:48

mat-hek approved these changes Apr 5, 2024

View reviewed changes

FelonEkonom merged commit f61b5ee into master Apr 5, 2024
3 checks passed

FelonEkonom deleted the basic-functionalities branch April 5, 2024 13:06

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implement basic functionalities #1

Implement basic functionalities #1

FelonEkonom commented Mar 19, 2024

mat-hek Mar 28, 2024

FelonEkonom Mar 29, 2024

mat-hek Mar 28, 2024 •

edited

Loading

FelonEkonom Apr 3, 2024

mat-hek Apr 4, 2024

mat-hek Mar 29, 2024

mat-hek Mar 29, 2024

mat-hek Mar 29, 2024

FelonEkonom Mar 29, 2024

mat-hek Apr 2, 2024

FelonEkonom Apr 2, 2024 •

edited

Loading

mat-hek Mar 29, 2024

mat-hek Apr 4, 2024

-  Makes the queue not return any buffer in `pop_batch/3`, until a buffer or end of stream arrival
-  from `pad_ref`.
-  """
-  @spec wait_on_pad(t(), Pad.ref()) :: t()
-  def wait_on_pad(%__MODULE__{} = timestamp_queue, pad_ref) do
+  Registers an input pad in the queue without pushing anything on that pad.
+  Once a pad is registered, the `pop_batch/3` function won't return buffers
+  until a `buffer` or `end_of_stream` is available on the registered pad.
+  Pushing a buffer on an unregistered pad automatically registers it.
+  """
+  @spec register_pad(t(), Pad.ref()) :: t()
+  def register_pad(%__MODULE__{} = timestamp_queue, pad_ref) do

Implement basic functionalities #1

Implement basic functionalities #1

Conversation

FelonEkonom commented Mar 19, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mat-hek Mar 28, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

FelonEkonom Apr 2, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mat-hek Mar 28, 2024 •

edited

Loading

FelonEkonom Apr 2, 2024 •

edited

Loading