Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
22 changes: 11 additions & 11 deletions daemon/media_socket.c
Original file line number Diff line number Diff line change
Expand Up @@ -2664,7 +2664,7 @@ const struct streamhandler *determine_sink_handler(struct packet_stream *in, str
static const char *__stream_ssrc_inout(struct packet_stream *ps, uint32_t ssrc,
uint32_t output_ssrc,
struct ssrc_entry_call **output, struct ssrc_hash *ssrc_hash,
const char *label)
const char *label, bool is_rtcp)
{
const char *ret = NULL;

Expand All @@ -2677,7 +2677,7 @@ static const char *__stream_ssrc_inout(struct packet_stream *ps, uint32_t ssrc,
mutex_unlock(&ssrc_hash->lock);

bool created = false;
struct ssrc_entry_call *se = first ?: get_ssrc_full(ssrc, ssrc_hash, &created);
struct ssrc_entry_call *se = first ?: get_ssrc_full(ssrc, ssrc_hash, &created, !is_rtcp);

if (created) {
ret = "SSRC changed";
Expand All @@ -2700,26 +2700,26 @@ static const char *__stream_ssrc_inout(struct packet_stream *ps, uint32_t ssrc,
// check and update input SSRC pointers
// returns non-null with reason string if stream should be removed from kernel
static const char *__stream_ssrc_in(struct packet_stream *in_srtp, uint32_t ssrc_bs,
struct ssrc_entry_call **ssrc_in_p, struct ssrc_hash *ssrc_hash)
struct ssrc_entry_call **ssrc_in_p, struct ssrc_hash *ssrc_hash, bool is_rtcp)
{
return __stream_ssrc_inout(in_srtp, ntohl(ssrc_bs),
0, ssrc_in_p, ssrc_hash, "ingress");
0, ssrc_in_p, ssrc_hash, "ingress", is_rtcp);
}
// check and update output SSRC pointers
// returns non-null with reason string if stream should be removed from kernel
static const char *__stream_ssrc_out(struct packet_stream *out_srtp, uint32_t ssrc_bs,
struct ssrc_entry_call *ssrc_in, struct ssrc_entry_call **ssrc_out_p,
struct ssrc_hash *ssrc_hash,
bool ssrc_change)
bool ssrc_change, bool is_rtcp)
{
if (ssrc_change)
return __stream_ssrc_inout(out_srtp, ssrc_in->ssrc_map_out,
ntohl(ssrc_bs), ssrc_out_p, ssrc_hash,
"egress (mapped)");
"egress (mapped)", is_rtcp);

return __stream_ssrc_inout(out_srtp, ntohl(ssrc_bs),
0, ssrc_out_p, ssrc_hash,
"egress (direct)");
"egress (direct)", is_rtcp);
}


Expand Down Expand Up @@ -3003,7 +3003,7 @@ static void media_packet_rtp_in(struct packet_handler_ctx *phc) {

if (phc->mp.rtp) {
unkern = __stream_ssrc_in(phc->in_srtp, phc->mp.rtp->ssrc, &phc->mp.ssrc_in,
ssrc_hash);
ssrc_hash, false);

if (G_LIKELY(phc->mp.ssrc_in))
payload_tracker_add(&phc->mp.ssrc_in->tracker, phc->payload_type);
Expand Down Expand Up @@ -3032,7 +3032,7 @@ static void media_packet_rtp_in(struct packet_handler_ctx *phc) {
}
else if (phc->rtcp) {
unkern = __stream_ssrc_in(phc->in_srtp, phc->mp.rtcp->ssrc, &phc->mp.ssrc_in,
ssrc_hash);
ssrc_hash, true);
}

