diff --git a/subprojects/gst-plugins-good/gst/rtpmanager/rtptwccstats.c b/subprojects/gst-plugins-good/gst/rtpmanager/rtptwccstats.c index 772dbda1c7..b78f39b56b 100644 --- a/subprojects/gst-plugins-good/gst/rtpmanager/rtptwccstats.c +++ b/subprojects/gst-plugins-good/gst/rtpmanager/rtptwccstats.c @@ -36,6 +36,20 @@ GST_DEBUG_CATEGORY_EXTERN (rtp_twcc_debug); #define GST_CAT_DEFAULT rtp_twcc_debug +/* Maximum number of distinct means of repair that can be credited with + * recovering a single packet. Two is enough for the current RTX + FEC setup; + * bump this if a third means of repair (e.g. a second FEC layer) is added. */ +#define TWCC_MAX_REPAIR 2 + +/* Identity of a single means of repair (e.g. RTX, a FEC layer, a second FEC + * layer). A means of repair is identified by the SSRC and payload type of its + * redundant stream. */ +typedef struct +{ + guint32 ssrc; + guint8 pt; +} TWCCRepairId; + typedef struct { GstClockTime local_ts; @@ -65,7 +79,17 @@ typedef struct GArray *protects_twcc_seqnums; gboolean stats_processed; + /* Directly-observed reception state only: UNKNOWN / LOST / RECEIVED. + RECOVERED is never stored here anymore -- it is derived from + (status == LOST && recovered_by_count > 0). */ TWCCPktState status; + + /* For data packets: the means of repair (TWCCRepairId) that were able to + recover this packet had it been lost. A packet recoverable by both RTX and + FEC lists both, which is how repair overlap is accounted for. Stored inline + (no allocation) to keep SentPacket cache-friendly on the sending path. */ + TWCCRepairId recovered_by[TWCC_MAX_REPAIR]; + guint8 recovered_by_count; } SentPacket; typedef struct @@ -73,6 +97,31 @@ typedef struct SentPacket *sentpkt; } StatsPktPtr; +/* Recovery accounting for a single means of repair within a window. */ +typedef struct +{ + guint32 ssrc; + guint8 pt; + guint recovered_count; + gdouble recovery_pct; + /* Raw bits this means of repair sent in the window (accumulated during the + pass), converted to @bitrate_sent once the window duration is known. */ + guint64 bits_sent; + guint bitrate_sent; +} TWCCRepairStats; + +/* Recovery accounting for the overlap of a pair of means of repair within a + window: lost packets recoverable by both repair a and repair b. */ +typedef struct +{ + guint32 ssrc_a; + guint8 pt_a; + guint32 ssrc_b; + guint8 pt_b; + guint count; + gdouble pct; +} TWCCOverlapStats; + typedef struct { GstVecDeque *pt_packets; @@ -90,6 +139,11 @@ typedef struct gint64 avg_delta_of_delta; gdouble delta_of_delta_growth; gdouble queueing_slope; + + /* Per-repair recovery (TWCCRepairStats) and pairwise repair overlap + (TWCCOverlapStats), recomputed every window. */ + GArray *recovery_by_repair; + GArray *recovery_overlap; } TWCCStatsCtx; struct _TWCCStatsManager @@ -139,7 +193,17 @@ static SentPacket *_find_sentpacket (TWCCStatsManager * statsman, #define SENT_PKT_UNLOCK(statsman) g_mutex_unlock (&(statsman)->sent_packets_lock) /******************************************************************************/ -typedef GArray *RedBlockKey; +/* A redundancy block is keyed by the set of protected data TWCC seqnums + together with the identity (SSRC + payload type) of the redundant stream + protecting them. Including the full repair identity keeps distinct means of + repair (e.g. two FEC layers, or RTX and FEC) protecting the same data set as + separate blocks instead of collapsing them into one. */ +typedef struct +{ + GArray *seqs; + TWCCRepairId redund; +} RedBlockKeyStruct; +typedef RedBlockKeyStruct *RedBlockKey; typedef struct { @@ -150,12 +214,16 @@ typedef struct GArray *fec_states; gsize num_redundant_packets; + + /* Identity of the means of repair this block represents. */ + TWCCRepairId redund; } RedBlock; static RedBlock *_redblock_new (GArray * seq, guint16 fec_seq, - guint16 idx_redundant_packets, guint16 num_redundant_packets); + guint16 idx_redundant_packets, guint16 num_redundant_packets, + TWCCRepairId redund); static void _redblock_free (RedBlock * block); -static RedBlockKey _redblock_key_new (GArray * seqs); +static RedBlockKey _redblock_key_new (GArray * seqs, TWCCRepairId redund); static void _redblock_key_free (RedBlockKey key); static guint redblock_2_key (GArray * seq); static guint _redund_hash (gconstpointer key); @@ -205,6 +273,61 @@ _pkt_status_s (TWCCPktState state) } } +/* A packet counts as recovered for statistics purposes if it was observed lost + but at least one means of repair was able to recover it. The ground truth + reception state (pkt->status) is kept separate from this derived notion so + that overlapping means of repair can be attributed independently. */ +static gboolean +_pkt_is_recovered (const SentPacket * pkt) +{ + return pkt->status == RTP_TWCC_FECBLOCK_PKT_LOST + && pkt->recovered_by_count > 0; +} + +/* Effective state used by windowed stats: RECOVERED is derived from the + recovered_by attribution rather than stored on the packet. */ +static TWCCPktState +_pkt_effective_state (const SentPacket * pkt) +{ + if (_pkt_is_recovered (pkt)) { + return RTP_TWCC_FECBLOCK_PKT_RECOVERED; + } + return pkt->status; +} + +/* A packet is a redundant (RTX/FEC) packet if it actually protects other + packets. This is the authoritative discriminator: the redundant_num meta + field is not reliable on its own (it can carry a non-zero value for plain + data packets), and a redundant packet whose protected-seqnum conversion + failed -- e.g. due to reordering -- is deliberately treated as a plain data + packet (see rtp_twcc_stats_sent_pkt). Keep this consistent with + _accumulate_recovery. */ +static gboolean +_pkt_is_redundant (const SentPacket * pkt) +{ + return pkt->protects_twcc_seqnums && pkt->protects_twcc_seqnums->len > 0; +} + +/* Credit means of repair @r with recovering data packet @pkt. The CALLER must + have already verified that @r actually repaired the packet (the RTX packet + was received, or the FEC block had enough received redundancy) -- this only + records the attribution. De-duplicated and bounded by TWCC_MAX_REPAIR. */ +static void +_pkt_add_recovered_by (SentPacket * pkt, TWCCRepairId r) +{ + for (guint8 i = 0; i < pkt->recovered_by_count; i++) { + if (pkt->recovered_by[i].ssrc == r.ssrc && pkt->recovered_by[i].pt == r.pt) { + return; + } + } + if (pkt->recovered_by_count >= TWCC_MAX_REPAIR) { + /* More means of repair recovered this packet than we track inline; the + pairwise overlap stays correct for the means we did record. */ + return; + } + pkt->recovered_by[pkt->recovered_by_count++] = r; +} + static StatsPktPtr * _sent_pkt_get (GstVecDeque * pkt_array, guint idx) { @@ -275,6 +398,7 @@ _sent_packet_init (SentPacket * packet, guint16 seqnum, RTPPacketInfo * pinfo, packet->protects_timestamps = protects_timestamps; packet->protects_twcc_seqnums = protect_twcc_seqnums_array; packet->stats_processed = FALSE; + packet->recovered_by_count = 0; } static void @@ -323,15 +447,14 @@ _rm_pkt_stats (TWCCStatsManager * statsman, SentPacket * pkt) if (g_hash_table_lookup_extended (seq_to_twcc, GUINT_TO_POINTER (pkt->orig_seqnum), NULL, &val) && GPOINTER_TO_UINT (val) == pkt->seqnum) { - g_hash_table_remove (seq_to_twcc, - GUINT_TO_POINTER (pkt->orig_seqnum)); + g_hash_table_remove (seq_to_twcc, GUINT_TO_POINTER (pkt->orig_seqnum)); } } GST_LOG_OBJECT (statsman->parent, - "Removing #%u from history, main ctx length: %" G_GSIZE_FORMAT ", pkt ts: %" - GST_TIME_FORMAT, pkt->seqnum, + "Removing #%u from history, main ctx length: %" G_GSIZE_FORMAT + ", pkt ts: %" GST_TIME_FORMAT, pkt->seqnum, gst_vec_deque_get_length (statsman->stats_ctx->pt_packets), GST_TIME_ARGS (_pkt_stats_ts (pkt))); @@ -450,7 +573,8 @@ _sent_pkt_keep_length (TWCCStatsManager * statsman, gsize max_len, max_history_duration)); if (new_packet) { gst_vec_deque_push_tail_struct (statsman->sent_packets, new_packet); - ret = (SentPacket *) gst_vec_deque_peek_tail_struct (statsman->sent_packets); + ret = + (SentPacket *) gst_vec_deque_peek_tail_struct (statsman->sent_packets); } SENT_PKT_UNLOCK (statsman); return ret; @@ -465,6 +589,10 @@ twcc_stats_ctx_new (void) MAX_STATS_PACKETS); ctx->last_pkt_fb = NULL; + ctx->recovery_by_repair = g_array_new (FALSE, FALSE, + sizeof (TWCCRepairStats)); + ctx->recovery_overlap = g_array_new (FALSE, FALSE, sizeof (TWCCOverlapStats)); + return ctx; } @@ -472,6 +600,8 @@ static void twcc_stats_ctx_free (TWCCStatsCtx * ctx) { gst_vec_deque_free (ctx->pt_packets); + g_array_free (ctx->recovery_by_repair, TRUE); + g_array_free (ctx->recovery_overlap, TRUE); g_free (ctx); } @@ -609,6 +739,80 @@ _linear_compute (LinearRegression * l, gdouble * slope, gdouble * intercept) } } +/* Find (or append) the per-repair recovery accumulator for (ssrc, pt). */ +static TWCCRepairStats * +_repair_stats_get (GArray * arr, guint32 ssrc, guint8 pt) +{ + for (guint i = 0; i < arr->len; i++) { + TWCCRepairStats *m = &g_array_index (arr, TWCCRepairStats, i); + if (m->ssrc == ssrc && m->pt == pt) { + return m; + } + } + { + TWCCRepairStats nm = { ssrc, pt, 0, 0.0, 0, 0 }; + g_array_append_val (arr, nm); + } + return &g_array_index (arr, TWCCRepairStats, arr->len - 1); +} + +/* Order two means of repair canonically so an unordered pair maps to one cell. */ +static gboolean +_repair_less (TWCCRepairId a, TWCCRepairId b) +{ + if (a.ssrc != b.ssrc) { + return a.ssrc < b.ssrc; + } + return a.pt < b.pt; +} + +/* Find (or append) the overlap accumulator for the unordered pair {a, b}. */ +static TWCCOverlapStats * +_overlap_stats_get (GArray * arr, TWCCRepairId a, TWCCRepairId b) +{ + if (!_repair_less (a, b)) { + TWCCRepairId tmp = a; + a = b; + b = tmp; + } + for (guint i = 0; i < arr->len; i++) { + TWCCOverlapStats *o = &g_array_index (arr, TWCCOverlapStats, i); + if (o->ssrc_a == a.ssrc && o->pt_a == a.pt + && o->ssrc_b == b.ssrc && o->pt_b == b.pt) { + return o; + } + } + { + TWCCOverlapStats no = { a.ssrc, a.pt, b.ssrc, b.pt, 0, 0.0 }; + g_array_append_val (arr, no); + } + return &g_array_index (arr, TWCCOverlapStats, arr->len - 1); +} + +/* Credit the means of repair that recovered @pkt into the per-repair and + pairwise-overlap accumulators. Only media (data) packets are accounted: a + means of repair recovers lost media, not the redundant packets it sends, so + recovered RTX/FEC packets must not inflate the repair stats. The attribution + itself was verified when it was recorded (see _redblock_reconsider). */ +static void +_accumulate_recovery (TWCCStatsCtx * ctx, const SentPacket * pkt) +{ + if (_pkt_is_redundant (pkt)) { + /* This packet protects others (i.e. it is itself an RTX/FEC packet), so it + is not recovered media -- don't let it inflate the repair stats. */ + return; + } + for (guint8 a = 0; a < pkt->recovered_by_count; a++) { + TWCCRepairId ra = pkt->recovered_by[a]; + _repair_stats_get (ctx->recovery_by_repair, ra.ssrc, + ra.pt)->recovered_count++; + for (guint8 b = a + 1; b < pkt->recovered_by_count; b++) { + TWCCRepairId rb = pkt->recovered_by[b]; + _overlap_stats_get (ctx->recovery_overlap, ra, rb)->count++; + } + } +} + static gboolean twcc_stats_ctx_calculate_windowed_stats (TWCCStatsCtx * ctx, GstClockTimeDiff start_time, GstClockTimeDiff end_time, gint pt) @@ -623,10 +827,10 @@ twcc_stats_ctx_calculate_windowed_stats (TWCCStatsCtx * ctx, guint packets_unknown = 0; guint i; - guint bits_sent = 0; - guint bits_recv = 0; - guint bits_sent_non_recovery = 0; - guint bits_recv_non_recovery = 0; + guint64 bits_sent = 0; + guint64 bits_recv = 0; + guint64 bits_sent_non_recovery = 0; + guint64 bits_recv_non_recovery = 0; GstClockTimeDiff delta_delta_sum = 0; guint delta_delta_count = 0; @@ -658,6 +862,9 @@ twcc_stats_ctx_calculate_windowed_stats (TWCCStatsCtx * ctx, ctx->recovery_pct = -1.0; ctx->queueing_slope = 0.; + g_array_set_size (ctx->recovery_by_repair, 0); + g_array_set_size (ctx->recovery_overlap, 0); + gboolean ret = _get_stats_packets_window (packets, start_time, end_time, &start_idx, &packets_overall); @@ -684,7 +891,7 @@ twcc_stats_ctx_calculate_windowed_stats (TWCCStatsCtx * ctx, GST_LOG ("STATS WINDOW: %u/%u: pkt #%u, pt: %u, size: %u, status: %s, " "local-ts: %" GST_TIME_FORMAT ", remote-ts %" GST_TIME_FORMAT, i + 1, packets_overall, pkt->seqnum, pkt->pt, pkt->size * 8, - _pkt_status_s (pkt->status), + _pkt_status_s (_pkt_effective_state (pkt)), GST_TIME_ARGS (_pkt_stats_ts (pkt)), GST_TIME_ARGS (pkt->remote_ts)); if (GST_CLOCK_TIME_IS_VALID (_pkt_stats_ts (pkt)) @@ -694,8 +901,13 @@ twcc_stats_ctx_calculate_windowed_stats (TWCCStatsCtx * ctx, first_local_pkt = pkt; } else { bits_sent += pkt->size * 8; - if (pkt->redundant_num <= 0) { + if (!_pkt_is_redundant (pkt)) { bits_sent_non_recovery += pkt->size * 8; + } else { + /* Redundant (RTX/FEC) packet: credit its bits to its own means of + repair, keyed by the redundant stream's SSRC and payload type. */ + _repair_stats_get (ctx->recovery_by_repair, pkt->ssrc, + pkt->pt)->bits_sent += pkt->size * 8; } } last_local_pkt = pkt; @@ -707,18 +919,19 @@ twcc_stats_ctx_calculate_windowed_stats (TWCCStatsCtx * ctx, first_remote_pkt = pkt; } else { bits_recv += pkt->size * 8; - if (pkt->redundant_num <= 0) { + if (!_pkt_is_redundant (pkt)) { bits_recv_non_recovery += pkt->size * 8; } } last_remote_pkt = pkt; packets_sent++; packets_recv++; - } else if (pkt->status == RTP_TWCC_FECBLOCK_PKT_RECOVERED) { + } else if (_pkt_is_recovered (pkt)) { GST_LOG ("Packet #%u is lost and recovered", pkt->seqnum); packets_sent++; packets_lost++; packets_recovered++; + _accumulate_recovery (ctx, pkt); } else if (pkt->status == RTP_TWCC_FECBLOCK_PKT_LOST) { GST_LOG ("Packet #%u is lost", pkt->seqnum); packets_sent++; @@ -774,14 +987,14 @@ twcc_stats_ctx_calculate_windowed_stats (TWCCStatsCtx * ctx, _pkt_stats_ts (last_local_pkt)); } if (first_remote_pkt && last_remote_pkt) { - if (GST_CLOCK_TIME_IS_VALID(first_remote_pkt->remote_ts) - && GST_CLOCK_TIME_IS_VALID(last_remote_pkt->remote_ts)){ + if (GST_CLOCK_TIME_IS_VALID (first_remote_pkt->remote_ts) + && GST_CLOCK_TIME_IS_VALID (last_remote_pkt->remote_ts)) { remote_duration = GST_CLOCK_DIFF (first_remote_pkt->remote_ts, last_remote_pkt->remote_ts); } else { - remote_duration = GST_CLOCK_DIFF(first_remote_pkt->local_ts, - last_remote_pkt->local_ts); + remote_duration = GST_CLOCK_DIFF (first_remote_pkt->local_ts, + last_remote_pkt->local_ts); } } @@ -792,6 +1005,20 @@ twcc_stats_ctx_calculate_windowed_stats (TWCCStatsCtx * ctx, if (packets_lost) { ctx->recovery_pct = (packets_recovered * 100) / (gfloat) packets_lost; ctx->recovery_pct = MIN (ctx->recovery_pct, 100); + + /* Per-repair and overlap percentages are expressed relative to the + total number of lost packets in the window. */ + for (i = 0; i < ctx->recovery_by_repair->len; i++) { + TWCCRepairStats *m = + &g_array_index (ctx->recovery_by_repair, TWCCRepairStats, i); + m->recovery_pct = + MIN ((m->recovered_count * 100) / (gdouble) packets_lost, 100.0); + } + for (i = 0; i < ctx->recovery_overlap->len; i++) { + TWCCOverlapStats *o = + &g_array_index (ctx->recovery_overlap, TWCCOverlapStats, i); + o->pct = MIN ((o->count * 100) / (gdouble) packets_lost, 100.0); + } } if (delta_delta_count) { @@ -813,18 +1040,30 @@ twcc_stats_ctx_calculate_windowed_stats (TWCCStatsCtx * ctx, ctx->bitrate_sent = gst_util_uint64_scale (bits_sent, GST_SECOND, local_duration); ctx->bitrate_sent_non_recovery = - gst_util_uint64_scale (bits_sent_non_recovery, GST_SECOND, local_duration); + gst_util_uint64_scale (bits_sent_non_recovery, GST_SECOND, + local_duration); + + /* Convert each means of repair's raw bits into a windowed bitrate using + the same local-clock duration as the aggregate sent bitrate. */ + for (i = 0; i < ctx->recovery_by_repair->len; i++) { + TWCCRepairStats *m = + &g_array_index (ctx->recovery_by_repair, TWCCRepairStats, i); + m->bitrate_sent = + gst_util_uint64_scale (m->bits_sent, GST_SECOND, local_duration); + } } if (remote_duration > 0) { ctx->bitrate_recv = gst_util_uint64_scale (bits_recv, GST_SECOND, remote_duration); ctx->bitrate_recv_non_recovery = - gst_util_uint64_scale (bits_recv_non_recovery, GST_SECOND, remote_duration); + gst_util_uint64_scale (bits_recv_non_recovery, GST_SECOND, + remote_duration); } _linear_compute (&dod_regression, &ctx->queueing_slope, NULL); - GST_INFO ("Got stats: bits_sent: %u, bits_recv: %u, packets_sent = %u, " + GST_INFO ("Got stats: bits_sent: %" G_GUINT64_FORMAT ", bits_recv: %" + G_GUINT64_FORMAT ", packets_sent = %u, " "packets_recv: %u, packetlost_pct = %lf, recovery_pct = %lf, " "recovered: %u, packets unknown: %u, dod-count: %u", bits_sent, bits_recv, packets_sent, packets_recv, ctx->packet_loss_pct, @@ -843,7 +1082,7 @@ twcc_stats_ctx_calculate_windowed_stats (TWCCStatsCtx * ctx, static GstStructure * twcc_stats_ctx_get_structure (TWCCStatsCtx * ctx) { - return gst_structure_new ("RTPTWCCStats", + GstStructure *s = gst_structure_new ("RTPTWCCStats", "packets-sent", G_TYPE_UINT, ctx->packets_sent, "packets-recv", G_TYPE_UINT, ctx->packets_recv, "bitrate-sent", G_TYPE_UINT, ctx->bitrate_sent, @@ -855,6 +1094,44 @@ twcc_stats_ctx_get_structure (TWCCStatsCtx * ctx) "avg-delta-of-delta", G_TYPE_INT64, ctx->avg_delta_of_delta, "delta-of-delta-growth", G_TYPE_DOUBLE, ctx->delta_of_delta_growth, "queueing-slope", G_TYPE_DOUBLE, ctx->queueing_slope, NULL); + + /* Per-repair recovery: one structure per means of repair (identified by the + redundant stream's SSRC, labelled with its payload type). */ + { + GValueArray *repairs = g_value_array_new (ctx->recovery_by_repair->len); + for (guint i = 0; i < ctx->recovery_by_repair->len; i++) { + TWCCRepairStats *m = + &g_array_index (ctx->recovery_by_repair, TWCCRepairStats, i); + GstStructure *ms = gst_structure_new ("RTPTWCCRepairStats", + "ssrc", G_TYPE_UINT, m->ssrc, + "pt", G_TYPE_UINT, (guint) m->pt, + "recovered-count", G_TYPE_UINT, m->recovered_count, + "recovery-pct", G_TYPE_DOUBLE, m->recovery_pct, + "bitrate-sent", G_TYPE_UINT, m->bitrate_sent, NULL); + _append_structure_to_value_array (repairs, ms); + } + _structure_take_value_array (s, "recovery-by-repair", repairs); + } + + /* Pairwise overlap: lost packets recoverable by both means of repair a and b. */ + { + GValueArray *overlaps = g_value_array_new (ctx->recovery_overlap->len); + for (guint i = 0; i < ctx->recovery_overlap->len; i++) { + TWCCOverlapStats *o = + &g_array_index (ctx->recovery_overlap, TWCCOverlapStats, i); + GstStructure *os = gst_structure_new ("RTPTWCCRecoveryOverlap", + "ssrc-a", G_TYPE_UINT, o->ssrc_a, + "pt-a", G_TYPE_UINT, (guint) o->pt_a, + "ssrc-b", G_TYPE_UINT, o->ssrc_b, + "pt-b", G_TYPE_UINT, (guint) o->pt_b, + "count", G_TYPE_UINT, o->count, + "pct", G_TYPE_DOUBLE, o->pct, NULL); + _append_structure_to_value_array (overlaps, os); + } + _structure_take_value_array (s, "recovery-overlap", overlaps); + } + + return s; } static gint @@ -914,10 +1191,12 @@ twcc_stats_ctx_add_packet (TWCCStatsManager * statsman, SentPacket * pkt) static RedBlock * _redblock_new (GArray * seq, guint16 fec_seq, - guint16 idx_redundant_packets, guint16 num_redundant_packets) + guint16 idx_redundant_packets, guint16 num_redundant_packets, + TWCCRepairId redund) { RedBlock *block = g_malloc0 (sizeof (RedBlock)); block->seqs = g_array_ref (seq); + block->redund = redund; block->states = g_array_new (FALSE, FALSE, sizeof (TWCCPktState)); g_array_set_size (block->states, seq->len); for (gsize i = 0; i < seq->len; i++) { @@ -957,15 +1236,19 @@ _redblock_free (RedBlock * block) } static RedBlockKey -_redblock_key_new (GArray * seqs) +_redblock_key_new (GArray * seqs, TWCCRepairId redund) { - return g_array_ref (seqs); + RedBlockKey key = g_new (RedBlockKeyStruct, 1); + key->seqs = g_array_ref (seqs); + key->redund = redund; + return key; } static void _redblock_key_free (RedBlockKey key) { - g_array_unref (key); + g_array_unref (key->seqs); + g_free (key); } static guint @@ -988,7 +1271,7 @@ static guint _redund_hash (gconstpointer key) { RedBlockKey bk = (RedBlockKey) key; - return redblock_2_key (bk); + return redblock_2_key (bk->seqs) ^ bk->redund.ssrc ^ bk->redund.pt; } static gboolean @@ -996,16 +1279,27 @@ _redund_equal (gconstpointer a, gconstpointer b) { RedBlockKey bk1 = (RedBlockKey) a; RedBlockKey bk2 = (RedBlockKey) b; - return bk1->len == bk2->len && - memcmp (bk1->data, bk2->data, bk1->len * sizeof (guint16)) + return bk1->redund.ssrc == bk2->redund.ssrc + && bk1->redund.pt == bk2->redund.pt && bk1->seqs->len == bk2->seqs->len + && memcmp (bk1->seqs->data, bk2->seqs->data, + bk1->seqs->len * sizeof (guint16)) == 0; } -/* Check if the block could be recovered: - * all packets have known states - * number of lost packets is less than redundant packets were originally sent +/* Check whether the block can recover its lost data packets: + * a data/fec packet is "missing" if not directly RECEIVED (LOST or UNKNOWN) + * the block can recover iff total missing <= number of redundant packets + (MDS / Reed-Solomon property); note total-missing <= total-fec is + algebraically equivalent to data-lost <= fec-received, i.e. it already + verifies that enough redundancy was actually received. - Returns the number of recoverd packets + Recovery is recorded by attributing this block's means of repair to each + recovered data packet's recovered_by list, WITHOUT mutating the packet's + ground-truth status. This keeps overlapping means of repair independent: a + packet recoverable by both RTX and FEC is attributed to both, since each block + evaluates the same ground truth on its own. + + Returns the number of newly-attributed recovered packets. */ static gsize _redblock_reconsider (TWCCStatsManager * statsman, RedBlock * block) @@ -1014,6 +1308,8 @@ _redblock_reconsider (TWCCStatsManager * statsman, RedBlock * block) gsize nrecovered = 0; gsize lost = 0; + const TWCCRepairId repair = block->redund; + gchar states_media[48]; gchar states_fec[16]; @@ -1034,13 +1330,19 @@ _redblock_reconsider (TWCCStatsManager * statsman, RedBlock * block) } SentPacket *redundant_pkt = _find_stats_sentpacket (statsman, g_array_index (block->fec_seqs, guint16, i)); - if (redundant_pkt && redundant_pkt->status == RTP_TWCC_FECBLOCK_PKT_RECEIVED) { + if (redundant_pkt + && redundant_pkt->status == RTP_TWCC_FECBLOCK_PKT_RECEIVED) { nrecovered++; break; } } - if (nrecovered == 1) { - media_pkt->status = RTP_TWCC_FECBLOCK_PKT_RECOVERED; + /* Only attribute a recovery to a packet we have confirmed lost. A media + packet still in UNKNOWN state may yet arrive, so we must not declare + it recovered (see test_twcc_stats_rtx_recover_not_lost_stuff). */ + if (nrecovered == 1 && media_pkt->status == RTP_TWCC_FECBLOCK_PKT_LOST) { + _pkt_add_recovered_by (media_pkt, repair); + } else { + nrecovered = 0; } } @@ -1060,14 +1362,10 @@ _redblock_reconsider (TWCCStatsManager * statsman, RedBlock * block) nreceived++; if (i < G_N_ELEMENTS (states_media)) states_media[i] = '+'; - } else if (pkt->status == RTP_TWCC_FECBLOCK_PKT_RECOVERED) { - nrecovered++; - if (i < G_N_ELEMENTS (states_media)) - states_media[i] = 'R'; - } else if (pkt->status == RTP_TWCC_FECBLOCK_PKT_LOST) { + } else { /* LOST (possibly already recovered by another block) */ lost++; if (i < G_N_ELEMENTS (states_media)) - states_media[i] = '-'; + states_media[i] = _pkt_is_recovered (pkt) ? 'R' : '-'; } } states_media[block->seqs->len] = '\0'; @@ -1089,21 +1387,17 @@ _redblock_reconsider (TWCCStatsManager * statsman, RedBlock * block) nreceived++; if (i < G_N_ELEMENTS (states_fec)) states_fec[i] = '+'; - } else if (pkt->status == RTP_TWCC_FECBLOCK_PKT_RECOVERED) { - nrecovered++; - if (i < G_N_ELEMENTS (states_fec)) - states_fec[i] = 'R'; - } else if (pkt->status == RTP_TWCC_FECBLOCK_PKT_LOST) { + } else { /* LOST */ lost++; if (i < G_N_ELEMENTS (states_fec)) - states_fec[i] = '-'; + states_fec[i] = _pkt_is_recovered (pkt) ? 'R' : '-'; } } states_fec[block->fec_seqs->len] = '\0'; - if ((lost + nreceived + nrecovered > block->seqs->len + block->fec_seqs->len)) { - GST_TRACE_OBJECT ("Media: %s; FEC: %s; recovered: %lu", states_media, - states_fec, nrecovered); + if ((lost + nreceived > block->seqs->len + block->fec_seqs->len)) { + GST_TRACE_OBJECT (statsman->parent, "Media: %s; FEC: %s", states_media, + states_fec); GST_ERROR ("The FEC block is partly recovered, abort: %lu lost, %lu/%lu received", lost, nreceived, block->seqs->len + block->fec_seqs->len); @@ -1111,7 +1405,8 @@ _redblock_reconsider (TWCCStatsManager * statsman, RedBlock * block) } if (lost > 0 && lost <= block->fec_seqs->len) { - /* We have enough packets to recover the block */ + /* We have enough packets to recover the block: attribute this means of + repair to every confirmed-lost data packet it covers. */ for (gsize i = 0; i < block->seqs->len; ++i) { const guint16 seqnum = g_array_index (block->seqs, guint16, i); SentPacket *pkt = _find_stats_sentpacket (statsman, seqnum); @@ -1119,8 +1414,10 @@ _redblock_reconsider (TWCCStatsManager * statsman, RedBlock * block) GST_WARNING_OBJECT (statsman->parent, "Packet #%u not found in stats", seqnum); } else if (pkt->status == RTP_TWCC_FECBLOCK_PKT_LOST) { - pkt->status = RTP_TWCC_FECBLOCK_PKT_RECOVERED; - nrecovered++; + if (!_pkt_is_recovered (pkt)) { + nrecovered++; + } + _pkt_add_recovered_by (pkt, repair); } } for (gsize i = 0; i < block->fec_seqs->len; ++i) { @@ -1133,8 +1430,10 @@ _redblock_reconsider (TWCCStatsManager * statsman, RedBlock * block) SentPacket *pkt = _find_stats_sentpacket (statsman, seqnum); if (pkt && pkt->status == RTP_TWCC_FECBLOCK_PKT_LOST) { - pkt->status = RTP_TWCC_FECBLOCK_PKT_RECOVERED; - nrecovered++; + if (!_pkt_is_recovered (pkt)) { + nrecovered++; + } + _pkt_add_recovered_by (pkt, repair); } } } @@ -1159,32 +1458,99 @@ _get_ctx_for_pt (TWCCStatsManager * statsman, guint pt) return ctx; } +static GPtrArray * +_seqnum_blocks_get (TWCCStatsManager * statsman, guint16 seqnum) +{ + return g_hash_table_lookup (statsman->seqnum_2_redblocks, + GUINT_TO_POINTER (seqnum)); +} + +/* Associate @block with @seqnum (a seqnum may participate in several blocks + when multiple means of repair protect it). De-duplicated. */ static void -_rm_redundancy_links_pkt (TWCCStatsManager * ctx, SentPacket * pkt) -{ - /* If this packet maps to a block in hash tables -- remove every links - leading to this block as well as this packet: as we will remove this packet - from the context, we will not be able to use this block anyways. */ - RedBlock *block = NULL; - if (pkt && g_hash_table_lookup_extended (ctx->seqnum_2_redblocks, - GUINT_TO_POINTER (pkt->seqnum), NULL, (gpointer *) & block)) { - RedBlockKey key = _redblock_key_new (block->seqs); +_seqnum_blocks_add (TWCCStatsManager * statsman, guint16 seqnum, + RedBlock * block) +{ + GPtrArray *blocks = _seqnum_blocks_get (statsman, seqnum); + if (!blocks) { + blocks = g_ptr_array_new (); + g_hash_table_insert (statsman->seqnum_2_redblocks, + GUINT_TO_POINTER (seqnum), blocks); + } + for (guint i = 0; i < blocks->len; i++) { + if (g_ptr_array_index (blocks, i) == block) { + return; + } + } + g_ptr_array_add (blocks, block); +} + +/* Remove @block from @seqnum's block list, dropping the hash entry entirely if + it becomes empty. Does not free the block itself. */ +static void +_seqnum_blocks_remove_block (TWCCStatsManager * statsman, guint16 seqnum, + RedBlock * block) +{ + GPtrArray *blocks = _seqnum_blocks_get (statsman, seqnum); + if (!blocks) { + return; + } + g_ptr_array_remove_fast (blocks, block); + if (blocks->len == 0) { + g_hash_table_remove (statsman->seqnum_2_redblocks, + GUINT_TO_POINTER (seqnum)); + } +} + +static void +_rm_redundancy_links_pkt (TWCCStatsManager * statsman, SentPacket * pkt) +{ + /* When this packet is evicted, every block it participates in becomes + unusable (one of its constituents is gone), so we destroy each such block + entirely: unlink it from all of its data and fec seqnums and drop it from + the redundancy hash tables. */ + GPtrArray *blocks; + GPtrArray *to_destroy; + + if (!pkt) { + return; + } + blocks = _seqnum_blocks_get (statsman, pkt->seqnum); + if (!blocks || blocks->len == 0) { + return; + } + + /* Destroying a block mutates the per-seqnum lists (including this one), so + iterate over a private copy of the block pointers. */ + to_destroy = g_ptr_array_new (); + for (guint i = 0; i < blocks->len; i++) { + g_ptr_array_add (to_destroy, g_ptr_array_index (blocks, i)); + } + + for (guint b = 0; b < to_destroy->len; b++) { + RedBlock *block = g_ptr_array_index (to_destroy, b); + for (gsize i = 0; i < block->seqs->len; i++) { - g_hash_table_remove (ctx->seqnum_2_redblocks, - GUINT_TO_POINTER (g_array_index (block->seqs, guint16, i))); + _seqnum_blocks_remove_block (statsman, + g_array_index (block->seqs, guint16, i), block); } for (gsize i = 0; i < block->fec_seqs->len; i++) { if (g_array_index (block->fec_states, TWCCPktState, i) == RTP_TWCC_FECBLOCK_PKT_UNKNOWN) { - /* This redundant packet hasn't been not processed yet */ + /* This redundant packet hasn't been processed yet */ continue; } - g_hash_table_remove (ctx->seqnum_2_redblocks, - GUINT_TO_POINTER (g_array_index (block->fec_seqs, guint16, i))); + _seqnum_blocks_remove_block (statsman, + g_array_index (block->fec_seqs, guint16, i), block); } - g_hash_table_remove (ctx->redund_2_redblocks, key); + + /* Frees the block via the redund_2_redblocks value destroy func. */ + RedBlockKey key = _redblock_key_new (block->seqs, block->redund); + g_hash_table_remove (statsman->redund_2_redblocks, key); _redblock_key_free (key); } + + g_ptr_array_free (to_destroy, TRUE); } static gint32 @@ -1208,7 +1574,8 @@ _lookup_seqnum (TWCCStatsManager * statsman, guint32 ssrc, guint16 seqnum) } static SentPacket * -_find_sentpacket (TWCCStatsManager * statsman, guint16 seqnum, const guint32 * timestamp) +_find_sentpacket (TWCCStatsManager * statsman, guint16 seqnum, + const guint32 * timestamp) { SentPacket *result = NULL; @@ -1229,7 +1596,7 @@ _find_sentpacket (TWCCStatsManager * statsman, guint16 seqnum, const guint32 * t SENT_PKT_UNLOCK (statsman); if (result && result->seqnum == seqnum - && (!timestamp || result->timestamp == *timestamp)) { + && (!timestamp || result->timestamp == *timestamp)) { return result; } else { return NULL; @@ -1246,16 +1613,18 @@ _process_pkt_feedback (SentPacket * pkt, TWCCStatsManager * statsman) _pkt_status_s (pkt->status)); if (pkt->stats_processed) { /* This packet was already added to stats structures, but we've got - one more feedback for it + one more feedback for it. Re-evaluate every block it participates in. */ - RedBlock *block; - if (g_hash_table_lookup_extended (statsman->seqnum_2_redblocks, - GUINT_TO_POINTER (pkt->seqnum), NULL, (gpointer *) & block)) { - const gsize packets_recovered = _redblock_reconsider (statsman, block); - if (packets_recovered > 0) { - GST_LOG_OBJECT (statsman->parent, - "Reconsider block because of packet #%u, " "recovered %lu pckt", - pkt->seqnum, packets_recovered); + GPtrArray *blocks = _seqnum_blocks_get (statsman, pkt->seqnum); + if (blocks) { + for (guint bi = 0; bi < blocks->len; bi++) { + RedBlock *block = g_ptr_array_index (blocks, bi); + const gsize packets_recovered = _redblock_reconsider (statsman, block); + if (packets_recovered > 0) { + GST_LOG_OBJECT (statsman->parent, + "Reconsider block because of packet #%u, " "recovered %lu pckt", + pkt->seqnum, packets_recovered); + } } } return; @@ -1288,8 +1657,12 @@ _process_pkt_feedback (SentPacket * pkt, TWCCStatsManager * statsman) g_assert_not_reached (); } - /* Check if this packet covers the same block that was already added. */ - RedBlockKey key = _redblock_key_new (pkt->protects_twcc_seqnums); + /* Check if this packet covers a block that already exists. The block key + includes the redundant stream's identity (SSRC + payload type), so + distinct means of repair protecting the same data set map to distinct + blocks (and may overlap). */ + const TWCCRepairId repair = { pkt->ssrc, pkt->pt }; + RedBlockKey key = _redblock_key_new (pkt->protects_twcc_seqnums, repair); RedBlock *block = NULL; if (g_hash_table_lookup_extended (statsman->redund_2_redblocks, key, NULL, (gpointer *) & block)) { @@ -1324,27 +1697,15 @@ _process_pkt_feedback (SentPacket * pkt, TWCCStatsManager * statsman) /* Link this seqnum to the block in order to be able to release the block once this packet leave its lifetime */ - g_hash_table_insert (statsman->seqnum_2_redblocks, - GUINT_TO_POINTER (pkt->seqnum), block); + _seqnum_blocks_add (statsman, pkt->seqnum, block); /* There is no such block, add a new one */ } else { - /* Verify all the packets this redundant packet is protecting - * are not covered by any other existing block, because if - * they do it means some weird state of the data structures. - */ - for (gsize i = 0; i < pkt->protects_twcc_seqnums->len; ++i) { - const guint16 data_key = g_array_index (pkt->protects_twcc_seqnums, - guint16, i); - RedBlock *data_block = NULL; - if (g_hash_table_lookup_extended (statsman->seqnum_2_redblocks, - GUINT_TO_POINTER (data_key), NULL, (gpointer *) & data_block)) { - _redblock_key_free (key); - return; - } - } - /* Add every data packet into seqnum_2_redblocks */ + /* Create a new block. Overlap with other means of repair's blocks is + expected and supported: a data packet may be covered by several blocks + (e.g. both RTX and FEC), which is exactly how repair overlap is + tracked. */ block = _redblock_new (pkt->protects_twcc_seqnums, pkt->seqnum, - pkt->redundant_idx, pkt->redundant_num); + pkt->redundant_idx, pkt->redundant_num, repair); g_array_index (block->fec_seqs, guint16, (gsize) pkt->redundant_idx) = pkt->seqnum; g_array_index (block->fec_states, TWCCPktState, @@ -1352,24 +1713,11 @@ _process_pkt_feedback (SentPacket * pkt, TWCCStatsManager * statsman) g_hash_table_insert (statsman->redund_2_redblocks, key, block); /* Link this seqnum to the block in order to be able to release the block once this packet outlives its lifetime */ - g_hash_table_insert (statsman->seqnum_2_redblocks, - GUINT_TO_POINTER (pkt->seqnum), block); + _seqnum_blocks_add (statsman, pkt->seqnum, block); for (gsize i = 0; i < pkt->protects_twcc_seqnums->len; ++i) { const guint16 data_key = g_array_index (pkt->protects_twcc_seqnums, guint16, i); - RedBlock *data_block = NULL; - if (!g_hash_table_lookup_extended (statsman->seqnum_2_redblocks, - GUINT_TO_POINTER (data_key), NULL, (gpointer *) & data_block)) { - - g_hash_table_insert (statsman->seqnum_2_redblocks, - GUINT_TO_POINTER (data_key), block); - } else if (block != data_block) { - /* Overlapped blocks are not supported yet */ - GST_WARNING_OBJECT (statsman->parent, - "Data packet %ld covered by two blocks", data_key); - g_hash_table_replace (statsman->seqnum_2_redblocks, - GUINT_TO_POINTER (data_key), block); - } + _seqnum_blocks_add (statsman, data_key, block); } } const gsize packets_recovered = _redblock_reconsider (statsman, block); @@ -1378,22 +1726,23 @@ _process_pkt_feedback (SentPacket * pkt, TWCCStatsManager * statsman) pkt->seqnum, packets_recovered); /* Neither RTX nor FEC */ } else { - RedBlock *block; - if (g_hash_table_lookup_extended (statsman->seqnum_2_redblocks, - GUINT_TO_POINTER (pkt->seqnum), NULL, (gpointer *) & block)) { - - for (gsize i = 0; i < block->seqs->len; ++i) { - if (g_array_index (block->seqs, guint16, i) == pkt->seqnum) { - g_array_index (block->states, TWCCPktState, i) = - _better_pkt_state (g_array_index (block->states, TWCCPktState, i), - pkt->status); - break; + GPtrArray *blocks = _seqnum_blocks_get (statsman, pkt->seqnum); + if (blocks) { + for (guint bi = 0; bi < blocks->len; bi++) { + RedBlock *block = g_ptr_array_index (blocks, bi); + for (gsize i = 0; i < block->seqs->len; ++i) { + if (g_array_index (block->seqs, guint16, i) == pkt->seqnum) { + g_array_index (block->states, TWCCPktState, i) = + _better_pkt_state (g_array_index (block->states, TWCCPktState, + i), pkt->status); + break; + } } + const gsize packets_recovered = _redblock_reconsider (statsman, block); + GST_LOG_OBJECT (statsman->parent, + "Reconsider block because of packet #%u, " "recovered %lu pckt", + pkt->seqnum, packets_recovered); } - const gsize packets_recovered = _redblock_reconsider (statsman, block); - GST_LOG_OBJECT (statsman->parent, - "Reconsider block because of packet #%u, " "recovered %lu pckt", - pkt->seqnum, packets_recovered); } } } @@ -1426,7 +1775,7 @@ rtp_twcc_stats_manager_new (GObject * parent) _redund_equal, (GDestroyNotify) _redblock_key_free, (GDestroyNotify) _redblock_free); statsman->seqnum_2_redblocks = g_hash_table_new_full (g_direct_hash, - g_direct_equal, NULL, NULL); + g_direct_equal, NULL, (GDestroyNotify) g_ptr_array_unref); statsman->first_fci_parse = TRUE; statsman->expected_parsed_seqnum = 0; @@ -1488,13 +1837,13 @@ rtp_twcc_stats_sent_pkt (TWCCStatsManager * statsman, const guint16 rtp_seqnum = g_array_index (protect_seqnums_array, guint16, i); const guint32 rtp_ts = - g_array_index (protects_timestamps_array, guint32, i); - gint32 twcc_sn = - _lookup_seqnum (statsman, protect_ssrc, rtp_seqnum); + g_array_index (protects_timestamps_array, guint32, i); + gint32 twcc_sn = _lookup_seqnum (statsman, protect_ssrc, rtp_seqnum); const guint16 twcc_sn16 = (guint16) twcc_sn; - SentPacket * protected_pkt = NULL; + SentPacket *protected_pkt = NULL; if (twcc_sn == -1 - || !(protected_pkt = _find_sentpacket(statsman, twcc_sn16, &rtp_ts))) { + || !(protected_pkt = + _find_sentpacket (statsman, twcc_sn16, &rtp_ts))) { /* RTX can start sending stuffing packets right after a data packet was pushed * through, so it's queue can have very few packets. If the data packet * and stuffing packets be reordered on their way downstream, TWCC @@ -1531,14 +1880,12 @@ rtp_twcc_stats_sent_pkt (TWCCStatsManager * statsman, for (guint i = 0; protect_seqnums_array && i < protect_seqnums_array->len; i++) { - const guint16 rtp_sn = - g_array_index (protect_seqnums_array, guint16, i); + const guint16 rtp_sn = g_array_index (protect_seqnums_array, guint16, i); const guint16 twcc_sn = protect_twcc_seqnums_array && i < protect_twcc_seqnums_array->len ? g_array_index (protect_twcc_seqnums_array, guint16, i) : 0; GST_DEBUG_OBJECT (statsman->parent, - "%u protects rtp-seqnum: %u (twcc: %u)", twcc_seqnum, rtp_sn, - twcc_sn); + "%u protects rtp-seqnum: %u (twcc: %u)", twcc_seqnum, rtp_sn, twcc_sn); } GST_DEBUG_OBJECT @@ -1729,8 +2076,9 @@ rtp_twcc_stats_check_for_lost_packets (TWCCStatsManager * statsman, goto done; } - const gint seqnum_diff = gst_rtp_buffer_compare_seqnum ( - statsman->expected_parsed_seqnum, base_seqnum); + const gint seqnum_diff = + gst_rtp_buffer_compare_seqnum (statsman->expected_parsed_seqnum, + base_seqnum); if (seqnum_diff < 0) { GST_DEBUG_OBJECT (statsman->parent, "twcc seqnum is older than expected (%u < %u)", base_seqnum, diff --git a/subprojects/gst-plugins-good/tests/check/elements/rtpsession.c b/subprojects/gst-plugins-good/tests/check/elements/rtpsession.c index ae82136255..7393d6b573 100644 --- a/subprojects/gst-plugins-good/tests/check/elements/rtpsession.c +++ b/subprojects/gst-plugins-good/tests/check/elements/rtpsession.c @@ -201,10 +201,12 @@ generate_rtx_buffer (guint rtx_seqnum, GstBuffer * buffer) gst_rtp_buffer_map (new_buffer, GST_MAP_WRITE, &new_rtp); gst_rtp_buffer_set_payload_type (&new_rtp, TEST_RTX_BUF_PT); gst_rtp_buffer_set_ssrc (&new_rtp, TEST_RTX_BUF_SSRC); - gst_rtp_buffer_set_timestamp (&new_rtp, orig_timestamp); gst_rtp_buffer_set_seq (&new_rtp, rtx_seqnum); + gst_rtp_buffer_set_timestamp (&new_rtp, orig_timestamp); gst_rtp_buffer_unmap (&new_rtp); + /* Copy over timestamps */ + gst_buffer_copy_into (new_buffer, buffer, GST_BUFFER_COPY_TIMESTAMPS, 0, -1); gst_rtp_repair_meta_add (new_buffer, 0, 1, orig_ssrc, &orig_seqnum, &orig_timestamp, 1); @@ -5383,25 +5385,137 @@ construct_initial_state_for_rtx (SessionHarness * h_send, } static gdouble -get_recovery_pct (SessionHarness * h) +twcc_stats_get_recovery_pct (GstStructure * stats) { gdouble stats_recovery_pct; - GstStructure *twcc_stats; + fail_unless (gst_structure_get (stats, + "recovery-pct", G_TYPE_DOUBLE, &stats_recovery_pct, NULL)); + return stats_recovery_pct; +} - twcc_stats = session_harness_get_twcc_stats_full (h, - 300 * GST_MSECOND, 100 * GST_MSECOND); +/* Look up a single means of repair (by redundant ssrc + pt) in the + * "recovery-by-repair" value array of a TWCC stats structure. */ +static gboolean +twcc_stats_get_recovery_by_repair (GstStructure * stats, guint32 ssrc, guint pt, + guint * recovered_count, gdouble * recovery_pct) +{ + GValueArray *arr = NULL; + gboolean found = FALSE; + guint i; - fail_unless (gst_structure_get (twcc_stats, - "recovery-pct", G_TYPE_DOUBLE, &stats_recovery_pct, NULL)); + fail_unless (gst_structure_get (stats, "recovery-by-repair", + G_TYPE_VALUE_ARRAY, &arr, NULL)); + if (!arr) + return FALSE; + + for (i = 0; i < arr->n_values; i++) { + GstStructure *m = g_value_get_boxed (g_value_array_get_nth (arr, i)); + guint m_ssrc = 0, m_pt = 0; + gst_structure_get (m, "ssrc", G_TYPE_UINT, &m_ssrc, + "pt", G_TYPE_UINT, &m_pt, NULL); + if (m_ssrc == ssrc && m_pt == pt) { + if (recovered_count) + gst_structure_get (m, "recovered-count", G_TYPE_UINT, recovered_count, + NULL); + if (recovery_pct) + gst_structure_get (m, "recovery-pct", G_TYPE_DOUBLE, recovery_pct, + NULL); + found = TRUE; + break; + } + } + g_value_array_free (arr); + return found; +} - gst_structure_free (twcc_stats); - return stats_recovery_pct; +/* Look up the overlap cell for the unordered means-of-repair pair {a, b} in the + * "recovery-overlap" value array of a TWCC stats structure. */ +static gboolean +twcc_stats_get_overlap (GstStructure * stats, + guint32 ssrc_a, guint pt_a, guint32 ssrc_b, guint pt_b, + guint * count, gdouble * pct) +{ + GValueArray *arr = NULL; + gboolean found = FALSE; + guint i; + + fail_unless (gst_structure_get (stats, "recovery-overlap", + G_TYPE_VALUE_ARRAY, &arr, NULL)); + if (!arr) + return FALSE; + + for (i = 0; i < arr->n_values; i++) { + GstStructure *o = g_value_get_boxed (g_value_array_get_nth (arr, i)); + guint o_sa = 0, o_pa = 0, o_sb = 0, o_pb = 0; + gst_structure_get (o, "ssrc-a", G_TYPE_UINT, &o_sa, "pt-a", G_TYPE_UINT, + &o_pa, "ssrc-b", G_TYPE_UINT, &o_sb, "pt-b", G_TYPE_UINT, &o_pb, NULL); + if ((o_sa == ssrc_a && o_pa == pt_a && o_sb == ssrc_b && o_pb == pt_b) + || (o_sa == ssrc_b && o_pa == pt_b && o_sb == ssrc_a && o_pb == pt_a)) { + if (count) + gst_structure_get (o, "count", G_TYPE_UINT, count, NULL); + if (pct) + gst_structure_get (o, "pct", G_TYPE_DOUBLE, pct, NULL); + found = TRUE; + break; + } + } + g_value_array_free (arr); + return found; } -static void -fail_unless_twcc_stats_recovery (SessionHarness * h, gdouble recovery_pct) +static guint +twcc_stats_num_repairs (GstStructure * stats) { - fail_unless_equals_float (recovery_pct, get_recovery_pct (h)); + GValueArray *arr = NULL; + guint n = 0; + gst_structure_get (stats, "recovery-by-repair", G_TYPE_VALUE_ARRAY, &arr, + NULL); + if (arr) { + n = arr->n_values; + g_value_array_free (arr); + } + return n; +} + +static guint +twcc_stats_num_overlaps (GstStructure * stats) +{ + GValueArray *arr = NULL; + guint n = 0; + gst_structure_get (stats, "recovery-overlap", G_TYPE_VALUE_ARRAY, &arr, NULL); + if (arr) { + n = arr->n_values; + g_value_array_free (arr); + } + return n; +} + + +/* Sum the per-repair "bitrate-sent" emitted in the "recovery-by-repair" array + for the given payload type. Returns 0 if no entry for @pt is present. */ +static guint +twcc_stats_get_repair_bitrate_sent (GstStructure * stats, guint pt) +{ + GValueArray *repairs = NULL; + guint total = 0; + + gst_structure_get (stats, "recovery-by-repair", G_TYPE_VALUE_ARRAY, + &repairs, NULL); + if (repairs) { + for (guint i = 0; i < repairs->n_values; i++) { + const GstStructure *rs = + gst_value_get_structure (g_value_array_get_nth (repairs, i)); + guint entry_pt = 0; + guint bitrate_sent = 0; + gst_structure_get (rs, "pt", G_TYPE_UINT, &entry_pt, + "bitrate-sent", G_TYPE_UINT, &bitrate_sent, NULL); + if (entry_pt == pt) + total += bitrate_sent; + } + g_value_array_free (repairs); + } + + return total; } static void @@ -5433,6 +5547,7 @@ test_twcc_stats_rtx_recovery (gboolean rtx_arrive, gdouble recovery_pct) SessionHarness *h_send = session_harness_new (); SessionHarness *h_recv = session_harness_new (); guint i, next_seqnum; + GstStructure *twcc_stats; session_harness_add_twcc_caps_for_pt (h_send, TEST_BUF_PT); session_harness_add_twcc_caps_for_pt (h_send, TEST_RTX_BUF_PT); @@ -5460,8 +5575,13 @@ test_twcc_stats_rtx_recovery (gboolean rtx_arrive, gdouble recovery_pct) session_harness_recv_rtcp (h_send, session_harness_produce_twcc (h_recv))); - fail_unless_twcc_stats_recovery (h_send, recovery_pct); + twcc_stats = session_harness_get_twcc_stats_full (h_send, + 300 * GST_MSECOND, 100 * GST_MSECOND); + fail_unless_equals_float (recovery_pct, + twcc_stats_get_recovery_pct (twcc_stats)); + + gst_structure_free (twcc_stats); session_harness_free (h_send); session_harness_free (h_recv); } @@ -5473,6 +5593,57 @@ GST_START_TEST (test_twcc_stats_rtx_recover_lost) GST_END_TEST; +/* When RTX packets are sent, the "recovery-by-repair" array must report a + non-zero per-repair "bitrate-sent" for the RTX payload type, and none for + the media payload type. */ +GST_START_TEST (test_twcc_stats_rtx_repair_bitrate) +{ + SessionHarness *h_send = session_harness_new (); + SessionHarness *h_recv = session_harness_new (); + guint i, next_seqnum; + GstStructure *twcc_stats; + + session_harness_add_twcc_caps_for_pt (h_send, TEST_BUF_PT); + session_harness_add_twcc_caps_for_pt (h_send, TEST_RTX_BUF_PT); + session_harness_set_twcc_recv_ext_id (h_recv, TEST_TWCC_EXT_ID); + + next_seqnum = construct_initial_state_for_rtx (h_send, h_recv); + + for (i = 0; i < 3; i++) { + GstBuffer *buf; + GstBuffer *rtx_buf; + + buf = generate_twcc_send_buffer (next_seqnum++, FALSE); + rtx_buf = generate_rtx_buffer (i, buf); + + /* the original is lost, the RTX retransmission arrives */ + send_recv_buffer (h_send, h_recv, buf, FALSE); + send_recv_buffer (h_send, h_recv, rtx_buf, TRUE); + } + + send_recv_buffer (h_send, h_recv, + generate_twcc_send_buffer (next_seqnum++, TRUE), TRUE); + + fail_unless_equals_int64 (GST_FLOW_OK, + session_harness_recv_rtcp (h_send, + session_harness_produce_twcc (h_recv))); + + /* The RTX payload type must carry redundant bitrate; the media PT must not + appear as a means of repair. */ + twcc_stats = session_harness_get_twcc_stats_full (h_send, + 300 * GST_MSECOND, 100 * GST_MSECOND); + fail_unless (twcc_stats_get_repair_bitrate_sent (twcc_stats, + TEST_RTX_BUF_PT) > 0); + fail_unless_equals_int (twcc_stats_get_repair_bitrate_sent (twcc_stats, + TEST_BUF_PT), 0); + + gst_structure_free (twcc_stats); + session_harness_free (h_send); + session_harness_free (h_recv); +} + +GST_END_TEST; + static GstBuffer * create_twcc_rtcp_from_fci (const guint8 * fci, guint fci_len, guint32 sender_ssrc, guint32 media_ssrc) @@ -5700,6 +5871,7 @@ GST_START_TEST (test_twcc_stats_long_rtx_recover) guint nframes = 100; guint nframes_to_skip = 10; guint pkt_in_frame = 10; + GstStructure *twcc_stats; GRand *rnd = g_rand_new_with_seed (101); @@ -5740,13 +5912,17 @@ GST_START_TEST (test_twcc_stats_long_rtx_recover) session_harness_recv_rtcp (h_send, session_harness_produce_twcc (h_recv))); + twcc_stats = session_harness_get_twcc_stats_full (h_send, + 300 * GST_MSECOND, 100 * GST_MSECOND); if (nframe > nframes_to_skip) { - const gdouble recovery_stats = get_recovery_pct (h_send); + const gdouble recovery_stats = twcc_stats_get_recovery_pct (twcc_stats); fail_if (recovery_stats != 100. && recovery_stats != -1.); } else { - const gdouble recovery_stats = get_recovery_pct (h_send); + const gdouble recovery_stats = twcc_stats_get_recovery_pct (twcc_stats); GST_ERROR ("Frame %d: recovery stats: %f", nframe, recovery_stats); } + gst_structure_free (twcc_stats); + } g_rand_free (rnd); @@ -5833,6 +6009,7 @@ GST_START_TEST (test_twcc_stats_block_fec_recover) guint i, next_seqnum; const guint window_size_ms = 400; const guint num_buffers = window_size_ms / TEST_BUF_MS + 1; + GstStructure *twcc_stats; guint16 fec_seqnum = 6666; gsize fec_num = 0; @@ -5878,7 +6055,8 @@ GST_START_TEST (test_twcc_stats_block_fec_recover) } protects_seqnums[protects_seqnums_i] = next_seqnum; - protects_timestamps[protects_seqnums_i] = next_seqnum * TEST_RTP_TS_DURATION; + protects_timestamps[protects_seqnums_i] = + next_seqnum * TEST_RTP_TS_DURATION; protects_seqnums_i += 1; if (protects_seqnums_i >= block_len) { protects_seqnums_i = 0; @@ -5903,7 +6081,258 @@ GST_START_TEST (test_twcc_stats_block_fec_recover) /* produce a twcc feedback to process those packets */ session_harness_recv_rtcp (h_send, session_harness_produce_twcc (h_recv)); - fail_unless_twcc_stats_recovery (h_send, 100.); + twcc_stats = session_harness_get_twcc_stats_full (h_send, + 300 * GST_MSECOND, 100 * GST_MSECOND); + + fail_unless_equals_float (100., twcc_stats_get_recovery_pct (twcc_stats)); + + gst_structure_free (twcc_stats); + session_harness_free (h_send); + session_harness_free (h_recv); + g_free (protects_seqnums); + g_free (protects_timestamps); +} + +GST_END_TEST; + + +/* A lost data packet that is retransmitted (RTX) AND protected by a FEC block, + * both of which are received, must be attributed to BOTH mechanisms and appear + * in the RTX-x-FEC overlap cell. */ +GST_START_TEST (test_twcc_stats_rtx_and_fec_overlap) +{ + SessionHarness *h_send = session_harness_new (); + SessionHarness *h_recv = session_harness_new (); + guint i; + guint next_seqnum; + GstStructure *twcc_stats; + + const guint32 fec_ssrc = 0x12345678; + const guint8 fec_pt = 127; + const gsize nblocks = 6; + const gsize block_len = 4; + guint16 fec_seqnum = 5000; + guint fec_num = 0; + + guint rtx_count = 0, fec_count = 0, overlap_count = 0; + gdouble rtx_pct = 0, fec_pct = 0, overlap_pct = 0; + + guint16 *protects_seqnums = g_malloc0 (sizeof (guint16) * block_len); + guint32 *protects_timestamps = g_malloc0 (sizeof (guint32) * block_len); + + session_harness_add_twcc_caps_for_pt (h_send, TEST_BUF_PT); + session_harness_add_twcc_caps_for_pt (h_send, TEST_RTX_BUF_PT); + session_harness_add_twcc_caps_for_pt (h_send, fec_pt); + session_harness_set_twcc_recv_ext_id (h_recv, TEST_TWCC_EXT_ID); + + next_seqnum = construct_initial_state_for_rtx (h_send, h_recv); + + for (gsize blk = 0; blk < nblocks; blk++) { + GstBuffer *fec_buf; + + for (i = 0; i < block_len; i++) { + const guint seqnum = next_seqnum++; + GstBuffer *buf = generate_twcc_send_buffer (seqnum, FALSE); + + protects_seqnums[i] = seqnum; + protects_timestamps[i] = (guint32) seqnum *(guint32) TEST_RTP_TS_DURATION; + + if (i == 0) { + /* Build the RTX for the lost packet first (does not consume buf), + then lose the data packet and deliver the RTX. */ + GstBuffer *rtx_buf = generate_rtx_buffer (fec_num, buf); + send_recv_buffer (h_send, h_recv, buf, FALSE); + send_recv_buffer (h_send, h_recv, rtx_buf, TRUE); + } else { + send_recv_buffer (h_send, h_recv, buf, TRUE); + } + } + + /* One FEC packet protecting the whole block (capacity 1 == lost 1). */ + fec_buf = generate_test_buffer_full (fec_num * TEST_BUF_DURATION, + fec_seqnum, fec_num * TEST_RTP_TS_DURATION, fec_ssrc, + blk == nblocks - 1, fec_pt, 0, 0, 0, 1, TEST_BUF_SSRC, + protects_seqnums, protects_timestamps, block_len); + send_recv_buffer (h_send, h_recv, fec_buf, TRUE); + fec_num++; + fec_seqnum++; + } + + fail_unless_equals_int64 (GST_FLOW_OK, + session_harness_recv_rtcp (h_send, + session_harness_produce_twcc (h_recv))); + + twcc_stats = session_harness_get_twcc_stats_full (h_send, + (nblocks * (block_len + 1) + 2) * TEST_BUF_DURATION, TEST_BUF_DURATION); + + /* Both means of repair recovered every lost packet, so both should report + 100% recovery, and the overlap cell between them should be present. */ + fail_unless (twcc_stats_get_recovery_by_repair (twcc_stats, TEST_RTX_BUF_SSRC, + TEST_RTX_BUF_PT, &rtx_count, &rtx_pct)); + fail_unless (twcc_stats_get_recovery_by_repair (twcc_stats, fec_ssrc, fec_pt, + &fec_count, &fec_pct)); + fail_unless (twcc_stats_get_overlap (twcc_stats, TEST_RTX_BUF_SSRC, + TEST_RTX_BUF_PT, fec_ssrc, fec_pt, &overlap_count, &overlap_pct)); + + fail_unless (rtx_count > 0); + fail_unless (fec_count > 0); + fail_unless (overlap_count > 0); + fail_unless_equals_float (rtx_pct, 100.0); + fail_unless_equals_float (fec_pct, 100.0); + fail_unless_equals_float (overlap_pct, 100.0); + + gst_structure_free (twcc_stats); + + session_harness_free (h_send); + session_harness_free (h_recv); + g_free (protects_seqnums); + g_free (protects_timestamps); +} + +GST_END_TEST; + +/* When only RTX recovers losses (no FEC in play), the per-repair map must + * list exactly the RTX repair and there must be no overlap cell. */ +GST_START_TEST (test_twcc_stats_rtx_repair_no_overlap) +{ + SessionHarness *h_send = session_harness_new (); + SessionHarness *h_recv = session_harness_new (); + guint i, next_seqnum; + GstStructure *twcc_stats; + guint rtx_count = 0; + gdouble rtx_pct = 0; + + session_harness_add_twcc_caps_for_pt (h_send, TEST_BUF_PT); + session_harness_add_twcc_caps_for_pt (h_send, TEST_RTX_BUF_PT); + session_harness_set_twcc_recv_ext_id (h_recv, TEST_TWCC_EXT_ID); + + next_seqnum = construct_initial_state_for_rtx (h_send, h_recv); + + for (i = 0; i < 3; i++) { + GstBuffer *buf = generate_twcc_send_buffer (next_seqnum++, FALSE); + GstBuffer *rtx_buf = generate_rtx_buffer (i, buf); + send_recv_buffer (h_send, h_recv, buf, FALSE); /* lost */ + send_recv_buffer (h_send, h_recv, rtx_buf, TRUE); /* recovered */ + } + + send_recv_buffer (h_send, h_recv, + generate_twcc_send_buffer (next_seqnum++, TRUE), TRUE); + + fail_unless_equals_int64 (GST_FLOW_OK, + session_harness_recv_rtcp (h_send, + session_harness_produce_twcc (h_recv))); + + twcc_stats = session_harness_get_twcc_stats_full (h_send, + 300 * GST_MSECOND, 100 * GST_MSECOND); + + fail_unless (twcc_stats_get_recovery_by_repair (twcc_stats, TEST_RTX_BUF_SSRC, + TEST_RTX_BUF_PT, &rtx_count, &rtx_pct)); + fail_unless (rtx_count > 0); + fail_unless_equals_float (rtx_pct, 100.0); + /* exactly one means of repair, and no overlap with anything */ + fail_unless_equals_int (1, twcc_stats_num_repairs (twcc_stats)); + fail_unless_equals_int (0, twcc_stats_num_overlaps (twcc_stats)); + + gst_structure_free (twcc_stats); + + session_harness_free (h_send); + session_harness_free (h_recv); +} + +GST_END_TEST; + +/* A FEC block that is over capacity (more losses than redundant packets) cannot + * recover anything, but RTX of the same lost packets can. The lost packets must + * be credited to RTX only -- FEC must NOT appear in the per-repair map and there + * must be no overlap cell. */ +GST_START_TEST (test_twcc_stats_rtx_recovers_fec_insufficient) +{ + SessionHarness *h_send = session_harness_new (); + SessionHarness *h_recv = session_harness_new (); + guint i; + guint next_seqnum; + GstStructure *twcc_stats; + + const guint32 fec_ssrc = 0x12345678; + const guint8 fec_pt = 127; + const gsize nblocks = 6; + const gsize block_len = 4; + const gsize losses_per_block = 2; /* > 1 FEC packet => FEC cannot recover */ + guint16 fec_seqnum = 5000; + guint fec_num = 0; + guint rtx_seqnum = 0; + + guint rtx_count = 0, fec_count = 0, overlap_count = 0; + gdouble rtx_pct = 0, fec_pct = 0, overlap_pct = 0; + + guint16 *protects_seqnums = g_malloc0 (sizeof (guint16) * block_len); + guint32 *protects_timestamps = g_malloc0 (sizeof (guint32) * block_len); + + session_harness_add_twcc_caps_for_pt (h_send, TEST_BUF_PT); + session_harness_add_twcc_caps_for_pt (h_send, TEST_RTX_BUF_PT); + session_harness_add_twcc_caps_for_pt (h_send, fec_pt); + session_harness_set_twcc_recv_ext_id (h_recv, TEST_TWCC_EXT_ID); + + next_seqnum = construct_initial_state_for_rtx (h_send, h_recv); + + for (gsize blk = 0; blk < nblocks; blk++) { + GstBuffer *fec_buf; + + for (i = 0; i < block_len; i++) { + const guint seqnum = next_seqnum++; + GstBuffer *buf = generate_twcc_send_buffer (seqnum, FALSE); + + protects_seqnums[i] = seqnum; + protects_timestamps[i] = (guint32) seqnum *(guint32) TEST_RTP_TS_DURATION; + + if (i < losses_per_block) { + /* Lose the data packet, but deliver an RTX for it. */ + GstBuffer *rtx_buf = generate_rtx_buffer (rtx_seqnum++, buf); + send_recv_buffer (h_send, h_recv, buf, FALSE); + send_recv_buffer (h_send, h_recv, rtx_buf, TRUE); + } else { + send_recv_buffer (h_send, h_recv, buf, TRUE); + } + } + + /* A single FEC packet (capacity 1) for a block with 2 losses: the FEC + block is over capacity and recovers nothing. */ + fec_buf = generate_test_buffer_full (fec_num * TEST_BUF_DURATION, + fec_seqnum, fec_num * TEST_RTP_TS_DURATION, fec_ssrc, + blk == nblocks - 1, fec_pt, 0, 0, 0, 1, TEST_BUF_SSRC, + protects_seqnums, protects_timestamps, block_len); + send_recv_buffer (h_send, h_recv, fec_buf, TRUE); + fec_num++; + fec_seqnum++; + } + + fail_unless_equals_int64 (GST_FLOW_OK, + session_harness_recv_rtcp (h_send, + session_harness_produce_twcc (h_recv))); + + twcc_stats = session_harness_get_twcc_stats_full (h_send, + (nblocks * (block_len + 1) + 2) * TEST_BUF_DURATION, TEST_BUF_DURATION); + + /* RTX recovered every lost packet; FEC recovered none. */ + fail_unless (twcc_stats_get_recovery_by_repair (twcc_stats, TEST_RTX_BUF_SSRC, + TEST_RTX_BUF_PT, &rtx_count, &rtx_pct)); + fail_unless (rtx_count > 0); + fail_unless_equals_float (rtx_pct, 100.0); + + /* FEC is present in the per-repair map (its sent bitrate is accounted for), + but it recovered nothing: zero recovered count and zero recovery pct. */ + fail_unless (twcc_stats_get_recovery_by_repair (twcc_stats, fec_ssrc, fec_pt, + &fec_count, &fec_pct)); + fail_unless_equals_int (0, fec_count); + fail_unless_equals_float (fec_pct, 0.0); + /* The bits FEC spent must still show up in the statistics. */ + fail_unless (twcc_stats_get_repair_bitrate_sent (twcc_stats, fec_pt) > 0); + /* No lost packet was recoverable by both means of repair, so no overlap. */ + fail_if (twcc_stats_get_overlap (twcc_stats, TEST_RTX_BUF_SSRC, + TEST_RTX_BUF_PT, fec_ssrc, fec_pt, &overlap_count, &overlap_pct)); + fail_unless_equals_int (0, twcc_stats_num_overlaps (twcc_stats)); + + gst_structure_free (twcc_stats); session_harness_free (h_send); session_harness_free (h_recv); @@ -6649,11 +7078,11 @@ GST_START_TEST (test_twcc_keep_queue_size) session_harness_produce_twcc (h_recv))); /* expected_parsed_seqnum = BASE_SEQNUM + 16 + 11 = 65307 - recovery base_seqnum = (65307 + 70000) mod 65536 + 11 = 4235 - gap = (gint16)(4235 - 65307) = 4464 packets marked LOST. + recovery base_seqnum = (65307 + 70000) mod 65536 + 11 = 4235 + gap = (gint16)(4235 - 65307) = 4464 packets marked LOST. Within the 500ms stats window (25 intervals at 20ms = 26 packets from TWCC 4220 to 4245): - 15 LOST flood packets (TWCC 4220-4234) + 15 LOST flood packets (TWCC 4220-4234) + 11 RECEIVED recovery (TWCC 4235-4245) = 26 total sent */ const guint gap_pkts_in_window = 15; @@ -6699,7 +7128,7 @@ GST_START_TEST (test_twcc_seqnum_wrap_gap_detection) 0x00, 0x00, 0x00, /* reference time: 0 */ 0x00, /* fb_pkt_count: 0 */ 0x20, 0x0A, /* run-length: received small_delta, count=10 */ - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 /* 10 recv-deltas, all 0 */ + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 /* 10 recv-deltas, all 0 */ }; guint8 fci2[] = { @@ -6708,14 +7137,14 @@ GST_START_TEST (test_twcc_seqnum_wrap_gap_detection) 0x00, 0x00, 0x00, /* reference time: 0 */ 0x01, /* fb_pkt_count: 1 */ 0x20, 0x0A, /* run-length: received small_delta, count=10 */ - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 /* 10 recv-deltas, all 0 */ + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 /* 10 recv-deltas, all 0 */ }; session_harness_add_twcc_caps_for_pt (h, TEST_BUF_PT); for (i = 0; i < n_packets; i++) { fail_unless_equals_int64 (GST_FLOW_OK, - session_harness_send_rtp (h, generate_twcc_send_buffer (i, FALSE))); + session_harness_send_rtp (h, generate_twcc_send_buffer (i, FALSE))); session_harness_advance_and_crank (h, TEST_BUF_DURATION); } @@ -6727,7 +7156,7 @@ GST_START_TEST (test_twcc_seqnum_wrap_gap_detection) /* Stats window 500ms/0delay covers: - Packet 240 (TWCC 65520, t=4800ms) through 265 (TWCC 9, t=5300ms) - 10 RECEIVED (65520-65529 from FCI #1) + 6 UNKNOWN (65530-65535 gap) - + 10 RECEIVED (0-9 from FCI #2) + + 10 RECEIVED (0-9 from FCI #2) - If gap detection worked: 6 LOST → loss_pct = 6/26 ≈ 23% - Bug: 6 UNKNOWN (not counted) → loss_pct = 0/20 = 0 @@ -6790,8 +7219,7 @@ _twcc_set_sock_ts_race_feedback_thread (gpointer data) { TwccSetSockTsRaceCtx *ctx = data; while (TRUE) { - GstBuffer *buf = - g_async_queue_timeout_pop (ctx->sent_bufs, 100 * 1000); + GstBuffer *buf = g_async_queue_timeout_pop (ctx->sent_bufs, 100 * 1000); if (buf) { GstTxFeedbackMeta *meta = gst_buffer_get_tx_feedback_meta (buf); if (meta) @@ -7435,6 +7863,9 @@ _nack_race_push_thread (gpointer data) static GstPadProbeReturn _drop_buffer_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data) { + (void) pad; + (void) info; + (void) user_data; return GST_PAD_PROBE_DROP; } @@ -7456,11 +7887,9 @@ GST_START_TEST (test_schedule_nack_hashtable_race) fail_unless (rtpsession != NULL); /* Request pads */ - recv_rtp_sink = - gst_element_request_pad_simple (rtpsession, "recv_rtp_sink"); + recv_rtp_sink = gst_element_request_pad_simple (rtpsession, "recv_rtp_sink"); fail_unless (recv_rtp_sink != NULL); - send_rtcp_src = - gst_element_request_pad_simple (rtpsession, "send_rtcp_src"); + send_rtcp_src = gst_element_request_pad_simple (rtpsession, "send_rtcp_src"); fail_unless (send_rtcp_src != NULL); /* Get the recv_rtp_src pad (created when recv_rtp_sink was requested) */ @@ -7493,20 +7922,20 @@ GST_START_TEST (test_schedule_nack_hashtable_race) { GstSegment segment; gst_segment_init (&segment, GST_FORMAT_TIME); - gst_pad_push_event (rtp_srcpad, - gst_event_new_stream_start ("test-stream")); + fail_unless (gst_pad_push_event (rtp_srcpad, + gst_event_new_stream_start ("test-stream"))); caps = gst_caps_new_simple ("application/x-rtp", - "clock-rate", G_TYPE_INT, 8000, - "payload", G_TYPE_INT, 96, NULL); - gst_pad_push_event (rtp_srcpad, gst_event_new_caps (caps)); + "clock-rate", G_TYPE_INT, 8000, "payload", G_TYPE_INT, 96, NULL); + fail_unless (gst_pad_push_event (rtp_srcpad, gst_event_new_caps (caps))); gst_caps_unref (caps); - gst_pad_push_event (rtp_srcpad, gst_event_new_segment (&segment)); + fail_unless (gst_pad_push_event (rtp_srcpad, + gst_event_new_segment (&segment))); } /* Establish the NACK source: receive a few packets from 0xDEADBEEF */ for (i = 0; i < 3; i++) { GstBuffer *buf = _make_rtp_buffer (i, 0xDEADBEEF); - gst_pad_push (rtp_srcpad, buf); + fail_unless_equals_int (gst_pad_push (rtp_srcpad, buf), GST_FLOW_OK); } /* Set up the push thread context */ @@ -7518,8 +7947,7 @@ GST_START_TEST (test_schedule_nack_hashtable_race) /* Start multiple push threads: continuously push packets from new SSRCs */ for (t = 0; t < G_N_ELEMENTS (push_threads); t++) { gchar *name = g_strdup_printf ("nack-race-push-%u", t); - push_threads[t] = - g_thread_new (name, _nack_race_push_thread, &ctx); + push_threads[t] = g_thread_new (name, _nack_race_push_thread, &ctx); g_free (name); } @@ -7536,8 +7964,7 @@ GST_START_TEST (test_schedule_nack_hashtable_race) "seqnum", G_TYPE_UINT, (guint) (i % 3), "delay", G_TYPE_UINT, (guint) 0, "deadline", G_TYPE_UINT, (guint) 100, - "avg-rtt", G_TYPE_UINT, (guint) 0, - NULL); + "avg-rtt", G_TYPE_UINT, (guint) 0, NULL); event = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM, s); /* Push upstream from the sinkpad connected to recv_rtp_src */ gst_pad_push_event (rtp_sinkpad, event); @@ -7662,11 +8089,15 @@ rtpsession_suite (void) tcase_add_test (tc_chain, test_twcc_reference_time_wrap); tcase_add_test (tc_chain, test_twcc_reference_time_wrap_start_negative); tcase_add_test (tc_chain, test_twcc_stats_rtx_recover_lost); + tcase_add_test (tc_chain, test_twcc_stats_rtx_repair_bitrate); tcase_add_test (tc_chain, test_twcc_parse_fci_robustness); tcase_add_test (tc_chain, test_twcc_stats_no_rtx_no_recover); tcase_add_test (tc_chain, test_twcc_stats_long_rtx_recover); tcase_add_test (tc_chain, test_twcc_stats_rtx_recover_not_lost_stuff); tcase_add_test (tc_chain, test_twcc_stats_block_fec_recover); + tcase_add_test (tc_chain, test_twcc_stats_rtx_and_fec_overlap); + tcase_add_test (tc_chain, test_twcc_stats_rtx_repair_no_overlap); + tcase_add_test (tc_chain, test_twcc_stats_rtx_recovers_fec_insufficient); tcase_add_test (tc_chain, test_twcc_stats_dod); tcase_add_test (tc_chain, test_twcc_feedback_max_sent_packets); tcase_add_test (tc_chain, test_twcc_feedback_wraparound);