serializer.c File Reference

#include "asterisk.h"
#include "asterisk/astobj2.h"
#include "asterisk/serializer.h"
#include "asterisk/taskprocessor.h"
#include "asterisk/threadpool.h"
#include "asterisk/utils.h"
#include "asterisk/vector.h"

Include dependency graph for serializer.c:

Go to the source code of this file.

Data Structures
struct	ast_serializer_pool

Functions
struct ast_serializer_pool *	ast_serializer_pool_create (const char *name, unsigned int size, struct ast_threadpool *threadpool, int timeout)
	Create a serializer pool. More...
int	ast_serializer_pool_destroy (struct ast_serializer_pool *pool)
	Destroy the serializer pool. More...
struct ast_taskprocessor *	ast_serializer_pool_get (struct ast_serializer_pool *pool)
	Retrieve a serializer from the pool. More...
const char *	ast_serializer_pool_name (const struct ast_serializer_pool *pool)
	Retrieve the base name of the serializer pool. More...
int	ast_serializer_pool_set_alerts (struct ast_serializer_pool *pool, long high, long low)
	Set taskprocessor alert levels for the serializers in the pool. More...

Function Documentation

ast_serializer_pool_create()

struct ast_serializer_pool * ast_serializer_pool_create	(	const char *	name,
		unsigned int	size,
		struct ast_threadpool *	threadpool,
		int	timeout
	)

Create a serializer pool.

Create a serializer pool with an optional shutdown group. If a timeout greater than -1 is specified then a shutdown group is enabled on the pool.

Parameters

name	The base name for the pool, and used when building taskprocessor(s)
size	The size of the pool
threadpool	The backing threadpool to use
timeout	The timeout used if using a shutdown group (-1 = disabled)

Returns

A newly allocated serializer pool object

Return values

NULL

on error

Definition at line 76 of file serializer.c.

{
    struct ast_serializer_pool *pool;
    char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1];
    size_t idx;
 
    ast_assert(size > 0);
 
    pool = ast_malloc(sizeof(*pool) + strlen(name) + 1);
    if (!pool) {
        return NULL;
    }
 
    strcpy(pool->name, name); /* safe */
 
    pool->shutdown_group_timeout = timeout;
    pool->shutdown_group = timeout > -1 ? ast_serializer_shutdown_group_alloc() : NULL;
 
    AST_VECTOR_RW_INIT(&pool->serializers, size);
 
    for (idx = 0; idx < size; ++idx) {
        struct ast_taskprocessor *tps;
 
        /* Create name with seq number appended. */
        ast_taskprocessor_name_append(tps_name, sizeof(tps_name), name);
 
        tps = ast_threadpool_serializer_group(tps_name, threadpool, pool->shutdown_group);
        if (!tps) {
            ast_serializer_pool_destroy(pool);
            ast_log(LOG_ERROR, "Pool create: unable to create named serializer '%s'\n",
                    tps_name);
            return NULL;
        }
 
        if (AST_VECTOR_APPEND(&pool->serializers, tps)) {
            ast_serializer_pool_destroy(pool);
            ast_log(LOG_ERROR, "Pool create: unable to append named serializer '%s'\n",
                    tps_name);
            return NULL;
        }
    }
 
    return pool;
}

References ast_assert, ast_log, ast_malloc, ast_serializer_pool_destroy(), ast_serializer_shutdown_group_alloc(), AST_TASKPROCESSOR_MAX_NAME, ast_taskprocessor_name_append(), ast_threadpool_serializer_group(), AST_VECTOR_APPEND, AST_VECTOR_RW_INIT, LOG_ERROR, name, NULL, ast_serializer_pool::shutdown_group, ast_serializer_pool::shutdown_group_timeout, and threadpool.

Referenced by AST_TEST_DEFINE(), and load_module().

ast_serializer_pool_destroy()

int ast_serializer_pool_destroy

(

struct ast_serializer_pool *

pool

)

Destroy the serializer pool.

Attempt to destroy the serializer pool. If a shutdown group has been enabled, and times out waiting for threads to complete, then this function will return the number of remaining threads, and the pool will not be destroyed.

Parameters

pool	The pool to destroy

Definition at line 39 of file serializer.c.

{
    if (!pool) {
        return 0;
    }
 
    /* Clear out the serializers */
    AST_VECTOR_RW_WRLOCK(&pool->serializers);
    AST_VECTOR_RESET(&pool->serializers, ast_taskprocessor_unreference);
    AST_VECTOR_RW_UNLOCK(&pool->serializers);
 
    /* If using a shutdown group then wait for all threads to complete */
    if (pool->shutdown_group) {
        int remaining;
 
        ast_debug(3, "Waiting on serializers before destroying pool '%s'\n", pool->name);
 
        remaining = ast_serializer_shutdown_group_join(
            pool->shutdown_group, pool->shutdown_group_timeout);
 
        if (remaining) {
            /* If we've timed out don't fully cleanup yet */
            ast_log(LOG_WARNING, "'%s' serializer pool destruction timeout. "
                "'%d' dependencies still processing.\n", pool->name, remaining);
            return remaining;
        }
 
        ao2_ref(pool->shutdown_group, -1);
        pool->shutdown_group = NULL;
    }
 
    AST_VECTOR_RW_FREE(&pool->serializers);
    ast_free(pool);
 
    return 0;
}

