rh264: the bench says how many pictures the pool decoded at once, and a pool oracle

A sixteen-core run had the pool's CPU time equal to the wall time:
however the pictures were posted, they decoded one at a time. The
decoder counts the most pictures its pool had decoding at one moment,
and the bench prints it beside the row waits. And rthreads gets a
sample that settles the pool by itself: four jobs each burning a spell
of CPU, posted to a pool of four, must run together and finish in
about one spell; jobs that never overlap fail it, and a machine that
gives the process one core is told apart from a pool that does not
wake its workers.

libretro-common/formats/h264/rh264.c

@@ -3055,6 +3055,8 @@ static void rh264_inter_clear_i4mode(rh264_frame *f, int mbx, int mby)
 static retro_atomic_int_t rh264_ref_wait_misses;
 /* process-wide: reference reads that waited, and the rows they were short */
 static retro_atomic_int_t rh264_row_waits, rh264_row_short_sum;
+/* process-wide: jobs on the pool at this moment, and the most at once */
+static retro_atomic_int_t rh264_jobs_running, rh264_jobs_running_max;
 
 int rh264_video_ref_wait_misses(void)
 {
@@ -10150,7 +10152,17 @@ static void rh264_ctx_run_slices(rh264_pic_ctx *c, const struct rh264_vlc_set *v
 static void rh264_ctx_job(void *arg)
 {
    rh264_pic_ctx *c = (rh264_pic_ctx*)arg;
+   {
+      int now = retro_atomic_fetch_add_int(&rh264_jobs_running, 1) + 1, m;
+      do
+      {
+         m = retro_atomic_load_acquire_int(&rh264_jobs_running_max);
+         if (now <= m)
+            break;
+      } while (!retro_atomic_cas_int(&rh264_jobs_running_max, m, now));
+   }
    rh264_ctx_run_slices(c, c->vlc);
+   retro_atomic_fetch_sub_int(&rh264_jobs_running, 1);
    if (c->job_rc != 0 && c->f.planes)
       rh264_block_publish_rows(c->f.planes, c->f.mbh); /* nobody waits on a refusal */
    slock_lock(rh264_rows_lock);
@@ -10385,6 +10397,13 @@ void rh264_video_stats(const rh264_video *v, int *posted, int *inflight_x100,
    if (pop_waits)     *pop_waits     = v->st_pop_waits;
 }
 
+int rh264_video_jobs_at_once(void)
+{
+   int m = retro_atomic_load_acquire_int(&rh264_jobs_running_max);
+   retro_atomic_store_release_int(&rh264_jobs_running_max, 0);
+   return m;
+}
+
 void rh264_video_row_wait_stats(int *waits, int *rows_short_x100)
 {
    int w = retro_atomic_load_acquire_int(&rh264_row_waits);

libretro-common/include/formats/rh264.h

@@ -129,6 +129,10 @@ void rh264_video_stats(const rh264_video *v, int *posted, int *inflight_x100,
  * stands when it does. Process-wide; reading resets. */
 void rh264_video_row_wait_stats(int *waits, int *rows_short_x100);
 
+/* The most pictures the pool had decoding at one moment since the last
+ * call; reading resets. Process-wide. One means they never overlapped. */
+int rh264_video_jobs_at_once(void);
+
 int rh264_video_bit_depth(const rh264_video *v);
 
 /* Borrow a decoded plane (0=Y, 1=U, 2=V). Valid until the next decode call. */

libretro-common/samples/rthreads/tpool_parallel_test/Makefile

@@ -0,0 +1,40 @@
+TARGET := tpool_parallel_test
+
+LIBRETRO_COMM_DIR := ../../..
+
+# tpool.c depends on rthreads.c.  Both are self-contained at the
+# libretro-common layer (rthreads.c is the platform abstraction over
+# pthreads / Win32 / etc.) so we just compile them in directly --
+# no other libretro-common sources are needed.
+SOURCES := \
+	tpool_parallel_test.c \
+	$(LIBRETRO_COMM_DIR)/rthreads/tpool.c \
+	$(LIBRETRO_COMM_DIR)/rthreads/rthreads.c \
+	$(LIBRETRO_COMM_DIR)/features/features_cpu.c
+
+OBJS := $(SOURCES:.c=.o)
+
+CFLAGS  += -Wall -pedantic -std=gnu99 -g -O0 -I$(LIBRETRO_COMM_DIR)/include
+LDFLAGS += -lpthread
+
+# rthreads.c uses clock_gettime + CLOCK_REALTIME on Linux glibc; on
+# older glibc those live in -lrt.  Harmless on newer glibc.
+LDFLAGS += -lrt
+
+ifneq ($(SANITIZER),)
+   CFLAGS  := -fsanitize=$(SANITIZER) -fno-omit-frame-pointer $(CFLAGS)
+   LDFLAGS := -fsanitize=$(SANITIZER) $(LDFLAGS)
+endif
+
+all: $(TARGET)
+
+%.o: %.c
+	$(CC) -c -o $@ $< $(CFLAGS)
+
+$(TARGET): $(OBJS)
+	$(CC) -o $@ $^ $(LDFLAGS)
+
+clean:
+	rm -f $(TARGET) $(OBJS)
+
+.PHONY: clean

libretro-common/samples/rthreads/tpool_parallel_test/tpool_parallel_test.c

@@ -0,0 +1,111 @@
+/* tpool_parallel_test.c -- work posted to a pool of N runs on N
+ * threads at once, not one after another.
+ *
+ * Four jobs, each spinning for a fixed spell and reporting the moment
+ * it started and stopped, posted to a pool of four. If the workers
+ * wake and take the work, the four run together and the whole lot
+ * lasts about one spell; if only one worker ever wakes - a lost
+ * wake-up in the pool or the condition variable beneath it - they run
+ * back to back and last four. The counts of jobs running at once say
+ * which, and the test fails on fewer than two: a decoder that posts
+ * its pictures here would then decode them one at a time and gain
+ * nothing from the cores.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+
+#include <rthreads/tpool.h>
+#include <rthreads/rthreads.h>
+#include <retro_atomic.h>
+#include <features/features_cpu.h>
+
+#define JOBS 4
+#define SPELL_US 300000
+
+static retro_atomic_int_t running;
+static retro_atomic_int_t running_max;
+static retro_atomic_int_t done;
+
+/* A spell of work that costs CPU time rather than waiting for the
+ * clock: a job that waited on the clock would finish on time however
+ * many cores it shared, and say nothing. */
+static volatile unsigned sink;
+static unsigned spins_per_spell;
+
+static void spell(void)
+{
+   unsigned i, x = 1;
+   for (i = 0; i < spins_per_spell; i++)
+      x = x * 1664525u + 1013904223u;
+   sink = x;
+}
+
+static void job(void *arg)
+{
+   int now_running = retro_atomic_fetch_add_int(&running, 1) + 1;
+   int m;
+   (void)arg;
+   /* record the most seen at once */
+   do
+   {
+      m = retro_atomic_load_acquire_int(&running_max);
+      if (now_running <= m)
+         break;
+   } while (!retro_atomic_cas_int(&running_max, m, now_running));
+   spell();
+   retro_atomic_fetch_sub_int(&running, 1);
+   retro_atomic_fetch_add_int(&done, 1);
+}
+
+int main(void)
+{
+   tpool_t *tp = tpool_create_with_stack_size(JOBS, 512 * 1024);
+   int64_t t0, ms;
+   int i, most, rc = 0;
+   if (!tp)
+   {
+      printf("tpool_parallel_test: no pool\n");
+      return 2;
+   }
+   /* size the spell to about SPELL_US on this machine, alone */
+   spins_per_spell = 1u << 24;
+   t0 = cpu_features_get_time_usec();
+   spell();
+   {
+      int64_t one = cpu_features_get_time_usec() - t0;
+      if (one < 1) one = 1;
+      spins_per_spell = (unsigned)((double)spins_per_spell * SPELL_US / (double)one);
+   }
+   t0 = cpu_features_get_time_usec();
+   for (i = 0; i < JOBS; i++)
+      if (!tpool_add_work(tp, job, NULL))
+      {
+         printf("tpool_parallel_test: add_work refused\n");
+         return 2;
+      }
+   tpool_wait(tp);
+   ms = (cpu_features_get_time_usec() - t0) / 1000;
+   most = retro_atomic_load_acquire_int(&running_max);
+   printf("%d jobs of %d ms on a pool of %d: %d ran at once at most, "
+         "all done in %lld ms (one core would take about %d)\n", JOBS,
+         SPELL_US / 1000, JOBS, most, (long long)ms, JOBS * SPELL_US / 1000);
+   if (retro_atomic_load_acquire_int(&done) != JOBS)
+   {
+      printf("FAIL: %d of %d jobs ran\n", retro_atomic_load_acquire_int(&done), JOBS);
+      rc = 1;
+   }
+   else if (most < 2)
+   {
+      printf("FAIL: the jobs ran one after another - the pool's workers "
+            "are not being woken\n");
+      rc = 1;
+   }
+   else if (ms > JOBS * SPELL_US / 1000 * 3 / 4)
+      printf("NOTE: they overlapped but took about a core's time: this "
+            "machine gave the process one core\n");
+   else
+      printf("PASS\n");
+   tpool_destroy(tp);
+   return rc;
+}

samples/gfx/gfx_thumbnail_preview/preview_decode_bench.c

@@ -126,8 +126,9 @@ static int run(const uint8_t *buf, size_t len, enum image_type_enum type,
          printf("      pipeline: %d posted, %d.%02d in flight on average, "
                "%d posts at the limit, %d joins waited, %d pops held, %d pops waited\n",
                posted, inflight / 100, inflight % 100, at_max, jw, ph, pw);
-         printf("      row waits: %d, short by %d.%02d rows on average\n",
-               rw, rs / 100, rs % 100);
+         printf("      row waits: %d, short by %d.%02d rows on average; "
+               "%d pictures decoding at once at most\n",
+               rw, rs / 100, rs % 100, rh264_video_jobs_at_once());
       }
    }