task_queue: budget retirement and unthreaded handlers per check

Two audit P1s, both about task_queue_check() taking as much of the
frame as it likes on the thread that drives the frame loop:

- Retirement had no bound. Finished tasks retire (progress push,
  callback, cleanup, free) on the checking thread, and a burst of
  completions - a thumbnail scan, a bulk download - retired in one
  check however long the callbacks took.
- On the unthreaded queue every running task's handler ran once per
  check, so the frame cost grew with the running list.

task_queue_set_budget(retire_usec, retire_max, handler_usec) bounds
one check. Retirement stops once retire_max tasks or retire_usec have
gone by, after at least one; what is left stays on the retiring list,
findable, for the next check (the threaded early-out is told there is
work pending). Handlers stop once handler_usec has gone by, after at
least one due handler; the tasks not run go back to the front of the
running list, the ones that ran behind them, so the list rotates
instead of starving its tail. The unthreaded gather now keeps FIFO
order rather than reversing the list each pass, which the rotation
needs. Zero means unbounded and is the default, so nothing changes
for a caller that does not opt in.

RetroArch sets 2 ms or 32 callbacks for retirement and 4 ms of
handler time per check, leaving most of a 16.7 ms frame to the core.
task_queue_wait() loops on the check, so a blocking wait is slowed,
never changed.

task_queue_budget_test covers the count and time retire budgets and
the unbounded case on the threaded queue, and the handler budget,
rotation fairness and the unbounded case on the unthreaded one. Four
of its seven lanes fail against the old queue. Wired into
Linux-libretro-common-samples.

.github/workflows/Linux-libretro-common-samples.yml

@@ -159,6 +159,7 @@ jobs:
             # Stops the worker with a bound when a detachable
             # handler is stuck; waits ~1.5 s by design.
             task_queue_detach_test
+            task_queue_budget_test
             tpool_wait_test
             # Four CPU-burning jobs on a pool of four must overlap;
             # on a one-core runner it notes that and still passes.

libretro-common/include/queues/task_queue.h

@@ -785,6 +785,35 @@ void task_queue_deinit(void);
  */
 void task_queue_init(bool threaded, retro_task_queue_msg_t msg_push);
 
+/**
+ * Bounds what one task_queue_check() may do on the calling thread,
+ * which is the thread that also drives the frame loop.
+ *
+ * Retirement: finished tasks are retired (progress push, callback,
+ * cleanup, free) on the checking thread, and a burst of completions -
+ * a thumbnail scan, a bulk download - used to retire in one call
+ * however long the callbacks took. With a budget, a check retires at
+ * least one task, then stops once @retire_max tasks or @retire_usec
+ * microseconds have gone by; the rest stay queued, still findable,
+ * for the next check.
+ *
+ * Handlers (unthreaded queue only): task handlers run on the checking
+ * thread, one call per running task per check. With a budget, a check
+ * runs at least one due handler, then stops once @handler_usec has
+ * gone by; the tasks not run keep their place at the front of the
+ * queue so the list rotates rather than starving its tail. On the
+ * threaded queue handlers run on the worker and this bound is unused.
+ *
+ * Zero for any bound means unbounded, which is the default and the
+ * behaviour before budgets existed.
+ *
+ * task_queue_wait() and task_queue_wait_timeout() loop on the check,
+ * so a budget slows a blocking wait but never changes what it waits
+ * for.
+ */
+void task_queue_set_budget(retro_time_t retire_usec, unsigned retire_max,
+      retro_time_t handler_usec);
+
 /**
  * Called when a task handler occupies the calling thread for longer
  * than the configured budget.

libretro-common/queues/task_queue.c

@@ -307,16 +307,51 @@ static void retro_task_internal_retire_cleanup(retro_task_t *task)
 #endif
 }
 
+/* Per-check budgets (task_queue_set_budget). Read and written on the
+ * thread that drives task_queue_check(); zero means unbounded, the
+ * behaviour before budgets existed. */
+static retro_time_t task_retire_budget_usec;
+static unsigned     task_retire_budget_max;
+static retro_time_t task_handler_budget_usec;
+
+void task_queue_set_budget(retro_time_t retire_usec, unsigned retire_max,
+      retro_time_t handler_usec)
+{
+   task_retire_budget_usec  = retire_usec;
+   task_retire_budget_max   = retire_max;
+   task_handler_budget_usec = handler_usec;
+}
+
+/* Whether one more retirement fits: always the first, then while
+ * under the count and time budgets. */
+static bool task_retire_budget_allows(unsigned done, retro_time_t started)
+{
+   if (!done)
+      return true;
+   if (task_retire_budget_max && done >= task_retire_budget_max)
+      return false;
+   if (task_retire_budget_usec
+         && cpu_features_get_time_usec() - started >= task_retire_budget_usec)
+      return false;
+   return true;
+}
+
 static void retro_task_internal_gather(void)
 {
-   retro_task_t *task = NULL;
-   while ((task = task_queue_get(&tasks_finished)))
+   retro_task_t *task   = NULL;
+   unsigned      done   = 0;
+   retro_time_t  started = (task_retire_budget_usec)
+      ? cpu_features_get_time_usec() : 0;
+
+   while (task_retire_budget_allows(done, started)
+         && (task = task_queue_get(&tasks_finished)))
    {
       /* Already popped off the queue, so it is unreachable either way
        * and the split is unobservable here. */
       retro_task_internal_retire_callback(task);
       retro_task_internal_retire_cleanup(task);
       free(task);
+      done++;
    }
 }
 
