thread/m_threadpool.c
/*! Insert a new task into the threadpool.  If there are no free slots, wait
 *  for one to open up.  Can insert multiple tasks at once.
 *  \param parent    initialized parent (user/consumer) of threadpool
 *  \param task      Callback for task to perform
 *  \param task_args Argument passed to task (array, one per task)
 *  \param num_tasks Number of tasks being inserted */
Insert a new task into the threadpool.  If there are no free slots, wait
for one to open up.  Can insert multiple tasks at once.
\param parent    initialized parent (user/consumer) of threadpool
\param task      Callback for task to perform
\param task_args Argument passed to task (array, one per task)
\param num_tasks Number of tasks being inserted
Insert a new task into the threadpool . If there are no free slots wait for one to open up . Can insert multiple tasks at once . \ param parent initialized parent ( user / consumer ) of threadpool \ param task Callback for task to perform \ param task_args Argument passed to task ( array one per task ) \ param num_tasks Number of tasks being inserted
static void M_threadpool_queue_insert(M_threadpool_parent_t *parent, void (*task)(void *), void **task_args, size_t num_tasks, void (*finished)(void *))
{
	M_bool          i_just_woke_up = M_FALSE;
	M_threadpool_t *pool           = parent->pool;
	M_thread_mutex_lock(pool->queue_lock);
	while (1) {
		if (pool->num_idle_threads <= M_llist_len(pool->queue) && pool->num_threads < pool->max_threads)
			M_threadpool_thread_spawn(pool);
		if (pool->queue_waiters == 0 || i_just_woke_up) {
			if (pool->queue_max_size > M_llist_len(pool->queue)) {
				M_threadpool_queue_t *q = M_malloc_zero(sizeof(*q));
				q->parent   = parent;
				q->task     = task;
				q->finished = finished;
				if (task_args != NULL)
					q->task_arg = *task_args;
				M_llist_insert(pool->queue, q);
				M_thread_cond_signal(pool->queue_ocond);
				if (task_args != NULL)
					task_args++;
				num_tasks--;
				if (num_tasks == 0) {
					break;
				} else {
					i_just_woke_up = M_TRUE;
					continue;
				}
			}
		}
		pool->queue_waiters++;
		M_thread_cond_wait(pool->queue_icond, pool->queue_lock);
		pool->queue_waiters--;
		i_just_woke_up = M_TRUE;
	}
	M_thread_mutex_unlock(pool->queue_lock);
}
static void M_threadpool_queue_insert ( M_threadpool_parent_t * parent , void ( * task ) ( void * ) , void * * task_args , size_t num_tasks , void ( * finished ) ( void * ) ) { M_bool i_just_woke_up = M_FALSE ; M_threadpool_t * pool = parent -> pool ; M_thread_mutex_lock ( pool -> queue_lock ) ; while ( 1 ) { if ( pool -> num_idle_threads <= M_llist_len ( pool -> queue ) && pool -> num_threads < pool -> max_threads ) M_threadpool_thread_spawn ( pool ) ; if ( pool -> queue_waiters == 0 || i_just_woke_up ) { if ( pool -> queue_max_size > M_llist_len ( pool -> queue ) ) { M_threadpool_queue_t * q = M_malloc_zero ( sizeof ( * q ) ) ; q -> parent = parent ; q -> task = task ; q -> finished = finished ; if ( task_args != NULL ) q -> task_arg = * task_args ; M_llist_insert ( pool -> queue , q ) ; M_thread_cond_signal ( pool -> queue_ocond ) ; if ( task_args != NULL ) task_args ++ ; num_tasks -- ; if ( num_tasks == 0 ) { break ; } else { i_just_woke_up = M_TRUE ; continue ; } } } pool -> queue_waiters ++ ; M_thread_cond_wait ( pool -> queue_icond , pool -> queue_lock ) ; pool -> queue_waiters -- ; i_just_woke_up = M_TRUE ; } M_thread_mutex_unlock ( pool -> queue_lock ) ; }
Insert a new task into the threadpool.
Insert a new task into the threadpool .
/* Spawn a new thread on demand if needed */ /* Wake up a thread waiting for things to be queued */ /* Allow us to keep looping while we have tasks to queue */