diff --git a/include/asterisk/taskprocessor.h b/include/asterisk/taskprocessor.h index 6477fc382b..3d27e09d2d 100644 --- a/include/asterisk/taskprocessor.h +++ b/include/asterisk/taskprocessor.h @@ -344,6 +344,15 @@ const char *ast_taskprocessor_name(struct ast_taskprocessor *tps); */ long ast_taskprocessor_size(struct ast_taskprocessor *tps); +/*! + * \brief Return whether the taskprocessor is currently executing a task + * + * \param tps taskprocessor + * \retval 0 if the taskprocessor is not executing a task. + * \retval non-zero if the taskprocessor is executing a task. + */ +unsigned int ast_taskprocessor_is_executing(const struct ast_taskprocessor *tps); + /*! * \brief Return the listener associated with the taskprocessor */ diff --git a/main/taskpool.c b/main/taskpool.c index ca6a423d93..77e2c772e1 100644 --- a/main/taskpool.c +++ b/main/taskpool.c @@ -79,6 +79,20 @@ struct ast_taskpool { /*! \brief The threshold for a taskprocessor at which we consider the pool needing to grow (50% of high water threshold) */ #define TASKPOOL_GROW_THRESHOLD (AST_TASKPROCESSOR_HIGH_WATER_LEVEL * 5) / 10 +/*! + * \internal + * \brief Effective load of a pool taskprocessor + * + * The queue size alone does not account for the task currently being executed, + * so a taskprocessor inside a long running task reports the same load as an + * idle one. + */ +static long taskpool_taskprocessor_load(struct taskpool_taskprocessor *tp) +{ + return ast_taskprocessor_size(tp->taskprocessor) + + ast_taskprocessor_is_executing(tp->taskprocessor); +} + /*! \brief Scheduler used for dynamic pool shrinking */ static struct ast_sched_context *sched; @@ -277,7 +291,7 @@ static int taskpool_dynamic_pool_shrink(const void *data) *taskprocessor = AST_VECTOR_GET(&taskprocessors->taskprocessors, taskprocessor_num); /* Check to see if this has reached the growth threshold */ - *growth_threshold_reached = (ast_taskprocessor_size((*taskprocessor)->taskprocessor) >= pool->options.growth_threshold) ? 1 : 0; + *growth_threshold_reached = (taskpool_taskprocessor_load(*taskprocessor) >= pool->options.growth_threshold) ? 1 : 0; } /*! @@ -288,6 +302,7 @@ static void taskpool_least_full_selector(struct ast_taskpool *pool, struct taskp struct taskpool_taskprocessor **taskprocessor, unsigned int *growth_threshold_reached) { struct taskpool_taskprocessor *least_full = NULL; + long least_full_load = 0; unsigned int i; if (!AST_VECTOR_SIZE(&taskprocessors->taskprocessors)) { @@ -300,21 +315,23 @@ static void taskpool_least_full_selector(struct ast_taskpool *pool, struct taskp for (i = 0; i < AST_VECTOR_SIZE(&taskprocessors->taskprocessors); i++) { struct taskpool_taskprocessor *tp = AST_VECTOR_GET(&taskprocessors->taskprocessors, i); + long load = taskpool_taskprocessor_load(tp); - /* If this taskprocessor has no outstanding tasks, it is the best choice */ - if (!ast_taskprocessor_size(tp->taskprocessor)) { + /* If this taskprocessor has nothing queued and nothing in flight, it is the best choice */ + if (!load) { *taskprocessor = tp; return; } /* If any of the taskprocessors have reached the growth threshold then we should grow the pool */ - if (ast_taskprocessor_size(tp->taskprocessor) >= pool->options.growth_threshold) { + if (load >= pool->options.growth_threshold) { *growth_threshold_reached = 1; } - /* The taskprocessor with the fewest tasks should be used */ - if (!least_full || ast_taskprocessor_size(tp->taskprocessor) < ast_taskprocessor_size(least_full->taskprocessor)) { + /* The taskprocessor with the lowest load should be used */ + if (!least_full || load < least_full_load) { least_full = tp; + least_full_load = load; } } diff --git a/main/taskprocessor.c b/main/taskprocessor.c index 1eb7e450bb..53b3648ba4 100644 --- a/main/taskprocessor.c +++ b/main/taskprocessor.c @@ -1087,6 +1087,22 @@ long ast_taskprocessor_size(struct ast_taskprocessor *tps) return (tps) ? tps->tps_queue_size : -1; } +unsigned int ast_taskprocessor_is_executing(const struct ast_taskprocessor *tps) +{ + if (!tps) { + return 0; + } + + /* + * Read without the object lock. This is called from the taskpool + * selector on every push, and taking the lock there costs more than the + * value is worth: a stale answer only means one push is routed as if the + * taskprocessor had just changed state, which the selector already + * tolerates. + */ + return tps->executing; +} + struct ast_taskprocessor_listener *ast_taskprocessor_listener(struct ast_taskprocessor *tps) { return tps ? tps->listener : NULL; diff --git a/tests/test_taskpool.c b/tests/test_taskpool.c index d2789878c9..a4f25a6636 100644 --- a/tests/test_taskpool.c +++ b/tests/test_taskpool.c @@ -1067,6 +1067,214 @@ end: return CLI_SUCCESS; } +#define SEL_POOL 4 + +struct sel_data { + ast_mutex_t lock; + ast_cond_t cond; + int blocker_running; + int blocker_release; + int blocker_done; + int probe_ran; +}; + +static int sel_blocking_task(void *data) +{ + struct sel_data *sd = data; + + ast_mutex_lock(&sd->lock); + sd->blocker_running = 1; + ast_cond_broadcast(&sd->cond); + while (!sd->blocker_release) { + ast_cond_wait(&sd->cond, &sd->lock); + } + sd->blocker_done = 1; + ast_cond_broadcast(&sd->cond); + ast_mutex_unlock(&sd->lock); + return 0; +} + +static int sel_probe_task(void *data) +{ + struct sel_data *sd = data; + + ast_mutex_lock(&sd->lock); + sd->probe_ran = 1; + ast_cond_broadcast(&sd->cond); + ast_mutex_unlock(&sd->lock); + return 0; +} + +/*! + * \internal + * \brief Run the scenario once. + * + * \param test The test being run. + * \param probe_first Create the probe serializer before the blocking task runs. + * \param label Description of the scenario, used in the test output. + * \retval 1 the probe task ran + * \retval 0 the probe task did not run within the timeout + */ +static int sel_run_case(struct ast_test *test, int probe_first, const char *label) +{ + struct ast_taskpool *pool = NULL; + struct sel_data *sd = NULL; + struct ast_taskpool_options options = { + .version = AST_TASKPOOL_OPTIONS_VERSION, + .idle_timeout = 0, + .auto_increment = 0, + .minimum_size = SEL_POOL, + .initial_size = SEL_POOL, + .max_size = SEL_POOL, + }; + struct ast_taskprocessor *blocker_ser = NULL; + struct ast_taskprocessor *probe_ser = NULL; + int ran = 0; + int running; + struct timeval start; + struct timespec end; + + sd = ast_calloc(1, sizeof(*sd)); + if (!sd) { + return 0; + } + ast_mutex_init(&sd->lock); + ast_cond_init(&sd->cond, NULL); + + pool = ast_taskpool_create(label, &options); + if (!pool) { + ast_test_status_update(test, "%s: could not create pool\n", label); + goto cleanup; + } + + blocker_ser = ast_taskpool_serializer("blocker", pool); + if (!blocker_ser) { + goto cleanup; + } + + if (probe_first) { + probe_ser = ast_taskpool_serializer("probe", pool); + if (!probe_ser) { + goto cleanup; + } + } + + if (ast_taskprocessor_push(blocker_ser, sel_blocking_task, sd)) { + goto cleanup; + } + + /* Wait until the blocker is definitely inside its task. */ + start = ast_tvnow(); + end.tv_sec = start.tv_sec + 5; + end.tv_nsec = start.tv_usec * 1000; + + ast_mutex_lock(&sd->lock); + while (!sd->blocker_running + && ast_cond_timedwait(&sd->cond, &sd->lock, &end) != ETIMEDOUT) { + } + running = sd->blocker_running; + ast_mutex_unlock(&sd->lock); + + if (!running) { + ast_test_status_update(test, "%s: blocking task never started\n", label); + goto release; + } + + /* + * With probe_first == 0 the probe serializer is created here, while one + * executor is already busy and three are free. + */ + if (!probe_first) { + probe_ser = ast_taskpool_serializer("probe", pool); + if (!probe_ser) { + goto release; + } + } + + if (ast_taskprocessor_push(probe_ser, sel_probe_task, sd)) { + goto release; + } + + start = ast_tvnow(); + end.tv_sec = start.tv_sec + 3; + end.tv_nsec = start.tv_usec * 1000; + + ast_mutex_lock(&sd->lock); + while (!sd->probe_ran + && ast_cond_timedwait(&sd->cond, &sd->lock, &end) != ETIMEDOUT) { + } + ran = sd->probe_ran; + ast_mutex_unlock(&sd->lock); + + ast_test_status_update(test, + "%s: one executor busy, %d of %d free -> probe %s\n", + label, SEL_POOL - 1, SEL_POOL, + ran ? "ran on a free executor" : "did NOT run within 3s"); + +release: + /* Wait for the blocking task to leave so the pool is idle before shutdown */ + start = ast_tvnow(); + end.tv_sec = start.tv_sec + 3; + end.tv_nsec = start.tv_usec * 1000; + + ast_mutex_lock(&sd->lock); + sd->blocker_release = 1; + ast_cond_broadcast(&sd->cond); + while (!sd->blocker_done + && ast_cond_timedwait(&sd->cond, &sd->lock, &end) != ETIMEDOUT) { + } + ast_mutex_unlock(&sd->lock); + +cleanup: + ast_taskprocessor_unreference(blocker_ser); + ast_taskprocessor_unreference(probe_ser); + if (pool) { + ast_taskpool_shutdown(pool); + } + if (sd) { + ast_mutex_destroy(&sd->lock); + ast_cond_destroy(&sd->cond); + ast_free(sd); + } + return ran; +} + +AST_TEST_DEFINE(taskpool_selector_skips_busy_executor) +{ + int ran_after; + int ran_before; + + switch (cmd) { + case TEST_INIT: + info->name = "selector_skips_busy_executor"; + info->category = "/main/taskpool/"; + info->summary = "A task should not be assigned to an executor that is busy"; + info->description = + "Occupies one executor of a four-executor pool with a task that " + "blocks until released, then pushes a trivial task on a different " + "serializer while three executors are free. Runs the scenario " + "twice: once with the second serializer created after the blocking " + "task is already running, and once with it created beforehand, so " + "the result cannot be attributed to both serializers having been " + "created while the pool was idle."; + return AST_TEST_NOT_RUN; + case TEST_EXECUTE: + break; + } + + ran_after = sel_run_case(test, 0, "serializer created while an executor is busy"); + ran_before = sel_run_case(test, 1, "serializer created while the pool is idle"); + + if (!ran_after || !ran_before) { + ast_test_status_update(test, + "a trivial task was queued behind the blocked executor while %d " + "executors were free\n", SEL_POOL - 1); + return AST_TEST_FAIL; + } + + return AST_TEST_PASS; +} + static struct ast_cli_entry cli[] = { AST_CLI_DEFINE(handle_cli_taskpool_push_efficiency, "Push tasks to a taskpool and measure efficiency"), AST_CLI_DEFINE(handle_cli_taskpool_push_serializer_efficiency, "Push tasks to a taskpool in serializers and measure efficiency"), @@ -1085,6 +1293,7 @@ static int unload_module(void) AST_TEST_UNREGISTER(taskpool_serializer_suspension); AST_TEST_UNREGISTER(taskpool_serializer_multiple_suspension); AST_TEST_UNREGISTER(taskpool_serializer_push_wait_while_suspended_from_other_serializer); + AST_TEST_UNREGISTER(taskpool_selector_skips_busy_executor); return 0; } @@ -1101,6 +1310,7 @@ static int load_module(void) AST_TEST_REGISTER(taskpool_serializer_suspension); AST_TEST_REGISTER(taskpool_serializer_multiple_suspension); AST_TEST_REGISTER(taskpool_serializer_push_wait_while_suspended_from_other_serializer); + AST_TEST_REGISTER(taskpool_selector_skips_busy_executor); return AST_MODULE_LOAD_SUCCESS; }