diff --git a/include/internal/task_def.h b/include/internal/task_def.h index 0e3b027..858b49e 100644 --- a/include/internal/task_def.h +++ b/include/internal/task_def.h @@ -38,6 +38,7 @@ extern "C" { struct task { uint32_t id; //!< Task identifier. bool is_orphan; //!< Flag indicating whether the task is orphaned. + bool auto_release; //!< Flag indicating whether the task must be released after its execution. struct task *next; //!< Next element of the current instance. struct task *prev; //!< Previous element of the current instance. void *(*task)(void *); //!< Task callback. diff --git a/include/pool_day.h b/include/pool_day.h index 7947dcd..f7f030d 100644 --- a/include/pool_day.h +++ b/include/pool_day.h @@ -81,7 +81,7 @@ uint32_t queued_tasks(pool_day_t pool); * * pool = create_pool(size); * if (pool) { - * task_t t = create_task(id, callback, (void *)param, param_size, NULL, NULL); + * task_t t = create_sync_task(id, callback, (void *)param, param_size); * * enqueue_task(pool, t); * @@ -100,6 +100,15 @@ uint32_t queued_tasks(pool_day_t pool); */ void *get_task_result(task_t task); +/** + * @brief Wait for the finish of a given task without retrieving its result. + * + * @note This function blocks the current thread. + * + * @param[in] task Task handle. + */ +void wait_task_finish(task_t task); + /** * @brief Abort the execution of incoming tasks. * diff --git a/include/task.h b/include/task.h index 4bdde85..40c3e53 100644 --- a/include/task.h +++ b/include/task.h @@ -24,6 +24,7 @@ #ifndef TASK_H_ #define TASK_H_ +#include #include #include @@ -34,23 +35,42 @@ extern "C" { typedef struct task *task_t; //!< Handle to the task. /** - * @brief Create a new task. + * @brief Create a new synchronous task. * - * @note The created task doesn't require a manual release once it's bound to a + * @note The created task requires a manual release once it's bound to a * pool. * * @param[in] id Task identifier. * @param[in] task Task callback. * @param[in] param Task parameter. * @param[in] param_size Size of the task parameter. + * + * @return Handle to the new task. + */ +task_t create_sync_task(uint32_t id, void *(*task)(void *), void *param, + size_t param_size); + +/** + * @brief Create a new asynchronous task. + * + * @note When the 'auto_release' parameter is set to false, the created task + * requires a manual release once it's bound to a pool. + * + * @param[in] id Task identifier. + * @param[in] task Task callback. + * @param[in] param Task parameter. + * @param[in] param_size Size of the task parameter. + * @param[in] auto_release Flag indicating whether the task must be released + * after its execution. * @param[in] start_cb Callback executed when the task starts. * @param[in] end_cb Callback executed when the task ends. * * @return Handle to the new task. */ -task_t create_task(uint32_t id, void *(*task)(void *), void *param, - size_t param_size, void (*start_cb)(uint32_t, void *), - void (*end_cb)(uint32_t, void *, void *)); +task_t create_async_task(uint32_t id, void *(*task)(void *), void *param, + size_t param_size, bool auto_release, + void (*start_cb)(uint32_t, void *), + void (*end_cb)(uint32_t, void *, void *)); /** * @brief Destroy a task. diff --git a/samples/c/main.c b/samples/c/main.c index d17f16c..a98f8cb 100644 --- a/samples/c/main.c +++ b/samples/c/main.c @@ -33,8 +33,8 @@ int main(void) { } char str[] = "foo"; - task_t task = create_task(DUMMY_TASK_ID, func, (void *)str, - sizeof(char) * strlen(str) + 1, NULL, NULL); + task_t task = create_sync_task(DUMMY_TASK_ID, func, (void *)str, + sizeof(char) * strlen(str) + 1); assert(enqueue_task(pool, task) == POOL_DAY_SUCCESS); char *ret = (char *)get_task_result(task); diff --git a/samples/cpp/main.cc b/samples/cpp/main.cc index cfa77d9..1502250 100644 --- a/samples/cpp/main.cc +++ b/samples/cpp/main.cc @@ -9,12 +9,10 @@ class Foo { public: Foo() : pool_{create_pool(2)}, - t1_{create_task(0, Foo::Cb1, (void *)"hello", - sizeof(char) * std::strlen("hello") + 1, nullptr, - nullptr)}, - t2_{create_task(1, Foo::Cb2, (void *)"hi", - sizeof(char) * std::strlen("hi") + 1, nullptr, - nullptr)} { + t1_{create_sync_task(0, Foo::Cb1, (void *)"hello", + sizeof(char) * std::strlen("hello") + 1)}, + t2_{create_sync_task(1, Foo::Cb2, (void *)"hi", + sizeof(char) * std::strlen("hi") + 1)} { } void RunTasks(void) { diff --git a/src/pool_day.c b/src/pool_day.c index d7a806f..fb6ca69 100644 --- a/src/pool_day.c +++ b/src/pool_day.c @@ -26,8 +26,8 @@ __static void *thread_func(void *param) { pool_day_t pool = (pool_day_t)param; while (!pool->must_stop) { + POOL_DAY_DEBUG("thread '0x%x' waiting...", pthread_self()); sem_wait(&pool->lock); - POOL_DAY_DEBUG("thread '0x%x' woke up", pthread_self()); if (pool->must_stop) { @@ -36,30 +36,36 @@ __static void *thread_func(void *param) { } task_t entry = dequeue(pool->tasks); - if (entry) { - entry->is_orphan = true; + if (!entry) { + POOL_DAY_DEBUG("thread '0x%x' found no task to execute", pthread_self()); + continue; + } + + entry->is_orphan = true; + + if (entry->on_task_start) { + POOL_DAY_DEBUG("thread '0x%x' executing start callback for task '0x%x'", + pthread_self(), entry->id); + entry->on_task_start(entry->id, entry->param); + } - if (entry->on_task_start) { - entry->on_task_start(entry->id, entry->param); - } + POOL_DAY_DEBUG("thread '0x%x' running the task '0x%x'", pthread_self(), + entry->id); + void *ret = entry->task(entry->param); + POOL_DAY_DEBUG("thread '0x%x' finished the task '0x%x'", pthread_self(), + entry->id); - POOL_DAY_DEBUG("thread '0x%x' running the task '0x%x'", pthread_self(), - entry->id); - void *ret = entry->task(entry->param); - POOL_DAY_DEBUG("thread '0x%x' finished the task '0x%x'", pthread_self(), - entry->id); + if (entry->on_task_end) { + POOL_DAY_DEBUG("thread '0x%x' executing end callback for task '0x%x'", + pthread_self(), entry->id); + entry->on_task_end(entry->id, entry->param, ret); + } + if (!entry->auto_release) { entry->ret_val = ret; - if (entry->on_task_end) { - POOL_DAY_DEBUG("thread '0x%x' running async teardown for task '0x%x'", - pthread_self(), entry->id); - entry->on_task_end(entry->id, entry->param, entry->ret_val); - destroy_task(entry); - } else { - POOL_DAY_DEBUG("thread '0x%x' running sync teardown for task '0x%x'", - pthread_self(), entry->id); - sem_post(&entry->ready); - } + sem_post(&entry->ready); + } else { + destroy_task(entry); } } @@ -180,7 +186,6 @@ uint32_t queued_tasks(pool_day_t pool) { return pool ? queue_size(pool->tasks) : 0; } -// cppcheck-suppress unusedFunction void *get_task_result(task_t task) { if (!task) { POOL_DAY_ERROR("null task provided"); @@ -193,3 +198,8 @@ void *get_task_result(task_t task) { return task->ret_val; } + +// cppcheck-suppress unusedFunction +void wait_task_finish(task_t task) { + (void)get_task_result(task); +} diff --git a/src/task.c b/src/task.c index b65f74e..788e3ce 100644 --- a/src/task.c +++ b/src/task.c @@ -6,9 +6,15 @@ #include "internal/task_def.h" // cppcheck-suppress unusedFunction -task_t create_task(uint32_t id, void *(*task)(void *), void *param, - size_t param_size, void (*start_cb)(uint32_t, void *), - void (*end_cb)(uint32_t, void *, void *)) { +task_t create_sync_task(uint32_t id, void *(*task)(void *), void *param, + size_t param_size) { + return create_async_task(id, task, param, param_size, false, NULL, NULL); +} + +task_t create_async_task(uint32_t id, void *(*task)(void *), void *param, + size_t param_size, bool auto_release, + void (*start_cb)(uint32_t, void *), + void (*end_cb)(uint32_t, void *, void *)) { task_t node; node = (task_t)calloc(1, sizeof(*node)); @@ -18,6 +24,7 @@ task_t create_task(uint32_t id, void *(*task)(void *), void *param, node->on_task_start = start_cb; node->on_task_end = end_cb; node->is_orphan = true; + node->auto_release = auto_release; if (param) { node->param = malloc(param_size); diff --git a/test/src/pool_day_test.cc b/test/src/pool_day_test.cc index 6a2c535..416df83 100644 --- a/test/src/pool_day_test.cc +++ b/test/src/pool_day_test.cc @@ -57,7 +57,7 @@ TEST_F(PoolDayTest, CreatePollWithInvalidSize) { * it, then it must be added to the pool's task queue. */ TEST_F(PoolDayTest, EnqueueSingleTaskWithPoolEmpty) { - auto task = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); + auto task = create_sync_task(0, nullptr, nullptr, 0); EXPECT_EQ(queued_tasks(pool_), 0); EXPECT_EQ(enqueue_task(pool_, task), POOL_DAY_SUCCESS); @@ -72,11 +72,11 @@ TEST_F(PoolDayTest, EnqueueSingleTaskWithPoolEmpty) { * it, then they must be added to the pool's task queue. */ TEST_F(PoolDayTest, EnqueueTaskWithPoolNotEmpty) { - auto t1 = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); - auto t2 = create_task(1, nullptr, nullptr, 0, nullptr, nullptr); - auto t3 = create_task(2, nullptr, nullptr, 0, nullptr, nullptr); - auto t4 = create_task(3, nullptr, nullptr, 0, nullptr, nullptr); - auto t5 = create_task(4, nullptr, nullptr, 0, nullptr, nullptr); + auto t1 = create_sync_task(0, nullptr, nullptr, 0); + auto t2 = create_sync_task(1, nullptr, nullptr, 0); + auto t3 = create_sync_task(2, nullptr, nullptr, 0); + auto t4 = create_sync_task(3, nullptr, nullptr, 0); + auto t5 = create_sync_task(4, nullptr, nullptr, 0); EXPECT_EQ(queued_tasks(pool_), 0); @@ -101,7 +101,7 @@ TEST_F(PoolDayTest, EnqueueTaskWithPoolNotEmpty) { * then nothing must happen and the suitable error code must be returned. */ TEST_F(PoolDayTest, EnqueueTaskWithNullPool) { - auto task = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); + auto task = create_sync_task(0, nullptr, nullptr, 0); EXPECT_EQ(enqueue_task(nullptr, task), POOL_DAY_ERROR_NULL_PARAM); free(task); @@ -112,29 +112,12 @@ TEST_F(PoolDayTest, EnqueueTaskWithNullPool) { * then nothing must happen and the suitable error code must be returned. */ TEST_F(PoolDayTest, EnqueueTaskWithNullTask) { - auto task = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); + auto task = create_sync_task(0, nullptr, nullptr, 0); EXPECT_EQ(enqueue_task(pool_, nullptr), POOL_DAY_ERROR_NULL_PARAM); free(task); } -/** - * @brief Given we have a valid pool and a task already bound to a pool, when we - * try to enqueue it, then nothing must happen and the suitable error code - * must be returned. - */ -TEST_F(PoolDayTest, EnqueueTaskAlreadyBound) { - auto another_pool = create_pool(1); - auto task = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); - - EXPECT_EQ(enqueue_task(pool_, task), POOL_DAY_SUCCESS); - - destroy_pool(&another_pool); - - // to force the task destruction on destroy_pool call - task->is_orphan = true; -} - /** * @brief Given we have a pool with no tasks, when we try to get the number of * queued tasks of the pool, then 0 must be returned. @@ -148,7 +131,7 @@ TEST_F(PoolDayTest, GetQueuedTasksCountWithNoTasks) { * the number of queued tasks in the pool, then 1 must be returned. */ TEST_F(PoolDayTest, GetQueuedTasksCountWithSingleTask) { - auto task = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); + auto task = create_sync_task(0, nullptr, nullptr, 0); enqueue_task(pool_, task); @@ -164,11 +147,11 @@ TEST_F(PoolDayTest, GetQueuedTasksCountWithSingleTask) { * returned. */ TEST_F(PoolDayTest, GetQueuedTasksCountWithSeveralTasks) { - auto t1 = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); - auto t2 = create_task(1, nullptr, nullptr, 0, nullptr, nullptr); - auto t3 = create_task(2, nullptr, nullptr, 0, nullptr, nullptr); - auto t4 = create_task(3, nullptr, nullptr, 0, nullptr, nullptr); - auto t5 = create_task(4, nullptr, nullptr, 0, nullptr, nullptr); + auto t1 = create_sync_task(0, nullptr, nullptr, 0); + auto t2 = create_sync_task(1, nullptr, nullptr, 0); + auto t3 = create_sync_task(2, nullptr, nullptr, 0); + auto t4 = create_sync_task(3, nullptr, nullptr, 0); + auto t5 = create_sync_task(4, nullptr, nullptr, 0); enqueue_task(pool_, t1); enqueue_task(pool_, t2); @@ -209,7 +192,7 @@ TEST_F(PoolDayTest, DestroyPollWithNullHandle) { * scheduled for execution, then the task's callback must be called. */ TEST_F(PoolDayTest, ExecuteTaskWithNullParameterWithSuccess) { - auto task = create_task(0, CbWrapper::TaskCb, nullptr, 0, nullptr, nullptr); + auto task = create_sync_task(0, CbWrapper::TaskCb, nullptr, 0); char ret_val[]{ "hello, i'm the return of the task" }; enqueue_task(pool_, task); @@ -241,8 +224,42 @@ TEST_F(PoolDayTest, ExecuteTaskWithNullParameterWithSuccess) { * suitable times. */ TEST_F(PoolDayTest, ExecuteTaskWithCallbacks) { - auto task = create_task(0, CbWrapper::TaskCb, nullptr, 0, - CbWrapper::OnTaskStartCb, CbWrapper::OnTaskEndCb); + auto task = create_async_task(0, CbWrapper::TaskCb, nullptr, 0, false, + CbWrapper::OnTaskStartCb, + CbWrapper::OnTaskEndCb); + char ret_val[]{ "hello, i'm the return of the task" }; + + enqueue_task(pool_, task); + + EXPECT_CALL(CbWrapper::mock(), OnTaskStartCb(task->id, task->param)) + .Times(1); + EXPECT_CALL(CbWrapper::mock(), TaskCb(nullptr)) + .Times(1) + .WillOnce( + InvokeWithoutArgs([&]() { + abort_tasks(pool_); // to break the thread loop + return ret_val; + }) + ); + EXPECT_CALL(CbWrapper::mock(), OnTaskEndCb(task->id, task->param, ret_val)) + .Times(1); + + auto ret = thread_func(pool_); + EXPECT_EQ(ret, nullptr); + + // cleanup + destroy_task(task); +} + +/** + * @brief Given the pool has one task with callbacks with 'auto release' set, + * when the task is scheduled for execution, then the task's callbacks must be + * called at the suitable times and the task must be destroyed. + */ +TEST_F(PoolDayTest, ExecuteTaskWithCallbacksAndAutoRelease) { + auto task = create_async_task(0, CbWrapper::TaskCb, nullptr, 0, true, + CbWrapper::OnTaskStartCb, + CbWrapper::OnTaskEndCb); char ret_val[]{ "hello, i'm the return of the task" }; enqueue_task(pool_, task); @@ -269,8 +286,8 @@ TEST_F(PoolDayTest, ExecuteTaskWithCallbacks) { * scheduled for execution, then the task's start callback must be called. */ TEST_F(PoolDayTest, ExecuteTaskWithStartCallbackOnly) { - auto task = create_task(0, CbWrapper::TaskCb, nullptr, 0, - CbWrapper::OnTaskStartCb, nullptr); + auto task = create_async_task(0, CbWrapper::TaskCb, nullptr, 0, false, + CbWrapper::OnTaskStartCb, nullptr); char ret_val[]{ "hello, i'm the return of the task" }; enqueue_task(pool_, task); @@ -301,8 +318,8 @@ TEST_F(PoolDayTest, ExecuteTaskWithStartCallbackOnly) { * scheduled for execution, then the task's end callback must be called. */ TEST_F(PoolDayTest, ExecuteTaskWithEndCallbackOnly) { - auto task = create_task(0, CbWrapper::TaskCb, nullptr, 0, nullptr, - CbWrapper::OnTaskEndCb); + auto task = create_async_task(0, CbWrapper::TaskCb, nullptr, 0, false, + nullptr, CbWrapper::OnTaskEndCb); char ret_val[]{ "hello, i'm the return of the task" }; enqueue_task(pool_, task); @@ -322,6 +339,9 @@ TEST_F(PoolDayTest, ExecuteTaskWithEndCallbackOnly) { auto ret = thread_func(pool_); EXPECT_EQ(ret, nullptr); + + // cleanup + destroy_task(task); } /** @@ -332,8 +352,9 @@ TEST_F(PoolDayTest, ExecuteTaskWithEndCallbackOnly) { TEST_F(PoolDayTest, ExecuteTaskWithParameterWithSuccess) { char param[]{ "param" }; char ret_val[]{ "hello, i'm the return of the task" }; - auto task = create_task(0, CbWrapper::TaskCb, param, - sizeof(char) * strlen(param) + 1, nullptr, nullptr); + auto task = create_async_task(0, CbWrapper::TaskCb, param, + sizeof(char) * strlen(param) + 1, false, + nullptr, nullptr); enqueue_task(pool_, task); @@ -363,7 +384,7 @@ TEST_F(PoolDayTest, ExecuteTaskWithParameterWithSuccess) { * task is scheduled for execution, then the task's callback must not be called. */ TEST_F(PoolDayTest, ExecuteTaskWithMustStopSet) { - auto task = create_task(0, CbWrapper::TaskCb, nullptr, 0, nullptr, nullptr); + auto task = create_sync_task(0, CbWrapper::TaskCb, nullptr, 0); enqueue_task(pool_, task); @@ -393,8 +414,8 @@ TEST_F(PoolDayTest, AbortTasksWithNullPoolHandle) { TEST_F(PoolDayTest, GetTaskResultWithSuccess) { int ret_val{1234}; char param[]{ "param" }; - auto task = create_task(0, CbWrapper::TaskCb, param, - sizeof(char) * strlen(param) + 1, nullptr, nullptr); + auto task = create_sync_task(0, CbWrapper::TaskCb, param, + sizeof(char) * strlen(param) + 1); { enqueue_task(pool_, task); @@ -435,7 +456,7 @@ TEST_F(PoolDayTest, GetTaskResultWithNullPoolTask) { * the finish of the task, then null must be returned. */ TEST_F(PoolDayTest, GetTaskResultWithWithUnboundTask) { - auto task = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); + auto task = create_sync_task(0, nullptr, nullptr, 0); EXPECT_EQ(get_task_result(task), nullptr); diff --git a/test/src/queue_test.cc b/test/src/queue_test.cc index a7b5ef7..cc4221b 100644 --- a/test/src/queue_test.cc +++ b/test/src/queue_test.cc @@ -33,11 +33,11 @@ TEST_F(QueueTest, GetQueueSizeWithEmptyQueue) { */ TEST_F(QueueTest, GetQueueSizeWithNotEmptyQueue) { { - auto t1 = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); - auto t2 = create_task(1, nullptr, nullptr, 0, nullptr, nullptr); - auto t3 = create_task(2, nullptr, nullptr, 0, nullptr, nullptr); - auto t4 = create_task(3, nullptr, nullptr, 0, nullptr, nullptr); - auto t5 = create_task(4, nullptr, nullptr, 0, nullptr, nullptr); + auto t1 = create_sync_task(0, nullptr, nullptr, 0); + auto t2 = create_sync_task(1, nullptr, nullptr, 0); + auto t3 = create_sync_task(2, nullptr, nullptr, 0); + auto t4 = create_sync_task(3, nullptr, nullptr, 0); + auto t5 = create_sync_task(4, nullptr, nullptr, 0); enqueue(queue_, t1); enqueue(queue_, t2); @@ -63,8 +63,8 @@ TEST_F(QueueTest, GetQueueSizeWithNullQueue) { */ TEST_F(QueueTest, EnqueueElementWithNotEmptyQueue) { { - auto t1 = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); - auto t2 = create_task(1, nullptr, nullptr, 0, nullptr, nullptr); + auto t1 = create_sync_task(0, nullptr, nullptr, 0); + auto t2 = create_sync_task(1, nullptr, nullptr, 0); enqueue(queue_, t1); enqueue(queue_, t2); @@ -72,7 +72,7 @@ TEST_F(QueueTest, EnqueueElementWithNotEmptyQueue) { EXPECT_EQ(queue_size(queue_), 2); } - auto t3 = create_task(3, nullptr, nullptr, 0, nullptr, nullptr); + auto t3 = create_sync_task(3, nullptr, nullptr, 0); enqueue(queue_, t3); @@ -95,8 +95,8 @@ TEST_F(QueueTest, EnqueueNullElementWithEmptyQueue) { */ TEST_F(QueueTest, EnqueueNullElementWithNotEmptyQueue) { { - auto t1 = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); - auto t2 = create_task(1, nullptr, nullptr, 0, nullptr, nullptr); + auto t1 = create_sync_task(0, nullptr, nullptr, 0); + auto t2 = create_sync_task(1, nullptr, nullptr, 0); enqueue(queue_, t1); enqueue(queue_, t2); @@ -112,7 +112,7 @@ TEST_F(QueueTest, EnqueueNullElementWithNotEmptyQueue) { * then nothing must happen. */ TEST_F(QueueTest, EnqueueElementWithNullQueue) { - auto t1 = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); + auto t1 = create_sync_task(0, nullptr, nullptr, 0); enqueue(nullptr, t1); free(t1); @@ -138,9 +138,9 @@ TEST_F(QueueTest, DequeueElementWithEmptyQueue) { * size of the queue must be decreased by 1. */ TEST_F(QueueTest, DequeueSingleElementWithNotEmptyQueue) { - auto t1 = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); - auto t2 = create_task(1, nullptr, nullptr, 0, nullptr, nullptr); - auto t3 = create_task(2, nullptr, nullptr, 0, nullptr, nullptr); + auto t1 = create_sync_task(0, nullptr, nullptr, 0); + auto t2 = create_sync_task(1, nullptr, nullptr, 0); + auto t3 = create_sync_task(2, nullptr, nullptr, 0); enqueue(queue_, t1); enqueue(queue_, t2); @@ -163,10 +163,10 @@ TEST_F(QueueTest, DequeueSingleElementWithNotEmptyQueue) { * the operations. */ TEST_F(QueueTest, DequeueMultipleElementsWithNotEmptyQueue) { - auto t1 = create_task(0, nullptr, nullptr, 0, nullptr, nullptr); - auto t2 = create_task(1, nullptr, nullptr, 0, nullptr, nullptr); - auto t3 = create_task(2, nullptr, nullptr, 0, nullptr, nullptr); - auto t4 = create_task(3, nullptr, nullptr, 0, nullptr, nullptr); + auto t1 = create_sync_task(0, nullptr, nullptr, 0); + auto t2 = create_sync_task(1, nullptr, nullptr, 0); + auto t3 = create_sync_task(2, nullptr, nullptr, 0); + auto t4 = create_sync_task(3, nullptr, nullptr, 0); task_t ret; enqueue(queue_, t1);