From 076e22509efa9e77a36f853c2551e695dbd34b2a Mon Sep 17 00:00:00 2001 From: Richard Fuchs Date: Thu, 4 Jun 2026 13:10:01 -0400 Subject: [PATCH] MT#55283 update to new CC ABI Change-Id: Iab7a930890d9b75d1aa08c2c5f122fc0e2180ebb --- daemon/statistics.c | 99 ++++++++++++++++++++++++++++++++++++++---- debian/control | 2 +- lib/auxlib.c | 5 +++ lib/auxlib.h | 2 + lib/codeclib.c | 102 ++++++++++++++++++++++++++++++++++++++++---- lib/codeclib.h | 39 +++++++++++++++++ 6 files changed, 232 insertions(+), 17 deletions(-) diff --git a/daemon/statistics.c b/daemon/statistics.c index 2129936e0..855571d63 100644 --- a/daemon/statistics.c +++ b/daemon/statistics.c @@ -951,10 +951,10 @@ stats_metric_q *statistics_gather_metrics(struct interface_sampled_rate_stats *i METRICs("pid", "%u", (unsigned int) pt->pid); METRICs("wakeups", "%" PRIu64, atomic64_get_na(&pt->wakeups)); PROM("poller_wakeups", "counter"); - PROMLAB("index=\"%u\",pid=\"%u\"", i, (unsigned int) pt->pid); + PROMLAB("index=\"%u\",pid=\"%u\",type=\"user\"", i, (unsigned int) pt->pid); METRICs("items", "%" PRIu64, atomic64_get_na(&pt->items)); PROM("poller_items", "counter"); - PROMLAB("index=\"%u\",pid=\"%u\"", i, (unsigned int) pt->pid); + PROMLAB("index=\"%u\",pid=\"%u\",type=\"user\"", i, (unsigned int) pt->pid); HEADER("}", NULL); } HEADER("]", NULL); @@ -977,7 +977,11 @@ stats_metric_q *statistics_gather_metrics(struct interface_sampled_rate_stats *i METRICs("index", "%u", i); METRICs("pid", "%u", (unsigned int) pt->pid); METRICs("wakeups", "%" PRIu64, atomic64_get_na(&pt->wakeups)); + PROM("poller_wakeups", "counter"); + PROMLAB("index=\"%u\",pid=\"%u\",type=\"kernel\"", i, (unsigned int) pt->pid); METRICs("items", "%" PRIu64, atomic64_get_na(&pt->items)); + PROM("poller_items", "counter"); + PROMLAB("index=\"%u\",pid=\"%u\",type=\"kernel\"", i, (unsigned int) pt->pid); HEADER("}", ""); } HEADER("]", NULL); @@ -990,16 +994,95 @@ stats_metric_q *statistics_gather_metrics(struct interface_sampled_rate_stats *i struct kernel_ring_buf *rb = &kernel_ring_bufs[i]; HEADER("{", ""); METRICs("index", "%u", i); - METRICs("errors", "%d", atomic_get_na(&rb->errors)); - METRICs("read_preempt", "%d", atomic_get_na(&rb->read_preempt)); - METRICs("write_preempt", "%d", atomic_get_na(&rb->write_preempt)); - METRICs("slots_full", "%d", atomic_get_na(&rb->slots_full)); - METRICs("buf_full", "%d", atomic_get_na(&rb->buf_full)); - HEADER("}", ""); + +#define M(x, f) \ + METRICs(#x, f, atomic_get_na(&rb->errors)); \ + PROM("ksender_" #x, "counter"); \ + PROMLAB("index=\"%u\"", i) + + M(errors, "%d"); + M(read_preempt, "%d"); + M(write_preempt, "%d"); + M(slots_full, "%d"); + M(buf_full, "%d"); +#undef M } HEADER("]", NULL); } +#ifdef HAVE_CODEC_CHAIN + { + codec_cc_stats_q q = codec_cc_stats(); + + if (q.length) { + HEADER("codec_chain", NULL); + HEADER("[", NULL); + + while (q.length) { + __auto_type e = t_queue_pop_head(&q); + + HEADER("{", ""); + METRICs("name", "\"%s\"", e->name); + +#define M(x, f) \ + METRICs(#x, f, e->x); \ + PROM("cc_" #x, "counter"); \ + PROMLAB("name=\"%s\"", e->name) + + M(async_busy, "%u"); + M(async_blocked, "%u"); + M(async_retry, "%u"); +#undef M + + HEADER("contexts", NULL); + HEADER("[", NULL); + + while (e->contexts.length) { + __auto_type c = t_queue_pop_head(&e->contexts); + + HEADER("{", ""); + +#define M(x, f) \ + METRICs(#x, f, c->x); \ + PROM("cc_" #x, "counter"); \ + PROMLAB("index=\"%u\",name=\"%s\"", c->ctx_idx, e->name) + + M(runs, "%u"); + M(slots, "%u"); + M(run_wait, "%" PRIu64); + M(writers_wait, "%" PRIu64); + M(compute_wait, "%" PRIu64); + M(readers_wait, "%" PRIu64); + + M(run_busy, "%u"); + M(write_busy, "%u"); + M(slots_full, "%u"); + M(buf_full, "%u"); + + M(ready_wait, "%" PRIu64); + M(callbacks_preempt, "%" PRIu64); + M(callbacks_fetch, "%" PRIu64); + M(callbacks_run, "%" PRIu64); + M(loop_barrier, "%" PRIu64); +#undef M + + HEADER("}", ""); + + g_free(c); + } + + HEADER("]", NULL); + + HEADER("}", ""); + + g_free(e); + } + + HEADER("]", NULL); + } + } +#endif + HEADER("}", NULL); return ret; diff --git a/debian/control b/debian/control index ee9fb3579..e7f4e8989 100644 --- a/debian/control +++ b/debian/control @@ -48,7 +48,7 @@ Build-Depends: libwebsockets-dev, libwww-perl, libxtables-dev (>= 1.4) | iptables-dev (>= 1.4), - ngcp-libcodec-chain-dev (>= 13.3) , + ngcp-libcodec-chain-dev (>= 26.1) , pandoc, pkgconf, pybuild-plugin-pyproject , diff --git a/lib/auxlib.c b/lib/auxlib.c index 6595e764d..639cd27ec 100644 --- a/lib/auxlib.c +++ b/lib/auxlib.c @@ -269,6 +269,8 @@ void config_load_ext(int *argc, char ***argv, GOptionEntry *app_entries, const c { "codec-chain-runners",0,0, G_OPTION_ARG_INT, &rtpe_common_config_ptr->codec_chain_runners,"Number of chain runners per codec","INT" }, { "codec-chain-concurrency",0,0,G_OPTION_ARG_INT, &rtpe_common_config_ptr->codec_chain_concurrency,"Max concurrent codec jobs per runner","INT" }, { "codec-chain-async",0,0, G_OPTION_ARG_INT, &rtpe_common_config_ptr->codec_chain_async,"Number of background callback threads","INT" }, + { "codec-chain-nonblock",0,0, G_OPTION_ARG_NONE, &rtpe_common_config_ptr->codec_chain_nonblock,"Non-blocking operation of async codecs",NULL}, + { "codec-chain-interval",0,0, G_OPTION_ARG_INT, &rtpe_common_config_ptr->codec_chain_interval,"Microseconds between runs", "INT" }, { "codec-chain-opus-application",0,0,G_OPTION_ARG_STRING,&opus_application, "Opus application", "default|VoIP|audio|low-delay" }, { "codec-chain-opus-complexity",0,0,G_OPTION_ARG_INT, &rtpe_common_config_ptr->codec_chain_opus_complexity,"Opus encoding complexity (0..10)","INT" }, #endif @@ -495,6 +497,9 @@ out: if (rtpe_common_config_ptr->codec_chain_async < 0) rtpe_common_config_ptr->codec_chain_async = 0; + if (rtpe_common_config_ptr->codec_chain_interval <= 0) + rtpe_common_config_ptr->codec_chain_interval = 10000; + if (rtpe_common_config_ptr->codec_chain_opus_complexity == -1) rtpe_common_config_ptr->codec_chain_opus_complexity = 10; if (rtpe_common_config_ptr->codec_chain_opus_complexity < 0 || rtpe_common_config_ptr->codec_chain_opus_complexity > 10) diff --git a/lib/auxlib.h b/lib/auxlib.h index f844ccad3..72f0f5f3e 100644 --- a/lib/auxlib.h +++ b/lib/auxlib.h @@ -44,6 +44,8 @@ struct rtpengine_common_config { int codec_chain_runners; int codec_chain_concurrency; int codec_chain_async; + gboolean codec_chain_nonblock; + int codec_chain_interval; int codec_chain_opus_application; int codec_chain_opus_complexity; }; diff --git a/lib/codeclib.c b/lib/codeclib.c index 4ef645477..a8922fb2e 100644 --- a/lib/codeclib.c +++ b/lib/codeclib.c @@ -157,7 +157,7 @@ static __typeof__(codec_chain_client_runner_free) *cc_client_runner_free; static __typeof__(codec_chain_client_async_runner_new) *cc_client_async_runner_new; static __typeof__(codec_chain_client_async_runner_free) *cc_client_async_runner_free; static __typeof__(codec_chain_runner_do) *cc_runner_do; -static __typeof__(codec_chain_async_runner_do_nonblock) *cc_async_runner_do_nonblock; +static __typeof__(codec_chain_async_runner_do) *cc_async_runner_do; static __typeof__(codec_chain_client_codec_new) *cc_client_codec_new; static __typeof__(codec_chain_client_codec_free) *cc_client_codec_free; @@ -172,6 +172,12 @@ static union { codec_chain_async_runner *async; } cc_runners[CODEC_CHAIN_ID_MAX]; +static struct { + unsigned int async_busy; + unsigned int async_blocked; + unsigned int async_retry; +} cc_stats[CODEC_CHAIN_ID_MAX]; + typedef enum { CCC_OK, @@ -191,6 +197,7 @@ struct codec_cc_s { codec_chain_runner *runner; codec_chain_async_runner *async_runner; }; + __typeof(&cc_stats[0]) stats; codec_chain_def *def; codec_chain_codec *codec; @@ -1473,8 +1480,15 @@ static void cc_dlsym_resolve(const char *fn) { cc_runner_do = dlsym_assert(cc_lib_handle, "codec_chain_runner_do", fn); - cc_async_runner_do_nonblock = dlsym_assert(cc_lib_handle, - "codec_chain_async_runner_do_nonblock", fn); + if (!rtpe_common_config_ptr->codec_chain_nonblock) { + cc_async_runner_do = dlsym_assert(cc_lib_handle, + "codec_chain_async_runner_do", fn); + } + else { + __typeof__(codec_chain_async_runner_do_nonblock) *nb = dlsym_assert(cc_lib_handle, + "codec_chain_async_runner_do_nonblock", fn); + cc_async_runner_do = nb; + } cc_client_codec_new = dlsym_assert(cc_lib_handle, "codec_chain_client_codec_new", fn); @@ -1561,7 +1575,7 @@ void cc_init_chain(codec_def_t *src, format_t *src_format, codec_def_t *dst, return; cc_runners[id].async = cc_client_async_runner_new(cc_client, id, rtpe_common_config_ptr->codec_chain_async, - 10000, + rtpe_common_config_ptr->codec_chain_interval, rtpe_common_config_ptr->codec_chain_runners, rtpe_common_config_ptr->codec_chain_concurrency); if (cc_runners[id].async) @@ -1573,7 +1587,7 @@ void cc_init_chain(codec_def_t *src, format_t *src_format, codec_def_t *dst, if (cc_runners[id].sync) return; cc_runners[id].sync = cc_client_runner_new(cc_client, id, - 10000, + rtpe_common_config_ptr->codec_chain_interval, rtpe_common_config_ptr->codec_chain_runners, rtpe_common_config_ptr->codec_chain_concurrency); if (cc_runners[id].sync) @@ -5073,13 +5087,16 @@ static bool __cc_async_check_busy_blocked_queue(codec_cc_t *c, const str *data, return false; } + atomic_inc_na(&c->stats->async_busy); + // codec is busy (either currently running or was blocked) // append to queue __cc_async_do_add_queue(c, data, ts, async_cb_obj); if (c->async_jobs.length > 20) { - ilog(LOG_WARN | LOG_FLAG_LIMIT, "Async job queue overflow (%u), dropping frames", - c->async_jobs.length); + ilog(LOG_WARN | LOG_FLAG_LIMIT, "Async job queue overflow (%u @ %s), dropping frames", + c->async_jobs.length, + c->def->name); do { __auto_type jj = t_queue_pop_head(&c->async_jobs); t_queue_push_tail(&overflow, jj); @@ -5098,8 +5115,11 @@ static bool __cc_async_check_busy_blocked_queue(codec_cc_t *c, const str *data, } if (j) { + atomic_inc_na(&c->stats->async_retry); + if (!run_async(c, &j->data, j->ts, j->async_cb_obj)) { // still blocked. return to queue + atomic_inc_na(&c->stats->async_blocked); LOCK(&c->async_lock); t_queue_push_head(&c->async_jobs, j); } @@ -5201,7 +5221,9 @@ static bool __cc_run_async(codec_cc_t *c, const str *data, unsigned long ts, voi c->ts = ts; c->async_cb_obj = async_cb_obj; - return cc_async_runner_do_nonblock(c->async_runner, c->codec, + return cc_async_runner_do(&c->async_runner->runner, + &c->async_runner->async, + c->codec, (unsigned char *) data->s, data->len, pkt->data, pkt->size, cc_run_callback, c); } @@ -5256,6 +5278,7 @@ static codec_cc_t *codec_cc_new_sync(codec_def_t *src, format_t *src_format, cod ret->clear = cc_clear; ret->clear_arg = ret->codec; ret->runner = cc_runners[id].sync; + ret->stats = &cc_stats[id]; ret->avpkt = av_packet_alloc(); ret->run = cc_run; @@ -5305,6 +5328,7 @@ static codec_cc_t *codec_cc_new_async(codec_def_t *src, format_t *src_format, co ret->clear = cc_clear; ret->clear_arg = ret->codec; ret->async_runner = cc_runners[id].async; + ret->stats = &cc_stats[id]; ret->run = cc_run_async; ret->avpkt_async = av_packet_alloc(); av_new_packet(ret->avpkt_async, 2048); @@ -5354,6 +5378,68 @@ void codec_cc_free(codec_cc_t **ccp) { } +codec_cc_stats_q codec_cc_stats(void) { + codec_cc_stats_q ret = TYPED_GQUEUE_INIT; + + for (unsigned int i = 0; i < CODEC_CHAIN_ID_MAX; i++) { + codec_chain_runner *r; + + if (rtpe_common_config_ptr->codec_chain_async) { + if (!cc_runners[i].async) + continue; + r = &cc_runners[i].async->runner; + } + else { + if (!cc_runners[i].sync) + continue; + r = cc_runners[i].sync; + } + + __auto_type q = r->queuer; + + __auto_type e = g_new0(codec_cc_stats_entry, 1); + t_queue_push_tail(&ret, e); + + g_strlcpy(e->name, r->def->name, sizeof(e->name)); + +#define LA(v) e->v = atomic_get_na(&cc_stats[i].v) + LA(async_busy); + LA(async_blocked); + LA(async_retry); +#undef LA + + for (unsigned int j = 0; j < r->num_contexts; j++) { + __auto_type c = g_new0(codec_cc_context_stats, 1); + t_queue_push_tail(&e->contexts, c); + + c->ctx_idx = j; + +#define LA(v) c->v = atomic_get_na(&q->contexts[j].v) + LA(runs); + LA(slots); + LA(run_wait); + LA(writers_wait); + LA(compute_wait); + LA(readers_wait); + + LA(run_busy); + LA(write_busy); + LA(slots_full); + LA(buf_full); + + LA(ready_wait); + LA(callbacks_preempt); + LA(callbacks_fetch); + LA(callbacks_run); + LA(loop_barrier); +#undef LA + } + } + + return ret; +} + + #endif AVPacket *codec_cc_input_data(codec_cc_t *c, const str *data, unsigned long ts, void *x, void *y, void *z) { diff --git a/lib/codeclib.h b/lib/codeclib.h index e4ca14c6c..d064c2219 100644 --- a/lib/codeclib.h +++ b/lib/codeclib.h @@ -454,6 +454,43 @@ void frame_fill_dtmf_samples(enum AVSampleFormat fmt, void *samples, unsigned in #ifdef HAVE_CODEC_CHAIN +typedef struct { + unsigned int ctx_idx; + + unsigned int runs; + unsigned int slots; + uint64_t run_wait; + uint64_t writers_wait; + uint64_t compute_wait; + uint64_t readers_wait; + + unsigned int run_busy; + unsigned int write_busy; + unsigned int slots_full; + unsigned int buf_full; + + uint64_t ready_wait; + uint64_t callbacks_preempt; + uint64_t callbacks_fetch; + uint64_t callbacks_run; + uint64_t loop_barrier; +} codec_cc_context_stats; + +TYPED_GQUEUE(codec_cc_context, codec_cc_context_stats); + +typedef struct { + char name[32]; + + unsigned int async_busy; + unsigned int async_blocked; + unsigned int async_retry; + + codec_cc_context_q contexts; +} codec_cc_stats_entry; + +TYPED_GQUEUE(codec_cc_stats, codec_cc_stats_entry); + + extern codec_cc_t *(*codec_cc_new)(codec_def_t *src, format_t *src_format, codec_def_t *dst, format_t *dst_format, int bitrate, int ptime, void *(*init_async)(void *, void *, void *), @@ -463,6 +500,8 @@ void cc_init_chain(codec_def_t *src, format_t *src_format, codec_def_t *dst, void codec_cc_stop(codec_cc_t *); void codec_cc_free(codec_cc_t **); +codec_cc_stats_q codec_cc_stats(void); + #else INLINE codec_cc_t *codec_cc_new(codec_def_t *src, format_t *src_format, codec_def_t *dst,