if (unkern)
Expand Down Expand Up @@ -3065,7 +3065,7 @@ static void media_packet_rtp_out(struct packet_handler_ctx *phc, struct sink_han
if (G_LIKELY(!phc->rtcp && phc->mp.rtp)) {
unkern = __stream_ssrc_out(phc->out_srtp, phc->mp.rtp->ssrc, phc->mp.ssrc_in,
&phc->mp.ssrc_out, &phc->mp.media_out->ssrc_hash_out,
sh->attrs.transcoding || fixed_ssrc);
sh->attrs.transcoding || fixed_ssrc, phc->rtcp);

/* a fixed egress SSRC outlives the ingress SSRC, so carry the
* sequence numbering across a change of source */
Expand All @@ -3087,7 +3087,7 @@ static void media_packet_rtp_out(struct packet_handler_ctx *phc, struct sink_han
else if (phc->rtcp && phc->mp.rtcp) {
unkern = __stream_ssrc_out(phc->out_srtp, phc->mp.rtcp->ssrc, phc->mp.ssrc_in,
&phc->mp.ssrc_out, &phc->mp.media_out->ssrc_hash_out,
sh->attrs.transcoding || fixed_ssrc);
sh->attrs.transcoding || fixed_ssrc, phc->rtcp);
}

if (unkern)
Expand Down
4 changes: 2 additions & 2 deletions daemon/rtcp.c
Original file line number Diff line number Diff line change
Expand Up @@ -1302,7 +1302,7 @@ static void transcode_rr(struct rtcp_process_ctx *ctx, struct report_block *rr)
return;

// reverse SSRC mapping
struct ssrc_entry_call *map_ctx = get_ssrc(ctx->scratch.rr.ssrc, &ctx->mp->media->ssrc_hash_out);
struct ssrc_entry_call *map_ctx = get_ssrc_no_move(ctx->scratch.rr.ssrc, &ctx->mp->media->ssrc_hash_out);
if (!map_ctx)
return;
rr->ssrc = htonl(map_ctx->ssrc_map_out);
Expand All @@ -1311,7 +1311,7 @@ static void transcode_rr(struct rtcp_process_ctx *ctx, struct report_block *rr)
return;

// for reception stats
struct ssrc_entry_call *input_ctx = get_ssrc(map_ctx->ssrc_map_out,
struct ssrc_entry_call *input_ctx = get_ssrc_no_move(map_ctx->ssrc_map_out,
&ctx->mp->media_out->ssrc_hash_in);
if (!input_ctx)
return;
Expand Down
32 changes: 18 additions & 14 deletions daemon/ssrc.c
Original file line number Diff line number Diff line change
Expand Up @@ -43,9 +43,12 @@ static struct ssrc_entry *create_ssrc_entry_call(void *uptr) {
ent->lost_bits = -1;
return &ent->h;
}
static void add_ssrc_entry(uint32_t ssrc, struct ssrc_entry *ent, struct ssrc_hash *ht) {
static void add_ssrc_entry(uint32_t ssrc, struct ssrc_entry *ent, struct ssrc_hash *ht, bool move_to_front) {
init_ssrc_entry(ent, ssrc);
g_queue_push_head_link(&ht->nq, &ent->link);
if (move_to_front)
g_queue_push_head_link(&ht->nq, &ent->link);
else
g_queue_push_tail_link(&ht->nq, &ent->link);
obj_hold(ent); // queue entry
}
static void free_sender_report(struct ssrc_sender_report_item *i) {
Expand Down Expand Up @@ -176,15 +179,16 @@ static void mos_calc_legacy(struct ssrc_stats_block *ssb) {
ssb->mos = mos_from_rx(r / 10); // e5
}

static void *find_ssrc(uint32_t ssrc, struct ssrc_hash *ht, unsigned int *iters) {
static void *find_ssrc(uint32_t ssrc, struct ssrc_hash *ht, unsigned int *iters, bool move_to_front) {
LOCK(&ht->lock);
for (GList *l = ht->nq.head; l; l = l->next) {
struct ssrc_entry *ret = l->data;
if (ret->ssrc == ssrc) {
obj_hold(ret);
// move to front
g_queue_unlink(&ht->nq, &ret->link);
g_queue_push_head_link(&ht->nq, &ret->link);
if (move_to_front) {
g_queue_unlink(&ht->nq, &ret->link);
g_queue_push_head_link(&ht->nq, &ret->link);
}
return ret;
}
}
Expand All @@ -194,15 +198,15 @@ static void *find_ssrc(uint32_t ssrc, struct ssrc_hash *ht, unsigned int *iters)
}

// returns a new reference
void *get_ssrc_full(uint32_t ssrc, struct ssrc_hash *ht, bool *created) {
void *get_ssrc_full(uint32_t ssrc, struct ssrc_hash *ht, bool *created, bool move_to_front) {
struct ssrc_entry *ent;

if (!ht->create_func)
return NULL;

while (true) {
unsigned int iters;
ent = find_ssrc(ssrc, ht, &iters);
ent = find_ssrc(ssrc, ht, &iters, move_to_front);
if (G_LIKELY(ent)) {
if (created)
*created = false;
Expand Down Expand Up @@ -244,7 +248,7 @@ void *get_ssrc_full(uint32_t ssrc, struct ssrc_hash *ht, bool *created) {
obj_put(ent);
continue;
}
add_ssrc_entry(ssrc, ent, ht);
add_ssrc_entry(ssrc, ent, ht, move_to_front);
if (created)
*created = true;
ht->iters++;
Expand Down Expand Up @@ -307,7 +311,7 @@ static void *__do_time_report_item(struct call_media *m, size_t struct_size, siz
sti->ntp_ts_lsw = ntp_lsw;
sti->ntp_ts_msw = ntp_msw;

e = get_ssrc(ssrc, &m->ssrc_hash_in);
e = get_ssrc_no_move(ssrc, &m->ssrc_hash_in);
if (G_UNLIKELY(!e)) {
free_func(sti);
return NULL;
Expand All @@ -331,7 +335,7 @@ static struct ssrc_entry_call *hunt_ssrc(struct call_media *media, uint32_t ssrc
return NULL;

IQUEUE_FOREACH(&media->media_subscriptions, ms) {
struct ssrc_entry_call *e = find_ssrc(ssrc, &ms->media->ssrc_hash_in, NULL);
struct ssrc_entry_call *e = find_ssrc(ssrc, &ms->media->ssrc_hash_in, NULL, false);
if (e)
return e;
}
Expand All @@ -350,7 +354,7 @@ static int64_t __calc_rtt(struct call_media *m, struct crtt_args a)
if (!a.ntp_middle_bits || !a.delay)
return 0;

struct ssrc_entry_call *e = find_ssrc(a.ssrc, a.ht, NULL);
struct ssrc_entry_call *e = find_ssrc(a.ssrc, a.ht, NULL, false);
if (G_UNLIKELY(!e))
return 0;

Expand Down Expand Up @@ -467,7 +471,7 @@ void ssrc_receiver_report(struct call_media *m, stream_fd *sfd, const struct ssr
.delay = rr->dlsr,
.reports_queue_offset = G_STRUCT_OFFSET(struct ssrc_entry_call, sender_reports));

struct ssrc_entry_call *other_e = get_ssrc(rr->from, &m->ssrc_hash_in);
struct ssrc_entry_call *other_e = get_ssrc_no_move(rr->from, &m->ssrc_hash_in);
if (G_UNLIKELY(!other_e))
goto out_nl;

Expand Down Expand Up @@ -614,7 +618,7 @@ void ssrc_voip_metrics(struct call_media *m, const struct ssrc_xr_voip_metrics *
vm->ext_r_factor, vm->mos_lq, vm->mos_cq, vm->rx_config, vm->jb_nom,
vm->jb_max, vm->jb_abs_max);

struct ssrc_entry_call *e = get_ssrc(vm->ssrc, &m->ssrc_hash_in);
struct ssrc_entry_call *e = get_ssrc_no_move(vm->ssrc, &m->ssrc_hash_in);
if (!e)
return;
e->last_rtt_xr = vm->rnd_trip_delay;
Expand Down
94 changes: 94 additions & 0 deletions fixtures/mixer-drift/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
# Conference admission latency regression

The regression recreates the missing control sequence from the October 4 conference
investigation: two AMR-WB phones, an Opus listener, and one shared mono L16/16000
recording endpoint. It measures decoded audio content age, not just RTP packet timing.

The original live call measured about 0.68 seconds of engine-local delay for one phone
and 2.58 seconds for the other. The old packet replay reached only 1.54 seconds and
omitted admission and private prompts. This test includes that control sequence using
synthetic audio; the original PCAP and recorded prompts are no longer available.
It is a reconstruction, not a byte-exact replay of the original call.

## Scenario

- Create the phone endpoints with permanent players, then answer them 330 ms and 630 ms
into the test while their RTP timestamp and sequence progression continues normally.
- Add them using DTT's fresh-tag-safe bidirectional join meshes. An established Opus
listener and recording anchor are present before the phone admissions.
- Change the L16 anchor from its discard destination to the recording socket, then
answer it again after the second phone answer.
- Play a 700-ms private join tone and staggered 7.5-second private welcome prompts with
`block egress`, followed by the second phone's join tone. These times approximate
the relative offsets recovered from the original NG-control log.
- Continue sending AMR-WB speech, SID updates and omitted no-data frames throughout.
- Send 20 seconds of SID-only silence between two 80-second fixture cycles. All input
RTP timestamps and sequence numbers remain continuous across that interval.
- Capture both phone outputs, the Opus output, and the shared L16 recording for the
complete call; collect a further four seconds for delayed output.

Private prompt tones are generated as temporary WAV files at 3000/3100 Hz, outside the
speech-marker frequency range. The test verifies welcome audio at its intended phone
and verifies that it does not leak into either shared output. It checks one stable
SSRC and continuous RTP sequence/timestamp progression after prompt playback ends.

FFmpeg independently decodes the actual AMR-WB and Opus output packets. The marker
oracle identifies each phone's tones independently in the mixed browser and recording
streams, checking six source/output paths in total. Markers before 12 seconds are
excluded because admission and private playback deliberately replace conversation.
Every later marker must arrive. Its age is measured from its actual input send time.

Each path must stay below 250 ms of absolute content age after private prompts, and
its later 20-second window medians may add less than 40 ms relative to its first
settled window (40–60 seconds). This is a fixed total allowance, not a per-window
allowance. A constant multi-second backlog therefore cannot pass the test.
The source pacing check must also remain within 20 ms; otherwise latency conclusions
are skipped and the separate pacing assertion fails.

## Running

From `t/`, after building the daemon and preload helper:

```sh
RTPENGINE_EXTENDED_TESTS=1 \
RTPENGINE_MIXER_DRIFT_ARTIFACTS=/tmp/conference-latency \
make daemon-tests-mixer-drift
```

The default replay lasts 184 seconds plus offline decoding/analysis.
Set `RTPENGINE_MIXER_DRIFT_DURATION=640` for a 10-minute-40-second source timeline
plus four seconds of output collection. This repeats the same fixture within one call,
with 20-second SID-only intervals at 80–100 and 420–440 seconds. Source admission
offsets of 330/630 ms also apply to these intervals. Every other latency and pacing
threshold stays the same, including the fixed 40-ms growth allowance for the whole call. Without the extended-test
flag it explicitly skips. It requires full AMR-WB and Opus support in the daemon and
AMR-WB/Opus reference decoders in FFmpeg. `RTPENGINE_TEST_FFMPEG` selects that executable.
Artifacts otherwise use an automatically removed temporary directory.
`RTPENGINE_CONFERENCE_LATENCY_ARTIFACTS` is an alternate artifacts-directory variable.

Artifacts include `sent.csv`, `results.json`, `daemon.log`, prompt WAVs, raw RTP
observations, and independently decoded PCM. Raw `.rtp` records contain a big-endian
IEEE-754 double (monotonic seconds since the send loop started), a big-endian 32-bit
packet length, then the RTP bytes. These are isolated Unix-datagram observations
through the test preload harness; no live service or physical phone is contacted.

## Synthetic source fixtures

`source-0.awb` and `source-1.awb` each contain 80 seconds of generated sine waves and
silence encoded with AMR-WB VAD/DTX. No human speech, call recording or music is included.
The generator and fixture data follow the repository's GPL-3.0 license.

Source A has 220-ms bursts every two seconds starting at 400 ms; source B has 160-ms
bursts every 2.6 seconds starting at 1.1 seconds. Source A uses 400–2350 Hz and source B
uses 3400–4900 Hz, in 50-Hz steps, so markers cannot be confused between sources in a
shared mix. SID updates are transmitted; FT=15 no-data frames advance
the source timeline without sending RTP. The first speech frame after silence carries
the talkspurt marker. Arrival jitter is deterministic and bounded to +/-15/20 ms.

`profiles.json` records hashes, marker frequencies and pacing parameters. Regenerate
with `python3 generate.py`, which requires FFmpeg with `libvo_amrwbenc`. The recorded
encoder versions are libavcodec 58.134.100 and libvo-amrwbenc 0.1.3. Other encoder
versions may produce different bytes; regenerate the manifest together with the streams.

The short `auto-daemon-tests-dtx-rtcp.pl` remains a separate, codec-light regression for
report-only SSRC handling, configured comfort noise, and actual RTP source changes.
30 changes: 30 additions & 0 deletions fixtures/mixer-drift/generate.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
from pathlib import Path
import math,struct,subprocess,json,collections,hashlib
p=Path(__file__).resolve().parent
profiles=[]
duration=80
for source in range(2):
period=(2.0,2.6)[source];first=(0.4,1.1)[source];speech=(0.22,0.16)[source]
bursts=[{'start':first+i*period,'end':first+i*period+speech,'frequency':(400 if source==0 else 3400)+i*50} for i in range(int((duration-first-speech)/period)+1)]
pcm=bytearray(duration*16000*2)
for burst in bursts:
for i in range(round(burst['start']*16000),round(burst['end']*16000)):
t=i/16000-burst['start'];envelope=min(1,t/0.01,(burst['end']-i/16000)/0.01)
value=round(11000*max(0,envelope)*math.sin(2*math.pi*burst['frequency']*i/16000))
struct.pack_into('<h',pcm,2*i,value)
raw=p/f'source-{source}.s16';raw.write_bytes(pcm)
awb=p/f'source-{source}.awb'
subprocess.run(['ffmpeg','-v','error','-y','-f','s16le','-ar','16000','-ac','1','-i',str(raw),'-c:a','libvo_amrwbenc','-b:a','12650','-dtx','1','-f','amr',str(awb)],check=True)
data=awb.read_bytes();assert data.startswith(b'#!AMR-WB\n');offset=9;counts=collections.Counter();frames=[];lengths=[17,23,32,36,40,46,50,58,60,5,0,0,0,0,0,0]
while offset<len(data):
toc=data[offset];ft=(toc>>3)&15;size=lengths[ft];body=data[offset:offset+1+size];counts[ft]+=1
if ft!=15: frames.append({'frame_index':sum(counts.values())-1,'data':body.hex()})
offset+=1+size
profiles.append({'source':source,'duration':duration,'bursts':bursts,'packets':frames,'frame_types':dict(counts)})
print(source,'frames',sum(counts.values()),'types',counts,'RTP packets',len(frames),'awb bytes',len(data))
for profile in profiles:
source=profile['source'];name=f'source-{source}.awb'
del profile['packets']
profile.update({'file':name,'sha256':hashlib.sha256((p/name).read_bytes()).hexdigest(),'jitter_amplitude':0.015+source*0.005,'jitter_step':0.37-source*0.11})
(p/f'source-{source}.s16').unlink()
(p/'profiles.json').write_text(json.dumps(profiles,indent=2)+'\n')
Loading
Loading