@@ -346,37 +381,66 @@ void task_queue_set_slow_handler_cb(retro_task_slow_handler_t cb,
 
 static void retro_task_regular_gather(void)
 {
-   retro_task_t *task  = NULL;
-   retro_task_t *queue = NULL;
-   retro_task_t *next  = NULL;
+   retro_task_t *task     = NULL;
+   retro_task_t *queue    = NULL;
+   retro_task_t *tail     = NULL;
+   retro_task_t *next     = NULL;
+   retro_task_t *ran      = NULL;
+   retro_task_t *ran_tail = NULL;
+   retro_time_t  started  = 0;
+   unsigned      n_ran    = 0;
+   bool          over     = false;
 
+   /* Detach the running list, in order. */
    while ((task = task_queue_get(&tasks_running)))
    {
-      task->next = queue;
-      queue = task;
+      if (tail)
+         tail->next = task;
+      else
+         queue      = task;
+      tail = task;
    }
 
+   if (task_handler_budget_usec)
+      started = cpu_features_get_time_usec();
+
    for (task = queue; task; task = next)
    {
       next = task->next;
 
+      /* Past the budget, the rest of the list waits for the next
+       * check. At least one due handler always runs, so a budget too
+       * small for any single handler still makes progress. */
+      if (   !over && n_ran && task_handler_budget_usec
+          && cpu_features_get_time_usec() - started >= task_handler_budget_usec)
+         over = true;
+
+      if (over)
+      {
+         /* Unrun tasks stay at the front, so a busy queue rotates
+          * through them rather than starving the tail. */
+         task_queue_put(&tasks_running, task);
+         continue;
+      }
+
       if (!task->when || task->when < cpu_features_get_time_usec())
       {
          /* Time the handler only when someone is listening: with no
           * callback registered this costs nothing at all. */
          if (task_slow_handler_cb)
          {
-            retro_time_t started = cpu_features_get_time_usec();
+            retro_time_t h_started = cpu_features_get_time_usec();
             retro_time_t took;
 
             task->handler(task);
 
-            took = cpu_features_get_time_usec() - started;
+            took = cpu_features_get_time_usec() - h_started;
             if (took > task_slow_handler_budget)
                task_slow_handler_cb(task, took);
          }
          else
             task->handler(task);
+         n_ran++;
       }
 
       /* No progress push here: the gather on the main thread pushes
@@ -386,7 +450,21 @@ static void retro_task_regular_gather(void)
       if ((task->flags & RETRO_TASK_FLG_FINISHED) > 0)
          task_queue_put(&tasks_finished, task);
       else
-         task_queue_put(&tasks_running, task);
+      {
+         task->next = NULL;
+         if (ran_tail)
+            ran_tail->next = task;
+         else
+            ran            = task;
+         ran_tail = task;
+      }
+   }
+
+   /* The tasks that ran go behind the ones that did not. */
+   for (task = ran; task; task = next)
+   {
+      next = task->next;
+      task_queue_put(&tasks_running, task);
    }
 
    retro_task_internal_gather();
@@ -623,9 +701,22 @@ static void retro_task_threaded_gather(void)
       /* Retire outside the lock (callbacks may push tasks, taking
        * running_lock; see the deadlock note above), pruning each
        * task from the retiring list - under the lock - before it
-       * is freed */
+       * is freed. Bounded by the retire budget: what is left on the
+       * retiring list stays findable and is picked up next check. */
+      {
+      unsigned     done    = 0;
+      retro_time_t started = (task_retire_budget_usec)
+         ? cpu_features_get_time_usec() : 0;
       for (;;)
       {
+         if (!task_retire_budget_allows(done, started))
+         {
+            /* Leftovers: keep the early-out at the top of this
+             * function from skipping the next check. */
+            retro_atomic_store_release_int(&tasks_finished_count, 1);
+            break;
+         }
+
          slock_lock(finished_lock);
          task = tasks_retiring.front;
          slock_unlock(finished_lock);
@@ -647,6 +738,8 @@ static void retro_task_threaded_gather(void)
          retro_task_internal_retire_cleanup(task);
 
          free(task);
+         done++;
+      }
       }
 
       in_gather = false;

libretro-common/samples/queues/task_queue_budget_test/Makefile

@@ -0,0 +1,39 @@
+TARGET := task_queue_budget_test
+
+LIBRETRO_COMM_DIR := ../../..
+
+# Built with HAVE_THREADS: the retire budget is exercised on the threaded
+# queue, the handler budget on the unthreaded one.
+SOURCES := \
+	task_queue_budget_test.c \
+	$(LIBRETRO_COMM_DIR)/queues/task_queue.c \
+	$(LIBRETRO_COMM_DIR)/rthreads/rthreads.c \
+	$(LIBRETRO_COMM_DIR)/features/features_cpu.c
+
+OBJS := $(SOURCES:.c=.o)
+
+CFLAGS  += -Wall -pedantic -std=gnu99 -g -O1 -fno-omit-frame-pointer \
+           -DHAVE_THREADS -I$(LIBRETRO_COMM_DIR)/include
+
+LIBS    += -lpthread
+
+ifneq ($(SANITIZER),)
+   CFLAGS  := -fsanitize=$(SANITIZER) $(CFLAGS)
+   LDFLAGS := -fsanitize=$(SANITIZER) $(LDFLAGS)
+endif
+
+all: $(TARGET)
+
+%.o: %.c
+	$(CC) -c -o $@ $< $(CFLAGS)
+
+$(TARGET): $(OBJS)
+	$(CC) -o $@ $^ $(LDFLAGS) $(LIBS)
+
+check: $(TARGET)
+	./$(TARGET)
+
+clean:
+	rm -f $(TARGET) $(OBJS)
+
+.PHONY: all check clean

libretro-common/samples/queues/task_queue_budget_test/task_queue_budget_test.c

@@ -0,0 +1,190 @@
+/* task_queue_budget_test: the per-check budgets of task_queue_check().
+ *
+ * Retirement (threaded queue): a burst of finished tasks used to be
+ * retired in one check however long their callbacks took. With a
+ * count budget a check retires that many; with a time budget it stops
+ * once the time has gone by; with none it retires everything.
+ *
+ * Handlers (unthreaded queue): every running task's handler used to
+ * run once per check. With a time budget a check runs a few, and the
+ * ones it did not run go first next time, so the list rotates rather
+ * than starving its tail.
+ *
+ * Against the old queue the budgeted checks retire or run every task
+ * at once and the first three lanes fail. */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <boolean.h>
+#include <retro_common_api.h>
+#include <retro_timers.h>
+#include <retro_atomic.h>
+#include <features/features_cpu.h>
+#include <queues/task_queue.h>
+
+static int fails;
+#define CHECK(cond, ...) do { if (!(cond)) { printf("FAIL: " __VA_ARGS__); printf("\n"); fails++; } else { printf("ok   " __VA_ARGS__); printf("\n"); } } while (0)
+
+static void busy_usec(retro_time_t usec)
+{
+   retro_time_t until = cpu_features_get_time_usec() + usec;
+   while (cpu_features_get_time_usec() < until) { }
+}
+
+/* --- retirement --- */
+
+static retro_atomic_int_t handled;
+static unsigned retired;
+
+static void finish_now(retro_task_t *task)
+{
+   task_set_flags(task, RETRO_TASK_FLG_FINISHED, true);
+   retro_atomic_fetch_add_int(&handled, 1);
+}
+
+static void slow_callback(retro_task_t *task, void *task_data,
+      void *user_data, const char *error)
+{
+   (void)task; (void)task_data; (void)user_data; (void)error;
+   busy_usec(500);
+   retired++;
+}
+
+static void push_finishers(unsigned n)
+{
+   unsigned i;
+   for (i = 0; i < n; i++)
+   {
+      retro_task_t *t = task_init();
+      t->handler  = finish_now;
+      t->callback = slow_callback;
+      task_queue_push(t);
+   }
+}
+
+/* Waits for the worker to finish every pushed task without retiring
+ * any: the retirement is what the lanes below measure. */
+static void wait_handled(unsigned n)
+{
+   retro_time_t until = cpu_features_get_time_usec() + 5000000;
+   while ((unsigned)retro_atomic_load_acquire_int(&handled) < n
+         && cpu_features_get_time_usec() < until)
+      retro_sleep(1);
+   retro_sleep(5);
+}
+
+static void lane_retire(void)
+{
+   unsigned before;
+   retro_time_t started, took;
+
+   task_queue_init(true, NULL);
+   retro_atomic_int_init(&handled, 0);
+   retired = 0;
+
+   /* Count budget: 8 of 64 per check. */
+   push_finishers(64);
+   wait_handled(64);
+   task_queue_set_budget(0, 8, 0);
+   task_queue_check();
+   CHECK(retired == 8, "count budget 8: retired %u of 64 in one check", retired);
+   task_queue_check();
+   CHECK(retired == 16, "count budget 8: next check retires the next 8 (%u)", retired);
+
+   /* Time budget: 2 ms of 500 us callbacks is a handful, not 48. */
+   before = retired;
+   task_queue_set_budget(2000, 0, 0);
+   started = cpu_features_get_time_usec();
+   task_queue_check();
+   took = cpu_features_get_time_usec() - started;
+   CHECK(retired - before >= 1 && retired - before <= 8 && took < 10000,
+         "time budget 2 ms: retired %u in %lld us", retired - before, (long long)took);
+
+   /* No budget: everything left goes in one check. */
+   task_queue_set_budget(0, 0, 0);
+   task_queue_check();
+   CHECK(retired == 64, "no budget: all 64 retired (%u)", retired);
+
+   task_queue_deinit();
+}
+
+/* --- handlers --- */
+
+#define N_SPIN 20
+static unsigned runs[N_SPIN];
+static bool     release;
+
+static void spin_handler(retro_task_t *task)
+{
+   unsigned idx = (unsigned)(size_t)task->user_data;
+   busy_usec(1000);
+   runs[idx]++;
+   if (release)
+      task_set_flags(task, RETRO_TASK_FLG_FINISHED, true);
+}
+
+static unsigned runs_total(void)
+{
+   unsigned i, n = 0;
+   for (i = 0; i < N_SPIN; i++)
+      n += runs[i];
+   return n;
+}
+
+static void lane_handlers(void)
+{
+   unsigned i, lo, hi, total;
+
+   /* init(false) after a threaded session: the threaded flag is
+    * sticky and the next check would re-init threaded, as RetroArch
+    * clears it through task_queue_unset_threaded() too. */
+   task_queue_unset_threaded();
+   task_queue_init(false, NULL);
+   memset(runs, 0, sizeof(runs));
+   release = false;
+
+   for (i = 0; i < N_SPIN; i++)
+   {
+      retro_task_t *t = task_init();
+      t->handler   = spin_handler;
+      t->user_data = (void*)(size_t)i;
+      task_queue_push(t);
+   }
+
+   /* 3 ms budget over 1 ms handlers: three or four per check. */
+   task_queue_set_budget(0, 0, 3000);
+   task_queue_check();
+   total = runs_total();
+   CHECK(total >= 1 && total <= 5, "handler budget 3 ms: %u of %u handlers ran in one check", total, N_SPIN);
+
+   /* Five more checks: the list rotates, so no task gets far ahead. */
+   for (i = 0; i < 5; i++)
+      task_queue_check();
+   lo = hi = runs[0];
+   for (i = 1; i < N_SPIN; i++)
+   {
+      if (runs[i] < lo) lo = runs[i];
+      if (runs[i] > hi) hi = runs[i];
+   }
+   CHECK(hi - lo <= 1, "rotation: after six checks runs range %u..%u", lo, hi);
+
+   /* No budget: one check runs every handler. */
+   task_queue_set_budget(0, 0, 0);
+   total = runs_total();
+   task_queue_check();
+   CHECK(runs_total() - total == N_SPIN, "no budget: %u handlers in one check", runs_total() - total);
+
+   release = true;
+   task_queue_wait(NULL, NULL);
+   task_queue_deinit();
+}
+
+int main(void)
+{
+   lane_retire();
+   lane_handlers();
+   printf("%s\n", fails ? "FAILED" : "PASSED");
+   return fails ? 1 : 0;
+}

retroarch.c

@@ -9096,6 +9096,16 @@ void retroarch_init_task_queue(void)
    else
       task_queue_set_slow_handler_cb(NULL, 0);
 #endif
+
+   /* What one task_queue_check() may spend of the frame it runs in.
+    * Retirement: a burst of completions (a thumbnail scan, a bulk
+    * download) retires over several frames instead of one - at most
+    * 32 callbacks or 2 ms per check, whichever comes first. Handlers
+    * (unthreaded queue): 4 ms of handler time per check, the rest of
+    * the running list waits its turn. Both leave most of a 16.7 ms
+    * frame to the core; a blocking task_queue_wait() still loops
+    * until its condition holds. */
+   task_queue_set_budget(2000, 32, 4000);
 }
 
 bool retroarch_ctl(enum rarch_ctl_state state, void *data)