381 lines
8.3 KiB
C
381 lines
8.3 KiB
C
#include <bh/internal/thread.h>
|
|
#include <stdlib.h>
|
|
|
|
/**
|
|
* \defgroup thread Multithreading
|
|
*
|
|
* Multithreading API
|
|
* \{
|
|
*/
|
|
|
|
/**
|
|
* Creates new task object with specified function \a func, \a data and \a
|
|
* flags.
|
|
*
|
|
* If the \a flags contain BH_THREAD_CLEANUP, upon successful completion task
|
|
* will be automaticly destroyed.
|
|
*
|
|
* \param func Task's function
|
|
* \param data Task's data
|
|
* \param flags Task's flags
|
|
*
|
|
* \return On success, returns new task object.
|
|
* \return On failure, returns null pointer.
|
|
*/
|
|
bh_task_t *bh_task_new(void (*func)(void *),
|
|
void *data,
|
|
int flags)
|
|
{
|
|
bh_task_t *result;
|
|
|
|
result = malloc(sizeof(bh_task_t));
|
|
if (result)
|
|
bh_task_init(result, func, data, flags);
|
|
|
|
return result;
|
|
}
|
|
|
|
/**
|
|
* Frees the \a task object.
|
|
*
|
|
* \param task Pointer to the task object.
|
|
*/
|
|
void bh_task_free(bh_task_t *task)
|
|
{
|
|
bh_task_destroy(task);
|
|
free(task);
|
|
}
|
|
|
|
/**
|
|
* Initializes the \a task object with specified function \a func, \a data and
|
|
* \a flags.
|
|
*
|
|
* If the \a flags contain BH_THREAD_CLEANUP, upon successful completion task
|
|
* will be automaticly destroyed.
|
|
*
|
|
* \warning This is an internal function.
|
|
*
|
|
* \param task Pointer to the task object
|
|
* \param func Task's function
|
|
* \param data Task's data
|
|
* \param flags Task's flags
|
|
*/
|
|
void bh_task_init(bh_task_t *task,
|
|
void (*func)(void *),
|
|
void *data,
|
|
int flags)
|
|
{
|
|
task->func = func;
|
|
task->data = data;
|
|
task->flags = flags;
|
|
}
|
|
|
|
/**
|
|
* Destroyes the \a task object.
|
|
*
|
|
* \warning This is an internal function.
|
|
*
|
|
* \param task Pointer to the task object
|
|
*/
|
|
void bh_task_destroy(bh_task_t *task)
|
|
{
|
|
(void)task;
|
|
}
|
|
|
|
/**
|
|
* Reuses the \a task object for different purpose.
|
|
*
|
|
* \param task Pointer to the task object
|
|
* \param func New task's function
|
|
* \param data New task's data
|
|
*/
|
|
void bh_task_reuse(bh_task_t *task,
|
|
void (*func)(void *),
|
|
void *data)
|
|
{
|
|
/* If the task is done - reinitilize it with new data */
|
|
if (task->flags & BH_THREAD_DONE)
|
|
bh_task_init(task, func, data, task->flags & ~(BH_THREAD_DONE));
|
|
}
|
|
|
|
/**
|
|
* Checks if the \a task is done.
|
|
*
|
|
* \param task Pointer to the task object
|
|
*
|
|
* \return If the task is done, returns non-zero.
|
|
* \return If the task is not done, returns zero.
|
|
*/
|
|
int bh_task_done(bh_task_t *task)
|
|
{
|
|
return (task->flags & BH_THREAD_DONE) != 0;
|
|
}
|
|
|
|
/**
|
|
* Creates the mutex object.
|
|
*
|
|
* \return On success, returns new mutex object.
|
|
* \return On failure, returns null pointer.
|
|
*/
|
|
bh_mutex_t *bh_mutex_new(void)
|
|
{
|
|
bh_mutex_t *result;
|
|
|
|
result = malloc(sizeof(*result));
|
|
if (result && bh_mutex_init(result))
|
|
{
|
|
free(result);
|
|
result = NULL;
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
/**
|
|
* Frees the \a mutex object.
|
|
*
|
|
* \param mutex Pointer to the mutex object
|
|
*/
|
|
void bh_mutex_free(bh_mutex_t *mutex)
|
|
{
|
|
bh_mutex_destroy(mutex);
|
|
free(mutex);
|
|
}
|
|
|
|
/**
|
|
* Creates the semaphore object with initial \a count value.
|
|
*
|
|
* \param count Initial semaphore value
|
|
*
|
|
* \return On success, returns new semaphore object.
|
|
* \return On failure, returns null pointer.
|
|
*
|
|
* \note Guranteed maximum \a count value is 32767.
|
|
*/
|
|
bh_semaphore_t *bh_semaphore_new(int count)
|
|
{
|
|
bh_semaphore_t *result;
|
|
|
|
result = malloc(sizeof(*result));
|
|
if (result && bh_semaphore_init(result, count))
|
|
{
|
|
free(result);
|
|
result = NULL;
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
/**
|
|
* Frees the \a semaphore object.
|
|
*
|
|
* \param semaphore Pointer to the semaphore object
|
|
*/
|
|
void bh_semaphore_free(bh_semaphore_t *semaphore)
|
|
{
|
|
bh_semaphore_destroy(semaphore);
|
|
free(semaphore);
|
|
}
|
|
|
|
/**
|
|
* Creates the condition variable object.
|
|
*
|
|
* \return On success, returns new condition variable object.
|
|
* \return On failure, returns null pointer.
|
|
*/
|
|
bh_cond_t *bh_cond_new(void)
|
|
{
|
|
bh_cond_t *result;
|
|
|
|
result = malloc(sizeof(*result));
|
|
if (result && bh_cond_init(result))
|
|
{
|
|
free(result);
|
|
result = NULL;
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
/**
|
|
* Frees the condition variable object \a cond.
|
|
*
|
|
* \param cond Pointer to the condition variable object
|
|
*/
|
|
void bh_cond_free(bh_cond_t *cond)
|
|
{
|
|
bh_cond_destroy(cond);
|
|
free(cond);
|
|
}
|
|
|
|
/**
|
|
* Adds \a task to the thread \a pool.
|
|
*
|
|
* \param pool Pointer to the thread pool object
|
|
* \param task Pointer to the task object
|
|
*
|
|
* \return On success, returns zero.
|
|
* \return On failure, returns error code.
|
|
*/
|
|
int bh_thread_pool_add(bh_thread_pool_t *pool,
|
|
bh_task_t *task)
|
|
{
|
|
void *iter;
|
|
bh_task_t *item;
|
|
|
|
/* Queue task for execution */
|
|
bh_mutex_lock(&pool->lock);
|
|
if (bh_queue_insert(&pool->tasks, task))
|
|
{
|
|
bh_mutex_unlock(&pool->lock);
|
|
return BH_ERROR;
|
|
}
|
|
|
|
/* Signal new job */
|
|
bh_mutex_unlock(&pool->lock);
|
|
bh_cond_signal(&pool->new_task);
|
|
|
|
return BH_OK;
|
|
}
|
|
|
|
/**
|
|
* Waits unitl all task in thread \a pool are complete.
|
|
*
|
|
* \param pool Pointer to the thread pool.
|
|
*
|
|
* \return On success, returns zero.
|
|
* \return On failure, returns error code.
|
|
*/
|
|
int bh_thread_pool_wait(bh_thread_pool_t *pool)
|
|
{
|
|
/* Lock and check if there is jobs in the queue */
|
|
bh_mutex_lock(&pool->lock);
|
|
while (!bh_queue_empty(&pool->tasks))
|
|
{
|
|
bh_task_t *task;
|
|
|
|
/* Fetch task from the queue */
|
|
task = bh_queue_front(&pool->tasks);
|
|
bh_queue_remove(&pool->tasks);
|
|
|
|
/* Unlock and do the task */
|
|
bh_mutex_unlock(&pool->lock);
|
|
|
|
task->func(task->data);
|
|
task->flags |= BH_THREAD_DONE;
|
|
if (task->flags & BH_THREAD_CLEANUP)
|
|
bh_task_free(task);
|
|
|
|
/* Lock and check for the next task */
|
|
bh_mutex_lock(&pool->lock);
|
|
}
|
|
|
|
/* Check until there is no active threads */
|
|
while (pool->active)
|
|
bh_cond_wait(&pool->done_task, &pool->lock);
|
|
|
|
/* Unlock */
|
|
bh_mutex_unlock(&pool->lock);
|
|
|
|
return BH_OK;
|
|
}
|
|
|
|
/**
|
|
* Destroyes the thread \a pool object.
|
|
*
|
|
* \warning This is an internal function.
|
|
*
|
|
* \param pool Pointer to the thread pool object
|
|
*/
|
|
void bh_thread_pool_destroy(bh_thread_pool_t *pool)
|
|
{
|
|
size_t i;
|
|
|
|
/* Broadcast shutdown */
|
|
pool->shutdown = 1;
|
|
bh_cond_broadcast(&pool->new_task);
|
|
|
|
/* Join and delete every thread */
|
|
for (i = 0; i < pool->size; i++)
|
|
bh_thread_join(pool->threads + i);
|
|
|
|
/* Destroy any tasks with cleanup flag */
|
|
while (!bh_queue_empty(&pool->tasks))
|
|
{
|
|
bh_task_t *task;
|
|
|
|
task = bh_queue_front(&pool->tasks);
|
|
bh_queue_remove(&pool->tasks);
|
|
|
|
if (task->flags & BH_THREAD_CLEANUP)
|
|
bh_task_free(task);
|
|
}
|
|
|
|
/* Destroy thread array, queue, lock and condition variables */
|
|
free(pool->threads);
|
|
bh_queue_destroy(&pool->tasks);
|
|
bh_cond_destroy(&pool->new_task);
|
|
bh_cond_destroy(&pool->done_task);
|
|
bh_mutex_destroy(&pool->lock);
|
|
}
|
|
|
|
/**
|
|
* Frees the thread \a pool object.
|
|
*
|
|
* \param pool Pointer to the thread pool object
|
|
*/
|
|
void bh_thread_pool_free(bh_thread_pool_t *pool)
|
|
{
|
|
bh_thread_pool_destroy(pool);
|
|
free(pool);
|
|
}
|
|
|
|
/**
|
|
* \warning This is an internal function.
|
|
* \warning Do not use this function directly!
|
|
*/
|
|
void bh_thread_pool_worker(void *arg)
|
|
{
|
|
bh_thread_pool_t *pool;
|
|
|
|
pool = (bh_thread_pool_t *)arg;
|
|
while (1)
|
|
{
|
|
bh_task_t *task;
|
|
|
|
/* Wait unitl there is a task or shutdown signal */
|
|
bh_mutex_lock(&pool->lock);
|
|
while (!pool->shutdown && bh_queue_empty(&pool->tasks))
|
|
bh_cond_wait(&pool->new_task, &pool->lock);
|
|
|
|
/* If its shutdown signal - exit the loop */
|
|
if (pool->shutdown)
|
|
{
|
|
bh_mutex_unlock(&pool->lock);
|
|
break;
|
|
}
|
|
|
|
/* Fetch task from the front of the queue, increase active counter */
|
|
task = bh_queue_front(&pool->tasks);
|
|
bh_queue_remove(&pool->tasks);
|
|
pool->active++;
|
|
bh_mutex_unlock(&pool->lock);
|
|
|
|
/* Do the task, mark as done, and if required - free it */
|
|
task->func(task->data);
|
|
task->flags |= BH_THREAD_DONE;
|
|
if (task->flags & BH_THREAD_CLEANUP)
|
|
bh_task_free(task);
|
|
|
|
/* Decrease active counter and broadcast that we are done */
|
|
bh_mutex_lock(&pool->lock);
|
|
pool->active--;
|
|
|
|
bh_mutex_unlock(&pool->lock);
|
|
bh_cond_broadcast(&pool->done_task);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* \}
|
|
*/ |