Lines Matching full:queue
48 util_queue_kill_threads(struct util_queue *queue, unsigned keep_num_threads,
86 add_to_atexit_list(struct util_queue *queue) in add_to_atexit_list() argument
91 list_add(&queue->head, &queue_list); in add_to_atexit_list()
96 remove_from_atexit_list(struct util_queue *queue) in remove_from_atexit_list() argument
102 if (iter == queue) { in remove_from_atexit_list()
230 struct util_queue *queue; member
237 struct util_queue *queue = ((struct thread_input*)input)->queue; in util_queue_thread_func() local
242 if (queue->flags & UTIL_QUEUE_INIT_SET_FULL_THREAD_AFFINITY) { in util_queue_thread_func()
255 if (queue->flags & UTIL_QUEUE_INIT_USE_MINIMUM_PRIORITY) { in util_queue_thread_func()
261 if (strlen(queue->name) > 0) { in util_queue_thread_func()
263 snprintf(name, sizeof(name), "%s%i", queue->name, thread_index); in util_queue_thread_func()
270 mtx_lock(&queue->lock); in util_queue_thread_func()
271 assert(queue->num_queued >= 0 && queue->num_queued <= queue->max_jobs); in util_queue_thread_func()
273 /* wait if the queue is empty */ in util_queue_thread_func()
274 while (thread_index < queue->num_threads && queue->num_queued == 0) in util_queue_thread_func()
275 cnd_wait(&queue->has_queued_cond, &queue->lock); in util_queue_thread_func()
278 if (thread_index >= queue->num_threads) { in util_queue_thread_func()
279 mtx_unlock(&queue->lock); in util_queue_thread_func()
283 job = queue->jobs[queue->read_idx]; in util_queue_thread_func()
284 memset(&queue->jobs[queue->read_idx], 0, sizeof(struct util_queue_job)); in util_queue_thread_func()
285 queue->read_idx = (queue->read_idx + 1) % queue->max_jobs; in util_queue_thread_func()
287 queue->num_queued--; in util_queue_thread_func()
288 cnd_signal(&queue->has_space_cond); in util_queue_thread_func()
290 queue->total_jobs_size -= job.job_size; in util_queue_thread_func()
291 mtx_unlock(&queue->lock); in util_queue_thread_func()
303 mtx_lock(&queue->lock); in util_queue_thread_func()
304 if (queue->num_threads == 0) { in util_queue_thread_func()
305 for (unsigned i = queue->read_idx; i != queue->write_idx; in util_queue_thread_func()
306 i = (i + 1) % queue->max_jobs) { in util_queue_thread_func()
307 if (queue->jobs[i].job) { in util_queue_thread_func()
308 if (queue->jobs[i].fence) in util_queue_thread_func()
309 util_queue_fence_signal(queue->jobs[i].fence); in util_queue_thread_func()
310 queue->jobs[i].job = NULL; in util_queue_thread_func()
313 queue->read_idx = queue->write_idx; in util_queue_thread_func()
314 queue->num_queued = 0; in util_queue_thread_func()
316 mtx_unlock(&queue->lock); in util_queue_thread_func()
321 util_queue_create_thread(struct util_queue *queue, unsigned index) in util_queue_create_thread() argument
325 input->queue = queue; in util_queue_create_thread()
328 if (thrd_success != u_thread_create(queue->threads + index, util_queue_thread_func, input)) { in util_queue_create_thread()
333 if (queue->flags & UTIL_QUEUE_INIT_USE_MINIMUM_PRIORITY) { in util_queue_create_thread()
344 pthread_setschedparam(queue->threads[index], SCHED_BATCH, &sched_param); in util_queue_create_thread()
351 util_queue_adjust_num_threads(struct util_queue *queue, unsigned num_threads, in util_queue_adjust_num_threads() argument
354 num_threads = MIN2(num_threads, queue->max_threads); in util_queue_adjust_num_threads()
358 mtx_lock(&queue->lock); in util_queue_adjust_num_threads()
360 unsigned old_num_threads = queue->num_threads; in util_queue_adjust_num_threads()
364 mtx_unlock(&queue->lock); in util_queue_adjust_num_threads()
369 util_queue_kill_threads(queue, num_threads, true); in util_queue_adjust_num_threads()
371 mtx_unlock(&queue->lock); in util_queue_adjust_num_threads()
380 queue->num_threads = num_threads; in util_queue_adjust_num_threads()
382 if (!util_queue_create_thread(queue, i)) { in util_queue_adjust_num_threads()
383 queue->num_threads = i; in util_queue_adjust_num_threads()
389 mtx_unlock(&queue->lock); in util_queue_adjust_num_threads()
393 util_queue_init(struct util_queue *queue, in util_queue_init() argument
412 const int max_chars = sizeof(queue->name) - 1; in util_queue_init()
421 memset(queue, 0, sizeof(*queue)); in util_queue_init()
424 snprintf(queue->name, sizeof(queue->name), "%.*s:%s", in util_queue_init()
427 snprintf(queue->name, sizeof(queue->name), "%s", name); in util_queue_init()
430 queue->create_threads_on_demand = true; in util_queue_init()
431 queue->flags = flags; in util_queue_init()
432 queue->max_threads = num_threads; in util_queue_init()
433 queue->num_threads = 1; in util_queue_init()
434 queue->max_jobs = max_jobs; in util_queue_init()
435 queue->global_data = global_data; in util_queue_init()
437 (void) mtx_init(&queue->lock, mtx_plain); in util_queue_init()
439 queue->num_queued = 0; in util_queue_init()
440 cnd_init(&queue->has_queued_cond); in util_queue_init()
441 cnd_init(&queue->has_space_cond); in util_queue_init()
443 queue->jobs = (struct util_queue_job*) in util_queue_init()
445 if (!queue->jobs) in util_queue_init()
448 queue->threads = (thrd_t*) calloc(queue->max_threads, sizeof(thrd_t)); in util_queue_init()
449 if (!queue->threads) in util_queue_init()
453 for (i = 0; i < queue->num_threads; i++) { in util_queue_init()
454 if (!util_queue_create_thread(queue, i)) { in util_queue_init()
460 queue->num_threads = i; in util_queue_init()
466 add_to_atexit_list(queue); in util_queue_init()
470 free(queue->threads); in util_queue_init()
472 if (queue->jobs) { in util_queue_init()
473 cnd_destroy(&queue->has_space_cond); in util_queue_init()
474 cnd_destroy(&queue->has_queued_cond); in util_queue_init()
475 mtx_destroy(&queue->lock); in util_queue_init()
476 free(queue->jobs); in util_queue_init()
479 memset(queue, 0, sizeof(*queue)); in util_queue_init()
484 util_queue_kill_threads(struct util_queue *queue, unsigned keep_num_threads, in util_queue_kill_threads() argument
489 mtx_lock(&queue->lock); in util_queue_kill_threads()
491 if (keep_num_threads >= queue->num_threads) { in util_queue_kill_threads()
493 mtx_unlock(&queue->lock); in util_queue_kill_threads()
497 unsigned old_num_threads = queue->num_threads; in util_queue_kill_threads()
501 queue->num_threads = keep_num_threads; in util_queue_kill_threads()
502 cnd_broadcast(&queue->has_queued_cond); in util_queue_kill_threads()
507 mtx_unlock(&queue->lock); in util_queue_kill_threads()
509 thrd_join(queue->threads[i], NULL); in util_queue_kill_threads()
511 mtx_lock(&queue->lock); in util_queue_kill_threads()
514 mtx_unlock(&queue->lock); in util_queue_kill_threads()
527 util_queue_destroy(struct util_queue *queue) in util_queue_destroy() argument
529 util_queue_kill_threads(queue, 0, false); in util_queue_destroy()
531 /* This makes it safe to call on a queue that failed util_queue_init. */ in util_queue_destroy()
532 if (queue->head.next != NULL) in util_queue_destroy()
533 remove_from_atexit_list(queue); in util_queue_destroy()
535 cnd_destroy(&queue->has_space_cond); in util_queue_destroy()
536 cnd_destroy(&queue->has_queued_cond); in util_queue_destroy()
537 mtx_destroy(&queue->lock); in util_queue_destroy()
538 free(queue->jobs); in util_queue_destroy()
539 free(queue->threads); in util_queue_destroy()
543 util_queue_add_job_locked(struct util_queue *queue, in util_queue_add_job_locked() argument
554 mtx_lock(&queue->lock); in util_queue_add_job_locked()
555 if (queue->num_threads == 0) { in util_queue_add_job_locked()
557 mtx_unlock(&queue->lock); in util_queue_add_job_locked()
567 assert(queue->num_queued >= 0 && queue->num_queued <= queue->max_jobs); in util_queue_add_job_locked()
570 if (queue->num_queued > 0 && in util_queue_add_job_locked()
571 queue->create_threads_on_demand && in util_queue_add_job_locked()
573 queue->num_threads < queue->max_threads) { in util_queue_add_job_locked()
574 util_queue_adjust_num_threads(queue, queue->num_threads + 1, true); in util_queue_add_job_locked()
577 if (queue->num_queued == queue->max_jobs) { in util_queue_add_job_locked()
578 if (queue->flags & UTIL_QUEUE_INIT_RESIZE_IF_FULL && in util_queue_add_job_locked()
579 queue->total_jobs_size + job_size < S_256MB) { in util_queue_add_job_locked()
580 /* If the queue is full, make it larger to avoid waiting for a free in util_queue_add_job_locked()
583 unsigned new_max_jobs = queue->max_jobs + 8; in util_queue_add_job_locked()
591 unsigned i = queue->read_idx; in util_queue_add_job_locked()
594 jobs[num_jobs++] = queue->jobs[i]; in util_queue_add_job_locked()
595 i = (i + 1) % queue->max_jobs; in util_queue_add_job_locked()
596 } while (i != queue->write_idx); in util_queue_add_job_locked()
598 assert(num_jobs == queue->num_queued); in util_queue_add_job_locked()
600 free(queue->jobs); in util_queue_add_job_locked()
601 queue->jobs = jobs; in util_queue_add_job_locked()
602 queue->read_idx = 0; in util_queue_add_job_locked()
603 queue->write_idx = num_jobs; in util_queue_add_job_locked()
604 queue->max_jobs = new_max_jobs; in util_queue_add_job_locked()
607 while (queue->num_queued == queue->max_jobs) in util_queue_add_job_locked()
608 cnd_wait(&queue->has_space_cond, &queue->lock); in util_queue_add_job_locked()
612 ptr = &queue->jobs[queue->write_idx]; in util_queue_add_job_locked()
615 ptr->global_data = queue->global_data; in util_queue_add_job_locked()
621 queue->write_idx = (queue->write_idx + 1) % queue->max_jobs; in util_queue_add_job_locked()
622 queue->total_jobs_size += ptr->job_size; in util_queue_add_job_locked()
624 queue->num_queued++; in util_queue_add_job_locked()
625 cnd_signal(&queue->has_queued_cond); in util_queue_add_job_locked()
627 mtx_unlock(&queue->lock); in util_queue_add_job_locked()
631 util_queue_add_job(struct util_queue *queue, in util_queue_add_job() argument
638 util_queue_add_job_locked(queue, job, fence, execute, cleanup, job_size, in util_queue_add_job()
644 * the queue. If the job has started execution, the function waits for it to
653 util_queue_drop_job(struct util_queue *queue, struct util_queue_fence *fence) in util_queue_drop_job() argument
660 mtx_lock(&queue->lock); in util_queue_drop_job()
661 for (unsigned i = queue->read_idx; i != queue->write_idx; in util_queue_drop_job()
662 i = (i + 1) % queue->max_jobs) { in util_queue_drop_job()
663 if (queue->jobs[i].fence == fence) { in util_queue_drop_job()
664 if (queue->jobs[i].cleanup) in util_queue_drop_job()
665 queue->jobs[i].cleanup(queue->jobs[i].job, queue->global_data, -1); in util_queue_drop_job()
668 memset(&queue->jobs[i], 0, sizeof(queue->jobs[i])); in util_queue_drop_job()
673 mtx_unlock(&queue->lock); in util_queue_drop_job()
685 util_queue_finish(struct util_queue *queue) in util_queue_finish() argument
694 mtx_lock(&queue->lock); in util_queue_finish()
697 if (!queue->num_threads) { in util_queue_finish()
698 mtx_unlock(&queue->lock); in util_queue_finish()
706 * enough space in the queue and wait for space. in util_queue_finish()
708 queue->create_threads_on_demand = false; in util_queue_finish()
710 fences = malloc(queue->num_threads * sizeof(*fences)); in util_queue_finish()
711 util_barrier_init(&barrier, queue->num_threads); in util_queue_finish()
713 for (unsigned i = 0; i < queue->num_threads; ++i) { in util_queue_finish()
715 util_queue_add_job_locked(queue, &barrier, &fences[i], in util_queue_finish()
718 queue->create_threads_on_demand = true; in util_queue_finish()
719 mtx_unlock(&queue->lock); in util_queue_finish()
721 for (unsigned i = 0; i < queue->num_threads; ++i) { in util_queue_finish()
730 util_queue_get_thread_time_nano(struct util_queue *queue, unsigned thread_index) in util_queue_get_thread_time_nano() argument
733 if (thread_index >= queue->num_threads) in util_queue_get_thread_time_nano()
736 return util_thread_get_time_nano(queue->threads[thread_index]); in util_queue_get_thread_time_nano()