Skip to content

upipe_audio_copy: fix outputted flow format #1156

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cmassiot merged 1 commit into from
Mar 25, 2026
Merged

upipe_audio_copy: fix outputted flow format #1156

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 71 additions & 31 deletions lib/upipe-modules/upipe_audio_copy.c
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
/*
* Copyright (C) 2018 OpenHeadend S.A.R.L.
* Copyright (C) 2020 EasyTools
* Copyright (C) 2020-2026 EasyTools
*
* Authors: Arnaud de Turckheim
*
Expand Down Expand Up @@ -130,7 +130,8 @@ static struct upipe *upipe_audio_copy_alloc(struct upipe_mgr *mgr,
upipe_audio_copy_from_upipe(upipe);
ulist_init(&upipe_audio_copy->buffers);
upipe_audio_copy->planes = 0;
upipe_audio_copy->samplerate = UINT64_MAX;
upipe_audio_copy->samplerate = 0;
upipe_audio_copy->sample_size = 0;
upipe_audio_copy->remain = 0;

upipe_throw_ready(upipe);
Expand All @@ -140,16 +141,14 @@ static struct upipe *upipe_audio_copy_alloc(struct upipe_mgr *mgr,
upipe_audio_copy->fps.num = 0;
upipe_audio_copy->fps.den = 1;
uref_pic_flow_get_fps(flow_def, &upipe_audio_copy->fps);
uref_free(flow_def);

if (unlikely(!upipe_audio_copy->samples && !upipe_audio_copy->fps.num)) {
upipe_err(upipe, "invalid flow def parameters");
uref_free(flow_def);
upipe_release(upipe);
return NULL;
}

upipe_audio_copy_store_flow_def_attr(upipe, flow_def);

return upipe;
}

Expand Down Expand Up @@ -220,31 +219,59 @@ static int upipe_audio_copy_set_flow_def_real(struct upipe *upipe,
struct upipe_audio_copy *upipe_audio_copy =
upipe_audio_copy_from_upipe(upipe);

UBASE_RETURN(uref_flow_match_def(flow_def, EXPECTED_FLOW_DEF));
UBASE_RETURN(uref_sound_flow_get_rate(flow_def,
&upipe_audio_copy->samplerate));
UBASE_RETURN(uref_sound_flow_get_planes(flow_def,
&upipe_audio_copy->planes));
UBASE_RETURN(uref_sound_flow_get_sample_size(
flow_def, &upipe_audio_copy->sample_size));
if (unlikely(!upipe_audio_copy->samplerate))
return UBASE_ERR_INVALID;
if (upipe_audio_copy_check_flow_def_input(upipe, flow_def)) {
/* nothing changed */
uref_free(flow_def);
return UBASE_ERR_NONE;
}
upipe_audio_copy_store_flow_def_input(upipe, flow_def);

uint64_t rate = 0;
uint8_t planes = 0;
uint8_t sample_size = 0;
uint64_t input_latency = 0;
uref_sound_flow_get_rate(flow_def, &rate);
uref_sound_flow_get_planes(flow_def, &planes);
uref_sound_flow_get_sample_size(flow_def, &sample_size);
uref_clock_get_latency(flow_def, &input_latency);

/* allocate new output flow def */
flow_def = upipe_audio_copy_make_flow_def(upipe);
if (unlikely(!flow_def)) {
upipe_audio_copy_store_flow_def_input(upipe, NULL);
return UBASE_ERR_ALLOC;
}