References ao2_ref, ast_debug, ast_free, ast_log, ast_serializer_shutdown_group_join(), ast_taskprocessor_unreference(), AST_VECTOR_RESET, AST_VECTOR_RW_FREE, AST_VECTOR_RW_UNLOCK, AST_VECTOR_RW_WRLOCK, LOG_WARNING, NULL, ast_serializer_pool::shutdown_group, and ast_serializer_pool::shutdown_group_timeout.

Referenced by ast_serializer_pool_create(), AST_TEST_DEFINE(), load_module(), and unload_module().

ast_serializer_pool_get()

struct ast_taskprocessor * ast_serializer_pool_get

(

struct ast_serializer_pool *

pool

)

Retrieve a serializer from the pool.

Parameters

pool	The pool object

Returns

A serializer/taskprocessor

Definition at line 127 of file serializer.c.

{
    struct ast_taskprocessor *res;
    size_t idx;
 
    if (!pool) {
        return NULL;
    }
 
    AST_VECTOR_RW_RDLOCK(&pool->serializers);
    if (AST_VECTOR_SIZE(&pool->serializers) == 0) {
        AST_VECTOR_RW_UNLOCK(&pool->serializers);
        return NULL;
    }
 
    res = AST_VECTOR_GET(&pool->serializers, 0);
 
    /* Choose the taskprocessor with the smallest queue */
    for (idx = 1; idx < AST_VECTOR_SIZE(&pool->serializers); ++idx) {
        struct ast_taskprocessor *cur = AST_VECTOR_GET(&pool->serializers, idx);
        if (ast_taskprocessor_size(cur) < ast_taskprocessor_size(res)) {
            res = cur;
        }
    }
 
    AST_VECTOR_RW_UNLOCK(&pool->serializers);
    return res;
}

References ast_taskprocessor_size(), AST_VECTOR_GET, AST_VECTOR_RW_RDLOCK, AST_VECTOR_RW_UNLOCK, AST_VECTOR_SIZE, and NULL.

Referenced by ast_sip_push_task(), ast_sip_push_task_wait_serializer(), AST_TEST_DEFINE(), load_module(), mwi_startup_event_cb(), mwi_stasis_cb(), send_contact_notify(), and send_notify().

ast_serializer_pool_name()

const char * ast_serializer_pool_name

(

const struct ast_serializer_pool *

pool

)

Retrieve the base name of the serializer pool.

Parameters

pool	The pool object

Returns

The base name given to the pool

Definition at line 122 of file serializer.c.

{
    return pool->name;
}

Referenced by AST_TEST_DEFINE().

ast_serializer_pool_set_alerts()

int ast_serializer_pool_set_alerts	(	struct ast_serializer_pool *	pool,
		long	high,
		long	low
	)

Set taskprocessor alert levels for the serializers in the pool.

Parameters

pool	The pool to destroy
high,low

Return values

0	on success.
-1	on error.

Definition at line 156 of file serializer.c.

{
    size_t idx;
    long tps_queue_high;
    long tps_queue_low;
 
    if (!pool) {
        return 0;
    }
 
    tps_queue_high = high;
    if (tps_queue_high <= 0) {
        ast_log(AST_LOG_WARNING, "Invalid '%s-*' taskprocessor high water alert "
                "trigger level '%ld'\n", pool->name, tps_queue_high);
        tps_queue_high = AST_TASKPROCESSOR_HIGH_WATER_LEVEL;
    }
 
    tps_queue_low = low;
    if (tps_queue_low < -1 || tps_queue_high < tps_queue_low) {
        ast_log(AST_LOG_WARNING, "Invalid '%s-*' taskprocessor low water clear alert "
                "level '%ld'\n", pool->name, tps_queue_low);
        tps_queue_low = -1;
    }
 
    for (idx = 0; idx < AST_VECTOR_SIZE(&pool->serializers); ++idx) {
        struct ast_taskprocessor *cur = AST_VECTOR_GET(&pool->serializers, idx);
        if (ast_taskprocessor_alert_set_levels(cur, tps_queue_low, tps_queue_high)) {
            ast_log(AST_LOG_WARNING, "Failed to set alert levels for serializer '%s'.\n",
                    ast_taskprocessor_name(cur));
        }
    }
 
    return 0;
}

References ast_log, AST_LOG_WARNING, ast_taskprocessor_alert_set_levels(), AST_TASKPROCESSOR_HIGH_WATER_LEVEL, ast_taskprocessor_name(), AST_VECTOR_GET, AST_VECTOR_SIZE, ast_taskprocessor::tps_queue_high, and ast_taskprocessor::tps_queue_low.

Referenced by AST_TEST_DEFINE(), and global_loaded().