TT#99621 support RTCP generation

Change-Id: Iff832eaa4148cce4d87d24d4dc3b908dfa361770

README.md

@@ -811,6 +811,13 @@ Optionally included keys are:
 		injection via the `play DTMF` control message. See `play DTMF` below for additional
 		information.
 
+	- `generate RTCP`
+
+		With this flag set, received RTCP packets will not simply be passed through as
+		usual, but instead will be consumed, and instead *rtpengine* will generate its own
+		RTCP packets to send to the RTP peers. This is currently supported only for
+		transcoded streams, and the flag will be effective for both sides of a call.
+
 * `replace`
 
 	Similar to the `flags` list. Controls which parts of the SDP body should be rewritten.

daemon/call.c

@@ -77,6 +77,8 @@ static struct timeval add_ongoing_calls_dur_in_interval(struct timeval *interval
 		struct timeval *interval_duration);
 static void __call_free(void *p);
 static void __call_cleanup(struct call *c);
+static void __monologue_stop(struct call_monologue *ml);
+static void media_stop(struct call_media *m);
 
 /* called with call->master_lock held in R */
 static int call_timer_delete_monologues(struct call *c) {
@@ -2122,6 +2124,11 @@ int monologue_offer_answer(struct call_monologue *other_ml, GQueue *streams,
 				ice_restart(other_media->ice_agent);
 		}
 
+		if (flags && flags->generate_rtcp) {
+			MEDIA_SET(media, RTCP_GEN);
+			MEDIA_SET(other_media, RTCP_GEN);
+		}
+
 		__update_media_protocol(media, other_media, sp, flags);
 		__update_media_id(media, other_media, sp, flags);
 		__endpoint_loop_protect(sp, other_media);
@@ -2385,13 +2392,13 @@ static void __call_cleanup(struct call *c) {
 	for (GList *l = c->medias.head; l; l = l->next) {
 		struct call_media *md = l->data;
 		ice_shutdown(&md->ice_agent);
-		t38_gateway_stop(md->t38_gateway);
+		media_stop(md);
 		t38_gateway_put(&md->t38_gateway);
 	}
 
 	for (GList *l = c->monologues.head; l; l = l->next) {
 		struct call_monologue *ml = l->data;
-		media_player_stop(ml->player);
+		__monologue_stop(ml);
 		media_player_put(&ml->player);
 	}
 
@@ -3039,13 +3046,18 @@ struct call_monologue *call_get_mono_dialogue(struct call *call, const str *from
 
 
 
-static void monologue_stop(struct call_monologue *ml) {
+static void media_stop(struct call_media *m) {
+	t38_gateway_stop(m->t38_gateway);
+	codec_handlers_stop(&m->codec_handlers_store);
+	m->rtcp_timer.tv_sec = 0;
+}
+static void __monologue_stop(struct call_monologue *ml) {
 	media_player_stop(ml->player);
-	for (GList *l = ml->medias.head; l; l = l->next) {
-		struct call_media *m = l->data;
-		t38_gateway_stop(m->t38_gateway);
-		codec_handlers_stop(&m->codec_handlers_store);
-	}
+}
+static void monologue_stop(struct call_monologue *ml) {
+	__monologue_stop(ml);
+	for (GList *l = ml->medias.head; l; l = l->next)
+		media_stop(l->data);
 }
 
 
@@ -3112,6 +3124,8 @@ do_delete:
 		ng_call_stats(c, fromtag, totag, output, NULL);
 
 	monologue_stop(ml);
+	if (ml->active_dialogue && ml->active_dialogue->active_dialogue == ml)
+		monologue_stop(ml->active_dialogue);
 
 	if (delete_delay > 0) {
 		ilog(LOG_INFO, "Scheduling deletion of call branch '" STR_FORMAT_M "' "

daemon/call_interfaces.c

@@ -801,6 +801,9 @@ static void call_ng_flags_flags(struct sdp_ng_flags *out, str *s, void *dummy) {
 		case CSH_LOOKUP("full-rtcp-attribute"):
 			out->full_rtcp_attr = 1;
 			break;
+		case CSH_LOOKUP("generate-RTCP"):
+			out->generate_rtcp = 1;
+			break;
 		case CSH_LOOKUP("loop-protect"):
 			out->loop_protect = 1;
 			break;

daemon/codec.c

@@ -131,9 +131,19 @@ struct codec_tracker {
 	GHashTable *supp_codecs; // telephone-event etc => hash table of clock rates
 };
 
+struct rtcp_timer_queue {
+	struct timerthread_queue ttq;
+};
+struct rtcp_timer {
+	struct timerthread_queue_entry ttq_entry;
+	struct call *call;
+	struct call_media *media;
+};
+
 
 
 static struct timerthread codec_timers_thread;
+static struct rtcp_timer_queue *rtcp_timer_queue;
 
 
 static codec_handler_func handler_func_passthrough_ssrc;
@@ -983,6 +993,80 @@ static int codec_handler_non_rtp_update(struct call_media *receiver, struct call
 }
 
 
+static void __rtcp_timer_free(void *p) {
+	struct rtcp_timer *rt = p;
+	if (rt->call)
+		obj_put(rt->call);
+	g_slice_free1(sizeof(*rt), rt);
+}
+// master lock held in W
+static void __codec_rtcp_timer_schedule(struct call_media *media) {
+	struct rtcp_timer *rt = g_slice_alloc0(sizeof(*rt));
+	rt->ttq_entry.when = media->rtcp_timer;
+	rt->call = obj_get(media->call);
+	rt->media = media;
+
+	timerthread_queue_push(&rtcp_timer_queue->ttq, &rt->ttq_entry);
+}
+// no lock held
+static void __rtcp_timer_run(struct timerthread_queue *q, void *p) {
+	struct rtcp_timer *rt = p;
+
+	// check scheduling
+	rwlock_lock_w(&rt->call->master_lock);
+	struct call_media *media = rt->media;
+	struct timeval rtcp_timer = media->rtcp_timer;
+
+	log_info_call(rt->call);
+
+	if (!rtcp_timer.tv_sec || timeval_diff(&rtpe_now, &rtcp_timer) < 0 || !proto_is_rtp(media->protocol)) {
+		__rtcp_timer_free(rt);
+		rwlock_unlock_w(&rt->call->master_lock);
+		goto out;
+	}
+	timeval_add_usec(&rtcp_timer, 5000000 + (random() % 2000000));
+	media->rtcp_timer = rtcp_timer;
+	__codec_rtcp_timer_schedule(media);
+
+	// switch locks to be more graceful
+	rwlock_unlock_w(&rt->call->master_lock);
+
+	rwlock_lock_r(&rt->call->master_lock);
+
+	struct ssrc_ctx *ssrc_out = NULL;
+	if (media->streams.head) {
+		struct packet_stream *ps = media->streams.head->data;
+		mutex_lock(&ps->out_lock);
+		ssrc_out = ps->ssrc_out;
+		if (ssrc_out)
+			obj_hold(&ssrc_out->parent->h);
+		mutex_unlock(&ps->out_lock);
+	}
+
+	if (ssrc_out)
+		rtcp_send_report(media, ssrc_out);
+
+	rwlock_unlock_r(&rt->call->master_lock);
+
+	if (ssrc_out)
+		obj_put(&ssrc_out->parent->h);
+
+	__rtcp_timer_free(rt);
+
+out:
+	log_info_clear();
+}
+// master lock held in W
+static void __codec_rtcp_timer(struct call_media *receiver) {
+	if (receiver->rtcp_timer.tv_sec) // already scheduled
+		return;
+
+	receiver->rtcp_timer = rtpe_now;
+	timeval_add_usec(&receiver->rtcp_timer, 5000000 + (random() % 2000000));
+	__codec_rtcp_timer_schedule(receiver);
+	// XXX unify with media player into a generic RTCP player
+}
+
 
 // call must be locked in W
 void codec_handlers_update(struct call_media *receiver, struct call_media *sink,
@@ -1242,6 +1326,10 @@ next:
 
 		// we have to translate RTCP packets
 		receiver->rtcp_handler = rtcp_transcode_handler;
+		if (MEDIA_ISSET(receiver, RTCP_GEN)) {
+			receiver->rtcp_handler = rtcp_sink_handler;
+			__codec_rtcp_timer(receiver);
+		}
 
 		__check_dtmf_injector(flags, receiver, pref_dest_codec, output_transcoders, dtmf_payload_type);
 
@@ -3067,6 +3155,8 @@ void codec_rtp_payload_types(struct call_media *media, struct call_media *other_
 void codecs_init(void) {
 #ifdef WITH_TRANSCODING
 	timerthread_init(&codec_timers_thread, timerthread_queue_run);
+	rtcp_timer_queue = timerthread_queue_new("rtcp_timer_queue", sizeof(*rtcp_timer_queue),
+			&codec_timers_thread, NULL, __rtcp_timer_run, NULL, __rtcp_timer_free);
 #endif
 }
 void codecs_cleanup(void) {

daemon/media_player.c

@@ -173,7 +173,7 @@ static void send_timer_rtcp(struct send_timer *st, struct ssrc_ctx *ssrc_out) {
 
 	// XXX missing locking?
 	ssrc_out->next_rtcp = rtpe_now;
-	timeval_add_usec(&ssrc_out->next_rtcp, 5000000);
+	timeval_add_usec(&ssrc_out->next_rtcp, 5000000 + (random() % 2000000));
 }
 
 

daemon/rtcp.c

@@ -311,6 +311,9 @@ static void transcode_common_wrap(struct rtcp_process_ctx *, struct rtcp_packet
 static void transcode_rr_wrap(struct rtcp_process_ctx *, struct report_block *);
 static void transcode_sr_wrap(struct rtcp_process_ctx *, struct sender_report_packet *);
 
+// RTCP sinks for local RTCP generation
+static void sink_common(struct rtcp_process_ctx *, struct rtcp_packet *);
+
 // homer functions
 static void homer_init(struct rtcp_process_ctx *);
 static void homer_sr(struct rtcp_process_ctx *, struct sender_report_packet *);
@@ -362,6 +365,9 @@ static struct rtcp_handler transcode_handlers = {
 	.rr = transcode_rr,
 	.sr = transcode_sr,
 };
+static struct rtcp_handler sink_handlers = {
+	.common = sink_common,
+};
 static struct rtcp_handler transcode_handlers_wrap = {
 	.common = transcode_common_wrap,
 	.rr = transcode_rr_wrap,
@@ -476,6 +482,7 @@ static const int min_xr_packet_sizes[] = {
 
 
 struct rtcp_handler *rtcp_transcode_handler = &transcode_handlers;
+struct rtcp_handler *rtcp_sink_handler = &sink_handlers;
 
 
 
@@ -1532,3 +1539,9 @@ void rtcp_send_report(struct call_media *media, struct ssrc_ctx *ssrc_out) {
 	socket_sendto(&ps->selected_sfd->socket, sr->str, sr->len, &ps->endpoint);
 	g_string_free(sr, TRUE);
 }
+
+
+
+static void sink_common(struct rtcp_process_ctx *ctx, struct rtcp_packet *common) {
+	ctx->discard = 1;
+}

include/call.h

@@ -149,6 +149,7 @@ enum call_stream_state {
 #define MEDIA_FLAG_RTCP_FB			SHARED_FLAG_RTCP_FB
 #define MEDIA_FLAG_GENERATOR			0x02000000
 #define MEDIA_FLAG_ICE_LITE_SELF		0x04000000
+#define MEDIA_FLAG_RTCP_GEN			0x08000000
 
 /* access macros */
 #define SP_ISSET(p, f)		bf_isset(&(p)->sp_flags, SP_FLAG_ ## f)
@@ -331,6 +332,7 @@ struct call_media {
 	GQueue			codec_handlers_store; // storage for struct codec_handler
 	struct codec_handler	*codec_handler_cache;
 	struct rtcp_handler	*rtcp_handler;
+	struct timeval		rtcp_timer;	// master lock for scheduling purposes
 	struct codec_handler	*dtmf_injector;
 	struct t38_gateway	*t38_gateway;
 	struct codec_handler	*t38_handler;

include/call_interfaces.h

@@ -75,6 +75,7 @@ struct sdp_ng_flags {
 	    rtcp_mux_reject:1,
 	    no_rtcp_attr:1,
 	    full_rtcp_attr:1,
+	    generate_rtcp:1,
 	    generate_mid:1,
 	    strict_source:1,
 	    media_handover:1,

include/rtcp.h

@@ -22,6 +22,7 @@ struct rtcp_parse_ctx {
 
 
 extern struct rtcp_handler *rtcp_transcode_handler;
+extern struct rtcp_handler *rtcp_sink_handler;
 
 
 int rtcp_avp2savp(str *, struct crypto_context *, struct ssrc_ctx *);

utils/rtpengine-ng-client

@@ -79,13 +79,14 @@ GetOptions(
 	'inject-DTMF'			=> \$options{'inject DTMF'},
 	'DTLS-fingerprint=s'		=> \$options{'DTLS-fingerprint'},
 	'ICE-lite=s'			=> \$options{'ICE-lite'},
+	'generate-RTCP'			=> \$options{'generate RTCP'},
 ) or die;
 
 my $cmd = shift(@ARGV) or die;
 
 my %packet = (command => $cmd);
 
-for my $x (split(/,/, 'from-tag,to-tag,call-id,transport protocol,media address,ICE,address family,DTLS,via-branch,media address,ptime,xmlrpc-callback,metadata,address,file,db-id,code,DTLS-fingerprint,ICE-lite')) {
+for my $x (split(/,/, 'from-tag,to-tag,call-id,transport protocol,media address,ICE,address family,DTLS,via-branch,media address,ptime,xmlrpc-callback,metadata,address,file,db-id,code,DTLS-fingerprint,ICE-lite,generate RTCP')) {
 	defined($options{$x}) and $packet{$x} = \$options{$x};
 }
 for my $x (split(/,/, 'TOS,delete-delay')) {