Lines Matching refs:queue
47 util_queue_kill_threads(struct util_queue *queue, unsigned keep_num_threads,
82 add_to_atexit_list(struct util_queue *queue) in add_to_atexit_list() argument
87 list_add(&queue->head, &queue_list); in add_to_atexit_list()
92 remove_from_atexit_list(struct util_queue *queue) in remove_from_atexit_list() argument
98 if (iter == queue) { in remove_from_atexit_list()
242 struct util_queue *queue; member
249 struct util_queue *queue = ((struct thread_input*)input)->queue; in util_queue_thread_func() local
254 if (queue->flags & UTIL_QUEUE_INIT_SET_FULL_THREAD_AFFINITY) { in util_queue_thread_func()
267 if (queue->flags & UTIL_QUEUE_INIT_USE_MINIMUM_PRIORITY) { in util_queue_thread_func()
273 if (strlen(queue->name) > 0) { in util_queue_thread_func()
275 snprintf(name, sizeof(name), "%s%i", queue->name, thread_index); in util_queue_thread_func()
282 mtx_lock(&queue->lock); in util_queue_thread_func()
283 assert(queue->num_queued >= 0 && queue->num_queued <= queue->max_jobs); in util_queue_thread_func()
286 while (thread_index < queue->num_threads && queue->num_queued == 0) in util_queue_thread_func()
287 cnd_wait(&queue->has_queued_cond, &queue->lock); in util_queue_thread_func()
290 if (thread_index >= queue->num_threads) { in util_queue_thread_func()
291 mtx_unlock(&queue->lock); in util_queue_thread_func()
295 job = queue->jobs[queue->read_idx]; in util_queue_thread_func()
296 memset(&queue->jobs[queue->read_idx], 0, sizeof(struct util_queue_job)); in util_queue_thread_func()
297 queue->read_idx = (queue->read_idx + 1) % queue->max_jobs; in util_queue_thread_func()
299 queue->num_queued--; in util_queue_thread_func()
300 cnd_signal(&queue->has_space_cond); in util_queue_thread_func()
302 queue->total_jobs_size -= job.job_size; in util_queue_thread_func()
303 mtx_unlock(&queue->lock); in util_queue_thread_func()
315 mtx_lock(&queue->lock); in util_queue_thread_func()
316 if (queue->num_threads == 0) { in util_queue_thread_func()
317 for (unsigned i = queue->read_idx; i != queue->write_idx; in util_queue_thread_func()
318 i = (i + 1) % queue->max_jobs) { in util_queue_thread_func()
319 if (queue->jobs[i].job) { in util_queue_thread_func()
320 if (queue->jobs[i].fence) in util_queue_thread_func()
321 util_queue_fence_signal(queue->jobs[i].fence); in util_queue_thread_func()
322 queue->jobs[i].job = NULL; in util_queue_thread_func()
325 queue->read_idx = queue->write_idx; in util_queue_thread_func()
326 queue->num_queued = 0; in util_queue_thread_func()
328 mtx_unlock(&queue->lock); in util_queue_thread_func()
333 util_queue_create_thread(struct util_queue *queue, unsigned index) in util_queue_create_thread() argument
337 input->queue = queue; in util_queue_create_thread()
340 if (thrd_success != u_thread_create(queue->threads + index, util_queue_thread_func, input)) { in util_queue_create_thread()
345 if (queue->flags & UTIL_QUEUE_INIT_USE_MINIMUM_PRIORITY) { in util_queue_create_thread()
356 pthread_setschedparam(queue->threads[index], SCHED_BATCH, &sched_param); in util_queue_create_thread()
363 util_queue_adjust_num_threads(struct util_queue *queue, unsigned num_threads) in util_queue_adjust_num_threads() argument
365 num_threads = MIN2(num_threads, queue->max_threads); in util_queue_adjust_num_threads()
368 simple_mtx_lock(&queue->finish_lock); in util_queue_adjust_num_threads()
369 unsigned old_num_threads = queue->num_threads; in util_queue_adjust_num_threads()
372 simple_mtx_unlock(&queue->finish_lock); in util_queue_adjust_num_threads()
377 util_queue_kill_threads(queue, num_threads, true); in util_queue_adjust_num_threads()
378 simple_mtx_unlock(&queue->finish_lock); in util_queue_adjust_num_threads()
387 queue->num_threads = num_threads; in util_queue_adjust_num_threads()
389 if (!util_queue_create_thread(queue, i)) { in util_queue_adjust_num_threads()
390 queue->num_threads = i; in util_queue_adjust_num_threads()
394 simple_mtx_unlock(&queue->finish_lock); in util_queue_adjust_num_threads()
398 util_queue_init(struct util_queue *queue, in util_queue_init() argument
417 const int max_chars = sizeof(queue->name) - 1; in util_queue_init()
426 memset(queue, 0, sizeof(*queue)); in util_queue_init()
429 snprintf(queue->name, sizeof(queue->name), "%.*s:%s", in util_queue_init()
432 snprintf(queue->name, sizeof(queue->name), "%s", name); in util_queue_init()
435 queue->flags = flags; in util_queue_init()
436 queue->max_threads = num_threads; in util_queue_init()
437 queue->num_threads = (flags & UTIL_QUEUE_INIT_SCALE_THREADS) ? 1 : num_threads; in util_queue_init()
438 queue->max_jobs = max_jobs; in util_queue_init()
439 queue->global_data = global_data; in util_queue_init()
441 (void) mtx_init(&queue->lock, mtx_plain); in util_queue_init()
442 (void) simple_mtx_init(&queue->finish_lock, mtx_plain); in util_queue_init()
444 queue->num_queued = 0; in util_queue_init()
445 cnd_init(&queue->has_queued_cond); in util_queue_init()
446 cnd_init(&queue->has_space_cond); in util_queue_init()
448 queue->jobs = (struct util_queue_job*) in util_queue_init()
450 if (!queue->jobs) in util_queue_init()
453 queue->threads = (thrd_t*) calloc(queue->max_threads, sizeof(thrd_t)); in util_queue_init()
454 if (!queue->threads) in util_queue_init()
458 for (i = 0; i < queue->num_threads; i++) { in util_queue_init()
459 if (!util_queue_create_thread(queue, i)) { in util_queue_init()
465 queue->num_threads = i; in util_queue_init()
471 add_to_atexit_list(queue); in util_queue_init()
475 free(queue->threads); in util_queue_init()
477 if (queue->jobs) { in util_queue_init()
478 cnd_destroy(&queue->has_space_cond); in util_queue_init()
479 cnd_destroy(&queue->has_queued_cond); in util_queue_init()
480 mtx_destroy(&queue->lock); in util_queue_init()
481 free(queue->jobs); in util_queue_init()
484 memset(queue, 0, sizeof(*queue)); in util_queue_init()
489 util_queue_kill_threads(struct util_queue *queue, unsigned keep_num_threads, in util_queue_kill_threads() argument
496 simple_mtx_lock(&queue->finish_lock); in util_queue_kill_threads()
498 if (keep_num_threads >= queue->num_threads) { in util_queue_kill_threads()
499 simple_mtx_unlock(&queue->finish_lock); in util_queue_kill_threads()
503 mtx_lock(&queue->lock); in util_queue_kill_threads()
504 unsigned old_num_threads = queue->num_threads; in util_queue_kill_threads()
508 queue->num_threads = keep_num_threads; in util_queue_kill_threads()
509 cnd_broadcast(&queue->has_queued_cond); in util_queue_kill_threads()
510 mtx_unlock(&queue->lock); in util_queue_kill_threads()
513 thrd_join(queue->threads[i], NULL); in util_queue_kill_threads()
516 simple_mtx_unlock(&queue->finish_lock); in util_queue_kill_threads()
528 util_queue_destroy(struct util_queue *queue) in util_queue_destroy() argument
530 util_queue_kill_threads(queue, 0, false); in util_queue_destroy()
533 if (queue->head.next != NULL) in util_queue_destroy()
534 remove_from_atexit_list(queue); in util_queue_destroy()
536 cnd_destroy(&queue->has_space_cond); in util_queue_destroy()
537 cnd_destroy(&queue->has_queued_cond); in util_queue_destroy()
538 simple_mtx_destroy(&queue->finish_lock); in util_queue_destroy()
539 mtx_destroy(&queue->lock); in util_queue_destroy()
540 free(queue->jobs); in util_queue_destroy()
541 free(queue->threads); in util_queue_destroy()
545 util_queue_add_job(struct util_queue *queue, in util_queue_add_job() argument
554 mtx_lock(&queue->lock); in util_queue_add_job()
555 if (queue->num_threads == 0) { in util_queue_add_job()
556 mtx_unlock(&queue->lock); in util_queue_add_job()
566 assert(queue->num_queued >= 0 && queue->num_queued <= queue->max_jobs); in util_queue_add_job()
569 if (queue->num_queued > 0 && in util_queue_add_job()
570 queue->flags & UTIL_QUEUE_INIT_SCALE_THREADS && in util_queue_add_job()
572 queue->num_threads < queue->max_threads) { in util_queue_add_job()
573 util_queue_adjust_num_threads(queue, queue->num_threads + 1); in util_queue_add_job()
576 if (queue->num_queued == queue->max_jobs) { in util_queue_add_job()
577 if (queue->flags & UTIL_QUEUE_INIT_RESIZE_IF_FULL && in util_queue_add_job()
578 queue->total_jobs_size + job_size < S_256MB) { in util_queue_add_job()
582 unsigned new_max_jobs = queue->max_jobs + 8; in util_queue_add_job()
590 unsigned i = queue->read_idx; in util_queue_add_job()
593 jobs[num_jobs++] = queue->jobs[i]; in util_queue_add_job()
594 i = (i + 1) % queue->max_jobs; in util_queue_add_job()
595 } while (i != queue->write_idx); in util_queue_add_job()
597 assert(num_jobs == queue->num_queued); in util_queue_add_job()
599 free(queue->jobs); in util_queue_add_job()
600 queue->jobs = jobs; in util_queue_add_job()
601 queue->read_idx = 0; in util_queue_add_job()
602 queue->write_idx = num_jobs; in util_queue_add_job()
603 queue->max_jobs = new_max_jobs; in util_queue_add_job()
606 while (queue->num_queued == queue->max_jobs) in util_queue_add_job()
607 cnd_wait(&queue->has_space_cond, &queue->lock); in util_queue_add_job()
611 ptr = &queue->jobs[queue->write_idx]; in util_queue_add_job()
614 ptr->global_data = queue->global_data; in util_queue_add_job()
620 queue->write_idx = (queue->write_idx + 1) % queue->max_jobs; in util_queue_add_job()
621 queue->total_jobs_size += ptr->job_size; in util_queue_add_job()
623 queue->num_queued++; in util_queue_add_job()
624 cnd_signal(&queue->has_queued_cond); in util_queue_add_job()
625 mtx_unlock(&queue->lock); in util_queue_add_job()
639 util_queue_drop_job(struct util_queue *queue, struct util_queue_fence *fence) in util_queue_drop_job() argument
646 mtx_lock(&queue->lock); in util_queue_drop_job()
647 for (unsigned i = queue->read_idx; i != queue->write_idx; in util_queue_drop_job()
648 i = (i + 1) % queue->max_jobs) { in util_queue_drop_job()
649 if (queue->jobs[i].fence == fence) { in util_queue_drop_job()
650 if (queue->jobs[i].cleanup) in util_queue_drop_job()
651 queue->jobs[i].cleanup(queue->jobs[i].job, queue->global_data, -1); in util_queue_drop_job()
654 memset(&queue->jobs[i], 0, sizeof(queue->jobs[i])); in util_queue_drop_job()
659 mtx_unlock(&queue->lock); in util_queue_drop_job()
671 util_queue_finish(struct util_queue *queue) in util_queue_finish() argument
680 simple_mtx_lock(&queue->finish_lock); in util_queue_finish()
683 if (!queue->num_threads) { in util_queue_finish()
684 simple_mtx_unlock(&queue->finish_lock); in util_queue_finish()
688 fences = malloc(queue->num_threads * sizeof(*fences)); in util_queue_finish()
689 util_barrier_init(&barrier, queue->num_threads); in util_queue_finish()
691 for (unsigned i = 0; i < queue->num_threads; ++i) { in util_queue_finish()
693 util_queue_add_job(queue, &barrier, &fences[i], in util_queue_finish()
697 for (unsigned i = 0; i < queue->num_threads; ++i) { in util_queue_finish()
701 simple_mtx_unlock(&queue->finish_lock); in util_queue_finish()
707 util_queue_get_thread_time_nano(struct util_queue *queue, unsigned thread_index) in util_queue_get_thread_time_nano() argument
710 if (thread_index >= queue->num_threads) in util_queue_get_thread_time_nano()
713 return util_thread_get_time_nano(queue->threads[thread_index]); in util_queue_get_thread_time_nano()