Skip to content

Instantly share code, notes, and snippets.

@moises-silva

moises-silva/mod_bert.c

Created August 29, 2018 12:25
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save moises-silva/a09ee85f58caec8e2c2054eef961d258 to your computer and use it in GitHub Desktop.
Save moises-silva/a09ee85f58caec8e2c2054eef961d258 to your computer and use it in GitHub Desktop.
Download ZIP
Latest mod_bert version
Raw
mod_bert.c
/*
* FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application
* Copyright (C) 2005-2011, Anthony Minessale II <anthm@freeswitch.org>
*
* Version: MPL 1.1
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is FreeSWITCH Modular Media Switching Software Library / Oreka Recording Module
*
* The Initial Developer of the Original Code is
* Moises Silva <moises.silva@gmail.com>
* Portions created by the Initial Developer are Copyright (C)
* the Initial Developer. All Rights Reserved.
*
* Contributor(s):
* Moises Silva <moises.silva@gmail.com>
*
* mod_bert -- Naive BERT tester
*
*/
#include <switch.h>
SWITCH_MODULE_LOAD_FUNCTION(mod_bert_load);
SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_bert_shutdown);
SWITCH_MODULE_DEFINITION(mod_bert, mod_bert_load, mod_bert_shutdown, NULL);
#define BERT_CHANNEL_PVT_KEY "__bert__"
#define BERT_STATS_VAR_SYNC_LOST "bert_stats_sync_lost"
#define BERT_STATS_VAR_SYNC_LOST_CNT "bert_stats_sync_lost_count"
#define BERT_EVENT_TIMEOUT "mod_bert::timeout"
#define BERT_EVENT_LOST_SYNC "mod_bert::lost_sync"
#define BERT_EVENT_IN_SYNC "mod_bert::in_sync"
#define BERT_EVENT_DELAY_PROBE_FIRED "mod_bert::delay_probe_fired"
#define BERT_EVENT_DELAY_PROBE_RECEIVED "mod_bert::delay_probe_received"
#define BERT_DEFAULT_WINDOW_MS 1000
#define BERT_DEFAULT_MAX_ERR 10.0
#define BERT_DEFAULT_TIMEOUT_MS 10000
#define G711_ULAW_IDLE_OCTET 0xFF
/* http://en.wikipedia.org/wiki/Digital_milliwatt */
static unsigned char ulaw_digital_milliwatt[8] = { 0x1e, 0x0b, 0x0b, 0x1e, 0x9e, 0x8b, 0x8b, 0x9e };
/* Same as digital milliwatt except for the last byte to be used as a marker */
static unsigned char ulaw_digital_marker[8] = { 0x1e, 0x0b, 0x0b, 0x1e, 0x9e, 0x8b, 0x8b, 0xFF };
/* Bert realms are used to coordinate more than two bert sessions (e.g bert testing on a conference,
* where there's a single producer of the bert pattern and several consumers) */
static switch_mutex_t *g_realms_mutex = NULL;
static switch_hash_t *g_realms = NULL;
typedef struct {
char *generator_uuid;
switch_hash_t *sessions;
switch_time_t delay_probe_start;
switch_memory_pool_t *pool;
const char *name;
uint32_t session_count;
} bert_realm_t;
typedef struct {
uint64_t total_processed_samples;
uint64_t next_delay_probe_samples;
uint64_t recv_probe_next_samples;
uint32_t processed_samples;
uint32_t err_samples;
uint32_t window_ms;
uint32_t window_samples;
uint32_t stats_sync_lost_cnt;
uint32_t stats_cng_cnt;
uint32_t delay_probe_samples;
uint32_t recv_probe_frame_count;
uint32_t recv_probe_count;
switch_time_t recv_probe_time_ref;
switch_time_t last_probe_delay;
float max_err;
float max_err_hit;
float max_err_ever;
uint8_t in_sync;
uint8_t hangup_on_error;
uint8_t realm_generator;
switch_time_t timeout;
FILE *input_debug_f;
FILE *output_debug_f;
switch_timer_t timer;
bert_realm_t *realm;
switch_codec_implementation_t *read_impl;
} bert_t;
// Interval for every delay probe
#define BERT_DELAY_PROBE_SECONDS_INTERVAL 5
// 1 second probe (it should always be less than BERT_DELAY_PROBE_INTERVAL_SAMPLES or we'll be sending a probe constantly)
#define BERT_DELAY_PROBE_SAMPLE_LEN(bert) ((bert)->read_impl->samples_per_second / 2)
#define BERT_DELAY_PROBE_INTERVAL_SAMPLES(bert) ((bert)->read_impl->samples_per_second * BERT_DELAY_PROBE_SECONDS_INTERVAL)
#define init_frame(buf, pattern, len) \
do { \
uint32_t i; \
for (i = 0; i < len; i += sizeof(pattern)) { \
memcpy(&buf[i], pattern, sizeof(pattern)); \
} \
} while (0)
static void bert_prepare_samples(switch_core_session_t *session, bert_t *bert, uint8_t *write_samples, uint32_t len)
{
if (bert->realm) {
if (!bert->realm_generator) {
// just keep the silence buffer
return;
}
// we have a realm, and we're a generator, initialize the probe sample count if not done already
if (!bert->next_delay_probe_samples) {
bert->next_delay_probe_samples = BERT_DELAY_PROBE_INTERVAL_SAMPLES(bert);
}
// check if it's time to send a delay probe
if (bert->total_processed_samples >= bert->next_delay_probe_samples) {
switch_event_t *event;
switch_channel_t *channel = switch_core_session_get_channel(session);
// time to send another delay probe, setup the marker frame
init_frame(write_samples, ulaw_digital_marker, len);
// set the next probe interval
bert->next_delay_probe_samples = bert->total_processed_samples + BERT_DELAY_PROBE_INTERVAL_SAMPLES(bert);
// thread-safe only because we assume only one realm generator at the time
// and only the realm generator is allowed to write the delay probe start time
bert->realm->delay_probe_start = switch_time_ref();
bert->delay_probe_samples = 0;
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Starting bert delay probe\n");
if (switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, BERT_EVENT_DELAY_PROBE_FIRED) == SWITCH_STATUS_SUCCESS) {
switch_channel_event_set_basic_data(channel, event);
switch_event_fire(&event);
}
}
// If a probe is in progress, check if it's time to stop it
if (!memcmp(write_samples, ulaw_digital_marker, sizeof(ulaw_digital_marker))) {
// If the delay probe is completed, restore the milliwatt pattern
if (bert->delay_probe_samples >= BERT_DELAY_PROBE_SAMPLE_LEN(bert)) {
init_frame(write_samples, ulaw_digital_milliwatt, len);
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Bert delay probe done\n");
} else {
// if the delay probe is not finished yet, just increase the counter
bert->delay_probe_samples += len;
}
} else if (memcmp(write_samples, ulaw_digital_milliwatt, sizeof(ulaw_digital_milliwatt))) {
// no probe in progress, and our signal is not a milliwat yet, initialize it
init_frame(write_samples, ulaw_digital_milliwatt, len);
}
} else if (memcmp(write_samples, ulaw_digital_milliwatt, sizeof(ulaw_digital_milliwatt))) {
// when there's no realm, we just keep sending the same pattern always
init_frame(write_samples, ulaw_digital_milliwatt, len);
}
}
#define bert_close_debug_streams(bert, session) \
do { \
int rc = 0; \
if (bert.input_debug_f) { \
rc = fclose(bert.input_debug_f); \
if (rc) { \
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Failed to close BERT input debug file!\n"); \
} \
bert.input_debug_f = NULL; \
} \
if (bert.output_debug_f) { \
rc = fclose(bert.output_debug_f); \
if (rc) { \
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Failed to close BERT output debug file!\n"); \
} \
bert.output_debug_f = NULL; \
} \
} while (0);
SWITCH_STANDARD_APP(bert_test_function)
{
switch_status_t status;
switch_frame_t *read_frame = NULL, write_frame = { 0 };
switch_codec_implementation_t read_impl = { 0 };
switch_channel_t *channel = NULL;
switch_event_t *event = NULL;
const char *var = NULL;
int synced = 0;
int32_t timeout_ms = 0;
int32_t interval = 20;
int32_t samples = 0;
uint8_t *write_samples = NULL;
uint8_t *m = NULL;
uint8_t *probe_frame = NULL;
const char *timer_name = NULL;
bert_realm_t *realm = NULL;
switch_bool_t clean_frame = SWITCH_FALSE;
bert_t bert = { 0 };
channel = switch_core_session_get_channel(session);
switch_channel_answer(channel);
switch_core_session_get_read_impl(session, &read_impl);
if (read_impl.ianacode != 0) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "This application only works when using ulaw codec\n");
goto done;
}
bert.read_impl = &read_impl;
bert.window_ms = BERT_DEFAULT_WINDOW_MS;
bert.window_samples = switch_samples_per_packet(read_impl.samples_per_second, bert.window_ms);
bert.max_err = BERT_DEFAULT_MAX_ERR;
timeout_ms = BERT_DEFAULT_TIMEOUT_MS;
/* check if there are user-defined overrides */
if ((var = switch_channel_get_variable(channel, "bert_realm"))) {
// when a realm is defined, we need to select only one generator per realm
switch_mutex_lock(g_realms_mutex);
realm = switch_core_hash_find(g_realms, var);
if (!realm) {
switch_memory_pool_t *pool;
status = switch_core_new_memory_pool(&pool);
if (status == SWITCH_STATUS_SUCCESS) {
realm = switch_core_alloc(pool, sizeof(*realm));
realm->pool = pool;
realm->name = switch_core_strdup(pool, var);
// Note this is allocated from the session pool
// but it's fine since the generator session cannot be destroyed with the realm
// still pointing to it because we detach the generator uuid when the session goes out of the bert app
realm->generator_uuid = switch_core_session_get_uuid(session);
bert.realm_generator = 1;
switch_core_hash_init(&realm->sessions);
// the first session joining is the generator (if it hangs up, another one will be picked up)
switch_core_hash_insert(g_realms, realm->name, realm);
}
}
if (realm) {
bert.realm = realm;
switch_core_hash_insert(realm->sessions, switch_core_session_get_uuid(session), session);
switch_channel_set_private(channel, BERT_CHANNEL_PVT_KEY, &bert);
realm->session_count++;
}
switch_mutex_unlock(g_realms_mutex);
if (!realm) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Failed to initialize realm %s\n", var);
goto done;
}
}
if ((var = switch_channel_get_variable(channel, "bert_window_ms"))) {
int tmp = atoi(var);
if (tmp > 0) {
bert.window_ms = tmp;
bert.window_samples = switch_samples_per_packet(read_impl.samples_per_second, bert.window_ms);
}
}
if ((var = switch_channel_get_variable(channel, "bert_timeout_ms"))) {
int tmp = atoi(var);
if (tmp > 0) {
timeout_ms = tmp;
}
}
if ((var = switch_channel_get_variable(channel, "bert_max_err"))) {
double tmp = atoi(var);
if (tmp > 0) {
bert.max_err = (float)tmp;
}
}
if ((var = switch_channel_get_variable(channel, "bert_hangup_on_error"))) {
if (switch_true(var)) {
bert.hangup_on_error = 1;
}
}
if ((var = switch_channel_get_variable(channel, "bert_debug_io_file"))) {
char debug_file[1024];
snprintf(debug_file, sizeof(debug_file), "%s.in", var);
bert.input_debug_f = fopen(debug_file, "w");
if (!bert.input_debug_f) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Failed to open input debug file %s\n", debug_file);
}
snprintf(debug_file, sizeof(debug_file), "%s.out", var);
bert.output_debug_f = fopen(debug_file, "w");
if (!bert.output_debug_f) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Failed to open output debug file %s\n", debug_file);
}
}
if ((var = switch_channel_get_variable(channel, "bert_timer_name"))) {
timer_name = var;
}
/* Setup the timer, so we can send audio at correct time frames even if we do not receive audio */
if (timer_name) {
interval = read_impl.microseconds_per_packet / 1000;
samples = switch_samples_per_packet(read_impl.samples_per_second, interval);
if (switch_core_timer_init(&bert.timer, timer_name, interval, samples, switch_core_session_get_pool(session)) == SWITCH_STATUS_SUCCESS) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Setup timer success interval: %u samples: %u\n", interval, samples);
} else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Timer Setup Failed. BERT cannot start!\n");
goto done;
}
switch_core_timer_sync(&bert.timer);
}
bert.timeout = (switch_micro_time_now() + (timeout_ms * 1000));
write_frame.codec = switch_core_session_get_read_codec(session);
write_frame.data = switch_core_session_alloc(session, SWITCH_RECOMMENDED_BUFFER_SIZE);
write_frame.buflen = SWITCH_RECOMMENDED_BUFFER_SIZE;
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "BERT Test Window=%ums/%u, MaxErr=%f%%, Timeout=%dms\n", bert.window_ms, bert.window_samples, bert.max_err, timeout_ms);
if (bert.window_samples <= 0) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Failed to compute BERT window samples!\n");
goto done;
}
switch_channel_set_variable(channel, BERT_STATS_VAR_SYNC_LOST_CNT, "0");
switch_channel_set_variable(channel, BERT_STATS_VAR_SYNC_LOST, "false");
write_samples = write_frame.data;
write_frame.datalen = read_impl.encoded_bytes_per_packet;
write_frame.samples = read_impl.samples_per_packet;
memset(write_samples, G711_ULAW_IDLE_OCTET, read_impl.encoded_bytes_per_packet);
probe_frame = switch_core_session_alloc(session, read_impl.encoded_bytes_per_packet);
init_frame(probe_frame, ulaw_digital_marker, read_impl.encoded_bytes_per_packet);
for (;;) {
if (!switch_channel_ready(channel)) {
break;
}
bert_prepare_samples(session, &bert, write_samples, read_impl.samples_per_packet);
switch_ivr_parse_all_events(session);
if (timer_name) {
if (switch_core_timer_next(&bert.timer) != SWITCH_STATUS_SUCCESS) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Failed to step on timer!\n");
break;
}
/* the playback() app does not set write_frame.timestamp unless a timer is used, what's the catch? does it matter? */
write_frame.timestamp = bert.timer.samplecount;
}
if (bert.output_debug_f) {
fwrite(write_frame.data, write_frame.datalen, 1, bert.output_debug_f);
}
status = switch_core_session_write_frame(session, &write_frame, SWITCH_IO_FLAG_NONE, 0);
if (!SWITCH_READ_ACCEPTABLE(status)) {
break;
}
/* Proceed to read and process the received frame ...
* Note that switch_core_session_read_frame is a blocking operation, we could do reathing in another thread like the playback() app
* does using switch_core_service_session() but OTOH that would lead to more load/cpu usage, extra threads being launched per call leg
* and most likely reduce the overall capacity of the test system */
status = switch_core_session_read_frame(session, &read_frame, SWITCH_IO_FLAG_NONE, 0);
if (!SWITCH_READ_ACCEPTABLE(status)) {
break;
}
// realm generators are not subject to syncing since they are expected to just generate, not sync/detect patterns
// which is mostly useful in conference room testing (only one speaker at the time)
if (bert.realm_generator) {
// skip all reading
goto sync_check_done;
}
if (bert.timeout && !synced) {
switch_time_t now = switch_micro_time_now();
if (now >= bert.timeout) {
bert.timeout = 0;
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_WARNING, "BERT Timeout (read_samples=%d, read_bytes=%d, expected_samples=%d, session=%s)\n",
read_frame->samples, read_frame->datalen, read_impl.samples_per_packet, switch_core_session_get_uuid(session));
if (switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, BERT_EVENT_TIMEOUT) == SWITCH_STATUS_SUCCESS) {
switch_channel_event_set_basic_data(channel, event);
switch_event_fire(&event);
}
if (bert.hangup_on_error) {
switch_channel_hangup(channel, SWITCH_CAUSE_MEDIA_TIMEOUT);
}
}
}
/* Treat CNG as silence */
if (switch_test_flag(read_frame, SFF_CNG)) {
read_frame->samples = read_impl.samples_per_packet;
read_frame->datalen = read_impl.samples_per_packet;
memset(read_frame->data, G711_ULAW_IDLE_OCTET, read_frame->datalen);
bert.stats_cng_cnt++;
}
if (read_frame->samples != read_impl.samples_per_packet || !read_frame->datalen) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_WARNING, "Read %d samples, expected %d!\n", read_frame->samples, read_impl.samples_per_packet);
continue;
}
if (bert.input_debug_f) {
size_t ret = fwrite(read_frame->data, read_frame->datalen, 1, bert.input_debug_f);
if (ret != 1) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Failed to write to BERT input debug file!\n");
}
}
if (bert.window_samples == bert.processed_samples) {
/* BERT err rate calculation */
float err = 0.0;
/* If the channel is going down, then it is expected we'll have errors, ignore them and bail out */
if (!switch_channel_ready(channel)) {
bert_close_debug_streams(bert, session);
break;
}
/* Calculate error rate */
err = ((float)((float)bert.err_samples / (float)bert.processed_samples) * 100.0);
if (err > bert.max_err) {
if (bert.in_sync) {
bert.in_sync = 0;
bert.stats_sync_lost_cnt++;
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_WARNING, "BERT Sync Lost: %f%% loss (count=%u, cng_count=%d, err_samples=%u, session=%s)\n",
err, bert.stats_sync_lost_cnt, bert.stats_cng_cnt, bert.err_samples, switch_core_session_get_uuid(session));
switch_channel_set_variable_printf(channel, BERT_STATS_VAR_SYNC_LOST_CNT, "%u", bert.stats_sync_lost_cnt);
switch_channel_set_variable(channel, BERT_STATS_VAR_SYNC_LOST, "true");
if (switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, BERT_EVENT_LOST_SYNC) == SWITCH_STATUS_SUCCESS) {
switch_channel_event_set_basic_data(channel, event);
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "sync_lost_percent", "%f", err);
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "sync_lost_count", "%u", bert.stats_sync_lost_cnt);
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "cng_count", "%u", bert.stats_cng_cnt);
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "err_samples", "%u", bert.err_samples);
switch_event_fire(&event);
}
if (bert.hangup_on_error) {
switch_channel_hangup(channel, SWITCH_CAUSE_MEDIA_TIMEOUT);
bert_close_debug_streams(bert, session);
}
}
} else if (!bert.in_sync) {
bert.in_sync = 1;
synced = 1;
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_WARNING, "BERT Sync Success\n");
bert.stats_cng_cnt = 0;
bert.timeout = 0;
if (switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, BERT_EVENT_IN_SYNC) == SWITCH_STATUS_SUCCESS) {
switch_channel_event_set_basic_data(channel, event);
switch_event_fire(&event);
}
}
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG10, "Err=%f%% (%u/%u)\n", err, bert.err_samples, bert.processed_samples);
if (synced && err > bert.max_err_hit) {
bert.max_err_hit = err;
}
if (err > bert.max_err_ever) {
bert.max_err_ever = err;
}
bert.processed_samples = 0;
bert.err_samples = 0;
}
if ((bert.in_sync || clean_frame) &&
!memcmp(read_frame->data, write_frame.data, read_frame->datalen)) {
goto sync_check_done;
}
/* We're not in sync, or we might be going out of sync */
/* Perform delay probe detection first if we're in sync */
if (realm && bert.in_sync && bert.total_processed_samples >= bert.recv_probe_next_samples) {
m = memmem(read_frame->data, read_frame->datalen, ulaw_digital_marker, sizeof(ulaw_digital_marker));
if (m) {
// we want to wait for a full frame to calculate the delay, but we want the time of the first
// marker to be there to try to be as accurate as possible without easily having false positives
uint8_t full_match = !memcmp(read_frame->data, probe_frame, read_frame->datalen) ? 1 : 0;
clean_frame = SWITCH_FALSE;
bert.recv_probe_frame_count++;
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG10, "Detected probe (samples=%lu, probe_next_samples=%lu)\n",
bert.total_processed_samples, bert.recv_probe_next_samples);
if (full_match) {
// XXX FIXME: Not thread safe XXX
// The delay probe start reference could be modified by the
// realm generator while we read it, but in practice
// that would only happen if the delay is so massive that
// the generator started a new probe and we're just receiving
// the previous probe. We should add a lock to the realm anyways ...
switch_event_t *event = NULL;
switch_time_t ref = realm->delay_probe_start;
if (bert.recv_probe_frame_count == 1) {
// lucky, full match on the first frame
bert.last_probe_delay = switch_time_ref() - ref;
} else if (bert.recv_probe_frame_count == 2) {
bert.last_probe_delay = bert.recv_probe_time_ref - ref;
}
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Measured delay: %lums\n", (bert.last_probe_delay / 1000));
// Do not start measuring delay again until the next interval (-1 second to not risk race conditions and try to be as accurate as possible)
bert.recv_probe_next_samples = bert.total_processed_samples + BERT_DELAY_PROBE_INTERVAL_SAMPLES(&bert) - read_impl.samples_per_second;
bert.recv_probe_count++;
// TODO: Add some sort of sequence number to the probe or some other id to make sure this is not an 'old' one??
if (switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, BERT_EVENT_DELAY_PROBE_RECEIVED) == SWITCH_STATUS_SUCCESS) {
switch_channel_event_set_basic_data(channel, event);
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "delay_us", "%lu", bert.last_probe_delay);
switch_event_fire(&event);
}
} else {
// not a match, if it's our first probe count, give it a break
// but if it's more than that then ignore, and restart the logic
if (bert.recv_probe_frame_count == 1) {
bert.recv_probe_time_ref = switch_time_ref();
} else {
bert.recv_probe_frame_count = 0;
}
}
goto sync_check_done;
} else {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG10, "No probe detected (samples=%lu, probe_next_samples=%lu)\n",
bert.total_processed_samples, bert.recv_probe_next_samples);
bert.recv_probe_frame_count = 0;
bert.recv_probe_time_ref = 0;
}
}
/* find the start of the milliwat pattern ... */
m = memmem(read_frame->data, read_frame->datalen, ulaw_digital_milliwatt, sizeof(ulaw_digital_milliwatt));
if (m) {
/* At least some bytes matched, let's find out the err sample count (could be zero if we're lucky and the whole frame matches) */
uint8_t *end = NULL;
size_t left = 0;
int cerrs = bert.err_samples;
/* Pattern found at least once in the frame, let's check if the rest of the frame also matches */
m += sizeof(ulaw_digital_milliwatt);
end = (uint8_t *)read_frame->data + read_frame->datalen;
left = (size_t)(end - m);
if (left && !memcmp(m, write_frame.data, left)) {
bert.err_samples += (m - (uint8_t *)read_frame->data - sizeof(ulaw_digital_milliwatt));
} else if (left) {
int s = 0;
bert.err_samples += (m - (uint8_t *)read_frame->data - sizeof(ulaw_digital_milliwatt));
/* count error samples */
for (s = 0; m != end; m++, s++) {
if (ulaw_digital_milliwatt[s%8] != *m) {
bert.err_samples++;
}
}
}
clean_frame = (cerrs == bert.err_samples) ? SWITCH_TRUE : SWITCH_FALSE;
} else {
/* the pattern was not found in the whole frame, then the whole frame is out of sync */
bert.err_samples += read_frame->samples;
clean_frame = SWITCH_FALSE;
}
sync_check_done:
bert.processed_samples += read_frame->samples;
bert.total_processed_samples += read_frame->samples;
}
done:
bert_close_debug_streams(bert, session);
if (bert.timer.interval) {
switch_core_timer_destroy(&bert.timer);
}
if (bert.realm) {
char *session_id = switch_core_session_get_uuid(session);
switch_mutex_lock(g_realms_mutex);
switch_core_hash_delete(realm->sessions, session_id);
realm->session_count--;
if (realm->session_count) {
// if the session is the generator and we still have sessions left, select a new generator
if (bert.realm_generator) {
const void *key;
void *val;
switch_hash_index_t *hi = switch_core_hash_first(realm->sessions);
bert.realm_generator = 0;
if (hi) {
switch_core_session_t *nsession = NULL;
switch_channel_t *nchannel = NULL;
bert_t *bert_data = NULL;
switch_core_hash_this(hi, &key, NULL, &val);
nsession = val;
nchannel = switch_core_session_get_channel(nsession);
bert_data = switch_channel_get_private(nchannel, BERT_CHANNEL_PVT_KEY);
switch_assert(bert_data);
realm->generator_uuid = switch_core_session_get_uuid(nsession);
bert_data->realm_generator = 1;
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "New generator for realm %s: %s\n", realm->name, realm->generator_uuid);
switch_safe_free(hi);
} else {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Failed to select a new generator for realm %s\n", realm->name);
realm->generator_uuid = NULL;
}
}
} else {
switch_memory_pool_t *pool = realm->pool;
realm->pool = NULL;
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Destroying bert realm %s\n", realm->name);
switch_core_hash_destroy(&realm->sessions);
switch_core_hash_delete(g_realms, realm->name);
switch_core_destroy_memory_pool(&pool);
}
switch_mutex_unlock(g_realms_mutex);
bert.realm = NULL;
}
switch_channel_set_private(channel, BERT_CHANNEL_PVT_KEY, NULL);
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "BERT Test Completed. MaxErr=%f%%\n", synced ? bert.max_err_hit : bert.max_err_ever);
}
SWITCH_MODULE_LOAD_FUNCTION(mod_bert_load)
{
switch_application_interface_t *app_interface = NULL;
if (switch_event_reserve_subclass(BERT_EVENT_TIMEOUT) != SWITCH_STATUS_SUCCESS) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't register subclass %s!\n", BERT_EVENT_TIMEOUT);
return SWITCH_STATUS_TERM;
}
if (switch_event_reserve_subclass(BERT_EVENT_LOST_SYNC) != SWITCH_STATUS_SUCCESS) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't register subclass %s!\n", BERT_EVENT_LOST_SYNC);
return SWITCH_STATUS_TERM;
}
if (switch_event_reserve_subclass(BERT_EVENT_IN_SYNC) != SWITCH_STATUS_SUCCESS) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't register subclass %s!\n", BERT_EVENT_IN_SYNC);
return SWITCH_STATUS_TERM;
}
switch_mutex_init(&g_realms_mutex, SWITCH_MUTEX_DEFAULT, pool);
switch_core_hash_init(&g_realms);
*module_interface = switch_loadable_module_create_module_interface(pool, modname);
SWITCH_ADD_APP(app_interface, "bert_test", "Start BERT Test", "Start BERT Test", bert_test_function, "", SAF_NONE);
return SWITCH_STATUS_SUCCESS;
}
SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_bert_shutdown)
{
switch_event_free_subclass(BERT_EVENT_TIMEOUT);
switch_event_free_subclass(BERT_EVENT_LOST_SYNC);
switch_event_free_subclass(BERT_EVENT_IN_SYNC);
switch_core_hash_destroy(&g_realms);
return SWITCH_STATUS_UNLOAD;
}
/* For Emacs:
* Local Variables:
* mode:c
* indent-tabs-mode:t
* tab-width:4
* c-basic-offset:4
* End:
* For VIM:
* vim:set softtabstop=4 shiftwidth=4 tabstop=4:
*/
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment