ben.carrasco/freeswitch
1
0
Fork 0
mirror of https://github.com/signalwire/freeswitch.git synced 2026-07-24 13:12:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Addition of scheduler engine and a few applications to use it.

Browse Source 
This patch adds a scheduler thread to the core and moves the heartbeat
event to use the new scheduler as an example.

Also The following features are implemented that use this scheduler:

sched_hangup dialplan application:

<action application="sched_hangup" data="+10 normal_clearing bleg"/>

** The cause code is optional and the optional bleg keyword will only hangup the
   channel the current channel is bridged to if the call is in a bridge.

sched_transfer dialplan application:

<action application="sched_transfer" data="+10 1000 XML default"/>

** The last 2 args (dialplan and context) are optional

sched_broadcast dialplan application:

<action application="sched_broadcast" data="+10 playback:/tmp/foo.wav"/>
<action application="sched_broadcast" data="+10 playback!normal_clearing:/tmp/foo.wav"/>

** The optional !<cause_code> can be added to make the channel hangup after broadcasting the file.


sched_hangup api function:

sched_hangup +10 <uuid_string> normal_clearing

** The cause code is optional

sched_transfer api function:

sched_transfer +10 <uuid_string> 1000 XML default

** The last 2 args (dialplan and context) are optional

sched_broadcast api function:

sched_broadcast +10 <uuid_str> playback:/tmp/foo.wav
sched_broadcast +10 <uuid_str> playback!normal_clearing:/tmp/foo.wav

** The optional !<cause_code> can be added to make the channel hangup after broadcasting the file.

The new C functions in the core are documented in the doxeygen.

*NOTE* This commit should satisfy at least 2 bounties on the wiki