struct uref *flow_def_dup = uref_dup(flow_def);
UBASE_ALLOC_RETURN(flow_def_dup);
struct uref *output_flow_def =
upipe_audio_copy_store_flow_def_input(upipe, flow_def_dup);
UBASE_ALLOC_RETURN(output_flow_def);
uint64_t latency = 0;
uref_clock_get_latency(flow_def, &latency);
latency += upipe_audio_copy->samples * UCLOCK_FREQ /
upipe_audio_copy->samplerate;
int ret = uref_clock_set_latency(output_flow_def, latency);
if (unlikely(!ubase_check(ret))) {
uref_free(output_flow_def);
return ret;
uref_sound_flow_delete_samples(flow_def);

/* compute maximum number of samples per output uref */
uint64_t samples = upipe_audio_copy->samples;
if (!samples) {
samples = rate * upipe_audio_copy->fps.den / upipe_audio_copy->fps.num;
if (samples * upipe_audio_copy->fps.num !=
rate * upipe_audio_copy->fps.den)
samples++;
else
uref_sound_flow_set_samples(flow_def, samples);
}
else
uref_sound_flow_set_samples(flow_def, samples);

/* compute latency */
uint64_t latency = samples * UCLOCK_FREQ / rate;
if (latency * rate != samples * UCLOCK_FREQ)
latency++;

uref_clock_set_latency(flow_def, input_latency + latency);

if (upipe_audio_copy->samplerate == rate &&
upipe_audio_copy->sample_size == sample_size &&
upipe_audio_copy->planes == planes) {
upipe_audio_copy_store_flow_def(upipe, flow_def);
return UBASE_ERR_NONE;
}

/* flush */
struct uchain *uchain;
while ((uchain = ulist_pop(&upipe_audio_copy->buffers))) {
upipe_warn(upipe, "delete retained buffer");
Expand All @@ -253,7 +280,11 @@ static int upipe_audio_copy_set_flow_def_real(struct upipe *upipe,
upipe_audio_copy->size = 0;
upipe_audio_copy->remain = 0;

upipe_audio_copy_require_ubuf_mgr(upipe, output_flow_def);
/* ask for a new ubuf manager */
upipe_audio_copy->samplerate = rate;
upipe_audio_copy->sample_size = sample_size;
upipe_audio_copy->planes = planes;
upipe_audio_copy_require_ubuf_mgr(upipe, flow_def);
return UBASE_ERR_NONE;
}

Expand Down Expand Up @@ -477,10 +508,19 @@ static void upipe_audio_copy_input(struct upipe *upipe,
static int upipe_audio_copy_set_flow_def(struct upipe *upipe,
struct uref *flow_def)
{
uint64_t rate = 0;
uint8_t planes = 0;
uint8_t sample_size = 0;

UBASE_RETURN(uref_flow_match_def(flow_def, EXPECTED_FLOW_DEF));
UBASE_RETURN(uref_sound_flow_get_rate(flow_def, NULL));
UBASE_RETURN(uref_sound_flow_get_planes(flow_def, NULL));
UBASE_RETURN(uref_sound_flow_get_sample_size(flow_def, NULL));
UBASE_RETURN(uref_sound_flow_get_rate(flow_def, &rate));
UBASE_RETURN(uref_sound_flow_get_planes(flow_def, &planes));
UBASE_RETURN(uref_sound_flow_get_sample_size(flow_def, &sample_size));
if (unlikely(!rate || !planes || !sample_size))
return UBASE_ERR_INVALID;

flow_def = uref_dup(flow_def);
UBASE_ALLOC_RETURN(flow_def);
upipe_input(upipe, flow_def, NULL);
return UBASE_ERR_NONE;
}
Expand Down
121 changes: 101 additions & 20 deletions tests/upipe_audio_copy_test.c
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
/*
* Copyright (C) 2018 OpenHeadend S.A.R.L.
* Copyright (C) 2020 EasyTools
* Copyright (C) 2020-2026 EasyTools
*
* Authors: Arnaud de Turckheim
*
Expand Down Expand Up @@ -43,13 +43,16 @@
#define RATE 48000
#define LIMIT 8
#define OUTPUT_SIZE 1024
#define INPUT_SIZE (3 * OUTPUT_SIZE + (OUTPUT_SIZE / LIMIT))
#define TOTAL_INPUT_SIZE (LIMIT * INPUT_SIZE)

static uint64_t last_pts = 0;

struct sink {
struct upipe upipe;
struct urefcount urefcount;
uint64_t count;
size_t size;
};

UPIPE_HELPER_UPIPE(sink, upipe, 0);
Expand All @@ -61,6 +64,7 @@ static void sink_free(struct upipe *upipe)
struct sink *sink = sink_from_upipe(upipe);

assert(sink->count == 3 * LIMIT + 1);
assert(sink->size <= TOTAL_INPUT_SIZE);

upipe_throw_dead(upipe);

Expand All @@ -80,6 +84,7 @@ static struct upipe *sink_alloc(struct upipe_mgr *mgr,

struct sink *sink = sink_from_upipe(upipe);
sink->count = 0;
sink->size = 0;

upipe_throw_ready(upipe);

Expand All @@ -97,6 +102,27 @@ static void sink_input(struct upipe *upipe, struct uref *uref,
size_t size;
ubase_assert(uref_sound_size(uref, &size, NULL));
assert(size == OUTPUT_SIZE);
sink->size += size;
uint64_t pts;
ubase_assert(uref_clock_get_pts_prog(uref, &pts));
assert(pts > last_pts);
last_pts = pts;
uref_free(uref);
}

static void sink_input4(struct upipe *upipe, struct uref *uref,
struct upump **upump_p)
{
struct sink *sink = sink_from_upipe(upipe);

sink->count = 3 * LIMIT + 1;
uref_dump(uref, upipe->uprobe);
assert(uref->ubuf);
size_t size;
ubase_assert(uref_sound_size(uref, &size, NULL));
upipe_notice_va(upipe, "size %zu", size);
assert(size == 1600 || size == 1602);
sink->size += size;
uint64_t pts;
ubase_assert(uref_clock_get_pts_prog(uref, &pts));
assert(pts > last_pts);
Expand All @@ -113,10 +139,20 @@ static int sink_set_flow_def1(struct upipe *upipe, struct uref *flow_def)
return UBASE_ERR_NONE;
}

static int sink_set_flow_def2(struct upipe *upipe, struct uref *flow_def)
static int sink_set_flow_def3(struct upipe *upipe, struct uref *flow_def)
{
uint64_t samples = 0;
ubase_assert(uref_flow_match_def(flow_def, UREF_SOUND_FLOW_DEF));
ubase_nassert(uref_sound_flow_get_samples(flow_def, NULL));
ubase_assert(uref_sound_flow_get_samples(flow_def, &samples));
assert(samples == OUTPUT_SIZE);
return UBASE_ERR_NONE;
}

static int sink_set_flow_def4(struct upipe *upipe, struct uref *flow_def)
{
uint64_t samples = 0;
ubase_assert(uref_flow_match_def(flow_def, UREF_SOUND_FLOW_DEF));
ubase_nassert(uref_sound_flow_get_samples(flow_def, &samples));
return UBASE_ERR_NONE;
}

Expand Down Expand Up @@ -152,6 +188,9 @@ static struct upipe_mgr sink_mgr = {

int main(int argc, char *argv[])
{
struct uref *flow_def;
struct urational fps;

struct uclock *uclock = uclock_std_alloc(0);

struct umem_mgr *umem_mgr = umem_alloc_mgr_alloc();
Expand Down Expand Up @@ -180,15 +219,15 @@ int main(int argc, char *argv[])
struct upipe_mgr *upipe_audio_copy_mgr = upipe_audio_copy_mgr_alloc();
assert(upipe_audio_copy_mgr);

struct uref *flow_def =
uref_sound_flow_alloc_def(uref_mgr, "s16.", CHANNELS, 2 * CHANNELS);
flow_def = uref_alloc_control(uref_mgr);
ubase_assert(uref_flow_set_def(flow_def, UREF_SOUND_FLOW_DEF));
ubase_assert(uref_sound_flow_set_samples(flow_def, OUTPUT_SIZE));

struct upipe *upipe_audio_copy =
upipe_flow_alloc(upipe_audio_copy_mgr,
uprobe_pfx_alloc(uprobe_use(uprobe),
UPROBE_LOG_LEVEL,
"frame"),
"frame 1"),
flow_def);
uref_free(flow_def);
upipe_mgr_release(upipe_audio_copy_mgr);
Expand All @@ -207,15 +246,15 @@ int main(int argc, char *argv[])
uref_sound_flow_alloc_def(uref_mgr, "s16.", CHANNELS, 2 * CHANNELS);
assert(flow_def);
ubase_assert(uref_sound_flow_set_rate(flow_def, RATE));
ubase_assert(uref_sound_flow_set_planes(flow_def, 2));
ubase_assert(uref_sound_flow_add_plane(flow_def, "l"));
ubase_assert(uref_sound_flow_add_plane(flow_def, "r"));
ubase_assert(upipe_set_flow_def(upipe_audio_copy, flow_def));
uref_free(flow_def);

uint64_t pts = 1000;
for (unsigned i = 0; i < LIMIT; i++) {
size_t size = 3 * OUTPUT_SIZE + (OUTPUT_SIZE / LIMIT) + (i ? 0 : 1);
struct uref *uref = uref_sound_alloc(uref_mgr, ubuf_mgr,
3 * OUTPUT_SIZE + (OUTPUT_SIZE / LIMIT) + (i ? 0 : 1));
size_t size = INPUT_SIZE + (i ? 0 : 1);
struct uref *uref = uref_sound_alloc(uref_mgr, ubuf_mgr, size);
uref_clock_set_pts_prog(uref, pts);
pts += 1000 + (UCLOCK_FREQ * size / RATE);
upipe_input(upipe_audio_copy, uref, NULL);
Expand All @@ -224,36 +263,80 @@ int main(int argc, char *argv[])
upipe_release(upipe_audio_copy);


flow_def =
uref_sound_flow_alloc_def(uref_mgr, "s16.", CHANNELS, 2 * CHANNELS);
flow_def = uref_alloc_control(uref_mgr);
ubase_assert(uref_flow_set_def(flow_def, UREF_SOUND_FLOW_DEF));
//ubase_assert(uref_sound_flow_set_samples(flow_def, OUTPUT_SIZE));

upipe_audio_copy =
upipe_flow_alloc(upipe_audio_copy_mgr,
uprobe_pfx_alloc(uprobe_use(uprobe),
UPROBE_LOG_LEVEL,
"frame"),
"frame 2"),
flow_def);
uref_free(flow_def);
upipe_mgr_release(upipe_audio_copy_mgr);
assert(upipe_audio_copy == NULL);

flow_def = uref_alloc_control(uref_mgr);
ubase_assert(uref_flow_set_def(flow_def, UREF_SOUND_FLOW_DEF));
fps.num = RATE;
fps.den = OUTPUT_SIZE;
ubase_assert(uref_pic_flow_set_fps(flow_def, fps));

upipe_audio_copy =
upipe_flow_alloc(upipe_audio_copy_mgr,
uprobe_pfx_alloc(uprobe_use(uprobe),
UPROBE_LOG_LEVEL,
"frame 3"),
flow_def);
uref_free(flow_def);
upipe_mgr_release(upipe_audio_copy_mgr);
assert(upipe_audio_copy);

sink_set_flow_def = sink_set_flow_def3;
sink =
upipe_void_alloc_output(upipe_audio_copy, &sink_mgr,
uprobe_pfx_alloc(uprobe_use(uprobe),
UPROBE_LOG_LEVEL,
"sink"));
assert(sink);
upipe_release(sink);

flow_def =
uref_sound_flow_alloc_def(uref_mgr, "s16.", CHANNELS, 2 * CHANNELS);
struct urational fps = { .num = RATE, .den = OUTPUT_SIZE };
assert(flow_def);
ubase_assert(uref_sound_flow_set_rate(flow_def, RATE));
ubase_assert(uref_sound_flow_add_plane(flow_def, "lr"));
ubase_assert(upipe_set_flow_def(upipe_audio_copy, flow_def));
uref_free(flow_def);

for (unsigned i = 0; i < LIMIT; i++) {
struct uref *uref = uref_sound_alloc(uref_mgr, ubuf_mgr, INPUT_SIZE);
uref_clock_set_pts_prog(uref, pts);
pts += (i ? 3 : 4) * UCLOCK_FREQ;
upipe_input(upipe_audio_copy, uref, NULL);
}

upipe_release(upipe_audio_copy);

flow_def = uref_alloc_control(uref_mgr);
ubase_assert(uref_flow_set_def(flow_def, UREF_SOUND_FLOW_DEF));
fps.num = 30000;
fps.den = 1001;
ubase_assert(uref_pic_flow_set_fps(flow_def, fps));

upipe_audio_copy =
upipe_flow_alloc(upipe_audio_copy_mgr,
uprobe_pfx_alloc(uprobe_use(uprobe),
UPROBE_LOG_LEVEL,
"frame"),
"frame 4"),
flow_def);
uref_free(flow_def);
upipe_mgr_release(upipe_audio_copy_mgr);
assert(upipe_audio_copy);

sink_set_flow_def = sink_set_flow_def2;
sink_set_flow_def = sink_set_flow_def4;
sink_mgr.upipe_input = sink_input4;
sink =
upipe_void_alloc_output(upipe_audio_copy, &sink_mgr,
uprobe_pfx_alloc(uprobe_use(uprobe),
Expand All @@ -266,14 +349,12 @@ int main(int argc, char *argv[])
uref_sound_flow_alloc_def(uref_mgr, "s16.", CHANNELS, 2 * CHANNELS);
assert(flow_def);
ubase_assert(uref_sound_flow_set_rate(flow_def, RATE));
ubase_assert(uref_sound_flow_set_planes(flow_def, 2));
ubase_assert(uref_sound_flow_add_plane(flow_def, "lr"));
ubase_assert(upipe_set_flow_def(upipe_audio_copy, flow_def));
uref_free(flow_def);

for (unsigned i = 0; i < LIMIT; i++) {
size_t size = 3 * OUTPUT_SIZE + (OUTPUT_SIZE / LIMIT);
struct uref *uref = uref_sound_alloc(
uref_mgr, ubuf_mgr, size);
struct uref *uref = uref_sound_alloc(uref_mgr, ubuf_mgr, INPUT_SIZE);
uref_clock_set_pts_prog(uref, pts);
pts += (i ? 3 : 4) * UCLOCK_FREQ;
upipe_input(upipe_audio_copy, uref, NULL);
Expand Down
Loading