/*
* maximum number of threads
+ * 0 means no real thread, only strict sync processing.
*/
- int max_threads;
+ unsigned max_threads;
/*
* Number of threads
*/
- int num_threads;
+ unsigned num_threads;
/*
* Number of idle threads
*/
- int num_idle;
+ unsigned num_idle;
/*
- * Condition variable indicating that we should quickly go
- * away making way for fork() without anybody waiting on
- * pool->condvar.
+ * Condition variable indicating that helper threads should
+ * quickly go away making way for fork() without anybody
+ * waiting on pool->condvar.
*/
pthread_cond_t *prefork_cond;
+
+ /*
+ * Waiting position for helper threads while fork is
+ * running. The forking thread will have locked it, and all
+ * idle helper threads will sit here until after the fork,
+ * where the forking thread will unlock it again.
+ */
+ pthread_mutex_t fork_mutex;
};
static pthread_mutex_t pthreadpools_mutex = PTHREAD_MUTEX_INITIALIZER;
return ret;
}
+ ret = pthread_mutex_init(&pool->fork_mutex, NULL);
+ if (ret != 0) {
+ pthread_cond_destroy(&pool->condvar);
+ pthread_mutex_destroy(&pool->mutex);
+ free(pool->jobs);
+ free(pool);
+ return ret;
+ }
+
pool->shutdown = false;
pool->num_threads = 0;
pool->max_threads = max_threads;
ret = pthread_mutex_lock(&pthreadpools_mutex);
if (ret != 0) {
+ pthread_mutex_destroy(&pool->fork_mutex);
pthread_cond_destroy(&pool->condvar);
pthread_mutex_destroy(&pool->mutex);
free(pool->jobs);
return 0;
}
+size_t pthreadpool_max_threads(struct pthreadpool *pool)
+{
+ return pool->max_threads;
+}
+
+size_t pthreadpool_queued_jobs(struct pthreadpool *pool)
+{
+ int res;
+ int unlock_res;
+ size_t ret;
+
+ res = pthread_mutex_lock(&pool->mutex);
+ if (res != 0) {
+ return 0;
+ }
+
+ ret = pool->num_jobs;
+
+ unlock_res = pthread_mutex_unlock(&pool->mutex);
+ assert(unlock_res == 0);
+ return ret;
+}
+
static void pthreadpool_prepare_pool(struct pthreadpool *pool)
{
- pthread_cond_t prefork_cond = PTHREAD_COND_INITIALIZER;
int ret;
+ ret = pthread_mutex_lock(&pool->fork_mutex);
+ assert(ret == 0);
+
ret = pthread_mutex_lock(&pool->mutex);
assert(ret == 0);
while (pool->num_idle != 0) {
+ unsigned num_idle = pool->num_idle;
+ pthread_cond_t prefork_cond;
+
+ ret = pthread_cond_init(&prefork_cond, NULL);
+ assert(ret == 0);
+
/*
- * Exit all idle threads, which are all blocked in
- * pool->condvar. In the child we can destroy the
- * pool, which would result in undefined behaviour in
- * the pthread_cond_destroy(pool->condvar). glibc just
+ * Push all idle threads off pool->condvar. In the
+ * child we can destroy the pool, which would result
+ * in undefined behaviour in the
+ * pthread_cond_destroy(pool->condvar). glibc just
* blocks here.
*/
pool->prefork_cond = &prefork_cond;
ret = pthread_cond_signal(&pool->condvar);
assert(ret == 0);
- ret = pthread_cond_wait(&prefork_cond, &pool->mutex);
- assert(ret == 0);
+ while (pool->num_idle == num_idle) {
+ ret = pthread_cond_wait(&prefork_cond, &pool->mutex);
+ assert(ret == 0);
+ }
pool->prefork_cond = NULL;
- }
- ret = pthread_cond_destroy(&prefork_cond);
- assert(ret == 0);
+ ret = pthread_cond_destroy(&prefork_cond);
+ assert(ret == 0);
+ }
/*
* Probably it's well-defined somewhere: What happens to
assert(ret == 0);
ret = pthread_mutex_unlock(&pool->mutex);
assert(ret == 0);
+ ret = pthread_mutex_unlock(&pool->fork_mutex);
+ assert(ret == 0);
}
ret = pthread_mutex_unlock(&pthreadpools_mutex);
ret = pthread_cond_init(&pool->condvar, NULL);
assert(ret == 0);
+
ret = pthread_mutex_unlock(&pool->mutex);
assert(ret == 0);
+
+ ret = pthread_mutex_unlock(&pool->fork_mutex);
+ assert(ret == 0);
}
ret = pthread_mutex_unlock(&pthreadpools_mutex);
static int pthreadpool_free(struct pthreadpool *pool)
{
- int ret, ret1;
+ int ret, ret1, ret2;
ret = pthread_mutex_lock(&pthreadpools_mutex);
if (ret != 0) {
ret = pthread_mutex_unlock(&pthreadpools_mutex);
assert(ret == 0);
+ ret = pthread_mutex_lock(&pool->mutex);
+ assert(ret == 0);
+ ret = pthread_mutex_unlock(&pool->mutex);
+ assert(ret == 0);
+
ret = pthread_mutex_destroy(&pool->mutex);
ret1 = pthread_cond_destroy(&pool->condvar);
+ ret2 = pthread_mutex_destroy(&pool->fork_mutex);
if (ret != 0) {
return ret;
if (ret1 != 0) {
return ret1;
}
+ if (ret2 != 0) {
+ return ret2;
+ }
free(pool->jobs);
free(pool);
return true;
}
+static void pthreadpool_undo_put_job(struct pthreadpool *p)
+{
+ p->num_jobs -= 1;
+}
+
static void *pthreadpool_server(void *arg)
{
struct pthreadpool *pool = (struct pthreadpool *)arg;
/*
* Me must allow fork() to continue
* without anybody waiting on
- * &pool->condvar.
+ * &pool->condvar. Tell
+ * pthreadpool_prepare_pool that we
+ * got that message.
*/
- pthread_cond_signal(pool->prefork_cond);
- pthreadpool_server_exit(pool);
- return NULL;
+
+ res = pthread_cond_signal(pool->prefork_cond);
+ assert(res == 0);
+
+ res = pthread_mutex_unlock(&pool->mutex);
+ assert(res == 0);
+
+ /*
+ * pthreadpool_prepare_pool has
+ * already locked this mutex across
+ * the fork. This makes us wait
+ * without sitting in a condvar.
+ */
+ res = pthread_mutex_lock(&pool->fork_mutex);
+ assert(res == 0);
+ res = pthread_mutex_unlock(&pool->fork_mutex);
+ assert(res == 0);
+
+ res = pthread_mutex_lock(&pool->mutex);
+ assert(res == 0);
}
if (res == ETIMEDOUT) {
}
}
-int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
- void (*fn)(void *private_data), void *private_data)
+static int pthreadpool_create_thread(struct pthreadpool *pool)
{
pthread_attr_t thread_attr;
pthread_t thread_id;
int res;
sigset_t mask, omask;
+ /*
+ * Create a new worker thread. It should not receive any signals.
+ */
+
+ sigfillset(&mask);
+
+ res = pthread_attr_init(&thread_attr);
+ if (res != 0) {
+ return res;
+ }
+
+ res = pthread_attr_setdetachstate(
+ &thread_attr, PTHREAD_CREATE_DETACHED);
+ if (res != 0) {
+ pthread_attr_destroy(&thread_attr);
+ return res;
+ }
+
+ res = pthread_sigmask(SIG_BLOCK, &mask, &omask);
+ if (res != 0) {
+ pthread_attr_destroy(&thread_attr);
+ return res;
+ }
+
+ res = pthread_create(&thread_id, &thread_attr, pthreadpool_server,
+ (void *)pool);
+
+ assert(pthread_sigmask(SIG_SETMASK, &omask, NULL) == 0);
+
+ pthread_attr_destroy(&thread_attr);
+
+ if (res == 0) {
+ pool->num_threads += 1;
+ }
+
+ return res;
+}
+
+int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
+ void (*fn)(void *private_data), void *private_data)
+{
+ int res;
+ int unlock_res;
+
res = pthread_mutex_lock(&pool->mutex);
if (res != 0) {
return res;
* Protect against the pool being shut down while
* trying to add a job
*/
- res = pthread_mutex_unlock(&pool->mutex);
- assert(res == 0);
+ unlock_res = pthread_mutex_unlock(&pool->mutex);
+ assert(unlock_res == 0);
return EINVAL;
}
+ if (pool->max_threads == 0) {
+ unlock_res = pthread_mutex_unlock(&pool->mutex);
+ assert(unlock_res == 0);
+
+ /*
+ * If no thread are allowed we do strict sync processing.
+ */
+ fn(private_data);
+ res = pool->signal_fn(job_id, fn, private_data,
+ pool->signal_fn_private_data);
+ return res;
+ }
+
/*
* Add job to the end of the queue
*/
if (!pthreadpool_put_job(pool, job_id, fn, private_data)) {
- pthread_mutex_unlock(&pool->mutex);
+ unlock_res = pthread_mutex_unlock(&pool->mutex);
+ assert(unlock_res == 0);
return ENOMEM;
}
* We have idle threads, wake one.
*/
res = pthread_cond_signal(&pool->condvar);
- pthread_mutex_unlock(&pool->mutex);
+ if (res != 0) {
+ pthreadpool_undo_put_job(pool);
+ }
+ unlock_res = pthread_mutex_unlock(&pool->mutex);
+ assert(unlock_res == 0);
return res;
}
- if ((pool->max_threads != 0) &&
- (pool->num_threads >= pool->max_threads)) {
+ if (pool->num_threads >= pool->max_threads) {
/*
* No more new threads, we just queue the request
*/
- pthread_mutex_unlock(&pool->mutex);
+ unlock_res = pthread_mutex_unlock(&pool->mutex);
+ assert(unlock_res == 0);
+ return 0;
+ }
+
+ res = pthreadpool_create_thread(pool);
+ if (res == 0) {
+ unlock_res = pthread_mutex_unlock(&pool->mutex);
+ assert(unlock_res == 0);
+ return 0;
+ }
+
+ if (pool->num_threads != 0) {
+ /*
+ * At least one thread is still available, let
+ * that one run the queued job.
+ */
+ unlock_res = pthread_mutex_unlock(&pool->mutex);
+ assert(unlock_res == 0);
return 0;
}
/*
- * Create a new worker thread. It should not receive any signals.
+ * No thread could be created to run job, fallback to sync
+ * call.
*/
+ pthreadpool_undo_put_job(pool);
- sigfillset(&mask);
+ unlock_res = pthread_mutex_unlock(&pool->mutex);
+ assert(unlock_res == 0);
- res = pthread_attr_init(&thread_attr);
- if (res != 0) {
- pthread_mutex_unlock(&pool->mutex);
- return res;
- }
+ return res;
+}
- res = pthread_attr_setdetachstate(
- &thread_attr, PTHREAD_CREATE_DETACHED);
- if (res != 0) {
- pthread_attr_destroy(&thread_attr);
- pthread_mutex_unlock(&pool->mutex);
- return res;
- }
+size_t pthreadpool_cancel_job(struct pthreadpool *pool, int job_id,
+ void (*fn)(void *private_data), void *private_data)
+{
+ int res;
+ size_t i, j;
+ size_t num = 0;
- res = pthread_sigmask(SIG_BLOCK, &mask, &omask);
+ res = pthread_mutex_lock(&pool->mutex);
if (res != 0) {
- pthread_attr_destroy(&thread_attr);
- pthread_mutex_unlock(&pool->mutex);
return res;
}
- res = pthread_create(&thread_id, &thread_attr, pthreadpool_server,
- (void *)pool);
- if (res == 0) {
- pool->num_threads += 1;
+ for (i = 0, j = 0; i < pool->num_jobs; i++) {
+ size_t idx = (pool->head + i) % pool->jobs_array_len;
+ size_t new_idx = (pool->head + j) % pool->jobs_array_len;
+ struct pthreadpool_job *job = &pool->jobs[idx];
+
+ if ((job->private_data == private_data) &&
+ (job->id == job_id) &&
+ (job->fn == fn))
+ {
+ /*
+ * Just skip the entry.
+ */
+ num++;
+ continue;
+ }
+
+ /*
+ * If we already removed one or more jobs (so j will be smaller
+ * then i), we need to fill possible gaps in the logical list.
+ */
+ if (j < i) {
+ pool->jobs[new_idx] = *job;
+ }
+ j++;
}
- assert(pthread_sigmask(SIG_SETMASK, &omask, NULL) == 0);
+ pool->num_jobs -= num;
- pthread_attr_destroy(&thread_attr);
+ res = pthread_mutex_unlock(&pool->mutex);
+ assert(res == 0);
- pthread_mutex_unlock(&pool->mutex);
- return res;
+ return num;
}
git.samba.org / garming / samba-autobuild / .git / blobdiff