[Mesa-dev] [PATCH 07/10] util/u_queue: add an option to resize the queue when it's full

[Marek Olšák]

From: Marek Olšák <marek.olsak at amd.com>

Consider the following situation:
  mtx_lock(mutex);
  do_something();
  util_queue_add_job(...);
  mtx_unlock(mutex);

If the queue is full, util_queue_add_job will wait for a free slot.
If the job which is currently being executed tries to lock the mutex,
it will be stuck forever, because util_queue_add_job is stuck.

The deadlock can be trivially resolved by increasing the queue size
(reallocating the queue) in util_queue_add_job if the queue is full.
Then util_queue_add_job becomes wait-free.

radeonsi will use it.
---
 src/util/u_queue.c | 37 ++++++++++++++++++++++++++++++++++---
 src/util/u_queue.h |  2 ++
 2 files changed, 36 insertions(+), 3 deletions(-)

diff --git a/src/util/u_queue.c b/src/util/u_queue.c
index cb59030..49361c3 100644
--- a/src/util/u_queue.c
+++ b/src/util/u_queue.c
@@ -197,20 +197,21 @@ bool
 util_queue_init(struct util_queue *queue,
                 const char *name,
                 unsigned max_jobs,
                 unsigned num_threads,
                 unsigned flags)
 {
    unsigned i;
 
    memset(queue, 0, sizeof(*queue));
    queue->name = name;
+   queue->flags = flags;
    queue->num_threads = num_threads;
    queue->max_jobs = max_jobs;
 
    queue->jobs = (struct util_queue_job*)
                  calloc(max_jobs, sizeof(struct util_queue_job));
    if (!queue->jobs)
       goto fail;
 
    (void) mtx_init(&queue->lock, mtx_plain);
 
@@ -322,23 +323,53 @@ util_queue_add_job(struct util_queue *queue,
       /* well no good option here, but any leaks will be
        * short-lived as things are shutting down..
        */
       return;
    }
 
    fence->signalled = false;
 
    assert(queue->num_queued >= 0 && queue->num_queued <= queue->max_jobs);
 
-   /* if the queue is full, wait until there is space */
-   while (queue->num_queued == queue->max_jobs)
-      cnd_wait(&queue->has_space_cond, &queue->lock);
+   if (queue->num_queued == queue->max_jobs) {
+      if (queue->flags & UTIL_QUEUE_INIT_RESIZE_IF_FULL) {
+         /* If the queue is full, make it larger to avoid waiting for a free
+          * slot.
+          */
+         unsigned new_max_jobs = queue->max_jobs + 8;
+         struct util_queue_job *jobs =
+            (struct util_queue_job*)calloc(new_max_jobs,
+                                           sizeof(struct util_queue_job));
+         assert(jobs);
+
+         /* Copy all queued jobs into the new list. */
+         unsigned num_jobs = 0;
+         unsigned i = queue->read_idx;
+
+         do {
+            jobs[num_jobs++] = queue->jobs[i];
+            i = (i + 1) % queue->max_jobs;
+         } while (i != queue->write_idx);
+
+         assert(num_jobs == queue->num_queued);
+
+         free(queue->jobs);
+         queue->jobs = jobs;
+         queue->read_idx = 0;
+         queue->write_idx = num_jobs;
+         queue->max_jobs = new_max_jobs;
+      } else {
+         /* Wait until there is a free slot. */
+         while (queue->num_queued == queue->max_jobs)
+            cnd_wait(&queue->has_space_cond, &queue->lock);
+      }
+   }
 
    ptr = &queue->jobs[queue->write_idx];
    assert(ptr->job == NULL);
    ptr->job = job;
    ptr->fence = fence;
    ptr->execute = execute;
    ptr->cleanup = cleanup;
    queue->write_idx = (queue->write_idx + 1) % queue->max_jobs;
 
    queue->num_queued++;
diff --git a/src/util/u_queue.h b/src/util/u_queue.h
index edd6bab..ff713ae 100644
--- a/src/util/u_queue.h
+++ b/src/util/u_queue.h
@@ -36,20 +36,21 @@
 #include <string.h>
 
 #include "util/list.h"
 #include "util/u_thread.h"
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
 #define UTIL_QUEUE_INIT_USE_MINIMUM_PRIORITY      (1 << 0)
+#define UTIL_QUEUE_INIT_RESIZE_IF_FULL            (1 << 1)
 
 /* Job completion fence.
  * Put this into your job structure.
  */
 struct util_queue_fence {
    mtx_t mutex;
    cnd_t cond;
    int signalled;
 };
 
@@ -62,20 +63,21 @@ struct util_queue_job {
    util_queue_execute_func cleanup;
 };
 
 /* Put this into your context. */
 struct util_queue {
    const char *name;
    mtx_t lock;
    cnd_t has_queued_cond;
    cnd_t has_space_cond;
    thrd_t *threads;
+   unsigned flags;
    int num_queued;
    unsigned num_threads;
    int kill_threads;
    int max_jobs;
    int write_idx, read_idx; /* ring buffer pointers */
    struct util_queue_job *jobs;
 
    /* for cleanup at exit(), protected by exit_mutex */
    struct list_head head;
 };
-- 
2.7.4