can occur between this thread and bridging (specifically when native bridging
attempts to do direct media) */
ast_channel_unlock(ast);
- res = ast_sip_push_task_synchronous(session->serializer, answer, session);
+ res = ast_sip_push_task_wait_serializer(session->serializer, answer, session);
if (res) {
if (res == -1) {
ast_log(LOG_ERROR,"Cannot answer '%s': Unable to push answer task to the threadpool.\n",
req_data.caps = cap;
req_data.dest = data;
- /* Default failure value in case ast_sip_push_task_synchronous() itself fails. */
+ /* Default failure value in case ast_sip_push_task_wait_servant() itself fails. */
req_data.cause = AST_CAUSE_FAILURE;
- if (ast_sip_push_task_synchronous(NULL, request, &req_data)) {
+ if (ast_sip_push_task_wait_servant(NULL, request, &req_data)) {
*cause = req_data.cause;
return NULL;
}
func_args.field = args.field;
func_args.buf = buf;
func_args.len = len;
- if (ast_sip_push_task_synchronous(func_args.session->serializer, read_pjsip, &func_args)) {
+ if (ast_sip_push_task_wait_serializer(func_args.session->serializer, read_pjsip, &func_args)) {
ast_log(LOG_WARNING, "Unable to read properties of channel %s: failed to push task\n", ast_channel_name(chan));
ao2_ref(func_args.session, -1);
return -1;
mdata.media_type = AST_MEDIA_TYPE_VIDEO;
}
- return ast_sip_push_task_synchronous(channel->session->serializer, media_offer_write_av, &mdata);
+ return ast_sip_push_task_wait_serializer(channel->session->serializer, media_offer_write_av, &mdata);
}
int pjsip_acf_dtmf_mode_read(struct ast_channel *chan, const char *cmd, char *data, char *buf, size_t len)
ast_channel_unlock(chan);
- return ast_sip_push_task_synchronous(channel->session->serializer, dtmf_mode_refresh_cb, &rdata);
+ return ast_sip_push_task_wait_serializer(channel->session->serializer, dtmf_mode_refresh_cb, &rdata);
}
static int refresh_write_cb(void *obj)
rdata.method = AST_SIP_SESSION_REFRESH_METHOD_UPDATE;
}
- return ast_sip_push_task_synchronous(channel->session->serializer, refresh_write_cb, &rdata);
+ return ast_sip_push_task_wait_serializer(channel->session->serializer, refresh_write_cb, &rdata);
}
int ast_sip_push_task(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data);
/*!
- * \brief Push a task to SIP servants and wait for it to complete
+ * \brief Push a task to SIP servants and wait for it to complete.
+ *
+ * Like \ref ast_sip_push_task except that it blocks until the task
+ * completes. If the current thread is a SIP servant thread then the
+ * task executes immediately. Otherwise, the specified serializer
+ * executes the task and the current thread waits for it to complete.
+ *
+ * \note PJPROJECT callbacks tend to have locks already held when
+ * called.
+ *
+ * \warning \b Never hold locks that may be acquired by a SIP servant
+ * thread when calling this function. Doing so may cause a deadlock
+ * if all SIP servant threads are blocked waiting to acquire the lock
+ * while the thread holding the lock is waiting for a free SIP servant
+ * thread.
+ *
+ * \warning \b Use of this function in an ao2 destructor callback is a
+ * bad idea. You don't have control over which thread executes the
+ * destructor. Attempting to shift execution to another thread with
+ * this function is likely to cause deadlock.
+ *
+ * \param serializer The SIP serializer to execute the task if the
+ * current thread is not a SIP servant. NULL if any of the default
+ * serializers can be used.
+ * \param sip_task The task to execute
+ * \param task_data The parameter to pass to the task when it executes
+ *
+ * \note The sip_task() return value may need to be distinguished from
+ * the failure to push the task.
+ *
+ * \return sip_task() return value on success.
+ * \retval -1 Failure to push the task.
+ */
+int ast_sip_push_task_wait_servant(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data);
+
+/*!
+ * \brief Push a task to SIP servants and wait for it to complete.
+ * \deprecated Replaced with ast_sip_push_task_wait_servant().
+ */
+int ast_sip_push_task_synchronous(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data);
+
+/*!
+ * \brief Push a task to the serializer and wait for it to complete.
+ *
+ * Like \ref ast_sip_push_task except that it blocks until the task is
+ * completed by the specified serializer. If the specified serializer
+ * is the current thread then the task executes immediately.
+ *
+ * \note PJPROJECT callbacks tend to have locks already held when
+ * called.
*
- * Like \ref ast_sip_push_task except that it blocks until the task completes.
+ * \warning \b Never hold locks that may be acquired by a SIP servant
+ * thread when calling this function. Doing so may cause a deadlock
+ * if all SIP servant threads are blocked waiting to acquire the lock
+ * while the thread holding the lock is waiting for a free SIP servant
+ * thread for the serializer to execute in.
*
- * \warning \b Never use this function in a SIP servant thread. This can potentially
- * cause a deadlock. If you are in a SIP servant thread, just call your function
- * in-line.
+ * \warning \b Never hold locks that may be acquired by the serializer
+ * when calling this function. Doing so will cause a deadlock.
*
- * \warning \b Never hold locks that may be acquired by a SIP servant thread when
- * calling this function. Doing so may cause a deadlock if all SIP servant threads
- * are blocked waiting to acquire the lock while the thread holding the lock is
- * waiting for a free SIP servant thread.
+ * \warning \b Never use this function in the pjsip monitor thread (It
+ * is a SIP servant thread). This is likely to cause a deadlock.
*
- * \param serializer The SIP serializer to which the task belongs. May be NULL.
+ * \warning \b Use of this function in an ao2 destructor callback is a
+ * bad idea. You don't have control over which thread executes the
+ * destructor. Attempting to shift execution to another thread with
+ * this function is likely to cause deadlock.
+ *
+ * \param serializer The SIP serializer to execute the task. NULL if
+ * any of the default serializers can be used.
* \param sip_task The task to execute
* \param task_data The parameter to pass to the task when it executes
- * \retval 0 Success
- * \retval -1 Failure
+ *
+ * \note It is generally better to call
+ * ast_sip_push_task_wait_servant() if you pass NULL for the
+ * serializer parameter.
+ *
+ * \note The sip_task() return value may need to be distinguished from
+ * the failure to push the task.
+ *
+ * \return sip_task() return value on success.
+ * \retval -1 Failure to push the task.
*/
-int ast_sip_push_task_synchronous(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data);
+int ast_sip_push_task_wait_serializer(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data);
/*!
* \brief Determine if the current thread is a SIP servant thread
int internal_sip_register_service(pjsip_module *module)
{
- return ast_sip_push_task_synchronous(NULL, register_service_noref, &module);
+ return ast_sip_push_task_wait_servant(NULL, register_service_noref, &module);
}
int ast_sip_register_service(pjsip_module *module)
{
- return ast_sip_push_task_synchronous(NULL, register_service, &module);
+ return ast_sip_push_task_wait_servant(NULL, register_service, &module);
}
static int unregister_service_noref(void *data)
int internal_sip_unregister_service(pjsip_module *module)
{
- return ast_sip_push_task_synchronous(NULL, unregister_service_noref, &module);
+ return ast_sip_push_task_wait_servant(NULL, unregister_service_noref, &module);
}
void ast_sip_unregister_service(pjsip_module *module)
{
- ast_sip_push_task_synchronous(NULL, unregister_service, &module);
+ ast_sip_push_task_wait_servant(NULL, unregister_service, &module);
}
static struct ast_sip_authenticator *registered_authenticator;
return CLI_SHOWUSAGE;
}
- ast_sip_push_task_synchronous(NULL, do_cli_dump_endpt, a);
+ ast_sip_push_task_wait_servant(NULL, do_cli_dump_endpt, a);
return CLI_SUCCESS;
}
return 0;
}
+static struct ast_taskprocessor *serializer_pool_pick(void)
+{
+ struct ast_taskprocessor *serializer;
+
+ unsigned int pos;
+
+ /*
+ * Pick a serializer to use from the pool.
+ *
+ * Note: We don't care about any reentrancy behavior
+ * when incrementing serializer_pool_pos. If it gets
+ * incorrectly incremented it doesn't matter.
+ */
+ pos = serializer_pool_pos++;
+ pos %= SERIALIZER_POOL_SIZE;
+ serializer = serializer_pool[pos];
+
+ return serializer;
+}
+
int ast_sip_push_task(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data)
{
if (!serializer) {
- unsigned int pos;
-
- /*
- * Pick a serializer to use from the pool.
- *
- * Note: We don't care about any reentrancy behavior
- * when incrementing serializer_pool_pos. If it gets
- * incorrectly incremented it doesn't matter.
- */
- pos = serializer_pool_pos++;
- pos %= SERIALIZER_POOL_SIZE;
- serializer = serializer_pool[pos];
+ serializer = serializer_pool_pick();
}
return ast_taskprocessor_push(serializer, sip_task, task_data);
/*
* Once we unlock std->lock after signaling, we cannot access
- * std again. The thread waiting within
- * ast_sip_push_task_synchronous() is free to continue and
- * release its local variable (std).
+ * std again. The thread waiting within ast_sip_push_task_wait()
+ * is free to continue and release its local variable (std).
*/
ast_mutex_lock(&std->lock);
std->complete = 1;
return ret;
}
-int ast_sip_push_task_synchronous(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data)
+static int ast_sip_push_task_wait(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data)
{
/* This method is an onion */
struct sync_task_data std;
- if (ast_sip_thread_is_servant()) {
- return sip_task(task_data);
- }
-
memset(&std, 0, sizeof(std));
ast_mutex_init(&std.lock);
ast_cond_init(&std.cond, NULL);
return std.fail;
}
+int ast_sip_push_task_wait_servant(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data)
+{
+ if (ast_sip_thread_is_servant()) {
+ return sip_task(task_data);
+ }
+
+ return ast_sip_push_task_wait(serializer, sip_task, task_data);
+}
+
+int ast_sip_push_task_synchronous(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data)
+{
+ return ast_sip_push_task_wait_servant(serializer, sip_task, task_data);
+}
+
+int ast_sip_push_task_wait_serializer(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data)
+{
+ if (!serializer) {
+ /* Caller doesn't care which PJSIP serializer the task executes under. */
+ serializer = serializer_pool_pick();
+ if (!serializer) {
+ /* No serializer picked to execute the task */
+ return -1;
+ }
+ }
+ if (ast_taskprocessor_is_task(serializer)) {
+ /*
+ * We are the requested serializer so we must execute
+ * the task now or deadlock waiting on ourself to
+ * execute it.
+ */
+ return sip_task(task_data);
+ }
+
+ return ast_sip_push_task_wait(serializer, sip_task, task_data);
+}
+
void ast_copy_pj_str(char *dest, const pj_str_t *src, size_t size)
{
size_t chars_to_copy = MIN(size - 1, pj_strlen(src));
* We must wait for the reload to complete so multiple
* reloads cannot happen at the same time.
*/
- if (ast_sip_push_task_synchronous(NULL, reload_configuration_task, NULL)) {
+ if (ast_sip_push_task_wait_servant(NULL, reload_configuration_task, NULL)) {
ast_log(LOG_WARNING, "Failed to reload PJSIP\n");
return -1;
}
/* The thread this is called from cannot call PJSIP/PJLIB functions,
* so we have to push the work to the threadpool to handle
*/
- ast_sip_push_task_synchronous(NULL, unload_pjsip, NULL);
+ ast_sip_push_task_wait_servant(NULL, unload_pjsip, NULL);
ast_sip_destroy_scheduler();
serializer_pool_shutdown();
ast_threadpool_shutdown(sip_threadpool);
void ast_sip_initialize_dns(void)
{
- ast_sip_push_task_synchronous(NULL, system_create_resolver_and_set_nameservers, NULL);
+ ast_sip_push_task_wait_servant(NULL, system_create_resolver_and_set_nameservers, NULL);
}
{
struct ast_sip_transport_state *state = obj;
- ast_sip_push_task_synchronous(NULL, destroy_sip_transport_state, state);
+ ast_sip_push_task_wait_servant(NULL, destroy_sip_transport_state, state);
}
/*! \brief Destructor for ast_sip_transport state information */
.type = "header_datastore",
};
-/*! \brief Data structure used for ast_sip_push_task_synchronous */
+/*! \brief Data structure used for ast_sip_push_task_wait_serializer */
struct header_data {
struct ast_sip_channel_pvt *channel;
char *header_name;
header_data.len = len;
if (!strcasecmp(args.action, "read")) {
- return ast_sip_push_task_synchronous(channel->session->serializer, read_header,
- &header_data);
+ return ast_sip_push_task_wait_serializer(channel->session->serializer,
+ read_header, &header_data);
} else if (!strcasecmp(args.action, "remove")) {
- return ast_sip_push_task_synchronous(channel->session->serializer, remove_header,
- &header_data);
+ return ast_sip_push_task_wait_serializer(channel->session->serializer,
+ remove_header, &header_data);
} else {
ast_log(AST_LOG_ERROR,
"Unknown action '%s' is not valid, must be 'read' or 'remove'.\n",
header_data.len = 0;
if (!strcasecmp(args.action, "add")) {
- return ast_sip_push_task_synchronous(channel->session->serializer, add_header,
- &header_data);
+ return ast_sip_push_task_wait_serializer(channel->session->serializer,
+ add_header, &header_data);
} else if (!strcasecmp(args.action, "update")) {
- return ast_sip_push_task_synchronous(channel->session->serializer, update_header,
- &header_data);
+ return ast_sip_push_task_wait_serializer(channel->session->serializer,
+ update_header, &header_data);
} else if (!strcasecmp(args.action, "remove")) {
- return ast_sip_push_task_synchronous(channel->session->serializer, remove_header,
- &header_data);
+ return ast_sip_push_task_wait_serializer(channel->session->serializer,
+ remove_header, &header_data);
} else {
ast_log(AST_LOG_ERROR,
"Unknown action '%s' is not valid, must be 'add', 'update', or 'remove'.\n",
ast_cli_unregister_multiple(cli_pjsip, ARRAY_LEN(cli_pjsip));
ast_sip_unregister_service(&logging_module);
- ast_sip_push_task_synchronous(NULL, clear_history_entries, NULL);
+ ast_sip_push_task_wait_servant(NULL, clear_history_entries, NULL);
AST_VECTOR_FREE(&vector_history);
ast_pjproject_caching_pool_destroy(&cachingpool);
static int initialize_publish_client(struct ast_sip_outbound_publish *publish,
struct ast_sip_outbound_publish_state *state)
{
- if (ast_sip_push_task_synchronous(state->client->serializer, sip_outbound_publish_client_alloc, state->client)) {
+ if (ast_sip_push_task_wait_serializer(state->client->serializer,
+ sip_outbound_publish_client_alloc, state->client)) {
ast_log(LOG_ERROR, "Unable to create client for outbound publish '%s'\n",
ast_sorcery_object_get_id(publish));
return -1;
return -1;
}
- if (ast_sip_push_task_synchronous(new_state->client_state->serializer,
+ if (ast_sip_push_task_wait_serializer(new_state->client_state->serializer,
sip_outbound_registration_regc_alloc, new_state)) {
return -1;
}
struct sip_ami_outbound *ami = arg;
ami->registration = obj;
- return ast_sip_push_task_synchronous(
- NULL, ami_outbound_registration_task, ami);
+ return ast_sip_push_task_wait_servant(NULL, ami_outbound_registration_task, ami);
}
static int ami_show_outbound_registrations(struct mansession *s,
destroy_subscriptions(sub_tree->root);
if (sub_tree->dlg) {
- ast_sip_push_task_synchronous(sub_tree->serializer, subscription_unreference_dialog, sub_tree);
+ ast_sip_push_task_wait_servant(sub_tree->serializer,
+ subscription_unreference_dialog, sub_tree);
}
ao2_cleanup(sub_tree->endpoint);
}
recreate_data.persistence = persistence;
recreate_data.rdata = &rdata;
- if (ast_sip_push_task_synchronous(serializer, sub_persistence_recreate, &recreate_data)) {
+ if (ast_sip_push_task_wait_serializer(serializer, sub_persistence_recreate,
+ &recreate_data)) {
ast_log(LOG_WARNING, "Failed recreating '%s' subscription: Could not continue under distributor serializer.\n",
persistence->endpoint);
ast_sorcery_delete(ast_sip_get_sorcery(), persistence);
/* It's possible that a task is waiting to remove us already, so bump the refcount of progress so it doesn't get destroyed */
ao2_ref(progress, +1);
pjsip_dlg_dec_lock(progress->dlg);
- ast_sip_push_task_synchronous(progress->serializer, refer_progress_terminate, progress);
+ /*
+ * XXX We are always going to execute this inline rather than
+ * in the serializer because this function is a PJPROJECT
+ * callback and thus has to be a SIP servant thread.
+ *
+ * The likely remedy is to push most of this function into
+ * refer_progress_terminate() with ast_sip_push_task().
+ */
+ ast_sip_push_task_wait_servant(progress->serializer, refer_progress_terminate, progress);
pjsip_dlg_inc_lock(progress->dlg);
ao2_ref(progress, -1);
invite.session = other_session;
- if (ast_sip_push_task_synchronous(other_session->serializer, invite_replaces, &invite)) {
+ if (ast_sip_push_task_wait_serializer(other_session->serializer, invite_replaces,
+ &invite)) {
response = 481;
goto inv_replace_failed;
}
create_data.ws_session = session;
- if (ast_sip_push_task_synchronous(serializer, transport_create, &create_data)) {
+ if (ast_sip_push_task_wait_serializer(serializer, transport_create, &create_data)) {
ast_log(LOG_ERROR, "Could not create WebSocket transport.\n");
ast_taskprocessor_unreference(serializer);
ast_websocket_unref(session);
}
if (opcode == AST_WEBSOCKET_OPCODE_TEXT || opcode == AST_WEBSOCKET_OPCODE_BINARY) {
- ast_sip_push_task_synchronous(serializer, transport_read, &read_data);
+ ast_sip_push_task_wait_serializer(serializer, transport_read, &read_data);
} else if (opcode == AST_WEBSOCKET_OPCODE_CLOSE) {
break;
}
}
- ast_sip_push_task_synchronous(serializer, transport_shutdown, transport);
+ ast_sip_push_task_wait_serializer(serializer, transport_shutdown, transport);
ast_taskprocessor_unreference(serializer);
ast_websocket_unref(session);
projects / thirdparty / asterisk.git / commitdiff
