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

MT#55283 update to new CC ABI

Browse Source 
Change-Id: Iab7a930890d9b75d1aa08c2c5f122fc0e2180ebb
pull/2123/head
Richard Fuchs 2 months ago
parent 154f4bf37c
commit 076e22509e
6 changed files with 232 additions and 17 deletions
Show Stats Download Patch File Download Diff File

99
daemon/statistics.c
Unescape View File

@ -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;

2
debian/control vendored
Unescape View File

@ -48,7 +48,7 @@ Build-Depends:
libwebsockets-dev,
libwww-perl,
libxtables-dev (>= 1.4) | iptables-dev (>= 1.4),
ngcp-libcodec-chain-dev (>= 13.3) <pkg.ngcp-rtpengine.codec-chain>,
ngcp-libcodec-chain-dev (>= 26.1) <pkg.ngcp-rtpengine.codec-chain>,
pandoc,
pkgconf,
pybuild-plugin-pyproject <pkg.ngcp-rtpengine.pysip-lite>,

5
lib/auxlib.c
Unescape View File

@ -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)

2
lib/auxlib.h
Unescape View File

@ -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;
};

102
lib/codeclib.c
Unescape View File

@ -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) {

39
lib/codeclib.h
Unescape View File

@ -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,

Write Preview
Loading…
Cancel
Save