X-Git-Url: https://git.libre-soc.org/?a=blobdiff_plain;f=src%2Futil%2Fu_queue.c;h=8f6dc08b3326a9d5f4475c238ccd9f8286fe0e56;hb=d9eaac02e53944799fbadf6ab7ff6cc725b0483a;hp=48c5c79552d1b6159a7c3e7532172005803b33d0;hpb=d99cdc9d5902da8437eabbd2f6981d01e4f7b78f;p=mesa.git diff --git a/src/util/u_queue.c b/src/util/u_queue.c index 48c5c79552d..8f6dc08b332 100644 --- a/src/util/u_queue.c +++ b/src/util/u_queue.c @@ -26,14 +26,26 @@ #include "u_queue.h" -#include +#include "c11/threads.h" #include "util/os_time.h" #include "util/u_string.h" #include "util/u_thread.h" #include "u_process.h" -static void util_queue_killall_and_wait(struct util_queue *queue); +#if defined(__linux__) +#include +#include +#include +#endif + + +/* Define 256MB */ +#define S_256MB (256 * 1024 * 1024) + +static void +util_queue_kill_threads(struct util_queue *queue, unsigned keep_num_threads, + bool finish_locked); /**************************************************************************** * Wait for all queues to assert idle when exit() is called. @@ -54,7 +66,7 @@ atexit_handler(void) mtx_lock(&exit_mutex); /* Wait for all queues to assert idle. */ LIST_FOR_EACH_ENTRY(iter, &queue_list, head) { - util_queue_killall_and_wait(iter); + util_queue_kill_threads(iter, 0, false); } mtx_unlock(&exit_mutex); } @@ -62,7 +74,7 @@ atexit_handler(void) static void global_init(void) { - LIST_INITHEAD(&queue_list); + list_inithead(&queue_list); atexit(atexit_handler); } @@ -72,7 +84,7 @@ add_to_atexit_list(struct util_queue *queue) call_once(&atexit_once_flag, global_init); mtx_lock(&exit_mutex); - LIST_ADD(&queue->head, &queue_list); + list_add(&queue->head, &queue_list); mtx_unlock(&exit_mutex); } @@ -84,7 +96,7 @@ remove_from_atexit_list(struct util_queue *queue) mtx_lock(&exit_mutex); LIST_FOR_EACH_ENTRY_SAFE(iter, tmp, &queue_list, head) { if (iter == queue) { - LIST_DEL(&iter->head); + list_del(&iter->head); break; } } @@ -253,9 +265,16 @@ util_queue_thread_func(void *input) } #endif +#if defined(__linux__) + if (queue->flags & UTIL_QUEUE_INIT_USE_MINIMUM_PRIORITY) { + /* The nice() function can only set a maximum of 19. */ + setpriority(PRIO_PROCESS, syscall(SYS_gettid), 19); + } +#endif + if (strlen(queue->name) > 0) { char name[16]; - util_snprintf(name, sizeof(name), "%s%i", queue->name, thread_index); + snprintf(name, sizeof(name), "%s%i", queue->name, thread_index); u_thread_setname(name); } @@ -266,10 +285,11 @@ util_queue_thread_func(void *input) assert(queue->num_queued >= 0 && queue->num_queued <= queue->max_jobs); /* wait if the queue is empty */ - while (!queue->kill_threads && queue->num_queued == 0) + while (thread_index < queue->num_threads && queue->num_queued == 0) cnd_wait(&queue->has_queued_cond, &queue->lock); - if (queue->kill_threads) { + /* only kill threads that are above "num_threads" */ + if (thread_index >= queue->num_threads) { mtx_unlock(&queue->lock); break; } @@ -280,6 +300,8 @@ util_queue_thread_func(void *input) queue->num_queued--; cnd_signal(&queue->has_space_cond); + if (job.job) + queue->total_jobs_size -= job.job_size; mtx_unlock(&queue->lock); if (job.job) { @@ -290,17 +312,19 @@ util_queue_thread_func(void *input) } } - /* signal remaining jobs before terminating */ + /* signal remaining jobs if all threads are being terminated */ mtx_lock(&queue->lock); - for (unsigned i = queue->read_idx; i != queue->write_idx; - i = (i + 1) % queue->max_jobs) { - if (queue->jobs[i].job) { - util_queue_fence_signal(queue->jobs[i].fence); - queue->jobs[i].job = NULL; + if (queue->num_threads == 0) { + for (unsigned i = queue->read_idx; i != queue->write_idx; + i = (i + 1) % queue->max_jobs) { + if (queue->jobs[i].job) { + util_queue_fence_signal(queue->jobs[i].fence); + queue->jobs[i].job = NULL; + } } + queue->read_idx = queue->write_idx; + queue->num_queued = 0; } - queue->read_idx = queue->write_idx; - queue->num_queued = 0; mtx_unlock(&queue->lock); return 0; } @@ -321,21 +345,55 @@ util_queue_create_thread(struct util_queue *queue, unsigned index) } if (queue->flags & UTIL_QUEUE_INIT_USE_MINIMUM_PRIORITY) { -#if defined(__linux__) && defined(SCHED_IDLE) +#if defined(__linux__) && defined(SCHED_BATCH) struct sched_param sched_param = {0}; /* The nice() function can only set a maximum of 19. - * SCHED_IDLE is the same as nice = 20. + * SCHED_BATCH gives the scheduler a hint that this is a latency + * insensitive thread. * * Note that Linux only allows decreasing the priority. The original * priority can't be restored. */ - pthread_setschedparam(queue->threads[index], SCHED_IDLE, &sched_param); + pthread_setschedparam(queue->threads[index], SCHED_BATCH, &sched_param); #endif } return true; } +void +util_queue_adjust_num_threads(struct util_queue *queue, unsigned num_threads) +{ + num_threads = MIN2(num_threads, queue->max_threads); + num_threads = MAX2(num_threads, 1); + + mtx_lock(&queue->finish_lock); + unsigned old_num_threads = queue->num_threads; + + if (num_threads == old_num_threads) { + mtx_unlock(&queue->finish_lock); + return; + } + + if (num_threads < old_num_threads) { + util_queue_kill_threads(queue, num_threads, true); + mtx_unlock(&queue->finish_lock); + return; + } + + /* Create threads. + * + * We need to update num_threads first, because threads terminate + * when thread_index < num_threads. + */ + queue->num_threads = num_threads; + for (unsigned i = old_num_threads; i < num_threads; i++) { + if (!util_queue_create_thread(queue, i)) + break; + } + mtx_unlock(&queue->finish_lock); +} + bool util_queue_init(struct util_queue *queue, const char *name, @@ -367,13 +425,14 @@ util_queue_init(struct util_queue *queue, memset(queue, 0, sizeof(*queue)); if (process_len) { - util_snprintf(queue->name, sizeof(queue->name), "%.*s:%s", - process_len, process_name, name); + snprintf(queue->name, sizeof(queue->name), "%.*s:%s", + process_len, process_name, name); } else { - util_snprintf(queue->name, sizeof(queue->name), "%s", name); + snprintf(queue->name, sizeof(queue->name), "%s", name); } queue->flags = flags; + queue->max_threads = num_threads; queue->num_threads = num_threads; queue->max_jobs = max_jobs; @@ -425,25 +484,40 @@ fail: } static void -util_queue_killall_and_wait(struct util_queue *queue) +util_queue_kill_threads(struct util_queue *queue, unsigned keep_num_threads, + bool finish_locked) { unsigned i; /* Signal all threads to terminate. */ + if (!finish_locked) + mtx_lock(&queue->finish_lock); + + if (keep_num_threads >= queue->num_threads) { + mtx_unlock(&queue->finish_lock); + return; + } + mtx_lock(&queue->lock); - queue->kill_threads = 1; + unsigned old_num_threads = queue->num_threads; + /* Setting num_threads is what causes the threads to terminate. + * Then cnd_broadcast wakes them up and they will exit their function. + */ + queue->num_threads = keep_num_threads; cnd_broadcast(&queue->has_queued_cond); mtx_unlock(&queue->lock); - for (i = 0; i < queue->num_threads; i++) + for (i = keep_num_threads; i < old_num_threads; i++) thrd_join(queue->threads[i], NULL); - queue->num_threads = 0; + + if (!finish_locked) + mtx_unlock(&queue->finish_lock); } void util_queue_destroy(struct util_queue *queue) { - util_queue_killall_and_wait(queue); + util_queue_kill_threads(queue, 0, false); remove_from_atexit_list(queue); cnd_destroy(&queue->has_space_cond); @@ -459,12 +533,13 @@ util_queue_add_job(struct util_queue *queue, void *job, struct util_queue_fence *fence, util_queue_execute_func execute, - util_queue_execute_func cleanup) + util_queue_execute_func cleanup, + const size_t job_size) { struct util_queue_job *ptr; mtx_lock(&queue->lock); - if (queue->kill_threads) { + if (queue->num_threads == 0) { mtx_unlock(&queue->lock); /* well no good option here, but any leaks will be * short-lived as things are shutting down.. @@ -477,7 +552,8 @@ util_queue_add_job(struct util_queue *queue, assert(queue->num_queued >= 0 && queue->num_queued <= queue->max_jobs); if (queue->num_queued == queue->max_jobs) { - if (queue->flags & UTIL_QUEUE_INIT_RESIZE_IF_FULL) { + if (queue->flags & UTIL_QUEUE_INIT_RESIZE_IF_FULL && + queue->total_jobs_size + job_size < S_256MB) { /* If the queue is full, make it larger to avoid waiting for a free * slot. */ @@ -516,7 +592,10 @@ util_queue_add_job(struct util_queue *queue, ptr->fence = fence; ptr->execute = execute; ptr->cleanup = cleanup; + ptr->job_size = job_size; + queue->write_idx = (queue->write_idx + 1) % queue->max_jobs; + queue->total_jobs_size += ptr->job_size; queue->num_queued++; cnd_signal(&queue->has_queued_cond); @@ -576,9 +655,7 @@ void util_queue_finish(struct util_queue *queue) { util_barrier barrier; - struct util_queue_fence *fences = malloc(queue->num_threads * sizeof(*fences)); - - util_barrier_init(&barrier, queue->num_threads); + struct util_queue_fence *fences; /* If 2 threads were adding jobs for 2 different barries at the same time, * a deadlock would happen, because 1 barrier requires that all threads @@ -586,9 +663,19 @@ util_queue_finish(struct util_queue *queue) */ mtx_lock(&queue->finish_lock); + /* The number of threads can be changed to 0, e.g. by the atexit handler. */ + if (!queue->num_threads) { + mtx_unlock(&queue->finish_lock); + return; + } + + fences = malloc(queue->num_threads * sizeof(*fences)); + util_barrier_init(&barrier, queue->num_threads); + for (unsigned i = 0; i < queue->num_threads; ++i) { util_queue_fence_init(&fences[i]); - util_queue_add_job(queue, &barrier, &fences[i], util_queue_finish_execute, NULL); + util_queue_add_job(queue, &barrier, &fences[i], + util_queue_finish_execute, NULL, 0); } for (unsigned i = 0; i < queue->num_threads; ++i) {