{
void *pop;
switch_queue_t *q = (switch_queue_t *) obj;
+ int my_id;
+
+ for (my_id = 0; my_id < mod_sofia_globals.msg_queue_len; my_id++) {
+ if (mod_sofia_globals.msg_queue[my_id] == q) {
+ break;
+ }
+ }
- switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "MSG Thread Started\n");
+
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "MSG Thread %d Started\n", my_id);
+
+#ifdef HAVE_CPU_SET_MACROS
+ {
+ cpu_set_t set;
+ CPU_ZERO(&set);
+ CPU_SET(my_id, &set);
+ sched_setaffinity(0, sizeof(set), &set);
+ }
+#endif
while(switch_queue_pop(q, &pop) == SWITCH_STATUS_SUCCESS && pop) {
return NULL;
}
-static int IDX = 0;
-
void sofia_msg_thread_start(int idx)
{
- if (idx >= SOFIA_MAX_MSG_QUEUE || (idx < mod_sofia_globals.msg_queue_len && mod_sofia_globals.msg_queue_thread[idx])) {
+ if (idx >= mod_sofia_globals.max_msg_queues ||
+ idx >= SOFIA_MAX_MSG_QUEUE || (idx < mod_sofia_globals.msg_queue_len && mod_sofia_globals.msg_queue_thread[idx])) {
return;
}
switch_threadattr_create(&thd_attr, mod_sofia_globals.pool);
switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE);
- //switch_threadattr_priority_increase(thd_attr);
+ switch_threadattr_priority_increase(thd_attr);
switch_thread_create(&mod_sofia_globals.msg_queue_thread[i],
thd_attr,
sofia_msg_thread_run,
static void sofia_queue_message(sofia_dispatch_event_t *de)
{
- int idx = 0;
+ int idx = 0, queued = 0;
- if (mod_sofia_globals.running == 0) {
+ if (mod_sofia_globals.running == 0 || !mod_sofia_globals.msg_queue[0]) {
sofia_process_dispatch_event(&de);
return;
}
again:
- switch_mutex_lock(mod_sofia_globals.mutex);
- idx = IDX;
- IDX++;
- if (IDX >= mod_sofia_globals.msg_queue_len) IDX = 0;
- switch_mutex_unlock(mod_sofia_globals.mutex);
-
- sofia_msg_thread_start(idx);
+ for (idx = 0; idx < mod_sofia_globals.msg_queue_len; idx++) {
+ if (switch_queue_trypush(mod_sofia_globals.msg_queue[idx], de) == SWITCH_STATUS_SUCCESS) {
+ queued++;
+ break;
+ }
+ }
- if (switch_queue_trypush(mod_sofia_globals.msg_queue[idx], de) != SWITCH_STATUS_SUCCESS) {
- if (mod_sofia_globals.msg_queue_len < SOFIA_MAX_MSG_QUEUE) {
- sofia_msg_thread_start(idx + 1);
+ if (!queued) {
+
+ if (mod_sofia_globals.msg_queue_len < mod_sofia_globals.max_msg_queues) {
+ sofia_msg_thread_start(mod_sofia_globals.msg_queue_len + 1);
goto again;
- } else {
- switch_queue_push(mod_sofia_globals.msg_queue[idx], de);
}
+
+ switch_queue_push(mod_sofia_globals.msg_queue[0], de);
}
+
}
NUTAG_AUTOACK(0),
NUTAG_AUTOALERT(0),
NUTAG_ENABLEMESSENGER(1),
+ NTATAG_EXTRA_100(0),
TAG_IF((profile->mflags & MFLAG_REGISTER), NUTAG_ALLOW("REGISTER")),
TAG_IF((profile->mflags & MFLAG_REFER), NUTAG_ALLOW("REFER")),
TAG_IF(!sofia_test_pflag(profile, PFLAG_DISABLE_100REL), NUTAG_ALLOW("PRACK")),
mod_sofia_globals.debug_sla = atoi(val);
} else if (!strcasecmp(var, "auto-restart")) {
mod_sofia_globals.auto_restart = switch_true(val);
- } else if (!strcasecmp(var, "message-threads")) {
- int num = atoi(val);
-
- if (num < 1 || num > SOFIA_MAX_MSG_QUEUE - 1) {
- switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "message-threads must be between 1 and %d", SOFIA_MAX_MSG_QUEUE -1);
- }
-
- if (num < 1) num = 1;
- if (num > SOFIA_MAX_MSG_QUEUE - 1) num = SOFIA_MAX_MSG_QUEUE -1;
-
- sofia_msg_thread_start(num);
-
} else if (!strcasecmp(var, "reg-deny-binding-fetch-and-no-lookup")) { /* backwards compatibility */
mod_sofia_globals.reg_deny_binding_fetch_and_no_lookup = switch_true(val); /* remove when noone complains about the extra lookup */
if (switch_true(val)) {
nua_respond(nh, 400, "Missing Contact Header", TAG_END());
goto fail;
}
-
+
sofia_glue_get_addr(de->data->e_msg, network_ip, sizeof(network_ip), &network_port);
if (sofia_test_pflag(profile, PFLAG_AGGRESSIVE_NAT_DETECTION)) {
int bind;
};
-#define MAX_DISPATCH_VAL 20
+#define MAX_DISPATCH_VAL 64
static unsigned int MAX_DISPATCH = MAX_DISPATCH_VAL;
static unsigned int SOFT_MAX_DISPATCH = 0;
static char guess_ip_v4[80] = "";
EVENT_DISPATCH_QUEUE_RUNNING[my_id] = 1;
switch_mutex_unlock(EVENT_QUEUE_MUTEX);
+#ifdef HAVE_CPU_SET_MACROS
+ {
+ cpu_set_t set;
+ CPU_ZERO(&set);
+ CPU_SET(my_id, &set);
+ sched_setaffinity(0, sizeof(set), &set);
+ }
+#endif
+
+
for (;;) {
void *pop = NULL;
switch_event_t *event = NULL;
switch_queue_t *queue = (switch_queue_t *) obj;
uint32_t index = 0;
int my_id = 0;
- int auto_pause = 0;
switch_mutex_lock(EVENT_QUEUE_MUTEX);
THREAD_COUNT++;
for (;;) {
void *pop = NULL;
switch_event_t *event = NULL;
- int loops = 0;
-
- if (auto_pause) {
- if (!--auto_pause) {
- switch_core_session_ctl(SCSC_PAUSE_INBOUND, &auto_pause);
- } else {
- switch_cond_next();
- }
- }
if (switch_queue_pop(queue, &pop) != SWITCH_STATUS_SUCCESS) {
break;
while (event) {
-
- if (++loops > 2) {
- if (auto_pause) {
- switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Event system *still* overloading.\n");
- } else {
- switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT,
- "Event system overloading. Taking a 10 second break\n");
- auto_pause = 10;
- switch_core_session_ctl(SCSC_PAUSE_INBOUND, &auto_pause);
- }
- switch_yield(1000000);
- }
-
for (index = 0; index < SOFT_MAX_DISPATCH; index++) {
if (switch_queue_trypush(EVENT_DISPATCH_QUEUE[index], event) == SWITCH_STATUS_SUCCESS) {
event = NULL;
launch_dispatch_threads(SOFT_MAX_DISPATCH + 1, DISPATCH_QUEUE_LEN, RUNTIME_POOL);
switch_mutex_unlock(EVENT_QUEUE_MUTEX);
} else {
- switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Out of event dispatch threads! Slowing things down.\n");
- switch_yield(1000000);
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Out of event dispatch threads! Resorting to a blocking push.... Look for laggy event consumers or event_socket connections!\n");
+ switch_queue_push(EVENT_DISPATCH_QUEUE[0], event);
}
}
}
SWITCH_DECLARE(void) switch_core_memory_reclaim_events(void)
{
+ int index;
+
+ for (index = 0; index < SOFT_MAX_DISPATCH; index++) {
+ if (EVENT_DISPATCH_QUEUE[index]) {
+ printf("%d size: %u\n", index, switch_queue_size(EVENT_DISPATCH_QUEUE[index]));
+ }
+ }
+
+
#ifdef SWITCH_EVENT_RECYCLE
void *pop;
switch_mutex_unlock(EVENT_QUEUE_MUTEX);
for (x = 0; x < 3; x++) {
- switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "Stopping event queue %d\n", x);
- switch_queue_trypush(EVENT_QUEUE[x], NULL);
- switch_queue_interrupt_all(EVENT_QUEUE[x]);
+ if (EVENT_QUEUE[x]) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "Stopping event queue %d\n", x);
+ switch_queue_trypush(EVENT_QUEUE[x], NULL);
+ switch_queue_interrupt_all(EVENT_QUEUE[x]);
+ }
}
for (x = 0; x < SOFT_MAX_DISPATCH; x++) {
switch_event_t *event = NULL;
switch_status_t st;
- switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "Stopping queue thread %d\n", x);
- switch_thread_join(&st, EVENT_QUEUE_THREADS[x]);
+ if (EVENT_QUEUE_THREADS[x]) {
- while (switch_queue_trypop(EVENT_QUEUE[x], &pop) == SWITCH_STATUS_SUCCESS && pop) {
- event = (switch_event_t *) pop;
- switch_event_destroy(&event);
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "Stopping queue thread %d\n", x);
+ switch_thread_join(&st, EVENT_QUEUE_THREADS[x]);
+
+ while (switch_queue_trypop(EVENT_QUEUE[x], &pop) == SWITCH_STATUS_SUCCESS && pop) {
+ event = (switch_event_t *) pop;
+ switch_event_destroy(&event);
+ }
}
}
switch_threadattr_priority_increase(thd_attr);
switch_thread_create(&EVENT_DISPATCH_QUEUE_THREADS[index], thd_attr, switch_event_dispatch_thread, EVENT_DISPATCH_QUEUE[index], pool);
while(--sanity && !EVENT_DISPATCH_QUEUE_RUNNING[index]) switch_yield(10000);
- switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Create event dispatch thread %d\n", index);
+ if (index == 1) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Create event dispatch thread %d\n", index);
+ } else {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "Create additional event dispatch thread %d\n", index);
+ }
launched++;
break;
}
SWITCH_DECLARE(switch_status_t) switch_event_init(switch_memory_pool_t *pool)
{
- switch_threadattr_t *thd_attr;;
+ //switch_threadattr_t *thd_attr;
/*
This statement doesn't do anything commenting it out for now.
switch_assert(switch_arraylen(EVENT_NAMES) == SWITCH_EVENT_ALL + 1);
*/
+ /* don't need any more dispatch threads than we have CPU's*/
+ MAX_DISPATCH = switch_core_cpu_count();
+
switch_assert(pool != NULL);
THRUNTIME_POOL = RUNTIME_POOL = pool;
SYSTEM_RUNNING = -1;
switch_mutex_unlock(EVENT_QUEUE_MUTEX);
- switch_threadattr_create(&thd_attr, pool);
+ //switch_threadattr_create(&thd_attr, pool);
switch_find_local_ip(guess_ip_v4, sizeof(guess_ip_v4), NULL, AF_INET);
switch_find_local_ip(guess_ip_v6, sizeof(guess_ip_v6), NULL, AF_INET6);
- switch_queue_create(&EVENT_QUEUE[0], POOL_COUNT_MAX + 10, THRUNTIME_POOL);
- switch_queue_create(&EVENT_QUEUE[1], POOL_COUNT_MAX + 10, THRUNTIME_POOL);
- switch_queue_create(&EVENT_QUEUE[2], POOL_COUNT_MAX + 10, THRUNTIME_POOL);
+ //switch_queue_create(&EVENT_QUEUE[0], POOL_COUNT_MAX + 10, THRUNTIME_POOL);
+ //switch_queue_create(&EVENT_QUEUE[1], POOL_COUNT_MAX + 10, THRUNTIME_POOL);
+ //switch_queue_create(&EVENT_QUEUE[2], POOL_COUNT_MAX + 10, THRUNTIME_POOL);
#ifdef SWITCH_EVENT_RECYCLE
switch_queue_create(&EVENT_RECYCLE_QUEUE, 250000, THRUNTIME_POOL);
switch_queue_create(&EVENT_HEADER_RECYCLE_QUEUE, 250000, THRUNTIME_POOL);
#endif
- switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE);
- switch_threadattr_priority_increase(thd_attr);
+ //switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE);
+ //switch_threadattr_priority_increase(thd_attr);
launch_dispatch_threads(1, DISPATCH_QUEUE_LEN, RUNTIME_POOL);
- switch_thread_create(&EVENT_QUEUE_THREADS[0], thd_attr, switch_event_thread, EVENT_QUEUE[0], RUNTIME_POOL);
- switch_thread_create(&EVENT_QUEUE_THREADS[1], thd_attr, switch_event_thread, EVENT_QUEUE[1], RUNTIME_POOL);
- switch_thread_create(&EVENT_QUEUE_THREADS[2], thd_attr, switch_event_thread, EVENT_QUEUE[2], RUNTIME_POOL);
+ //switch_thread_create(&EVENT_QUEUE_THREADS[0], thd_attr, switch_event_thread, EVENT_QUEUE[0], RUNTIME_POOL);
+ //switch_thread_create(&EVENT_QUEUE_THREADS[1], thd_attr, switch_event_thread, EVENT_QUEUE[1], RUNTIME_POOL);
+ //switch_thread_create(&EVENT_QUEUE_THREADS[2], thd_attr, switch_event_thread, EVENT_QUEUE[2], RUNTIME_POOL);
while (!THREAD_COUNT) {
switch_cond_next();
SWITCH_DECLARE(switch_status_t) switch_event_fire_detailed(const char *file, const char *func, int line, switch_event_t **event, void *user_data)
{
- int index;
-
switch_assert(BLOCK != NULL);
switch_assert(RUNTIME_POOL != NULL);
switch_assert(EVENT_QUEUE_MUTEX != NULL);
(*event)->event_user_data = user_data;
}
+ if (!EVENT_QUEUE_THREADS[(*event)->priority] && (*event)->priority < 3) {
+ switch_threadattr_t *thd_attr;
+
+ switch_queue_create(&EVENT_QUEUE[(*event)->priority], POOL_COUNT_MAX + 10, THRUNTIME_POOL);
+ switch_threadattr_create(&thd_attr, THRUNTIME_POOL);
+ switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE);
+ switch_threadattr_priority_increase(thd_attr);
+ switch_thread_create(&EVENT_QUEUE_THREADS[(*event)->priority], thd_attr, switch_event_thread, EVENT_QUEUE[(*event)->priority], RUNTIME_POOL);
+ }
+
for (;;) {
- for (index = (*event)->priority; index < 3; index++) {
- if (switch_queue_trypush(EVENT_QUEUE[index], *event) == SWITCH_STATUS_SUCCESS) {
- goto end;
- }
+ if (switch_queue_trypush(EVENT_QUEUE[(*event)->priority], *event) == SWITCH_STATUS_SUCCESS) {
+ goto end;
}
+
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Event queue is full!\n");
switch_yield(100000);
}
projects / thirdparty / freeswitch.git / commitdiff
