From d4a8203b4b008ea813f6f0ffb200131460bb35d4 Mon Sep 17 00:00:00 2001 From: Peter Kelly Date: Fri, 6 Sep 2024 16:55:41 +0100 Subject: [PATCH] MT#55283 Added decoder mix "slots" concept ... ... so that the desired wav channel can be controlled when producing a mixed audio file When a mixed wav file is created, the channels in the wav container are currently allocated in the same order as each SSRC is received, meaning it is impossible to know which channels have been allocated to the offer or answer side of the call. Furthermore if there is a reinvite or media file played, these are also allocated in the order that SSRC is received - so an "answer" could end up sharing a channel with an "offer" with no way of knowing this. This patch allows you to specify how many channel slots should be allocated within the mixer, and allows you to then specify which slot is assigned to each media in the call (this will usually be 2 slots in total, slot 1 for answer, slot 2 for offer or vice versa). Ported from https://github.com/sipwise/rtpengine/pull/1852 Closes #1857 Closes #1852 Change-Id: I010208427cabc3a48d6ef7bd3a84e9a5bdcfd492 --- daemon/call.c | 9 +++++++++ daemon/call_interfaces.c | 12 +++++++++++ daemon/recording.c | 23 +++++++++++++++++++-- include/call.h | 2 ++ include/call_interfaces.h | 3 +++ recording-daemon/decoder.c | 2 +- recording-daemon/metafile.c | 10 ++++++---- recording-daemon/mix.c | 40 ++++++++++++++++++++++++++++++++----- recording-daemon/mix.h | 3 ++- recording-daemon/stream.c | 6 +++++- recording-daemon/stream.h | 2 +- recording-daemon/types.h | 1 + 12 files changed, 98 insertions(+), 15 deletions(-) diff --git a/daemon/call.c b/daemon/call.c index da3841847..d6b6522da 100644 --- a/daemon/call.c +++ b/daemon/call.c @@ -2630,6 +2630,9 @@ static void __call_monologue_init_from_flags(struct call_monologue *ml, struct c call->last_signal = rtpe_now.tv_sec; call->deleted = 0; + call->media_rec_slots = (flags->media_rec_slots > 0 && call->media_rec_slots == 0) + ? flags->media_rec_slots + : call->media_rec_slots; // consume session attributes t_queue_clear_full(&ml->sdp_attributes, sdp_attr_free); @@ -2886,6 +2889,12 @@ static void __media_init_from_flags(struct call_media *other_media, struct call_ media->desired_family = sp->desired_family; } + if (flags->opmode == OP_OFFER) { + ilog(LOG_DEBUG, "setting other slot to %u, setting slot to %u", flags->media_rec_slot_offer, flags->media_rec_slot_answer); + other_media->media_rec_slot = flags->media_rec_slot_offer; + media->media_rec_slot = flags->media_rec_slot_answer; + } + /* bandwidth */ other_media->bandwidth_as = sp->media_session_as; other_media->bandwidth_rr = sp->media_session_rr; diff --git a/daemon/call_interfaces.c b/daemon/call_interfaces.c index 26f7b4858..9d690066f 100644 --- a/daemon/call_interfaces.c +++ b/daemon/call_interfaces.c @@ -1759,6 +1759,18 @@ void call_ng_main_flags(const ng_parser_t *parser, str *key, parser_arg value, h case CSH_LOOKUP("recording file"): out->recording_file = s; break; + case CSH_LOOKUP("recording-media-slot-offer"): + // This needs to be > 0 + //out->media_rec_slot_offer = bencode_get_integer_str(value, out->media_rec_slot_offer); + out->media_rec_slot_offer = parser->get_int_str(value, out->media_rec_slot_offer); + break; + case CSH_LOOKUP("recording-media-slot-answer"): + // This needs to be > 0 + out->media_rec_slot_answer = parser->get_int_str(value, out->media_rec_slot_answer); + break; + case CSH_LOOKUP("recording-media-slots"): + out->media_rec_slots = parser->get_int_str(value, out->media_rec_slots); + break; case CSH_LOOKUP("passthrough"): case CSH_LOOKUP("passthru"): switch (__csh_lookup(&s)) { diff --git a/daemon/recording.c b/daemon/recording.c index ae7391ae0..021e3293a 100644 --- a/daemon/recording.c +++ b/daemon/recording.c @@ -935,6 +935,8 @@ static void setup_stream_proc(struct packet_stream *stream) { struct recording *recording = call->recording; char buf[128]; int len; + unsigned int media_rec_slot; + unsigned int media_rec_slots; if (!recording) return; @@ -945,9 +947,26 @@ static void setup_stream_proc(struct packet_stream *stream) { if (ML_ISSET(ml, NO_RECORDING)) return; - len = snprintf(buf, sizeof(buf), "TAG %u MEDIA %u TAG-MEDIA %u COMPONENT %u FLAGS %" PRIu64 " MEDIA-SDP-ID %i", + ilog(LOG_INFO, "media_rec_slot=%u, media_rec_slots=%u, stream=%u", media->media_rec_slot, call->media_rec_slots, stream->unique_id); + + // If no slots have been specified or someone has tried to use slott 0 then we set the variables up so that the mix + // channels will be used in sequence as each SSRC is seen. (see mix.c for the algorithm) + if(call->media_rec_slots < 1 || media->media_rec_slot < 1) { + media_rec_slot = 1; + media_rec_slots = 1; + } else { + media_rec_slot = media->media_rec_slot; + media_rec_slots = call->media_rec_slots; + } + + if(media_rec_slot > media_rec_slots) { + ilog(LOG_ERR, "slot %i is greater than the total number of slots available %i, setting to slot %i", media->media_rec_slot, call->media_rec_slots, media_rec_slots); + media_rec_slot = media_rec_slots; + } + + len = snprintf(buf, sizeof(buf), "TAG %u MEDIA %u TAG-MEDIA %u COMPONENT %u FLAGS %" PRIu64 " MEDIA-SDP-ID %i MEDIA-REC-SLOT %i MEDIA-REC-SLOTS %i", ml->unique_id, media->unique_id, media->index, stream->component, - atomic64_get_na(&stream->ps_flags), media->media_sdp_id); + atomic64_get_na(&stream->ps_flags), media->media_sdp_id, media_rec_slot, media_rec_slots); append_meta_chunk(recording, buf, len, "STREAM %u details", stream->unique_id); len = snprintf(buf, sizeof(buf), "tag-%u-media-%u-component-%u-%s-id-%u", diff --git a/include/call.h b/include/call.h index bcd885862..1d10c0aa6 100644 --- a/include/call.h +++ b/include/call.h @@ -480,6 +480,7 @@ struct call_media { struct dtls_fingerprint fingerprint; /* as received */ const struct dtls_hash_func *fp_hash_func; /* outgoing */ str tls_id; + unsigned int media_rec_slot; packet_stream_q streams; /* normally RTP + RTCP */ endpoint_map_q endpoint_maps; @@ -751,6 +752,7 @@ struct call { enum block_dtmf_mode block_dtmf; atomic64 call_flags; + unsigned int media_rec_slots; }; diff --git a/include/call_interfaces.h b/include/call_interfaces.h index c491622d9..1762a06c6 100644 --- a/include/call_interfaces.h +++ b/include/call_interfaces.h @@ -118,6 +118,9 @@ struct sdp_ng_flags { int trigger_end_digits; int trigger_end_ms; int dtmf_delay; + int media_rec_slot_offer; + int media_rec_slot_answer; + int media_rec_slots; int repeat_times; str file; str blob; diff --git a/recording-daemon/decoder.c b/recording-daemon/decoder.c index 0a0632c93..a4ba07d91 100644 --- a/recording-daemon/decoder.c +++ b/recording-daemon/decoder.c @@ -115,7 +115,7 @@ static int decoder_got_frame(decoder_t *dec, AVFrame *frame, void *sp, void *dp) if (metafile->mix_out) { dbg("adding packet from stream #%lu to mix output", stream->id); if (G_UNLIKELY(deco->mixer_idx == (unsigned int) -1)) - deco->mixer_idx = mix_get_index(metafile->mix, ssrc, stream->media_sdp_id); + deco->mixer_idx = mix_get_index(metafile->mix, ssrc, stream->media_sdp_id, stream->channel_slot); format_t actual_format; if (output_config(metafile->mix_out, &dec->dest_format, &actual_format)) goto no_mix_out; diff --git a/recording-daemon/metafile.c b/recording-daemon/metafile.c index b5cfca38d..a8652dd87 100644 --- a/recording-daemon/metafile.c +++ b/recording-daemon/metafile.c @@ -110,12 +110,14 @@ static void meta_stream_interface(metafile_t *mf, unsigned long snum, char *cont // mf is locked static void meta_stream_details(metafile_t *mf, unsigned long snum, char *content) { dbg("stream %lu details %s", snum, content); - unsigned int tag, media, tm, cmp, media_sdp_id; + unsigned int tag, media, tm, cmp, media_sdp_id, media_rec_slot, media_rec_slots; uint64_t flags; - if (sscanf_match(content, "TAG %u MEDIA %u TAG-MEDIA %u COMPONENT %u FLAGS %" PRIu64 " MEDIA-SDP-ID %i", - &tag, &media, &tm, &cmp, &flags, &media_sdp_id) != 6) + if (sscanf_match(content, "TAG %u MEDIA %u TAG-MEDIA %u COMPONENT %u FLAGS %" PRIu64 " MEDIA-SDP-ID %i MEDIA-REC-SLOT %i MEDIA-REC-SLOTS %i", + &tag, &media, &tm, &cmp, &flags, &media_sdp_id, &media_rec_slot, &media_rec_slots) != 8) return; - stream_details(mf, snum, tag, media_sdp_id); + + mix_set_channel_slots(mf->mix, media_rec_slots); + stream_details(mf, snum, tag, media_sdp_id, media_rec_slot-1); } diff --git a/recording-daemon/mix.c b/recording-daemon/mix.c index 02d2c0e59..c50f9364d 100644 --- a/recording-daemon/mix.c +++ b/recording-daemon/mix.c @@ -31,7 +31,8 @@ struct mix_s { CH_LAYOUT_T channel_layout[MIX_MAX_INPUTS]; AVFilterContext *amix_ctx; AVFilterContext *sink_ctx; - unsigned int next_idx; + unsigned int next_idx[MIX_MAX_INPUTS]; //slots can never exceed MIN_MAX_INPUTS by definition + unsigned int channel_slots; AVFrame *sink_frame; resample_t resample; @@ -74,6 +75,18 @@ void mix_destroy(mix_t *mix) { g_slice_free1(sizeof(*mix), mix); } +void mix_set_channel_slots(mix_t *mix, unsigned int channel_slots) { + if(!mix) + return; + + if(channel_slots > mix_num_inputs) { + ilog(LOG_ERR, "channel_slots specified %u is higher than the maximum available %u", channel_slots, mix_num_inputs); + } + //ensures that mix->channel_slots will always be within the range of 1 to mix_max_inputs + mix->channel_slots = channel_slots < 1 ? 1 : (channel_slots > mix_num_inputs ? mix_num_inputs : channel_slots); + ilog(LOG_DEBUG, "setting slots %i", mix->channel_slots); +} + static void mix_input_reset(mix_t *mix, unsigned int idx) { mix->pts_offs[idx] = (uint64_t) -1LL; @@ -83,13 +96,19 @@ static void mix_input_reset(mix_t *mix, unsigned int idx) { } -unsigned int mix_get_index(mix_t *mix, void *ptr, unsigned int media_sdp_id) { - unsigned int next = mix->next_idx++; +unsigned int mix_get_index(mix_t *mix, void *ptr, unsigned int media_sdp_id, unsigned int stream_channel_slot) { + unsigned int next; + if (mix_output_per_media) { next = media_sdp_id; if (next >= mix_num_inputs) { ilog(LOG_WARNING, "Error with mix_output_per_media sdp_label next %i is bigger than mix_num_inputs %i", next, mix_num_inputs ); } + } else { + ilog(LOG_DEBUG, "getting mix input index for slot %u. channel slots for this mix are %u", stream_channel_slot, mix->channel_slots); + next = mix->next_idx[stream_channel_slot]; + mix->next_idx[stream_channel_slot] += mix->channel_slots; + ilog(LOG_DEBUG, "mix input index chosen is #%u", next); } if (next < mix_num_inputs) { @@ -98,19 +117,23 @@ unsigned int mix_get_index(mix_t *mix, void *ptr, unsigned int media_sdp_id) { return next; } + ilog(LOG_DEBUG, "mix input index #%u too high, cycling to find one to re-use", next); + // too many inputs - find one to re-use struct timeval earliest = {0,}; next = 0; for (unsigned int i = 0; i < mix_num_inputs; i++) { - if (earliest.tv_sec == 0 || timeval_cmp(&earliest, &mix->last_use[i]) > 0) { + if ((earliest.tv_sec == 0 || timeval_cmp(&earliest, &mix->last_use[i]) > 0) && + i % mix->channel_slots == stream_channel_slot) { next = i; earliest = mix->last_use[i]; } } - ilog(LOG_DEBUG, "Re-using mix input index #%u", next); + ilog(LOG_DEBUG, "requested slot is %u, Re-using mix input index #%u", stream_channel_slot, next); mix_input_reset(mix, next); mix->input_ref[next] = ptr; + mix->next_idx[stream_channel_slot] = next; return next; } @@ -235,6 +258,13 @@ mix_t *mix_new(void) { for (unsigned int i = 0; i < mix_num_inputs; i++) mix->pts_offs[i] = (uint64_t) -1LL; + for (unsigned int i = 0; i < mix_num_inputs; i++) { + // initialise with the first mixer channel to use for each slot. This is set to mix_num_inputs+1 + // so that we can detect first use and also if the maximum use has been reached. + //mix->next_idx[i] = mix_num_inputs+1; + mix->next_idx[i] = i; + } + return mix; } diff --git a/recording-daemon/mix.h b/recording-daemon/mix.h index 35d7432a3..a162f36ee 100644 --- a/recording-daemon/mix.h +++ b/recording-daemon/mix.h @@ -8,8 +8,9 @@ mix_t *mix_new(void); void mix_destroy(mix_t *mix); +void mix_set_channel_slots(mix_t *mix, unsigned int); int mix_config(mix_t *, const format_t *format); int mix_add(mix_t *mix, AVFrame *frame, unsigned int idx, void *, output_t *output); -unsigned int mix_get_index(mix_t *, void *, unsigned int); +unsigned int mix_get_index(mix_t *, void *, unsigned int, unsigned int); #endif diff --git a/recording-daemon/stream.c b/recording-daemon/stream.c index 0aa611dc6..503edaf0a 100644 --- a/recording-daemon/stream.c +++ b/recording-daemon/stream.c @@ -138,10 +138,14 @@ void stream_open(metafile_t *mf, unsigned long id, char *name) { epoll_add(stream->fd, EPOLLIN, &stream->handler); } -void stream_details(metafile_t *mf, unsigned long id, unsigned int tag, unsigned int media_sdp_id) { +void stream_details(metafile_t *mf, unsigned long id, unsigned int tag, unsigned int media_sdp_id, unsigned int channel_slot) { stream_t *stream = stream_get(mf, id); stream->tag = tag; stream->media_sdp_id = media_sdp_id; + if(channel_slot > mix_num_inputs) { + ilog(LOG_ERR, "Channel slot %u is greater than the maximum number of inputs %u, setting to %u", channel_slot, mix_num_inputs, mix_num_inputs); + } + stream->channel_slot = channel_slot > mix_num_inputs ? mix_num_inputs : channel_slot; } void stream_forwarding_on(metafile_t *mf, unsigned long id, unsigned int on) { diff --git a/recording-daemon/stream.h b/recording-daemon/stream.h index e03f594f9..a1a3c9384 100644 --- a/recording-daemon/stream.h +++ b/recording-daemon/stream.h @@ -4,7 +4,7 @@ #include "types.h" void stream_open(metafile_t *mf, unsigned long id, char *name); -void stream_details(metafile_t *mf, unsigned long id, unsigned int tag, unsigned int media_sdp_id); +void stream_details(metafile_t *mf, unsigned long id, unsigned int tag, unsigned int media_sdp_id, unsigned int channel_slot); void stream_forwarding_on(metafile_t *mf, unsigned long id, unsigned int on); void stream_sdp_label(metafile_t *mf, unsigned long id, unsigned long *label); void stream_close(stream_t *stream); diff --git a/recording-daemon/types.h b/recording-daemon/types.h index 7e3985443..226ac6387 100644 --- a/recording-daemon/types.h +++ b/recording-daemon/types.h @@ -58,6 +58,7 @@ struct stream_s { unsigned int forwarding_on:1; double start_time; unsigned int media_sdp_id; + unsigned int channel_slot; }; typedef struct stream_s stream_t;