[Asterisk-code-review] serializer: move/add asterisk serializer pool functionality (...asterisk[13])
Kevin Harwell
asteriskteam at digium.com
Wed Oct 2 11:18:24 CDT 2019
Kevin Harwell has uploaded this change for review. ( https://gerrit.asterisk.org/c/asterisk/+/12996
Change subject: serializer: move/add asterisk serializer pool functionality
......................................................................
serializer: move/add asterisk serializer pool functionality
Serializer pools have previously existed in Asterisk. However, for the most
part the code has been duplicated across modules. This patch abstracts the
code into an 'ast_serializer_pool' object. As well the code is now centralized
in serializer.c/h.
In addition serializer pools can now optionally be monitored by a shutdown
group. This will prevent the pool from being destroyed until all serializers
have completed.
Change-Id: Ib1e906144b90ffd4d5ed9826f0b719ca9c6d2971
---
A include/asterisk/serializer.h
M include/asterisk/taskprocessor.h
A main/serializer.c
M main/taskprocessor.c
4 files changed, 295 insertions(+), 1 deletion(-)
git pull ssh://gerrit.asterisk.org:29418/asterisk refs/changes/96/12996/1
diff --git a/include/asterisk/serializer.h b/include/asterisk/serializer.h
new file mode 100644
index 0000000..f872f3a
--- /dev/null
+++ b/include/asterisk/serializer.h
@@ -0,0 +1,85 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2019, Digium, Inc.
+ *
+ * Kevin Harwell <kharwell at digium.com>
+ *
+ * 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.
+ */
+
+#ifndef _AST_SERIALIZER_H
+#define _AST_SERIALIZER_H
+
+struct ast_threadpool;
+
+/*!
+ * Maintains a named pool of thread pooled taskprocessors. Also if configured
+ * a shutdown group can be enabled that will ensure all serializers have
+ * completed any assigned task before destruction.
+ */
+struct ast_serializer_pool;
+
+/*!
+ * \brief 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.
+ *
+ * \param pool The pool to destroy
+ */
+int ast_serializer_pool_destroy(struct ast_serializer_pool *pool);
+
+/*!
+ * \brief Create a serializer pool.
+ *
+ * Create a serializer pool with an optional shutdown group. If a timeout greater
+ * than zero is specified then a shutdown group is enabled on the pool.
+ *
+ * \param name The base name for the pool, and used when building taskprocessor(s)
+ * \param size The size of the pool
+ * \param threadpool The backing threadpool to use
+ * \param timeout The timeout used if using a shutdown group
+ *
+ * \retval A newly allocated serializer pool object, or NULL on error
+ */
+struct ast_serializer_pool *ast_serializer_pool_create(const char *name,
+ unsigned int size, struct ast_threadpool *threadpool, unsigned int timeout);
+
+/*!
+ * \brief Retrieve the base name of the serializer pool.
+ *
+ * \param pool The pool object
+ *
+ * \retval The base name given to the pool
+ */
+const char *ast_serializer_pool_name(const struct ast_serializer_pool *pool);
+
+/*!
+ * \brief Retrieve a serializer from the pool.
+ *
+ * \param pool The pool object
+ *
+ * \retval A serializer/taskprocessor
+ */
+struct ast_taskprocessor *ast_serializer_pool_get(struct ast_serializer_pool *pool);
+
+/*!
+ * \brief Set taskprocessor alert levels for the serializers in the pool.
+ *
+ * \param pool The pool to destroy
+ *
+ * \retval 0 on success, or -1 on error.
+ */
+int ast_serializer_pool_set_alerts(struct ast_serializer_pool *pool, long high, long low);
+
+#endif /* _AST_SERIALIZER_H */
diff --git a/include/asterisk/taskprocessor.h b/include/asterisk/taskprocessor.h
index 2f49e47..5145565 100644
--- a/include/asterisk/taskprocessor.h
+++ b/include/asterisk/taskprocessor.h
@@ -305,6 +305,15 @@
unsigned int ast_taskprocessor_seq_num(void);
/*!
+ * \brief Append the next sequence number to the given string, and copy into the buffer.
+ *
+ * \param buf Where to copy the appended taskprocessor name.
+ * \param size How large is buf including null terminator.
+ * \param name A name to append the sequence number to.
+ */
+void ast_taskprocessor_name_append(char *buf, unsigned int size, const char *name);
+
+/*!
* \brief Build a taskprocessor name with a sequence number on the end.
* \since 13.8.0
*
diff --git a/main/serializer.c b/main/serializer.c
new file mode 100644
index 0000000..4d7c5df
--- /dev/null
+++ b/main/serializer.c
@@ -0,0 +1,189 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2019, Digium, Inc.
+ *
+ * Kevin Harwell <kharwell at digium.com>
+ *
+ * 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_CALLBACK_VOID(&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, unsigned 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 > 0 ? 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 = 1; 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;
+}
diff --git a/main/taskprocessor.c b/main/taskprocessor.c
index 13bdd2b..6466b05 100644
--- a/main/taskprocessor.c
+++ b/main/taskprocessor.c
@@ -1282,11 +1282,22 @@
return (unsigned int) ast_atomic_fetchadd_int(&seq_num, +1);
}
+#define SEQ_STR_SIZE (1 + 8 + 1) /* Dash plus 8 hex digits plus null terminator */
+
+void ast_taskprocessor_name_append(char *buf, unsigned int size, const char *name)
+{
+ int final_size = strlen(name) + SEQ_STR_SIZE;
+
+ ast_assert(buf != NULL && name != NULL);
+ ast_assert(final_size <= size);
+
+ snprintf(buf, final_size, "%s-%08x", name, ast_taskprocessor_seq_num());
+}
+
void ast_taskprocessor_build_name(char *buf, unsigned int size, const char *format, ...)
{
va_list ap;
int user_size;
-#define SEQ_STR_SIZE (1 + 8 + 1) /* Dash plus 8 hex digits plus null terminator */
ast_assert(buf != NULL);
ast_assert(SEQ_STR_SIZE <= size);
--
To view, visit https://gerrit.asterisk.org/c/asterisk/+/12996
To unsubscribe, or for help writing mail filters, visit https://gerrit.asterisk.org/settings
Gerrit-Project: asterisk
Gerrit-Branch: 13
Gerrit-Change-Id: Ib1e906144b90ffd4d5ed9826f0b719ca9c6d2971
Gerrit-Change-Number: 12996
Gerrit-PatchSet: 1
Gerrit-Owner: Kevin Harwell <kharwell at digium.com>
Gerrit-MessageType: newchange
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.digium.com/pipermail/asterisk-code-review/attachments/20191002/ef4837f3/attachment-0001.html>
More information about the asterisk-code-review
mailing list