serializer.c

Go to the documentation of this file.

/*
 * Asterisk -- An open source telephony toolkit.
 *
 * Copyright (C) 2019, Sangoma Technologies Corporation
 *
 * Kevin Harwell <[email protected]>
 *
 * See http://www.asterisk.org for more information about
 * the Asterisk project. Please do not directly contact
 * any of the maintainers of this project for assistance;
 * the project provides a web site, mailing lists and IRC
 * channels for your use.
 *
 * This program is free software, distributed under the terms of
 * the GNU General Public License Version 2. See the LICENSE file
 * at the top of the source tree.
 */

#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"

struct ast_serializer_pool {
 /*! Shutdown group to monitor serializers. */
 struct ast_serializer_shutdown_group *shutdown_group;
 /*! Time to wait if using a shutdown group. */
 int shutdown_group_timeout;
 /*! A pool of taskprocessor(s) */
 AST_VECTOR_RW(, struct ast_taskprocessor *) serializers;
 /*! Base name for the pool */
 char name[];
};

int ast_serializer_pool_destroy(struct ast_serializer_pool *pool)
{
 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;
}

struct ast_serializer_pool *ast_serializer_pool_create(const char *name,
 unsigned int size, struct ast_threadpool *threadpool, int timeout)
{
 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;
}

const char *ast_serializer_pool_name(const struct ast_serializer_pool *pool)
{
 return pool->name;
}

struct ast_taskprocessor *ast_serializer_pool_get(struct ast_serializer_pool *pool)
{
 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;
}

int ast_serializer_pool_set_alerts(struct ast_serializer_pool *pool, long high, long low)
{
 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;
}