Something went wrong. Try again.
🎥 Command line media player
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224/* 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 <pthread.h>
#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;
pthread_mutex_t lock; pthread_cond_t wakeup;
// --- the following fields are protected by lock
pthread_t *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 void *worker_thread(void *arg){ struct mp_thread_pool *pool = arg;
mpthread_set_name("worker");
pthread_mutex_lock(&pool->lock);
struct timespec ts = {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 (!ts.tv_sec && !ts.tv_nsec) ts = mp_rel_time_to_timespec(DESTROY_TIMEOUT); if (pthread_cond_timedwait(&pool->wakeup, &pool->lock, &ts)) got_timeout = pool->num_threads > pool->min_threads; } else { pthread_cond_wait(&pool->wakeup, &pool->lock); } continue; }
pool->busy_threads += 1; pthread_mutex_unlock(&pool->lock);
work.fn(work.fn_ctx);
pthread_mutex_lock(&pool->lock); pool->busy_threads -= 1;
ts = (struct timespec){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 (pthread_equal(pool->threads[n], pthread_self())) { pthread_detach(pthread_self()); MP_TARRAY_REMOVE_AT(pool->threads, pool->num_threads, n); pthread_mutex_unlock(&pool->lock); return NULL; } } assert(0); }
pthread_mutex_unlock(&pool->lock); return NULL;}
static void thread_pool_dtor(void *ctx){ struct mp_thread_pool *pool = ctx;
pthread_mutex_lock(&pool->lock);
pool->terminate = true; pthread_cond_broadcast(&pool->wakeup);
pthread_t *threads = pool->threads; int num_threads = pool->num_threads;
pool->threads = NULL; pool->num_threads = 0;
pthread_mutex_unlock(&pool->lock);
for (int n = 0; n < num_threads; n++) pthread_join(threads[n], NULL);
assert(pool->num_work == 0); assert(pool->num_threads == 0); pthread_cond_destroy(&pool->wakeup); pthread_mutex_destroy(&pool->lock);}
static bool add_thread(struct mp_thread_pool *pool){ pthread_t thread;
if (pthread_create(&thread, NULL, 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){ assert(min_threads >= 0); assert(init_threads <= min_threads); 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);
pthread_mutex_init(&pool->lock, NULL); pthread_cond_init(&pool->wakeup, NULL);
pool->min_threads = min_threads; pool->max_threads = max_threads;
pthread_mutex_lock(&pool->lock); for (int n = 0; n < init_threads; n++) add_thread(pool); bool ok = pool->num_threads >= init_threads; pthread_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;
assert(fn);
pthread_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); pthread_cond_signal(&pool->wakeup); }
pthread_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);}