git-svn-id: http://svn.freeswitch.org/svn/freeswitch/trunk@4785 d0543943-73ff-0310-b7d9-9358b9ac24b2
This commit is contained in:
Anthony Minessale
2007-03-28 23:37:12 +00:00
parent 24b1a46ceb
commit 8a4406ece2
9 changed files with 855 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/switch_core.c
+279 -38
View File
@@ -147,6 +147,19 @@ struct switch_core_session {
SWITCH_DECLARE_DATA switch_directories SWITCH_GLOBAL_dirs = {0};
struct switch_core_scheduler_task_container {
switch_core_scheduler_task_t task;
time_t executed;
int in_thread;
int destroyed;
switch_core_scheduler_func_t func;
switch_memory_pool_t *pool;
uint32_t flags;
char *desc;
struct switch_core_scheduler_task_container *next;
};
typedef struct switch_core_scheduler_task_container switch_core_scheduler_task_container_t;
struct switch_core_runtime {
switch_time_t initiated;
uint32_t session_id;
@@ -169,6 +182,10 @@ struct switch_core_runtime {
uint32_t shutting_down;
uint8_t running;
char uuid_str[SWITCH_UUID_FORMATTED_LENGTH + 1];
switch_core_scheduler_task_container_t *task_list;
switch_mutex_t *task_mutex;
uint32_t task_id;
int task_thread_running;
};
/* Prototypes */
@@ -196,6 +213,238 @@ static void db_pick_path(char *dbname, char *buf, switch_size_t size)
}
}
static void send_heartbeat(void)
{
switch_event_t *event;
switch_core_time_duration_t duration;
switch_core_measure_time(switch_core_uptime(), &duration);
if (switch_event_create(&event, SWITCH_EVENT_HEARTBEAT) == SWITCH_STATUS_SUCCESS) {
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "Event-Info", "System Ready");
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "Up-Time",
"%u year%s, "
"%u day%s, "
"%u hour%s, "
"%u minute%s, "
"%u second%s, "
"%u millisecond%s, "
"%u microsecond%s\n",
duration.yr, duration.yr == 1 ? "" : "s",
duration.day, duration.day == 1 ? "" : "s",
duration.hr, duration.hr == 1 ? "" : "s",
duration.min, duration.min == 1 ? "" : "s",
duration.sec, duration.sec == 1 ? "" : "s",
duration.ms, duration.ms == 1 ? "" : "s",
duration.mms, duration.mms == 1 ? "" : "s"
);
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "Session-Count", "%u", switch_core_session_count());
switch_event_fire(&event);
}
}
static void heartbeat_callback(switch_core_scheduler_task_t *task)
{
send_heartbeat();
/* reschedule this task */
task->runtime += 20;
}
static void switch_core_scheduler_execute(switch_core_scheduler_task_container_t *tp)
{
//switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Executing task %u %s (%s)\n", tp->task.task_id, tp->desc, switch_str_nil(tp->task.group));
tp->func(&tp->task);
if (tp->task.runtime > tp->executed) {
tp->executed = 0;
} else {
tp->destroyed = 1;
}
}
static void *SWITCH_THREAD_FUNC task_own_thread(switch_thread_t *thread, void *obj)
{
switch_core_scheduler_task_container_t *tp = (switch_core_scheduler_task_container_t *) obj;
switch_memory_pool_t *pool;
pool = tp->pool;
tp->pool = NULL;
switch_core_scheduler_execute(tp);
switch_core_destroy_memory_pool(&pool);
tp->in_thread = 0;
return NULL;
}
static int task_thread_loop(int done)
{
switch_core_scheduler_task_container_t *tofree, *tp, *last = NULL;
switch_mutex_lock(runtime.task_mutex);
for (tp = runtime.task_list; tp; tp = tp->next) {
if (done) {
tp->destroyed = 1;
} else {
time_t now = time(NULL);
if (now >= tp->task.runtime && !tp->in_thread) {
tp->executed = now;
if (switch_test_flag(tp, SSHF_OWN_THREAD)) {
switch_thread_t *thread;
switch_threadattr_t *thd_attr;
assert(switch_core_new_memory_pool(&tp->pool) == SWITCH_STATUS_SUCCESS);
switch_threadattr_create(&thd_attr, tp->pool);
switch_threadattr_detach_set(thd_attr, 1);
tp->in_thread = 1;
switch_thread_create(&thread, thd_attr, task_own_thread, tp, tp->pool);
} else {
switch_core_scheduler_execute(tp);
}
}
}
}
switch_mutex_unlock(runtime.task_mutex);
switch_mutex_lock(runtime.task_mutex);
for (tp = runtime.task_list; tp; ) {
if (tp->destroyed && !tp->in_thread) {
tofree = tp;
tp = tp->next;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Deleting task %u %s (%s)\n",
tofree->task.task_id, tofree->desc, switch_str_nil(tofree->task.group));
if (last) {
last->next = tofree->next;
} else {
runtime.task_list = tofree->next;
}
switch_safe_free(tofree->task.group);
if (tofree->task.cmd_arg && switch_test_flag(tofree, SSHF_FREE_ARG)) {
free(tofree->task.cmd_arg);
}
switch_safe_free(tofree->desc);
free(tofree);
} else {
last = tp;
tp = tp->next;
}
}
switch_mutex_unlock(runtime.task_mutex);
return done;
}
static void *SWITCH_THREAD_FUNC switch_core_task_thread(switch_thread_t *thread, void *obj)
{
runtime.task_thread_running = 1;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Starting task thread\n");
while(runtime.task_thread_running == 1) {
if (task_thread_loop(0)) {
break;
}
switch_yield(500000);
}
task_thread_loop(1);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Task thread ending\n");
runtime.task_thread_running = 0;
return NULL;
}
static void switch_core_task_thread_launch(void)
{
switch_thread_t *thread;
switch_threadattr_t *thd_attr;
switch_threadattr_create(&thd_attr, runtime.memory_pool);
switch_threadattr_detach_set(thd_attr, 1);
switch_thread_create(&thread, thd_attr, switch_core_task_thread, NULL, runtime.memory_pool);
}
SWITCH_DECLARE(uint32_t) switch_core_scheduler_add_task(time_t task_runtime,
switch_core_scheduler_func_t func,
char *desc,
char *group,
uint32_t cmd_id,
void *cmd_arg,
switch_scheduler_flag_t flags)
{
switch_core_scheduler_task_container_t *container, *tp;
switch_mutex_lock(runtime.task_mutex);
assert((container = malloc(sizeof(*container))));
memset(container, 0, sizeof(*container));
assert(func);
container->func = func;
time(&container->task.created);
container->task.runtime = task_runtime;
container->task.group = strdup(group ? group : "none");
container->task.cmd_id = cmd_id;
container->task.cmd_arg = cmd_arg;
container->flags = flags;
container->desc = strdup(desc ? desc : "none");
for (tp = runtime.task_list; tp && tp->next; tp = tp->next);
if (tp) {
tp->next = container;
} else {
runtime.task_list = container;
}
for (container->task.task_id = 0; !container->task.task_id; container->task.task_id = ++runtime.task_id);
switch_mutex_unlock(runtime.task_mutex);
tp = container;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Added task %u %s (%s) to run at %ld\n",
tp->task.task_id, tp->desc, switch_str_nil(tp->task.group), task_runtime);
return container->task.task_id;
}
SWITCH_DECLARE(switch_status_t) switch_core_scheduler_del_task_id(uint32_t task_id)
{
switch_core_scheduler_task_container_t *tp;
switch_status_t status = SWITCH_STATUS_FALSE;
switch_mutex_lock(runtime.task_mutex);
for (tp = runtime.task_list; tp; tp = tp->next) {
if (tp->task.task_id == task_id) {
tp->destroyed++;
status = SWITCH_STATUS_SUCCESS;
break;
}
}
switch_mutex_unlock(runtime.task_mutex);
return status;
}
SWITCH_DECLARE(switch_status_t) switch_core_scheduler_del_task_group(char *group)
{
switch_core_scheduler_task_container_t *tp;
switch_status_t status = SWITCH_STATUS_FALSE;
switch_mutex_lock(runtime.task_mutex);
for (tp = runtime.task_list; tp; tp = tp->next) {
if (!strcmp(tp->task.group, group)) {
tp->destroyed++;
status = SWITCH_STATUS_SUCCESS;
}
}
switch_mutex_unlock(runtime.task_mutex);
return status;
}
static void switch_core_media_bug_destroy(switch_media_bug_t *bug)
{
switch_buffer_destroy(&bug->raw_read_buffer);
@@ -2148,7 +2397,8 @@ SWITCH_DECLARE(switch_status_t) switch_core_session_read_frame(switch_core_sessi
}
if (!session->read_codec && (*frame)->codec) {
need_codec = TRUE;
status = SWITCH_STATUS_FALSE;
goto done;
}
if (session->bugs && !need_codec) {
@@ -2422,7 +2672,7 @@ SWITCH_DECLARE(switch_status_t) switch_core_session_write_frame(switch_core_sess
}
if (!session->write_codec && frame->codec) {
need_codec = TRUE;
return SWITCH_STATUS_FALSE;
}
if (session->bugs && !need_codec) {
@@ -3153,11 +3403,10 @@ static void switch_core_standard_on_execute(switch_core_session_t *session)
switch_event_t *event;
const switch_application_interface_t *application_interface;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Standard EXECUTE\n");
if ((extension = switch_channel_get_caller_extension(session->channel)) == 0) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "No Extension!\n");
switch_channel_hangup(session->channel, SWITCH_CAUSE_DESTINATION_OUT_OF_ORDER);
switch_channel_hangup(session->channel, SWITCH_CAUSE_NORMAL_CLEARING);
return;
}
@@ -3731,6 +3980,8 @@ SWITCH_DECLARE(void) switch_core_session_destroy(switch_core_session_t **session
switch_event_t *event;
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Close Channel %s\n", switch_channel_get_name((*session)->channel));
switch_core_scheduler_del_task_group((*session)->uuid_str);
switch_mutex_lock(runtime.session_table_mutex);
switch_core_hash_delete(runtime.session_table, (*session)->uuid_str);
@@ -4143,7 +4394,6 @@ static void *SWITCH_THREAD_FUNC switch_core_sql_thread(switch_thread_t *thread,
char *sqlbuf = (char *) malloc(sql_len);
char *sql;
switch_size_t newlen;
uint32_t loops = 0;
if (!runtime.event_db) {
runtime.event_db = switch_core_db_handle();
@@ -4198,38 +4448,6 @@ static void *SWITCH_THREAD_FUNC switch_core_sql_thread(switch_thread_t *thread,
*sqlbuf = '\0';
}
if (loops++ >= 5000) {
switch_event_t *event;
switch_core_time_duration_t duration;
switch_core_measure_time(switch_core_uptime(), &duration);
if (switch_event_create(&event, SWITCH_EVENT_HEARTBEAT) == SWITCH_STATUS_SUCCESS) {
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "Event-Info", "System Ready");
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "Up-Time",
"%u year%s, "
"%u day%s, "
"%u hour%s, "
"%u minute%s, "
"%u second%s, "
"%u millisecond%s, "
"%u microsecond%s\n",
duration.yr, duration.yr == 1 ? "" : "s",
duration.day, duration.day == 1 ? "" : "s",
duration.hr, duration.hr == 1 ? "" : "s",
duration.min, duration.min == 1 ? "" : "s",
duration.sec, duration.sec == 1 ? "" : "s",
duration.ms, duration.ms == 1 ? "" : "s",
duration.mms, duration.mms == 1 ? "" : "s"
);
switch_event_add_header(event, SWITCH_STACK_BOTTOM, "Session-Count", "%u", switch_core_session_count());
switch_event_fire(&event);
}
loops = 0;
}
if (nothing_in_queue) {
switch_yield(1000);
}
@@ -4672,11 +4890,17 @@ SWITCH_DECLARE(switch_status_t) switch_core_init(char *console, const char **err
runtime.running = 1;
switch_core_hash_init(&runtime.session_table, runtime.memory_pool);
switch_mutex_init(&runtime.session_table_mutex, SWITCH_MUTEX_NESTED, runtime.memory_pool);
switch_mutex_init(&runtime.task_mutex, SWITCH_MUTEX_NESTED, runtime.memory_pool);
#ifdef CRASH_PROT
switch_core_hash_init(&runtime.stack_table, runtime.memory_pool);
#endif
switch_core_task_thread_launch();
runtime.initiated = switch_time_now();
switch_core_scheduler_add_task(time(NULL), heartbeat_callback, "heartbeat", "core", 0, NULL, SSHF_NONE);
switch_uuid_get(&uuid);
switch_uuid_format(runtime.uuid_str, &uuid);
@@ -4836,11 +5060,28 @@ SWITCH_DECLARE(switch_status_t) switch_core_destroy(void)
while (switch_queue_size(runtime.sql_queue) > 0) {
switch_yield(10000);
}
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "Stopping Task Thread\n");
if (runtime.task_thread_running == 1) {
int sanity = 0;
runtime.task_thread_running = -1;
while(runtime.task_thread_running) {
switch_yield(100000);
if (++sanity > 10) {
break;
}
}
}
switch_core_db_close(runtime.db);
switch_core_db_close(runtime.event_db);
switch_xml_destroy();
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "Finalizing Shutdown.\n");
switch_log_shutdown();
if(runtime.console != stdout && runtime.console != stderr) {
fclose(runtime.console);