sipwise
/
rtpengine
mirror of https://github.com/sipwise/rtpengine.git
1
0
Fork 0
Code Issues Releases Wiki Activity

MT#55283 iqueue for packet streams

Browse Source 
Change-Id: I0269d36be8d9578c1ec825fcf4a1ea0a29a00356
master
Richard Fuchs 4 days ago
parent 66e42183a8
commit c5afd60d28
23 changed files with 174 additions and 185 deletions
Show Stats Download Patch File Download Diff File

125
daemon/call.c
Unescape View File

@ -126,7 +126,6 @@ static void call_timer_iterator(call_t *c, struct iterator_helper *hlp) {
bool do_update = false;
bool has_srtp = false;
bool recv_checked = false;
struct packet_stream *ps;
int tmp_t_reason = UNKNOWN;
enum call_stream_state css;
int64_t timestamp;
@ -174,9 +173,7 @@ static void call_timer_iterator(call_t *c, struct iterator_helper *hlp) {
ice_fragments_cleanup(c->sdp_fragments, false);
for (__auto_type it = c->streams.head; it; it = it->next) {
ps = it->data;
IQUEUE_FOREACH(&c->streams, ps) {
timestamp = packet_stream_last_packet(ps);
if (!ps->media)
@ -899,9 +896,7 @@ static struct endpoint_map *__get_endpoint_map(struct call_media *media, unsigne
static void __assign_stream_fds(struct call_media *media, sfd_intf_list_q *intf_sfds) {
int reset_ice = 0;
for (__auto_type k = media->streams.head; k; k = k->next) {
struct packet_stream *ps = k->data;
IQUEUE_FOREACH(&media->streams, ps) {
// use opaque pointer to detect changes
void *old_selected_sfd = ps->selected_sfd;
@ -967,7 +962,7 @@ TYPED_GHASHTABLE_IMPL(rtp_stats_ht, g_direct_hash, g_direct_equal, NULL, __rtp_s
struct packet_stream *__packet_stream_new(call_t *call) {
struct packet_stream *stream;
stream = uid_alloc(&call->streams);
stream = iuid_alloc(&call->streams);
mutex_init(&stream->lock);
stream->call = call;
atomic64_set_na(&stream->last_packet_us, rtpe_now);
@ -996,12 +991,12 @@ static int __num_media_streams(struct call_media *media, unsigned int num_ports)
while (media->streams.length < num_ports) {
stream = __packet_stream_new(call);
stream->media = media;
t_queue_push_tail(&media->streams, stream);
i_queue_push_tail(&media->streams, stream);
stream->component = media->streams.length;
ret++;
}
t_queue_truncate(&media->streams, num_ports);
i_queue_truncate(&media->streams, num_ports);
return ret;
}
@ -1156,8 +1151,8 @@ enum call_stream_state call_stream_state_machine(struct packet_stream *ps) {
}
void call_media_state_machine(struct call_media *m) {
for (__auto_type l = m->streams.head; l; l = l->next)
call_stream_state_machine(l->data);
IQUEUE_FOREACH(&m->streams, ps)
call_stream_state_machine(ps);
}
bool __init_stream(struct packet_stream *ps) {
@ -1262,8 +1257,7 @@ void __add_sink_handler(sink_handler_q *q, struct packet_stream *sink, const str
// called once before calling __streams_set_sinks once for each sink
static void __reset_streams(struct call_media *media) {
for (__auto_type l = media->streams.head; l; l = l->next) {
struct packet_stream *ps = l->data;
IQUEUE_FOREACH(&media->streams, ps) {
t_queue_clear_full(&ps->rtp_sinks, free_sink_handler);
t_queue_clear_full(&ps->rtcp_sinks, free_sink_handler);
t_queue_clear_full(&ps->rtp_mirrors, free_sink_handler);
@ -1280,9 +1274,7 @@ static bool __init_streams(struct call_media *A, const struct stream_params *sp,
dbg_int("Stream set flags media %u", A->index);
for (__auto_type l = A->streams.head; l; l = l->next) {
__auto_type a = l->data;
IQUEUE_FOREACH(&A->streams, a) {
/* RTP */
PS_SET(a, RTP); /* XXX technically not correct, could be udptl too */
@ -1309,9 +1301,8 @@ static bool __init_streams(struct call_media *A, const struct stream_params *sp,
/* if muxing, this is the fallback RTCP port. it also contains the RTCP
* crypto context */
l = l->next;
assert(l != NULL);
a = l->data;
a = IQUEUE_NEXT(&A->streams, a);
assert(a != NULL);
PS_CLEAR(a, RTP);
PS_SET(a, RTCP);
@ -1353,19 +1344,17 @@ static bool __init_streams(struct call_media *A, const struct stream_params *sp,
*/
__attribute__((nonnull(1, 2, 4)))
static bool __streams_set_sinks(struct call_media *A, struct call_media *B,
const sdp_ng_flags *flags, const struct sink_attrs *attrs) {
__auto_type la = A->streams.head;
__auto_type lb = B->streams.head;
const sdp_ng_flags *flags, const struct sink_attrs *attrs)
{
__auto_type a_rtp = A->streams.head;
__auto_type b_rtp = B->streams.head;
dbg_int("Sink init media %u -> %u", A->index, B->index);
while (la) {
if (!lb)
while (a_rtp) {
if (!b_rtp)
break; // nothing left to do
__auto_type a_rtp = la->data;
__auto_type b_rtp = lb->data;
/* RTP */
// reflect media - pretend reflection also for blackhole, as otherwise
// we get SSRC flip-flops on the opposite side
@ -1390,15 +1379,13 @@ static bool __streams_set_sinks(struct call_media *A, struct call_media *B,
}
/* RTCP */
lb = lb->next;
assert(lb != NULL);
__auto_type b_rtcp = lb->data;
__auto_type b_rtcp = IQUEUE_NEXT(&B->streams, b_rtp);
assert(b_rtcp != NULL);
/* if muxing, this is the fallback RTCP port. it also contains the RTCP
* crypto context */
la = la->next;
assert(la != NULL);
__auto_type a_rtcp = la->data;
__auto_type a_rtcp = IQUEUE_NEXT(&A->streams, a_rtp);
assert(a_rtcp != NULL);
if (attrs->egress)
goto no_rtcp;
@ -1423,8 +1410,8 @@ static bool __streams_set_sinks(struct call_media *A, struct call_media *B,
}
no_rtcp:
la = la->next;
lb = lb->next;
a_rtp = IQUEUE_NEXT(&A->streams, a_rtcp);
b_rtp = IQUEUE_NEXT(&B->streams, b_rtcp);
}
return true;
@ -2010,14 +1997,11 @@ del_next:
static void __disable_streams(struct call_media *media, unsigned int num_ports) {
struct packet_stream *ps;
media->endpoint_map = NULL;
__num_media_streams(media, num_ports);
for (__auto_type l = media->streams.head; l; l = l->next) {
ps = l->data;
IQUEUE_FOREACH(&media->streams, ps) {
t_queue_clear(&ps->sfds);
ps->selected_sfd = NULL;
}
@ -2094,10 +2078,8 @@ static void __rtcp_mux_logic(sdp_ng_flags *flags, struct call_media *media,
}
static void __dtls_restart(struct call_media *m) {
struct packet_stream *ps;
for (__auto_type l = m->streams.head; l; l = l->next) {
ps = l->data;
IQUEUE_FOREACH(&m->streams, ps) {
PS_CLEAR(ps, FINGERPRINT_VERIFIED);
dtls_shutdown(ps);
__init_stream(ps);
@ -2652,8 +2634,8 @@ static void monologue_media_start(struct call_monologue *ml) {
if (!media)
continue;
for (__auto_type l = media->streams.head; l; l = l->next)
__init_stream(l->data);
IQUEUE_FOREACH(&media->streams, ps)
__init_stream(ps);
if (media->bundle && media->bundle != media && MEDIA_ISSET(media, BUNDLE_ONLY))
continue;
@ -3398,9 +3380,9 @@ static struct call_media *monologue_add_zero_media(struct call_monologue *sender
struct packet_stream *get_media_component(struct call_media *media, unsigned int component) {
// XXX maybe turn into array?
for (__auto_type l = media->streams.head; l; l = l->next) {
if (l->data->component == component)
return l->data;
IQUEUE_FOREACH(&media->streams, ps) {
if (ps->component == component)
return ps;
}
return NULL;
}
@ -3548,13 +3530,10 @@ static void monologue_bundle_set_fds(struct call_monologue *ml) {
if (media->ice_agent)
ice_shutdown(&media->ice_agent);
__auto_type msl = media->streams.head;
__auto_type bsl = bundle->streams.head;
while (msl) {
__auto_type ms = msl->data;
__auto_type bs = bsl->data;
__auto_type ms = media->streams.head;
__auto_type bs = bundle->streams.head;
while (ms) {
dtls_shutdown(ms);
// XXX close sockets that are not needed?
@ -3567,8 +3546,8 @@ static void monologue_bundle_set_fds(struct call_monologue *ml) {
ms->selected_sfd = bs->selected_sfd;
msl = msl->next;
bsl = bsl->next;
ms = IQUEUE_NEXT(&media->streams, ms);
bs = IQUEUE_NEXT(&bundle->streams, bs);
}
}
}
@ -3598,8 +3577,7 @@ static void monologue_bundle_set_sinks(struct call_monologue *ml) {
__auto_type media = ml->medias->pdata[i];
if (!media)
continue;
for (__auto_type l = media->streams.head; l; l = l->next) {
__auto_type ps = l->data;
IQUEUE_FOREACH(&media->streams, ps) {
monologue_bundle_set_sink_handlers(&ps->rtp_sinks);
monologue_bundle_set_sink_handlers(&ps->rtcp_sinks);
monologue_bundle_set_sink_handlers(&ps->rtp_mirrors);
@ -5115,7 +5093,7 @@ const rtp_payload_type *__rtp_stats_codec(struct call_media *m) {
if (!m->streams.head)
return NULL;
ps = m->streams.head->data;
ps = m->streams.head;
__auto_type iter = t_hash_table_iter(ps->rtp_stats);
struct rtp_stats *rs, *top = NULL;
@ -5167,9 +5145,7 @@ next:
}
static void __call_cleanup(call_t *c) {
for (__auto_type l = c->streams.head; l; l = l->next) {
struct packet_stream *ps = l->data;
IQUEUE_FOREACH(&c->streams, ps) {
send_timer_put(&ps->send_timer);
jb_put(&ps->jb);
__unkernelize(ps, "final call cleanup");
@ -5228,7 +5204,6 @@ static bool __remove_call_id_from_hash(str *callid, call_t *c) {
/* called lock-free, but must hold a reference to the call */
void call_destroy(call_t *c) {
struct packet_stream *ps=0;
struct call_monologue *ml;
struct call_media *md;
GList *k;
@ -5328,9 +5303,7 @@ void call_destroy(call_t *c) {
STR_FMT(&md->format_str));
}
for (__auto_type o = md->streams.head; o; o = o->next) {
ps = o->data;
IQUEUE_FOREACH(&md->streams, ps) {
// stats output only - no cleanups
if (PS_ISSET(ps, FALLBACK_RTCP))
@ -5458,7 +5431,6 @@ void media_subscription_free(struct media_subscription *p) {
void call_media_free(struct call_media *md) {
crypto_params_sdes_queue_clear(&md->sdes_in);
crypto_params_sdes_queue_clear(&md->sdes_out);
t_queue_clear(&md->streams);
t_queue_clear(&md->endpoint_maps);
codec_store_cleanup(&md->codecs);
codec_store_cleanup(&md->offered_codecs);
@ -5539,7 +5511,7 @@ static void __call_free(call_t *c) {
t_queue_clear(&c->callid_aliases);
while (c->streams.head) {
ps = t_queue_pop_head(&c->streams);
ps = i_queue_pop_head(&c->streams);
crypto_cleanup(&ps->crypto);
t_queue_clear(&ps->sfds);
t_hash_table_destroy(ps->rtp_stats);
@ -5795,12 +5767,12 @@ static bool call_merge(call_t *call, call_t *call2) {
t_hash_table_foreach_remove(call2->sdp_fragments, fragment_move, call);
last_id = call->streams.head->data->unique_id;
last_id = call->streams.head->unique_id;
while (call2->streams.head) {
__auto_type stream = t_queue_pop_head(&call2->streams);
__auto_type stream = i_queue_pop_head(&call2->streams);
stream->unique_id = ++last_id;
stream->call = call;
t_queue_push_tail(&call->streams, stream);
i_queue_push_tail(&call->streams, stream);
}
last_id = call->stream_fds.head->data->unique_id;
@ -6010,8 +5982,7 @@ void __media_unconfirm(struct call_media *media, const char *reason) {
if (!media)
return;
for (__auto_type m = media->streams.head; m; m = m->next) {
struct packet_stream *stream = m->data;
IQUEUE_FOREACH(&media->streams, stream) {
__stream_unconfirm(stream, reason);
__unconfirm_sinks(&stream->rtp_sinks, reason);
__unconfirm_sinks(&stream->rtcp_sinks, reason);
@ -6031,8 +6002,7 @@ static void __unkernelize_sinks(sink_handler_q *q, const char *reason) {
void call_media_unkernelize(struct call_media *media, const char *reason) {
if (!media)
return;
for (__auto_type m = media->streams.head; m; m = m->next) {
struct packet_stream *stream = m->data;
IQUEUE_FOREACH(&media->streams, stream) {
unkernelize(stream, reason);
__unkernelize_sinks(&stream->rtp_sinks, reason);
__unkernelize_sinks(&stream->rtcp_sinks, reason);
@ -6070,8 +6040,7 @@ void monologue_destroy(struct call_monologue *monologue) {
struct call_media *m = monologue->medias->pdata[i];
if (!m)
continue;
for (__auto_type k = m->streams.head; k; k = k->next) {
struct packet_stream *ps = k->data;
IQUEUE_FOREACH(&m->streams, ps) {
if (ps->selected_sfd && ps->selected_sfd->socket.local.port)
ps->last_local_endpoint = ps->selected_sfd->socket.local;
ps->selected_sfd = NULL;
@ -6717,7 +6686,7 @@ bool monologue_transform(struct call_monologue *ml, sdp_ng_flags *flags, medias_
// subscribe to itself
add_media_subscription(m, m, NULL);
__auto_type ps = m->streams.head->data;
__auto_type ps = m->streams.head;
ps->advertised_endpoint = ps->endpoint = media->destination;
__add_sink_handler(&ps->rtp_sinks, ps, NULL);

11
daemon/call_interfaces.c
Unescape View File

@ -78,7 +78,7 @@ static str streams_print(medias_arr *s, int start, int end, const char *prefix,
ilog(LOG_WARNING, "Media has no streams");
break;
}
ps = media->streams.head->data;
ps = media->streams.head;
if (format == SAF_TCP)
call_stream_address(o, ps, format, NULL, true);
@ -928,7 +928,6 @@ static void ng_stats_media(ng_command_ctx_t *ctx, parser_arg list, const struct
struct call_stats *totals, parser_arg ssrc)
{
parser_arg dict, streams = {0}, flags;
struct packet_stream *ps;
const rtp_payload_type *rtp_pt = NULL;
if (!ctx)
@ -985,10 +984,8 @@ static void ng_stats_media(ng_command_ctx_t *ctx, parser_arg list, const struct
ng_stats_ssrc(parser, NULL, parser->dict_add_list(dict, "egress SSRCs"), &m->ssrc_hash_out);
stats:
for (auto_iter(l, m->streams.head); l; l = l->next) {
ps = l->data;
IQUEUE_FOREACH(&m->streams, ps)
ng_stats_stream(ctx, streams, ps, totals);
}
}
static void ng_stats_monologue(ng_command_ctx_t *ctx, parser_arg dict, const struct call_monologue *ml,
@ -1419,7 +1416,7 @@ static const char *media_match(call_t *call, struct call_monologue **monologue,
continue;
if (!media->streams.head)
continue;
struct packet_stream *ps = media->streams.head->data;
struct packet_stream *ps = media->streams.head;
if (!sockaddr_eq(&addr, &ps->advertised_endpoint.address))
continue;
ilog(LOG_DEBUG, "Matched address %s%s%s to tag '" STR_FORMAT_M "'",
@ -2590,7 +2587,7 @@ const char *call_transform_ng(ng_command_ctx_t *ctx) {
parser->dict_add_str_dup(dict, "id", &m->media_id);
if (!m->streams.head)
continue;
__auto_type ps = m->streams.head->data;
__auto_type ps = m->streams.head;
if (!ps->selected_sfd)
continue;
__auto_type sfd = ps->selected_sfd;

5
daemon/cdr.c
Unescape View File

@ -39,7 +39,6 @@ void cdr_update_entry(call_t * c) {
g_autoptr(GString) cdr = g_string_new("");
struct call_media *md;
const rtp_payload_type *rtp_pt;
struct packet_stream *ps=0;
if (IS_FOREIGN_CALL(c))
return;
@ -107,9 +106,7 @@ void cdr_update_entry(call_t * c) {
g_string_append_printf(cdr, "payload_type=unknown, ");
}
for (__auto_type o = md->streams.head; o; o = o->next) {
ps = o->data;
IQUEUE_FOREACH(&md->streams, ps) {
if (PS_ISSET(ps, FALLBACK_RTCP))
continue;

5
daemon/cli.c
Unescape View File

@ -724,7 +724,6 @@ static void cli_list_call_info(struct cli_writer *cw, call_t *c) {
static void cli_list_tag_info(struct cli_writer *cw, struct call_monologue *ml) {
struct call_media *md;
struct packet_stream *ps;
int64_t tim_result_duration;
int64_t now;
char *local_addr;
@ -795,9 +794,7 @@ static void cli_list_tag_info(struct cli_writer *cw, struct call_monologue *ml)
else
cw->cw_printf(cw, STR_FORMAT "\n", STR_FMT(&rtp_pt->encoding_with_params));
for (__auto_type o = md->streams.head; o; o = o->next) {
ps = o->data;
IQUEUE_FOREACH(&md->streams, ps) {
if (PS_ISSET(ps, FALLBACK_RTCP))
continue;

4
daemon/codec.c
Unescape View File

@ -676,7 +676,7 @@ static const char *__make_transform_handler(struct codec_handler *handler) {
g_string_append(req, "e6:outputd5:codec");
append_pt(req, &handler->dest_pt);
__auto_type ps = tfh->transform_media->streams.head->data;
__auto_type ps = tfh->transform_media->streams.head;
__auto_type sock = &ps->selected_sfd->socket;
g_string_append(req, "eee11:destinationd");
@ -744,7 +744,7 @@ static const char *__make_transform_handler(struct codec_handler *handler) {
return "'transform' response media contained invalid 'address'";
tfh->remote.port = port;
ps = tfh->transform_media->streams.head->data;
ps = tfh->transform_media->streams.head;
ps->advertised_endpoint = ps->endpoint = tfh->remote;
PS_SET(ps, FILLED);

4
daemon/dtmf.c
Unescape View File

@ -605,7 +605,7 @@ void dtmf_dsp_event(const struct dtmf_event *new_event, struct dtmf_event *cur_e
return;
// we don't have a real fsin so just use the stream address
struct packet_stream *ps = media->streams.head->data;
struct packet_stream *ps = media->streams.head;
LOCK(&media->dtmf_lock);
@ -801,7 +801,7 @@ const char *dtmf_inject(struct call_media *media, int code, int volume, int dura
if (!media->streams.head)
return "Media doesn't have an RTP stream";
struct packet_stream *ps = media->streams.head->data;
struct packet_stream *ps = media->streams.head;
struct ssrc_entry_call *ssrc_in = call_get_first_ssrc(&media->ssrc_hash_in);
if (!ssrc_in)
return "No SSRC context present for DTMF injection"; // XXX fall back to generating stream

24
daemon/ice.c
Unescape View File

@ -481,7 +481,7 @@ void ice_update(struct ice_agent *ag, struct stream_params *sp, bool allow_reset
struct call_media *media;
call_t *call;
unsigned int comps;
struct packet_stream *components[MAX_COMPONENTS], *ps;
struct packet_stream *components[MAX_COMPONENTS];
candidate_q *candidates;
if (!ag)
@ -517,8 +517,11 @@ void ice_update(struct ice_agent *ag, struct stream_params *sp, bool allow_reset
/* get our component streams */
ZERO(components);
comps = 0;
for (__auto_type l = media->streams.head; l && comps < MAX_COMPONENTS; l = l->next)
components[comps++] = l->data;
IQUEUE_FOREACH(&media->streams, ps) {
if (comps >= MAX_COMPONENTS)
break;
components[comps++] = ps;
}
if (comps == 2 && (MEDIA_ISSET(media, RTCP_MUX) || !proto_is_rtp(media->protocol)))
components[1] = NULL;
@ -534,7 +537,7 @@ void ice_update(struct ice_agent *ag, struct stream_params *sp, bool allow_reset
/* skip invalid */
if (!cand->component_id || cand->component_id > G_N_ELEMENTS(components))
continue;
ps = components[cand->component_id - 1];
__auto_type ps = components[cand->component_id - 1];
if (ps) /* only count active components */
comps = MAX(comps, cand->component_id);
@ -608,8 +611,11 @@ static void __ice_pairings(struct ice_agent *ag) {
struct packet_stream *components[MAX_COMPONENTS] = {0};
unsigned int comps = 0;
for (__auto_type l = media->streams.head; l && comps < MAX_COMPONENTS; l = l->next)
components[comps++] = l->data;
IQUEUE_FOREACH(&media->streams, ps) {
if (comps >= MAX_COMPONENTS)
break;
components[comps++] = ps;
}
if (comps == 2 && (MEDIA_ISSET(media, RTCP_MUX) || !proto_is_rtp(media->protocol)))
components[1] = NULL;
@ -1215,7 +1221,6 @@ found:
static int __check_valid(struct ice_agent *ag) {
struct call_media *media;
struct packet_stream *ps;
packet_stream_list *l;
candidate_pair_list *k;
candidate_pair_q all_compos;
struct ice_candidate_pair *pair;
@ -1251,8 +1256,9 @@ static int __check_valid(struct ice_agent *ag) {
AGENT_SET(ag, USABLE);
}
for (l = media->streams.head, k = all_compos.head; l && k; l = l->next, k = k->next) {
ps = l->data;
for (ps = media->streams.head, k = all_compos.head; ps && k;
ps = IQUEUE_NEXT(&media->streams, ps), k = k->next)
{
pair = k->data;
LOCK(&ps->lock);

9
daemon/media_player.c
Unescape View File

@ -1243,7 +1243,7 @@ static bool media_player_read_packet(struct media_player *mp) {
void media_player_set_sink(struct media_player *mp) {
struct call_media *media = mp->media;
if (media->streams.head) {
mp->sink.sink = media->streams.head->data;
mp->sink.sink = media->streams.head;
sink_handler_set_generic(&mp->sink);
}
if (!mp->ssrc_out || mp->ssrc_out->h.ssrc != mp->ssrc) {
@ -1658,11 +1658,7 @@ static void call_ml_moh_handle_flags(struct call_monologue *from_ml, struct call
/* check zero-connection */
if (ML_ISSET(moh_ml, MOH_ZEROCONN)) {
struct packet_stream *ps;
__auto_type msl = audio->streams.head;
while (msl)
{
ps = msl->data;
IQUEUE_FOREACH(&audio->streams, ps) {
if (PS_ISSET(ps, RTP)) { /* find RTP stream, and don't touch RTCP */
ilog(LOG_DEBUG, "Forced packet stream of '"STR_FORMAT"' (media index: '%d')"
"to zero_addr due to MoH zero-connection.",
@ -1670,7 +1666,6 @@ static void call_ml_moh_handle_flags(struct call_monologue *from_ml, struct call
PS_SET(ps, ZERO_ADDR);
goto check_next; /* stop */
}
msl = msl->next;
}
}
check_next:

6
daemon/media_socket.c
Unescape View File

@ -2818,7 +2818,7 @@ static void media_packet_reset_media(struct packet_handler_ctx *phc, struct call
// reset media, stream, sinks, in_srtp
phc->mp.media = media;
phc->mp.stream = media->streams.head->data;
phc->mp.stream = media->streams.head;
phc->in_srtp = phc->mp.stream;
phc->sinks = &phc->mp.stream->rtp_sinks;
}
@ -3815,10 +3815,8 @@ out:
if (phc->unkernelize_subscriptions) {
IQUEUE_FOREACH(&phc->mp.media->media_subscriptions, ms) {
__auto_type sub_media = ms->media;
for (__auto_type m = sub_media->streams.head; m; m = m->next) {
struct packet_stream *sub_ps = m->data;
IQUEUE_FOREACH(&sub_media->streams, sub_ps)
unkernelize(sub_ps, "subscriptions modified");
}
}
}

2
daemon/mqtt.c
Unescape View File

@ -390,7 +390,7 @@ static void mqtt_media_stats(struct call_media *media, JsonBuilder *json) {
json_builder_add_string_value(json, "inactive");
}
struct packet_stream *ps = media->streams.head ? media->streams.head->data : NULL;
struct packet_stream *ps = media->streams.head ?: NULL;
mutex_lock(&media->ssrc_hash_in.lock);

11
daemon/recording.c
Unescape View File

@ -333,11 +333,9 @@ static void update_flags_proc(call_t *call, bool streams) {
update_metadata_call(call, NULL);
if (!streams)
return;
for (__auto_type l = call->streams.head; l; l = l->next) {
struct packet_stream *ps = l->data;
IQUEUE_FOREACH(&call->streams, ps)
append_meta_chunk_null(call->recording, "STREAM %u FORWARDING %u",
ps->unique_id, ML_ISSET(ps->media->monologue, REC_FORWARDING) ? 1 : 0);
}
}
static void recording_update_flags(call_t *call, bool streams) {
_rm(update_flags, call, streams);
@ -399,8 +397,7 @@ void recording_start_daemon(call_t *call) {
struct call_media *m = l->data;
recording_setup_media(m);
}
for (__auto_type l = call->streams.head; l; l = l->next) {
struct packet_stream *ps = l->data;
IQUEUE_FOREACH(&call->streams, ps) {
recording_setup_stream(ps);
__unkernelize(ps, "recording start");
__reset_sink_handlers(ps);
@ -1015,10 +1012,8 @@ static void finish_proc(call_t *call, bool discard) {
recording->proc.call_idx = UNINIT_IDX;
for (__auto_type l = call->streams.head; l; l = l->next) {
struct packet_stream *ps = l->data;
IQUEUE_FOREACH(&call->streams, ps)
ps->recording.proc.stream_idx = UNINIT_IDX;
}
if (!recording->proc.meta_filepath)
return;

28
daemon/redis.c
Unescape View File

@ -39,7 +39,7 @@ typedef union {
stream_fd_q *sfds_q;
medias_arr *ma;
sfd_intf_list_q *siq;
packet_stream_q *psq;
streams_in_media_q *psq;
endpoint_map_q *emq;
} callback_arg_t __attribute__ ((__transparent_union__));
@ -1327,6 +1327,13 @@ static int rbl_cb_simple(str *s, GQueue *q, struct redis_list *list, void *ptr)
return 0;
}
static int rbl_cb_ps_list(str *s, streams_in_media_q *q, struct redis_list *list, void *ptr) {
int j;
j = str_to_i(s, 0);
i_queue_push_tail(q, redis_list_get_idx_ptr(list, (unsigned) j));
return 0;
}
static int rbpa_cb_simple(str *s, medias_arr *pa, struct redis_list *list, void *ptr) {
int j;
j = str_to_i(s, 0);
@ -1340,6 +1347,12 @@ static int json_build_list(callback_arg_t q, call_t *c, const char *key,
return json_build_list_cb(q, c, key, idx, list, rbl_cb_simple, NULL, arg);
}
static int json_build_ps_list(callback_arg_t q, call_t *c, const char *key,
unsigned int idx, struct redis_list *list, parser_arg arg)
{
return json_build_list_cb(q, c, key, idx, list, rbl_cb_ps_list, NULL, arg);
}
static int json_build_ptra(medias_arr *q, call_t *c, const char *key,
unsigned int idx, struct redis_list *list, parser_arg arg)
{
@ -2030,7 +2043,7 @@ static int json_link_medias(call_t *c, struct redis_list *medias,
struct call_media *med = medias->ptrs[i];
if (!med || !med->monologue)
continue;
if (json_build_list(&med->streams, c, "streams", i, streams, arg))
if (json_build_ps_list(&med->streams, c, "streams", i, streams, arg))
return -1;
if (json_build_list(&med->endpoint_maps, c, "maps", i, maps, arg))
return -1;
@ -2759,9 +2772,7 @@ static str redis_encode_json(ng_parser_ctx_t *ctx, call_t *c, void **to_free,
} // --- for
for (__auto_type l = c->streams.head; l; l = l->next) {
struct packet_stream *ps = l->data;
IQUEUE_FOREACH(&c->streams, ps) {
if (!ps->media || !ml_in_scope(scope, ps->media->monologue))
continue;
@ -3024,10 +3035,8 @@ static str redis_encode_json(ng_parser_ctx_t *ctx, call_t *c, void **to_free,
if (!scope) {
snprintf(tmp, sizeof(tmp), "streams-%u", media->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (__auto_type m = media->streams.head; m; m = m->next) {
struct packet_stream *ps = m->data;
IQUEUE_FOREACH(&media->streams, ps)
JSON_ADD_LIST_STRING("%u", ps->unique_id);
}
}
if (!scope) {
@ -3439,8 +3448,7 @@ static void snapshot_apply_streams(call_t *c, struct call_monologue *ml, parser_
struct call_media *m = ml->medias->pdata[j];
if (!m)
continue;
for (__auto_type l = m->streams.head; l; l = l->next) {
struct packet_stream *ps = l->data;
IQUEUE_FOREACH(&m->streams, ps) {
struct redis_hash rh;
if (json_get_hash(&rh, "stream", ps->unique_id, root))
continue;

4
daemon/rtcp.c
Unescape View File

@ -1627,9 +1627,9 @@ void rtcp_send_report(struct call_media *media, struct ssrc_entry_call *ssrc_out
const struct packet_stream *locked)
{
// figure out where to send it
struct packet_stream *ps = media->streams.head->data;
struct packet_stream *ps = media->streams.head;
// crypto context is held separately
struct packet_stream *rtcp_ps = media->streams.head->next ? media->streams.head->next->data : ps;
struct packet_stream *rtcp_ps = IQUEUE_NEXT(&media->streams, ps) ?: ps;
if (MEDIA_ISSET(media, RTCP_MUX))
;

34
daemon/sdp.c
Unescape View File

@ -2832,19 +2832,13 @@ static void append_int_tagged_str_attr_to_gstring(GString *s, const str *name, u
__attr_end(&state);
}
static struct packet_stream *print_rtcp(GString *s, struct call_media *media, packet_stream_list *rtp_ps_link,
static struct packet_stream *print_rtcp(GString *s, struct call_media *media, struct packet_stream *ps,
const sdp_ng_flags *flags)
{
struct packet_stream *ps = rtp_ps_link->data;
struct packet_stream *ps_rtcp = NULL;
if (ps->rtcp_sibling) {
if (ps->rtcp_sibling)
ps_rtcp = ps->rtcp_sibling;
__auto_type rtcp_ps_link = rtp_ps_link->next;
if (!rtcp_ps_link)
return NULL;
assert(rtcp_ps_link->data == ps_rtcp);
}
if (proto_is_rtp(media->protocol)) {
if (MEDIA_ISSET(media, RTCP_MUX) &&
@ -2886,12 +2880,12 @@ static void sdp_out_print_information(GString *out, const str *s) {
}
/* TODO: rework an appending of parameters in terms of sdp attribute manipulations */
__attribute__((nonnull(1, 2, 3, 6, 7, 8)))
__attribute__((nonnull(1, 2, 3, 6, 7)))
static void print_sdp_media_section(GString *s, struct call_media *media,
const endpoint_t *address, struct call_media *copy_media,
struct call_media *source_media,
struct packet_stream *rtp_ps,
packet_stream_list *rtp_ps_link, sdp_ng_flags *flags)
sdp_ng_flags *flags)
{
struct packet_stream *ps_rtcp = NULL;
bool inactive_media = (!address->port || !rtp_ps->selected_sfd); /* audio is accepted? */
@ -2955,7 +2949,7 @@ static void print_sdp_media_section(GString *s, struct call_media *media,
}
}
ps_rtcp = print_rtcp(s, media, rtp_ps_link, flags);
ps_rtcp = print_rtcp(s, media, rtp_ps, flags);
if (proto_is_rtp(media->protocol)) {
insert_crypto(s, media, flags);
@ -3342,11 +3336,11 @@ static bool sdp_out_add_media(GString *out, struct call_media *media,
return true;
}
__attribute__((nonnull(1, 2, 4, 6, 7, 8)))
__attribute__((nonnull(1, 2, 4, 6, 7)))
static void sdp_out_handle_osrtp1(GString *out, struct call_media *media,
struct call_media *source_media,
const endpoint_t *address, const struct transport_protocol *prtp,
struct packet_stream *rtp_ps, packet_stream_list *rtp_ps_link,
struct packet_stream *rtp_ps,
sdp_ng_flags *flags)
{
if (!prtp)
@ -3361,7 +3355,7 @@ static void sdp_out_handle_osrtp1(GString *out, struct call_media *media,
sdp_out_add_osrtp_media(out, media, prtp, address);
/* print media level attributes */
print_sdp_media_section(out, media, address, NULL, source_media, rtp_ps, rtp_ps_link, flags);
print_sdp_media_section(out, media, address, NULL, source_media, rtp_ps, flags);
media->protocol = proto;
}
@ -3423,7 +3417,7 @@ static struct call_media *sdp_out_set_source_media_address(struct call_media *me
/* cases with message, force relay and pass through */
if (media->type_id == MT_MESSAGE || flags->ice_option == ICE_FORCE_RELAY || MEDIA_ISSET(media, PASSTHRU)) {
if (source_media->streams.head) {
__auto_type sub_ps = source_media->streams.head->data;
__auto_type sub_ps = source_media->streams.head;
*sdp_address = sub_ps->advertised_endpoint;
}
return source_media;
@ -3476,7 +3470,7 @@ bool sdp_create(str *out, struct call_monologue *monologue, sdp_ng_flags *flags)
continue;
if (!media->streams.head)
continue;
first_ps = media->streams.head->data;
first_ps = media->streams.head;
if (!first_ps->selected_sfd)
continue;
break;
@ -3534,9 +3528,7 @@ bool sdp_create(str *out, struct call_monologue *monologue, sdp_ng_flags *flags)
if (!media->streams.length)
goto err;
__auto_type rtp_ps_link = media->streams.head;
struct packet_stream *rtp_ps = rtp_ps_link->data;
__auto_type rtp_ps = media->streams.head;
__auto_type media_ms = call_media_get_top_ms(media);
__auto_type source_media = media_ms ? media_ms->media : NULL;
@ -3555,7 +3547,7 @@ bool sdp_create(str *out, struct call_monologue *monologue, sdp_ng_flags *flags)
prtp = &transport_protocols[media->protocol->rtp_proto];
/* handle first OSRTP part */
sdp_out_handle_osrtp1(s, media, source_media, &sdp_address, prtp, rtp_ps, rtp_ps_link, flags);
sdp_out_handle_osrtp1(s, media, source_media, &sdp_address, prtp, rtp_ps, flags);
/* set: media type, port, protocol (e.g. RTP/SAVP) */
err = "Unknown media protocol";
@ -3566,7 +3558,7 @@ bool sdp_create(str *out, struct call_monologue *monologue, sdp_ng_flags *flags)
/* print media level attributes */
print_sdp_media_section(s, media, &sdp_address, copy_media, source_media,
rtp_ps, rtp_ps_link, flags);
rtp_ps, flags);
/* handle second OSRTP part */
sdp_out_handle_osrtp2(s, media, prtp);

4
daemon/ssrc.c
Unescape View File

@ -718,8 +718,8 @@ void ssrc_collect_metrics(struct call_media *media) {
}
if (media->streams.head) {
LOCK(&media->streams.head->data->lock);
RTPE_SAMPLE_SFD(jitter_measured, s->jitter, media->streams.head->data->selected_sfd);
LOCK(&media->streams.head->lock);
RTPE_SAMPLE_SFD(jitter_measured, s->jitter, media->streams.head->selected_sfd);
}
}
}

6
daemon/statistics.c
Unescape View File

@ -138,9 +138,9 @@ void statistics_update_oneway(call_t * c) {
if (!md)
continue;
for (__auto_type o = md->streams.head; o; o = o->next) {
ps = o->data;
if (PS_ISSET(ps, RTP)) {
IQUEUE_FOREACH(&md->streams, xps) {
if (PS_ISSET(xps, RTP)) {
ps = xps;
// --- only RTP is interesting
goto found;
}

10
daemon/t38.c
Unescape View File

@ -213,7 +213,7 @@ static int t38_gateway_handler(t38_core_state_t *stat, void *user_data, const ui
// send our packet if we can
struct packet_stream *ps = NULL;
if (tg->t38_media && tg->t38_media->streams.head)
ps = tg->t38_media->streams.head->data;
ps = tg->t38_media->streams.head;
stream_fd *sfd = NULL;
if (ps) {
@ -261,8 +261,8 @@ static bool t38_pcm_player(struct media_player *mp) {
return true;
if (tg->pcm_media && tg->pcm_media->streams.head
&& ((struct packet_stream *) tg->pcm_media->streams.head->data)->selected_sfd)
log_info_stream_fd(((struct packet_stream *) tg->pcm_media->streams.head->data)->selected_sfd);
&& tg->pcm_media->streams.head->selected_sfd)
log_info_stream_fd(tg->pcm_media->streams.head->selected_sfd);
ilog(LOG_DEBUG, "Generating T.38 PCM samples");
@ -494,10 +494,10 @@ void t38_gateway_start(struct t38_gateway *tg, str_case_value_ht codec_set) {
return;
struct packet_stream *ps;
ps = tg->pcm_media->streams.head->data;
ps = tg->pcm_media->streams.head;
if (!PS_ISSET(ps, FILLED))
return;
ps = tg->t38_media->streams.head->data;
ps = tg->t38_media->streams.head;
if (!PS_ISSET(ps, FILLED))
return;

12
include/call.h
Unescape View File

@ -421,6 +421,8 @@ struct packet_stream {
* Preempted by call->master_lock held in W.
*/
mutex_t lock;
IQUEUE_LINK media_link;
IQUEUE_LINK call_link;
struct call_media *media; /* RO */
call_t *call; /* RO */
@ -466,6 +468,10 @@ struct packet_stream {
atomic64 ps_flags;
};
typedef IQUEUE(struct packet_stream, media_link) streams_in_media_q;
typedef IQUEUE(struct packet_stream, call_link) streams_in_call_q;
INLINE int64_t packet_stream_last_packet(const struct packet_stream *ps) {
int64_t lp1 = (int64_t)atomic64_get_na(&ps->last_packet_us);
int64_t lp2 = (int64_t)atomic64_get_na(&ps->stats_in->last_packet_us);
@ -544,7 +550,7 @@ struct call_media {
candidate_q ice_candidates; /* slice-alloc'd, as received */
unsigned int media_rec_slot;
packet_stream_q streams; /* normally RTP + RTCP */
streams_in_media_q streams; /* normally RTP + RTCP */
struct endpoint_map *endpoint_map;
endpoint_map_q endpoint_maps;
struct ssrc_hash ssrc_hash_in;
@ -798,7 +804,7 @@ struct call {
str_ml_ht sdps;
endpoint_ml_ht endpoints;
fragments_ht sdp_fragments;
packet_stream_q streams;
streams_in_call_q streams;
stream_fd_q stream_fds; /* stream_fd */
endpoint_map_q endpoint_maps;
struct dtls_cert *dtls_cert; /* for outgoing */
@ -974,7 +980,6 @@ void call_media_unkernelize(struct call_media *media, const char *reason);
void __monologue_unconfirm(struct call_monologue *monologue, const char *);
void __media_unconfirm(struct call_media *media, const char *);
__attribute__((nonnull(1)))
/* one monologue's state from before an offer, held as a call record snapshot */
struct call_checkpoint {
bool pending;
@ -986,6 +991,7 @@ void call_checkpoint_answer(call_t *, struct call_monologue *, struct call_monol
int call_checkpoint_rollback(call_t *, struct call_monologue *, struct call_monologue *);
void call_checkpoint_free_all(call_t *);
__attribute__((nonnull(1)))
void update_init_monologue_subscribers(struct call_monologue *ml, enum ng_opmode opmode);
__attribute__((nonnull(1)))

1
include/rtp.h
Unescape View File

@ -11,6 +11,7 @@ struct rtp_header;
struct ssrc_hash;
struct ssrc_entry_call;
struct codec_store;
struct packet_stream;
typedef GString crypto_debug_string;

3
include/types.h
Unescape View File

@ -80,9 +80,6 @@ struct codec_pipeline_index;
TYPED_GHASHTABLE_PROTO(transcode_config_ht, struct codec_pipeline_index, struct transcode_config)
TYPED_GQUEUE(transcode_config, struct transcode_config)
struct packet_stream;
TYPED_GQUEUE(packet_stream, struct packet_stream)
struct sink_handler;
TYPED_GQUEUE(sink_handler, struct sink_handler)

7
lib/helpers.h
Unescape View File

@ -264,5 +264,12 @@ void thread_create_looper(enum thread_looper_action (*f)(void), const char *sche
__ret; \
})
#define iuid_alloc(q) ({ \
__typeof__((q)->head) __ret = memory_arena_alloc0(__typeof__(*(q)->head)); \
__ret->unique_id = (q)->length; \
i_queue_push_tail(q, __ret); \
__ret; \
})
#endif

34
lib/iqueue.h
Unescape View File

@ -62,6 +62,20 @@
})
#define i_queue_pop_tail(list) ({ \
__auto_type __ret = (list)->tail; \
if (__ret) { \
__auto_type __link = (__typeof((list)->offset)) __ret; \
(list)->tail = __link->link.prev; \
__link->link.prev = NULL; \
(list)->length--; \
if (!(list)->tail) \
(list)->head = NULL; \
} \
__ret; \
})
#define i_queue_push_tail(list, ele) do { \
if ((list)->tail) { \
__auto_type __link = (__typeof((list)->offset)) (list)->tail; \
@ -107,13 +121,25 @@
} while (0)
#define i_queue_truncate(list, num) do { \
while ((list)->length > (num)) \
i_queue_pop_tail(list); \
} while (0)
#define IQUEUE_NEXT(list, var) ({ \
__typeof ((list)->head) __ret = ((__typeof((list)->offset)) var)->link.next; \
__ret; \
})
#define IQUEUE_FOREACH(list, var) \
for (__typeof__ ( ({ __typeof__ (*(list)->head) __t; &__t; }) ) var = (list)->head; \
for (__typeof ( ({ __typeof (*(list)->head) __t; &__t; }) ) var = (list)->head; \
var; var = ((__typeof((list)->offset)) var)->link.next)
#define IQUEUE_FOREACH_SAFE(list, var) \
for (__typeof__ ( ({ __typeof__ (*(list)->head) __t; &__t; }) ) var = (list)->head, \
for (__typeof ( ({ __typeof (*(list)->head) __t; &__t; }) ) var = (list)->head, \
__next ## var = var ? ((__typeof((list)->offset)) var)->link.next : NULL; \
var; \
var = __next ## var, \
@ -128,8 +154,8 @@
#define i_queue_find(list, fn) ({ \
__typeof__ ((list)->head) __ret = NULL; \
bool (*__fn)(__typeof__ ((list)->const_ele)) = (fn); \
__typeof ((list)->head) __ret = NULL; \
bool (*__fn)(__typeof ((list)->const_ele)) = (fn); \
IQUEUE_FOREACH(list, __ele) { \
if (__fn(__ele)) { \
__ret = __ele; \

10
t/test-transcode.c
Unescape View File

@ -105,8 +105,8 @@ static void __start(const char *file, int line) {
ml_B = __monologue_create(&call, &call.callid);
media_A = call_media_new(&call); // originator
media_B = call_media_new(&call); // output destination
t_queue_push_tail(&media_A->streams, ps_new(media_A));
t_queue_push_tail(&media_B->streams, ps_new(media_B));
i_queue_push_tail(&media_A->streams, ps_new(media_A));
i_queue_push_tail(&media_B->streams, ps_new(media_B));
ml_A->tag = STR("tag_A");
ml_A->label = STR("label_A");
media_A->monologue = ml_A;
@ -272,9 +272,9 @@ static void __packet_seq_ts(const char *file, int line, struct call_media *media
.media_out = other_media,
.ssrc_in = get_ssrc(ssrc, &media->ssrc_hash_in),
.sfd = &sfd,
.sink = { .sink = other_media->streams.head->data },
.sink = { .sink = other_media->streams.head },
};
determine_sink_handler(media->streams.head->data, &mp.sink);
determine_sink_handler(media->streams.head, &mp.sink);
// from __stream_ssrc()
if (!MEDIA_ISSET(media, TRANSCODING))
mp.ssrc_in->ssrc_map_out = ntohl(ssrc);
@ -384,8 +384,6 @@ static void __packet_seq_ts(const char *file, int line, struct call_media *media
static void end(void) {
g_hash_table_destroy(rtp_ts_ht);
g_hash_table_destroy(rtp_seq_ht);
t_queue_clear(&media_A->streams);
t_queue_clear(&media_B->streams);
call_media_free(media_A);
call_media_free(media_B);
t_hash_table_destroy(call.tags);

Write Preview
Loading…
Cancel
Save