From 30e4c71bbe6c7c5ab80fc4be1f726577446ad154 Mon Sep 17 00:00:00 2001 From: Anthony Minessale Date: Tue, 8 Jan 2013 13:01:07 -0600 Subject: [PATCH] FS-4996 --resolve --- .../mod_erlang_event/handle_msg.c | 7 ++- .../mod_erlang_event/mod_erlang_event.c | 52 +++++++++---------- 2 files changed, 30 insertions(+), 29 deletions(-) diff --git a/src/mod/event_handlers/mod_erlang_event/handle_msg.c b/src/mod/event_handlers/mod_erlang_event/handle_msg.c index 618b7eb829..21e7376e98 100644 --- a/src/mod/event_handlers/mod_erlang_event/handle_msg.c +++ b/src/mod/event_handlers/mod_erlang_event/handle_msg.c @@ -660,11 +660,13 @@ static switch_status_t handle_msg_api(listener_t *listener, erlang_msg * msg, in } } +#define ARGLEN 2048 static switch_status_t handle_msg_bgapi(listener_t *listener, erlang_msg * msg, int arity, ei_x_buff * buf, ei_x_buff * rbuf) { char api_cmd[MAXATOMLEN]; - char arg[1024]; - if (arity < 3 || ei_decode_atom(buf->buff, &buf->index, api_cmd) || ei_decode_string_or_binary(buf->buff, &buf->index, 1023, arg)) { + char arg[ARGLEN]; + + if (arity < 3 || ei_decode_atom(buf->buff, &buf->index, api_cmd) || ei_decode_string_or_binary(buf->buff, &buf->index, ARGLEN - 1, arg)) { ei_x_encode_tuple_header(rbuf, 2); ei_x_encode_atom(rbuf, "error"); ei_x_encode_atom(rbuf, "badarg"); @@ -1146,6 +1148,7 @@ static switch_status_t handle_ref_tuple(listener_t *listener, erlang_msg * msg, se->spawn_reply->pid = switch_core_alloc(se->pool, sizeof(erlang_pid)); switch_assert(se->spawn_reply->pid != NULL); memcpy(se->spawn_reply->pid, &pid, sizeof(erlang_pid)); + switch_thread_cond_signal(se->spawn_reply->ready_or_found); switch_mutex_unlock(se->spawn_reply->mutex); diff --git a/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c b/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c index 9b34bfa8fc..bf5b0b6f71 100644 --- a/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c +++ b/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c @@ -330,11 +330,12 @@ static void destroy_session_elem(session_elem_t *session_element) switch_thread_rwlock_wrlock(session_element->rwlock); switch_thread_rwlock_unlock(session_element->rwlock); - if ((session = switch_core_session_locate(session_element->uuid_str))) { + if ((session = switch_core_session_force_locate(session_element->uuid_str))) { switch_channel_t *channel = switch_core_session_get_channel(session); switch_channel_set_private(channel, "_erlang_session_", NULL); switch_channel_clear_flag(channel, CF_CONTROLLED); + switch_core_session_soft_unlock(session); switch_core_session_rwunlock(session); } switch_core_destroy_memory_pool(&session_element->pool); @@ -795,21 +796,8 @@ static void handle_exit(listener_t *listener, erlang_pid * pid) remove_binding(NULL, pid); /* TODO - why don't we pass the listener as the first argument? */ if ((s = find_session_elem_by_pid(listener, pid))) { - switch_core_session_t *session = NULL; - - if ((session = switch_core_session_locate(s->uuid_str))) { - switch_channel_t *channel = switch_core_session_get_channel(session); - - if (switch_channel_get_state(channel) < CS_HANGUP) { - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "Outbound session exited unexpectedly %s!\n", s->uuid_str); - } - - switch_channel_hangup(channel, SWITCH_CAUSE_DESTINATION_OUT_OF_ORDER); - switch_core_session_rwunlock(session); - } - + switch_set_flag_locked(s, LFLAG_SESSION_COMPLETE); switch_thread_rwlock_unlock(s->rwlock); - } if (listener->log_process.type == ERLANG_PID && !ei_compare_pids(&listener->log_process.pid, pid)) { @@ -1295,9 +1283,6 @@ static switch_status_t state_handler(switch_core_session_t *session) * we can throw this away */ switch_set_flag_locked(session_element, LFLAG_SESSION_COMPLETE); } - } else { - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "unable to update channel state for %s to %s\n", switch_core_session_get_uuid(session), - switch_channel_state_name(state)); } return SWITCH_STATUS_SUCCESS; @@ -1307,7 +1292,7 @@ session_elem_t *session_elem_create(listener_t *listener, switch_core_session_t { /* create a session list element */ switch_memory_pool_t *session_elem_pool; - session_elem_t *session_element; /* = malloc(sizeof(*session_element)); */ + session_elem_t *session_element; switch_channel_t *channel = switch_core_session_get_channel(session); int x; @@ -1339,6 +1324,7 @@ session_elem_t *session_elem_create(listener_t *listener, switch_core_session_t session_element->event_list[SWITCH_EVENT_ALL] = 1; /* defaults to everything */ switch_channel_set_private(channel, "_erlang_session_", session_element); + switch_core_session_soft_lock(session, 5); switch_core_event_hook_add_state_change(session, state_handler); @@ -1399,7 +1385,8 @@ session_elem_t *attach_call_to_spawned_process(listener_t *listener, char *modul /* attach the session to the listener */ add_session_elem_to_listener(listener, session_element); - + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Added session to listener\n"); + if (!strcmp(function, "!")) { /* send a message to request a pid */ ei_x_buff rbuf; @@ -1421,14 +1408,20 @@ session_elem_t *attach_call_to_spawned_process(listener_t *listener, char *modul } else { switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "rpc call: %s:%s(Ref)\n", module, function); /* should lock with mutex? */ + switch_mutex_lock(listener->sock_mutex); ei_pid_from_rpc(listener->ec, listener->sockfd, &ref, module, function); + switch_mutex_unlock(listener->sock_mutex); /* char *argv[1]; ei_spawn(listener->ec, listener->sockfd, &ref, module, function, 0, argv); */ } + + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Waiting for reply %s %s\n", hash, session_element->uuid_str); + switch_mutex_lock(p->mutex); switch_thread_cond_timedwait(p->ready_or_found, p->mutex, 5000000); + switch_mutex_unlock(p->mutex); if (!p->pid) { switch_channel_t *channel = switch_core_session_get_channel(session); @@ -1534,16 +1527,23 @@ SWITCH_STANDARD_APP(erlang_outbound_function) } } else { switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Using existing listener for session\n"); + } if (listener) { - if (module && function) { - switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Creating new spawned session for listener\n"); - session_element = attach_call_to_spawned_process(listener, module, function, session); + if ((session_element = find_session_elem_by_uuid(listener, switch_core_session_get_uuid(session)))) { + switch_thread_rwlock_unlock(session_element->rwlock); + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Session already exists\n"); + } else { - switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Creating new registered session for listener\n"); - session_element = attach_call_to_registered_process(listener, reg_name, session); + if (module && function) { + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Creating new spawned session for listener\n"); + session_element = attach_call_to_spawned_process(listener, module, function, session); + } else { + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Creating new registered session for listener\n"); + session_element = attach_call_to_registered_process(listener, reg_name, session); + } } switch_thread_rwlock_unlock(listener->rwlock); @@ -1552,8 +1552,6 @@ SWITCH_STANDARD_APP(erlang_outbound_function) switch_ivr_park(session, NULL); } - } else { - switch_thread_rwlock_unlock(globals.listener_rwlock); } switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "exit erlang_outbound_function\n");