+/**
+ * Remove a queued job. If the job hasn't started execution, it's removed from
+ * the queue. If the job has started execution, the function waits for it to
+ * complete.
+ *
+ * In all cases, the fence is signalled when the function returns.
+ *
+ * The function can be used when destroying an object associated with the job
+ * when you don't care about the job completion state.
+ */
+void
+util_queue_drop_job(struct util_queue *queue, struct util_queue_fence *fence)
+{
+ bool removed = false;
+
+ if (util_queue_fence_is_signalled(fence))
+ return;
+
+ mtx_lock(&queue->lock);
+ for (unsigned i = queue->read_idx; i != queue->write_idx;
+ i = (i + 1) % queue->max_jobs) {
+ if (queue->jobs[i].fence == fence) {
+ if (queue->jobs[i].cleanup)
+ queue->jobs[i].cleanup(queue->jobs[i].job, -1);
+
+ /* Just clear it. The threads will treat as a no-op job. */
+ memset(&queue->jobs[i], 0, sizeof(queue->jobs[i]));
+ removed = true;
+ break;
+ }
+ }
+ mtx_unlock(&queue->lock);
+
+ if (removed)
+ util_queue_fence_signal(fence);
+ else
+ util_queue_fence_wait(fence);
+}
+
+static void
+util_queue_finish_execute(void *data, int num_thread)
+{
+ util_barrier *barrier = data;
+ util_barrier_wait(barrier);
+}
+
+/**
+ * Wait until all previously added jobs have completed.
+ */
+void
+util_queue_finish(struct util_queue *queue)
+{
+ util_barrier barrier;
+ 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
+ * wait for it exclusively.
+ */
+ 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, 0);
+ }
+
+ for (unsigned i = 0; i < queue->num_threads; ++i) {
+ util_queue_fence_wait(&fences[i]);
+ util_queue_fence_destroy(&fences[i]);
+ }
+ mtx_unlock(&queue->finish_lock);
+
+ util_barrier_destroy(&barrier);
+
+ free(fences);
+}
+