/* Copyright (C) 2018 the mpv developers * * Permission to use, copy, modify, and/or distribute this software for any * purpose with or without fee is hereby granted, provided that the above * copyright notice and this permission notice appear in all copies. * * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. */ #include "common/common.h"#include "osdep/threads.h"#include "osdep/timer.h" #include "thread_pool.h" // Threads destroy themselves after this many seconds, if there's no new work// and the thread count is above the configured minimum.#define DESTROY_TIMEOUT 10 struct work { void (*fn)(void *ctx); void *fn_ctx;}; struct mp_thread_pool { int min_threads, max_threads; mp_mutex lock; mp_cond wakeup; // --- the following fields are protected by lock mp_thread *threads; int num_threads; // Number of threads which have taken up work and are still processing it. int busy_threads; bool terminate; struct work *work; int num_work;}; static MP_THREAD_VOID worker_thread(void *arg){ struct mp_thread_pool *pool = arg; mp_thread_set_name("worker"); mp_mutex_lock(&pool->lock); int64_t destroy_deadline = 0; bool got_timeout = false; while (1) { struct work work = {0}; if (pool->num_work > 0) { work = pool->work[pool->num_work - 1]; pool->num_work -= 1; } if (!work.fn) { if (got_timeout || pool->terminate) break; if (pool->num_threads > pool->min_threads) { if (!destroy_deadline) destroy_deadline = mp_time_ns() + MP_TIME_S_TO_NS(DESTROY_TIMEOUT); if (mp_cond_timedwait_until(&pool->wakeup, &pool->lock, destroy_deadline)) got_timeout = pool->num_threads > pool->min_threads; } else { mp_cond_wait(&pool->wakeup, &pool->lock); } continue; } pool->busy_threads += 1; mp_mutex_unlock(&pool->lock); work.fn(work.fn_ctx); mp_mutex_lock(&pool->lock); pool->busy_threads -= 1; destroy_deadline = 0; got_timeout = false; } // If no termination signal was given, it must mean we died because of a // timeout, and nobody is waiting for us. We have to remove ourselves. if (!pool->terminate) { for (int n = 0; n < pool->num_threads; n++) { if (mp_thread_id_equal(mp_thread_get_id(pool->threads[n]), mp_thread_current_id())) { mp_thread_detach(pool->threads[n]); MP_TARRAY_REMOVE_AT(pool->threads, pool->num_threads, n); mp_mutex_unlock(&pool->lock); MP_THREAD_RETURN(); } } MP_ASSERT_UNREACHABLE(); } mp_mutex_unlock(&pool->lock); MP_THREAD_RETURN();} static void thread_pool_dtor(void *ctx){ struct mp_thread_pool *pool = ctx; mp_mutex_lock(&pool->lock); pool->terminate = true; mp_cond_broadcast(&pool->wakeup); mp_thread *threads = pool->threads; int num_threads = pool->num_threads; pool->threads = NULL; pool->num_threads = 0; mp_mutex_unlock(&pool->lock); for (int n = 0; n < num_threads; n++) mp_thread_join(threads[n]); mp_assert(pool->num_work == 0); mp_assert(pool->num_threads == 0); mp_cond_destroy(&pool->wakeup); mp_mutex_destroy(&pool->lock);} static bool add_thread(struct mp_thread_pool *pool){ mp_thread thread; if (mp_thread_create(&thread, worker_thread, pool) != 0) return false; MP_TARRAY_APPEND(pool, pool->threads, pool->num_threads, thread); return true;} struct mp_thread_pool *mp_thread_pool_create(void *ta_parent, int init_threads, int min_threads, int max_threads){ mp_assert(min_threads >= 0); mp_assert(init_threads <= min_threads); mp_assert(max_threads > 0 && max_threads >= min_threads); struct mp_thread_pool *pool = talloc_zero(ta_parent, struct mp_thread_pool); talloc_set_destructor(pool, thread_pool_dtor); mp_mutex_init(&pool->lock); mp_cond_init(&pool->wakeup); pool->min_threads = min_threads; pool->max_threads = max_threads; mp_mutex_lock(&pool->lock); for (int n = 0; n < init_threads; n++) add_thread(pool); bool ok = pool->num_threads >= init_threads; mp_mutex_unlock(&pool->lock); if (!ok) TA_FREEP(&pool); return pool;} static bool thread_pool_add(struct mp_thread_pool *pool, void (*fn)(void *ctx), void *fn_ctx, bool allow_queue){ bool ok = true; mp_assert(fn); mp_mutex_lock(&pool->lock); struct work work = {fn, fn_ctx}; // If there are not enough threads to process all at once, but we can // create a new thread, then do so. If work is queued quickly, it can // happen that not all available threads have picked up work yet (up to // num_threads - busy_threads threads), which has to be accounted for. if (pool->busy_threads + pool->num_work + 1 > pool->num_threads && pool->num_threads < pool->max_threads) { if (!add_thread(pool)) { // If we can queue it, it'll get done as long as there is 1 thread. ok = allow_queue && pool->num_threads > 0; } } if (ok) { MP_TARRAY_INSERT_AT(pool, pool->work, pool->num_work, 0, work); mp_cond_signal(&pool->wakeup); } mp_mutex_unlock(&pool->lock); return ok;} bool mp_thread_pool_queue(struct mp_thread_pool *pool, void (*fn)(void *ctx), void *fn_ctx){ return thread_pool_add(pool, fn, fn_ctx, true);} bool mp_thread_pool_run(struct mp_thread_pool *pool, void (*fn)(void *ctx), void *fn_ctx){ return thread_pool_add(pool, fn, fn_ctx, false);}
/* Copyright (C) 2018 the mpv developers
*
* Permission to use, copy, modify, and/or distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#include "common/common.h"
#include "osdep/threads.h"
#include "osdep/timer.h"
#include "thread_pool.h"
// Threads destroy themselves after this many seconds, if there's no new work
// and the thread count is above the configured minimum.
#define DESTROY_TIMEOUT 10
struct work {
void (*fn)(void *ctx);
void *fn_ctx;
};
struct mp_thread_pool {
int min_threads, max_threads;
mp_mutex lock;
mp_cond wakeup;
// --- the following fields are protected by lock
mp_thread *threads;
int num_threads;
// Number of threads which have taken up work and are still processing it.
int busy_threads;
bool terminate;
struct work *work;
int num_work;
};
static MP_THREAD_VOID worker_thread(void *arg)
{
struct mp_thread_pool *pool = arg;
mp_thread_set_name("worker");
mp_mutex_lock(&pool->lock);
int64_t destroy_deadline = 0;
bool got_timeout = false;
while (1) {
struct work work = {0};
if (pool->num_work > 0) {
work = pool->work[pool->num_work - 1];
pool->num_work -= 1;
}
if (!work.fn) {
if (got_timeout || pool->terminate)
break;
if (pool->num_threads > pool->min_threads) {
if (!destroy_deadline)
destroy_deadline = mp_time_ns() + MP_TIME_S_TO_NS(DESTROY_TIMEOUT);
if (mp_cond_timedwait_until(&pool->wakeup, &pool->lock, destroy_deadline))
got_timeout = pool->num_threads > pool->min_threads;
} else {
mp_cond_wait(&pool->wakeup, &pool->lock);
}
continue;
}
pool->busy_threads += 1;
mp_mutex_unlock(&pool->lock);
work.fn(work.fn_ctx);
mp_mutex_lock(&pool->lock);
pool->busy_threads -= 1;
destroy_deadline = 0;
got_timeout = false;
}
// If no termination signal was given, it must mean we died because of a
// timeout, and nobody is waiting for us. We have to remove ourselves.
if (!pool->terminate) {
for (int n = 0; n < pool->num_threads; n++) {
if (mp_thread_id_equal(mp_thread_get_id(pool->threads[n]),
mp_thread_current_id()))
{
mp_thread_detach(pool->threads[n]);
MP_TARRAY_REMOVE_AT(pool->threads, pool->num_threads, n);
mp_mutex_unlock(&pool->lock);
MP_THREAD_RETURN();
}
}
MP_ASSERT_UNREACHABLE();
}
mp_mutex_unlock(&pool->lock);
MP_THREAD_RETURN();
}
static void thread_pool_dtor(void *ctx)
{
struct mp_thread_pool *pool = ctx;
mp_mutex_lock(&pool->lock);
pool->terminate = true;
mp_cond_broadcast(&pool->wakeup);
mp_thread *threads = pool->threads;
int num_threads = pool->num_threads;
pool->threads = NULL;
pool->num_threads = 0;
mp_mutex_unlock(&pool->lock);
for (int n = 0; n < num_threads; n++)
mp_thread_join(threads[n]);
mp_assert(pool->num_work == 0);
mp_assert(pool->num_threads == 0);
mp_cond_destroy(&pool->wakeup);
mp_mutex_destroy(&pool->lock);
}
static bool add_thread(struct mp_thread_pool *pool)
{
mp_thread thread;
if (mp_thread_create(&thread, worker_thread, pool) != 0)
return false;
MP_TARRAY_APPEND(pool, pool->threads, pool->num_threads, thread);
return true;
}
struct mp_thread_pool *mp_thread_pool_create(void *ta_parent, int init_threads,
int min_threads, int max_threads)
{
mp_assert(min_threads >= 0);
mp_assert(init_threads <= min_threads);
mp_assert(max_threads > 0 && max_threads >= min_threads);
struct mp_thread_pool *pool = talloc_zero(ta_parent, struct mp_thread_pool);
talloc_set_destructor(pool, thread_pool_dtor);
mp_mutex_init(&pool->lock);
mp_cond_init(&pool->wakeup);
pool->min_threads = min_threads;
pool->max_threads = max_threads;
mp_mutex_lock(&pool->lock);
for (int n = 0; n < init_threads; n++)
add_thread(pool);
bool ok = pool->num_threads >= init_threads;
mp_mutex_unlock(&pool->lock);
if (!ok)
TA_FREEP(&pool);
return pool;
}
static bool thread_pool_add(struct mp_thread_pool *pool, void (*fn)(void *ctx),
void *fn_ctx, bool allow_queue)
{
bool ok = true;
mp_assert(fn);
mp_mutex_lock(&pool->lock);
struct work work = {fn, fn_ctx};
// If there are not enough threads to process all at once, but we can
// create a new thread, then do so. If work is queued quickly, it can
// happen that not all available threads have picked up work yet (up to
// num_threads - busy_threads threads), which has to be accounted for.
if (pool->busy_threads + pool->num_work + 1 > pool->num_threads &&
pool->num_threads < pool->max_threads)
{
if (!add_thread(pool)) {
// If we can queue it, it'll get done as long as there is 1 thread.
ok = allow_queue && pool->num_threads > 0;
}
}
if (ok) {
MP_TARRAY_INSERT_AT(pool, pool->work, pool->num_work, 0, work);
mp_cond_signal(&pool->wakeup);
}
mp_mutex_unlock(&pool->lock);
return ok;
}
bool mp_thread_pool_queue(struct mp_thread_pool *pool, void (*fn)(void *ctx),
void *fn_ctx)
{
return thread_pool_add(pool, fn, fn_ctx, true);
}
bool mp_thread_pool_run(struct mp_thread_pool *pool, void (*fn)(void *ctx),
void *fn_ctx)
{
return thread_pool_add(pool, fn, fn_ctx, false);
}