Make the manual content scan non-blocking: budgeted BEGIN/END states, resumable dir walk, I/O-free DAT parser (#19407)

* rxml: add rxml_load_document_owned

A public owned-buffer entry point next to the copying string one:
the caller hands over a NUL-terminated heap buffer and the document
keeps it for its lifetime (the tree points into it), freeing it in
rxml_free_document - or immediately on a failed parse.  This is
exactly the contract rxml_load_document's tail already had with its
own file read; exposing it lets a caller that read the document
bytes itself (e.g. a task handler filling a buffer under a time
budget across gathers) parse without doubling peak memory the way
rxml_load_document_string's mandatory copy does.  DAT files reach
hundreds of megabytes, so the copy is not a rounding error there.

First consumer is logiqx_dat, which is losing its file I/O in a
following commit.

Verified: strict C89 syntax pass (-std=c89 -ansi -pedantic
-Werror=declaration-after-statement); full Linux build links; the
rxml sample parses; series tip swept under ASan+UBSan+LSan and TSan
separately via samples/tasks/database/scan_begin_budget_test
(threaded lane and a 14-point cancellation sweep included), clean.

* lists/dir_list: add a resumable directory iterator

dir_list_new() is a blocking recursive walk: whoever calls it pays
for the whole tree in that call, which is why the manual content
scanner freezes the frame that starts a scan of a large library
when Threaded Tasks is off.  dir_list_iter_new()/_step()/_free()
is the same walk with the recursion made explicit - one frame per
open directory on a growable stack - so it can stop between any
two entries and resume on the next gather under a caller-supplied
budget callback, with a floor of one entry per step so a spent
budget still makes progress.  Entries land in the caller's list as
they are read: it is the only walk, nothing is enumerated twice.

The enumeration is a transliteration of dir_list_read_ctx(), and
two behaviours of the recursion that read like accidents are
contract and preserved deliberately: the recursive call's return
value is ignored, so an unreadable subdirectory is skipped silently
while its owed post-order append still happens; and a directory
itself is appended (under include_dirs) after its contents, because
the append sits after the recursive call.  The platform special
cases (System Volume Information, the IOS/OSX .framework-as-file
rule) are carried over verbatim.  Equivalence with dir_list_new()
on the same tree - same entries, same order, pre-sort - is pinned
by samples/tasks/database/scan_begin_budget_test, whose playlist
parity lane fails on any drift.

Verified: strict C89 syntax pass; full Linux build links; series
tip swept under ASan+UBSan+LSan and TSan separately (iterator
lifecycle covered mid-walk by the harness's cancellation sweep,
which frees iterators at every phase), clean.

* tasks: extract the shared I/O window into task_nbio_slice.c

The per-frame window that bounds file-transfer work per gather -
at most 4 ms of budgeted work per 16.6 ms of wall time, shared by
every participating task rather than handed to each one, with a
floor of one work item per open so a spent window still makes
progress - moves from file-local statics in task_file_transfer.c
into a translation unit of its own, and its type plus the three
entry points (task_nbio_slice_open/close/within_budget) are
exported through tasks_internal.h.

Two reasons.  The manual content scanner's BEGIN and END states
are about to become budgeted, and per the window's own design
rationale they must share this window, not invent a parallel
per-task budget that multiplies with the task count.  And a TU
whose only dependencies are the clock and the task queue lets test
harnesses link the real window instead of a reimplementation -
samples/tasks/file_transfer already compiles the shipping transfer
TU and now links this one with it; the scanner oracle does the
same.

Behaviour is unchanged: same constants, same threaded-queue
opt-out (each task gets a whole slice on the worker thread, where
there is no frame to protect and a shared static would be a race),
same floor-then-deadline accounting.  task_file_transfer_test
passes unmodified against the extracted TU, including its
budget-exhaustion and concurrency lanes.

Verified: strict C89 syntax pass on both TUs; full Linux build
links (Makefile.common + griffin wiring included); series tip
swept under ASan+UBSan+LSan and TSan separately, clean.

* task_database: fix empty-path strdup leak on scan handle teardown

content_database_path and playlist_directory are strdup'd
unconditionally when a manual content scan task is pushed, but the
teardown freed each only when it was non-empty - and strdup("") is
a real allocation.  Any scan pushed while the content database
path setting is empty (every scan in a bare configuration, and
every scan in the new oracle's fixture) leaked both, 16 bytes per
push under glibc.

Found by LeakSanitizer under the cancellation sweep of
samples/tasks/database/scan_begin_budget_test, which pushes 16
scans per run: 16 identical 16-byte leaks, all from
task_push_manual_content_scan.  The leak predates the scanner
rework this series performs; it lands first so the sweeps behind
the rest of the series run clean on their own merits.

Verified: strict C89 syntax pass; full Linux build links; series
tip swept under ASan+UBSan+LSan (leak detection on) and TSan
separately, clean - the sweep that found this reports zero leaks
with it applied.

* logiqx_dat: make the parser I/O-free

logiqx_dat.c performed two kinds of file I/O: logiqx_dat_path_is_valid()
stat'd the path for existence and size, and logiqx_dat_init(path)
read the whole document off disk (blocking, via rxml_load_document)
before parsing it.  The second is why the manual content scanner's
BEGIN state could stall a frame for the duration of a DAT read -
DAT files reach hundreds of megabytes - with no way for the task
layer to budget it: the read was buried inside the parser.

The module now performs no I/O of any kind.  Its API is:

  logiqx_dat_extension_is_valid()  - the extension whitelist (.dat /
      .xml, case-insensitive), a pure string check.  The whitelist is
      logiqx domain knowledge, so it stays here; existence and size
      checks belong to whoever owns the path.
  logiqx_dat_init_owned(buf, len)  - parses a document the caller
      read, taking ownership of the buffer (heap, len + 1 bytes,
      NUL-terminated) via rxml_load_document_owned: kept for the
      document's lifetime on success, freed here on failure, and no
      copy either way.

Call sites move accordingly.  manual_content_scan.c gets a static
content_scan_dat_path_is_valid() with exactly the old validation
semantics (extension, path_is_valid, path_get_size > 0, optional
size out-param for the free-memory check).  task_database.c
validates then reads the file itself and hands the buffer over -
still one blocking read in this commit, kept deliberately so this
change is purely a custody move; a following commit spreads that
read across gathers under the shared I/O window, which is the point
of the custody move.

Verified: strict C89 syntax pass on all four touched TUs; full
Linux build links (no other callers of the removed entry points
exist, griffin included); series tip swept under ASan+UBSan+LSan
and TSan separately - the DAT ownership transfer, including the
parse-failure path that frees the handed-over buffer, is exercised
by the scanner oracle's cancellation sweep - clean.

* logiqx_dat: index game names for logiqx_dat_search

logiqx_dat_search() walked every child of the root per call.  The
manual content scanner calls it once per content file to resolve
the playlist label, so a DAT-backed scan was O(files x games):
gprof put logiqx_dat_is_game_node at 29% of a 5000-file scan's
wall time, and the product is quadratic into MAME-sized lists.

The search now builds, lazily on first use, an index of every game
node keyed by its 'name' attribute - pointers into the document's
own storage, so the index owns no strings - sorted by (name,
document position) and answered by a leftmost-equal binary search.
Sorting by position second means duplicate names resolve to the
first occurrence, which is the answer the linear scan gave; nodes
without a non-empty name are exactly the nodes the linear matcher
could never match, so omitting them changes no answer either.  On
any allocation failure the index is marked failed and the linear
walk remains authoritative, and logiqx_dat_get_next()/reset()
iteration is untouched.

Cost: 24 bytes per game (~120 KB for the 5000-game fixture, a few
MB for a full MAME list), and the build-plus-qsort runs once inside
whichever gather issues the first search - milliseconds at fixture
scale; on a very large DAT under a non-threaded queue it is a
single one-off hitch far below the per-file scan it replaces, noted
in the scanner oracle alongside the parse-tick residual.

Measured on the oracle's fixture (5000 files, 5000-game DAT,
regular queue, release build): total scan 364.9 ms -> 183.6 ms,
which is 0.87x the blocking dir_list_new walk + whole-file DAT
parse reference - the incremental scanner is now faster end to end
than the code it replaced, not merely smoother.

Verified: strict C89 syntax pass; full Linux build links; oracle
passes all lanes including playlist parity; series tip swept under
ASan+UBSan+LSan and TSan separately, clean.

* database_info: add database_info_dir_init_from_list

database_info_dir_init() owns three things: the blocking recursive
dir_list_new() walk, the cue/gdi-prioritising sort, and the handle
construction.  The scanner's BEGIN split needs the walk performed
incrementally by the task under a time budget, but the sort and the
handle belong with the database layer - the comparator
(dir_entry_compare) is static here for a reason.

database_info_dir_init_from_list() is the second and third pieces
over a content list the caller already built: same sort, same
handle fields, borrows the list without taking ownership, NULL only
on allocation failure.  The directory variant remains for callers
that want the one-call blocking form.

Verified: strict C89 syntax pass; full Linux build links; series
tip swept under ASan+UBSan+LSan and TSan separately, clean.

* task_database: split MANUAL_SCAN_BEGIN into budgeted sub-states

MANUAL_SCAN_BEGIN performed the entire recursive directory walk
(dir_list_new via database_info_dir_init or the blocking content
list getter), the whole DAT file load and the playlist setup in a
single handler invocation.  Under the regular task queue that
invocation runs on the frame thread, so starting a scan of a large
library froze the frame for the duration - hundreds of
milliseconds at 5000 files, unbounded above - and the freeze is
exactly what Threaded Tasks off is supposed to avoid paying.

BEGIN is now a sequence: BEGIN (setup and branch selection) ->
BEGIN_DIR_LIST (the walk, via the resumable dir_list iterator) ->
BEGIN_DAT_LOAD (a chunked 256 KiB-per-read fill of a single
buffer, then one owned-buffer parse) -> BEGIN_PLAYLIST (playlist
open, parameter recording, and the final status selection, moved
verbatim).  Every sub-state runs under the shared per-frame I/O
window (task_nbio_slice_*, the same window the file-transfer spine
uses - deliberately shared rather than a parallel per-task budget,
which would multiply with the task count), each with a floor of
one work item so a spent window still progresses.  Completed
sub-states collapse into one invocation while the window lasts, so
a small library finishes the whole BEGIN sequence in a single tick
- the latency it had before the split.  The task's public identity
(handler, state) is unchanged, so duplicate-scan suppression via
task_queue_find() is unaffected.

The same total work is performed: the walk is incremental rather
than done twice, the DAT is read into one buffer whose ownership
passes to logiqx_dat_init_owned, and the DB branch's rdb-list
setup moves verbatim into the setup sub-state.  The non-DB
branch's content list creation moves to
manual_content_scan_content_list_iter_new(), which shares its
extension/compressed-inclusion policy with the blocking getter
through one helper so the two cannot drift; the caller owes the
completed list the same empty-reject and dir_list_sort the getter
applied.  Error semantics match the old code path for path: same
messages, same terminal exits, including dir walk allocation
failure mirroring dir_list_new() returning NULL.

Cancellation mid-walk or mid-read is now possible, so the handle
teardown releases the in-flight resources: the directory iterator
(which closes any open directory handles), the DAT stream, and the
read buffer when its ownership has not yet transferred.

Residual, documented in the oracle: the rxml parse of a completed
DAT buffer and the (following commit's) search-index build remain
one operation inside a single gather - budgeting the read bounds
the dominant I/O latency, but a pathological multi-hundred-MB DAT
still parses in one tick on a non-threaded queue.  rxml is not
resumable today; the sub-state structure leaves the seam for it.

Measured on the scanner oracle's fixture (5000 files across 50
dirs, 5000-game DAT, regular queue, release build): worst gather
attributable to the scanner's begin path drops from the full
walk+load+setup cost to inside the 4 ms window plus slack; first
task feedback 0.01 ms after push; playlist parity exact.

Verified: strict C89 syntax pass (HAVE_LIBRETRODB both ways); full
Linux build links; database_scan_test links and passes against
this commit standalone; series tip swept under ASan+UBSan+LSan and
TSan separately, including a 14-point cancellation sweep landing
in setup, mid-walk, mid-DAT-read, iteration and flush, clean.

* task_database: make the MANUAL_SCAN_END flush resumable

With BEGIN split, the largest single freeze left in a scan was the
gather that reached MANUAL_SCAN_END: it added every accumulated
scan result to the playlist - each add a linear
playlist_entry_exists() probe over the growing entry array, so the
batch is quadratic - then sorted and wrote the file, all in one
handler invocation.  Measured at 166 ms for the 5000-file fixture
on a release build; gprof attributes 48% of the whole scan's wall
time to playlist_path_matches_entry alone.

scan_results_batch_update_playlists() becomes
manual_scan_end_flush_tick(): the same loop with its locals -
position, the open playlist for the group being written, the group
key, the added counter, the path scratch buffer - promoted to the
scan handle, yielding on the shared per-frame window between any
two results with a floor of one result per gather.  The
MANUAL_SCAN_END case drives it until completion before running its
terminal message block.  Group-close writes for multi-playlist
scans and the final sort+write remain single blocking operations:
a playlist file write is atomic by design, and a partial write
would corrupt the playlist on a cancelled scan.

One behavioural fix rides along, forced by the loop conversion:
the old for loop's continue on a playlist that failed to open
advanced the index implicitly; the while loop must advance the
position itself, or the same unopenable playlist would be retried
every gather forever.

Cancellation mid-flush is now possible, so the handle teardown
closes the flush's open playlist - before manual_scan->playlist is
released, because in single-playlist mode flush_playlist borrows
that handle and the identity comparison preventing a double free
needs the pointer still live to say so - and frees the scratch.

Measured on the oracle's fixture (regular queue, release build):
worst gather over the whole scan 166 ms -> 5.6 ms, inside the 4 ms
window plus bounded slack; total scan time unchanged by this
commit (the quadratic probe cost still runs, spread out - its
removal is the playlist-side dedup index left as follow-up, with
the gprof profile attached to the oracle).

Verified: strict C89 syntax pass; full Linux build links; series
tip swept under ASan+UBSan+LSan and TSan separately, with
mid-flush cancellation covered by the sweep's tail points, clean.

* samples/tasks/database: add scan_begin_budget_test

Pacing + throughput oracle for the manual content scanner's
BEGIN/END split, compiled from the tree so it exercises the
shipping translation units.  It builds its own fixture - 5000
content files across 50 subdirectories plus a Logiqx DAT naming
each - so it needs nothing but a writable directory (honours
TMPDIR).

Three wall-clock assertions guard the plain build, because any one
alone can be satisfied by a scanner that is worse: pacing (no
task_queue_check() attributable to the scanner over the shared
4 ms window times a 4x slack multiplier - the pre-split tree fails
this at >160 ms), first feedback within two frames of the push,
and total scan time within 1.60x of the blocking walk + DAT parse
+ 40 us/entry floor it replaced, so a scanner that no longer
freezes but takes twice as long fails.  Playlist parity - every
fixture file present exactly once - is asserted in every mode, and
is also the equivalence pin for the dir_list iterator against
dir_list_new().  Measured at the series tip (release, regular
queue): total 183.6 ms = 0.87x reference, max gather 5.6 ms, first
feedback 0.01 ms.

A 'sanitize' mode runs every lane but skips the wall-clock
assertions a sanitized build cannot honestly meet; the sanitizers
are the assertion there.  Its lanes are where the split's
in-flight resources are made to die at every phase: a threaded
lane runs the same scan under the threaded queue (the TSan target
- the shared window deliberately opts threaded queues out of its
static state, and a mistake in that opt-out surfaces here), and a
14-point cancellation sweep lands a cancel in setup, mid-walk,
mid-DAT-read, content iteration and mid-flush, with LeakSanitizer
deciding whether teardown released everything.  The sweep already
paid for itself: it found the pre-existing empty-path strdup leak
fixed earlier in this series.

A 'bench' mode prints the metrics and skips assertions, for
regression-guard sampling.

The CI workflow gains the corresponding step (plain assertions,
then ASan+UBSan and TSan in sanitize mode) and its C89
syntax-check list gains the translation units this series touched.

Known residuals, deliberate and documented in the source: the rxml
parse of a completed DAT buffer and the logiqx index build each
run inside a single gather; playlist-side entry dedup remains
quadratic (48% of scan wall time per gprof) with the flush now
spreading rather than removing it - the fix is a path-hash dedup
index in playlist.c verifying candidates with the existing fuzzy
matcher, left as follow-up.

* samples/tasks/image_decode_slice: link task_nbio_slice

The window extraction moved task_nbio_slice_open/close/
within_budget out of task_file_transfer.c into their own
translation unit, and updated the two sample Makefiles built
during that work - file_transfer and database.  This oracle also
compiles the transfer spine (for the fill path its decode window
budgets), so it now fails to link on the three slice symbols,
which is exactly how CI reported it.

Add the slice TU to its source list.  No flag or include changes
are needed: the slice TU's includes are a subset of the transfer
TU's, and its external calls resolve against what the sample
already links - task_queue_is_threaded from task_queue.o, and
cpu_features_get_time_usec from the sample's own fake clock.  The
latter is not an accident to paper over but the point: the
oracle's budgets are deterministic because the clock is the test's
instrument, and the extracted window now runs against that same
instrument, as the in-tree window did before the extraction.

A repo-wide sweep confirms these were the only three sample
Makefiles compiling the transfer TU, so no further consumer is
left to fail one CI job at a time.

Verified: image_decode_slice_test builds and passes in both its CI
lanes against this Makefile - plain and ASan+UBSan (leak detection
and stack-use-after-return on).

* logiqx_dat: make the parser fully path-agnostic

logiqx_dat_extension_is_valid() performed no I/O - it was a pure
string check - but it was still the module reasoning about file
paths, and it forced logiqx_dat.c to keep including file/file_path.h
for path_get_extension().  With the file system checks already moved
to the callers, the .dat/.xml whitelist was the last piece of path
knowledge left in a parser whose contract is bytes in, structs out.

The whitelist moves into manual_content_scan.c, folded into the
validator introduced when the parser lost its I/O, which is now
exported as manual_content_scan_dat_path_is_valid(): extension,
existence, size > 0, optional size out-param.  Both menu-side call
sites use it, and task_database.c's budgeted DAT loader validates
through it instead of composing the three checks itself - so the
task can no longer drift from the acceptance rules the menu applied
when the path was chosen, which is a small correctness win on top
of the layering one.  logiqx_dat.c drops the function and the
file_path.h include; its remaining includes serve the parse alone.

Verified: strict C89 syntax pass (-std=c89 -ansi -pedantic
-Werror=declaration-after-statement) on all three touched TUs; full
Linux build links; scan_begin_budget_test passes all lanes
(playlist parity, threaded, 14-point cancellation sweep) and was
swept under ASan+UBSan+LSan and TSan separately, clean; repo-wide
grep confirms no caller of the removed entry point remains.

* samples/tasks/database: build the budget oracle from all; pace on CPU time

The new CI step follows the workflow's house pattern - `make clean
all`, then probe for the binary, then run it - but `all` in this
Makefile only built database_task, so the step's silent
`test -x scan_begin_budget_test` failed under set -eu right after
a successful database_task link, which is exactly the shape of the
CI log: link echo, no error text, exit 1.  It passed locally only
because a stale binary from an earlier explicit
`make scan_begin_budget_test` was still sitting in the directory.
The oracle joins the all target, so every build invocation in the
step - plain and both sanitizer configs - produces it, and plain
`make` in this directory builds everything runnable.

While correcting the step, the pacing lane switches its per-gather
measurement from wall time to thread CPU time
(CLOCK_THREAD_CPUTIME_ID, falling back to the wall clock where
unavailable).  The property under test is how much work the
scanner chooses to do in one gather; on a shared CI runner a cheap
gather can be preempted mid-flight and charged tens of wall-clock
milliseconds it never spent, failing the lane spuriously, while
the regression the lane exists for - the pre-split tree doing the
whole walk, DAT load and flush in one invocation - burns its
>160 ms as CPU and is caught by either clock.  Wall time remains
the clock for the total/throughput and first-feedback lanes, where
elapsed time is the property.

Verified with the step's literal command sequence in all three
build configurations: plain (`make clean all`, test -x, full
assertion run - PASS, max gather 8.2 ms CPU), ASan+UBSan+LSan
sanitize mode, and TSan sanitize mode, all clean.

* comments: describe the code as it is, not as it was

Review feedback on the scanner series: comments narrating how the
code previously behaved add nothing to reading it now - that story
lives in the commit messages.  Rewrite the affected comments in
present tense (the BEGIN-family and flush headers, the enum and
handle-field notes, the logiqx index rationale, the DAT-path
validator, the shared-window deadline note), keeping only what
states a live contract: the parity targets against the still-
present blocking entry points, the atomic-playlist-write floor,
the teardown ordering constraint, and the advance-before-continue
invariant in the flush loop.  The leak-fix guards lose their
comments entirely - a NULL-checked free needs no explanation - and
the database sample Makefile's comments go with them.

The regression oracle and the CI workflow keep their historical
descriptions: there, the previous behaviour is the documented
subject.

Comment-only change apart from the Makefile whitespace; verified
no object-code effect is possible, strict C89 pass on all touched
TUs, and the oracle's full plain run passes.

---------

Co-authored-by: LibretroAdmin <[email protected]>

.github/workflows/Linux-samples-tasks.yml

@@ -434,6 +434,68 @@ jobs:
              ./image_decode_slice_test
           echo "[pass] image_decode_slice_test (ASan)"
 
+      - name: Build and run scan_begin_budget_test (plain, ASan + UBSan, TSan)
+        shell: bash
+        working-directory: samples/tasks/database
+        run: |
+          set -eu
+          # Pacing + throughput oracle for the manual content scanner's
+          # BEGIN/END split in tasks/task_database.c, compiled from the
+          # tree so it exercises the shipping translation units.
+          #
+          # The regression it pins: MANUAL_SCAN_BEGIN used to perform
+          # the entire recursive directory walk, the whole DAT load and
+          # the playlist setup in one handler invocation, and
+          # MANUAL_SCAN_END flushed every accumulated result (each add
+          # a linear playlist_entry_exists probe) plus a sort and a
+          # write in another - 166 ms measured for the 5000-file
+          # fixture, a hard frame freeze whenever Threaded Tasks is
+          # off.  Both are now sequences of sub-states under the shared
+          # per-frame I/O window (tasks/task_nbio_slice.c, the same
+          # window the file-transfer spine uses), with completed
+          # sub-states collapsing into one gather so small libraries
+          # keep single-tick latency.
+          #
+          # Three wall-clock assertions guard the plain build, because
+          # any one alone can be satisfied by a scanner that is worse:
+          # pacing (no gather over the window plus bounded slack),
+          # first feedback within two frames of the push, and total
+          # scan time within a factor of the blocking walk + DAT parse
+          # it replaced - a scanner that no longer freezes but takes
+          # twice as long fails the third lane.  Playlist parity (every
+          # fixture file present exactly once) is asserted in every
+          # mode.
+          #
+          # The sanitizer runs use the 'sanitize' mode, which runs
+          # every lane but skips the wall-clock assertions a sanitized
+          # build cannot honestly meet.  Those lanes are where the
+          # in-flight resources the split introduced - an open
+          # directory iterator, an open DAT stream over a half-filled
+          # buffer, an open flush playlist - are made to die at every
+          # phase: a 14-point cancellation sweep lands a cancel in
+          # setup, mid-walk, mid-DAT-read, content iteration and
+          # mid-flush, and LeakSanitizer decides whether the teardown
+          # released everything (it already found one 16-byte-per-scan
+          # leak that predates the split).  The threaded lane runs the
+          # same scan under the threaded queue and is the TSan target:
+          # the shared window deliberately opts threaded queues out of
+          # its static state, and this is where a mistake in that
+          # opt-out would surface.
+          make clean all
+          test -x scan_begin_budget_test
+          timeout 600 ./scan_begin_budget_test
+          echo "[pass] scan_begin_budget_test"
+          make clean all SANITIZER=address,undefined
+          ASAN_OPTIONS=detect_leaks=1:detect_stack_use_after_return=1 \
+             UBSAN_OPTIONS=print_stacktrace=1 timeout 900 \
+             ./scan_begin_budget_test sanitize
+          echo "[pass] scan_begin_budget_test (ASan)"
+          make clean all SANITIZER=thread
+          TSAN_OPTIONS=halt_on_error=1 timeout 900 \
+             ./scan_begin_budget_test sanitize
+          make clean
+          echo "[pass] scan_begin_budget_test (TSan)"
+
       - name: Build and run save_state_io_test (ASan + UBSan, TSan)
         shell: bash
         working-directory: samples/tasks/save_state
@@ -604,7 +666,13 @@ jobs:
                    tasks/task_pl_thumbnail_download.c \
                    tasks/task_core_backup.c \
                    core_backup.c \
-                   libretro-common/streams/interface_stream.c; do
+                   libretro-common/streams/interface_stream.c \
+                   tasks/task_file_transfer.c \
+                   tasks/task_nbio_slice.c \
+                   libretro-common/lists/dir_list.c \
+                   libretro-common/formats/logiqx_dat/logiqx_dat.c \
+                   libretro-common/formats/xml/rxml.c \
+                   database_info.c; do
             if gcc -fsyntax-only $C89 $INC $DEF "$f"; then
               echo "[pass] C89 $f"
             else

Makefile.common

@@ -265,6 +265,7 @@ OBJ += \
        tasks/task_save.o \
        tasks/task_movie.o \
        tasks/task_file_transfer.o \
+       tasks/task_nbio_slice.o \
        tasks/task_content_prefetch.o \
        tasks/task_image.o \
        tasks/task_playlist_manager.o \

database_info.c

@@ -809,6 +809,26 @@ database_info_handle_t *database_info_dir_init(const char *dir,
    return db;
 }
 
+database_info_handle_t *database_info_dir_init_from_list(
+      enum database_type type, struct string_list *list)
+{
+   database_info_handle_t *db = (database_info_handle_t*)
+      malloc(sizeof(*db));
+
+   if (!db)
+      return NULL;
+
+   /* dir list prioritize - same cue/gdi ordering as
+    * database_info_dir_init() */
+   qsort(list->elems, list->size, sizeof(*list->elems),
+         dir_entry_compare);
+
+   db->status = DATABASE_STATUS_ITERATE;
+   db->type   = type;
+
+   return db;
+}
+
 database_info_handle_t *database_info_file_init(const char *path,
       enum database_type type, retro_task_t *task, struct string_list **content_list)
 {

database_info.h

@@ -226,6 +226,15 @@ database_info_handle_t *database_info_dir_init(const char *dir,
       bool show_hidden_files, bool recursive, bool include_archive, 
       struct string_list **content_list);
 
+/* As database_info_dir_init(), but over a content list the caller has
+ * already built (e.g. incrementally via dir_list_iter_step() so the
+ * walk could be spread across task gathers).  Applies the same
+ * cue/gdi-prioritising sort the directory variant applies and borrows
+ * @list without taking ownership.  Returns NULL only on allocation
+ * failure. */
+database_info_handle_t *database_info_dir_init_from_list(
+      enum database_type type, struct string_list *list);
+
 database_info_handle_t *database_info_file_init(const char *path,
       enum database_type type, retro_task_t *task, struct string_list **content_list);
 

griffin/griffin.c

@@ -1418,6 +1418,7 @@ DATA RUNLOOP
 #include "../tasks/task_content_prefetch.c"
 #include "../tasks/task_image.c"
 #include "../tasks/task_file_transfer.c"
+#include "../tasks/task_nbio_slice.c"
 #include "../tasks/task_playlist_manager.c"
 #include "../tasks/task_core_backup.c"
 #ifdef HAVE_TRANSLATE

libretro-common/formats/logiqx_dat/logiqx_dat.c

@@ -20,17 +20,36 @@
  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  */
 
-#include <file/file_path.h>
 #include <string/stdstring.h>
 #include <formats/rxml.h>
 
 #include <formats/logiqx_dat.h>
 
+/* One entry of the lazy search index: a game node keyed by its
+ * 'name' attribute, with its position in the document so ties
+ * between duplicate names resolve to the first occurrence - the
+ * same answer the linear scan gives. */
+typedef struct
+{
+   const char *name;   /* points into the document's own storage */
+   rxml_node_t *node;
+   size_t seq;
+} logiqx_dat_index_entry_t;
+
 /* Holds all internal DAT file data */
 struct logiqx_dat
 {
    rxml_document_t *data;
    rxml_node_t *current_node;
+   /* Lazy search index, built on the first logiqx_dat_search() and
+    * sorted by name for binary search: without it every search
+    * walks every child of the root, which is O(files x games) for
+    * the scanner's per-file label lookups and quadratic into
+    * MAME-sized lists.  On any allocation failure index_failed is
+    * set and the linear walk remains the answer. */
+   logiqx_dat_index_entry_t *index;
+   size_t index_size;
+   bool index_failed;
 };
 
 /* List of HTML formatting codes that must
@@ -45,73 +64,38 @@ const char *logiqx_dat_html_code_list[][2] = {
 
 #define LOGIQX_DAT_HTML_CODE_LIST_SIZE 5
 
-/* Validation */
+/* Initialisation/de-initialisation */
 
-/* Performs rudimentary validation of the specified
- * Logiqx XML DAT file path (not rigorous - just
- * enough to prevent obvious errors).
- * Also provides access to file size (DAT files can
- * be very large, so it is useful to have this information
- * on hand - i.e. so we can check that the system has
- * enough free memory to load the file). */
-bool logiqx_dat_path_is_valid(const char *path, uint64_t *file_size)
-{
-   const char *file_ext = NULL;
-   int64_t file_size_int;
-
-   if (!path || !*path)
-      return false;
-
-   /* Check file extension */
-   file_ext = path_get_extension(path);
-
-   if (!file_ext || !*file_ext)
-      return false;
-
-   if (   !string_is_equal_noncase(file_ext, "dat")
-       && !string_is_equal_noncase(file_ext, "xml"))
-      return false;
-
-   /* Ensure file exists */
-   if (!path_is_valid(path))
-      return false;
-
-   /* Get file size */
-   file_size_int = path_get_size(path);
-
-   if (file_size_int <= 0)
-      return false;
-
-   if (file_size)
-      *file_size = (uint64_t)file_size_int;
-
-   return true;
-}
-
-/* File initialisation/de-initialisation */
-
-/* Loads specified Logiqx XML DAT file from disk.
- * Returned logiqx_dat_t object must be free'd using
- * logiqx_dat_free().
- * Returns NULL if file is invalid or a read error
- * occurs. */
-logiqx_dat_t *logiqx_dat_init(const char *path)
+/* Parses the specified Logiqx XML DAT document held in
+ * memory.  Takes ownership of @xml_data (heap, len + 1
+ * bytes, NUL-terminated at len); it is released by
+ * logiqx_dat_free(), or here on failure.  Reading the
+ * document from disk is the caller's job - this module
+ * performs no file I/O.
+ * Returns NULL if the document is not valid Logiqx/MAME
+ * XML. */
+logiqx_dat_t *logiqx_dat_init_owned(char *xml_data, size_t len)
 {
    logiqx_dat_t *dat_file = NULL;
    rxml_node_t *root_node = NULL;
 
-   /* Check file path */
-   if (!logiqx_dat_path_is_valid(path, NULL))
-      goto error;
+   if (!xml_data)
+      return NULL;
 
-   /* Create logiqx_dat_t object */
-   dat_file = (logiqx_dat_t*)calloc(1, sizeof(*dat_file));
+   /* Create logiqx_dat_t object.  On this one failure
+    * path the buffer has not yet been handed to rxml,
+    * so it is freed here to honour the ownership
+    * contract. */
+   if (!(dat_file = (logiqx_dat_t*)calloc(1, sizeof(*dat_file))))
+   {
+      free(xml_data);
+      return NULL;
+   }
 
-   if (!dat_file)
-      goto error;
-
-   /* Read file from disk */
-   dat_file->data = rxml_load_document(path);
+   /* Parse document.  rxml_load_document_owned takes
+    * the buffer: the tree points into it on success,
+    * and it is freed on failure. */
+   dat_file->data = rxml_load_document_owned(xml_data, len);
 
    if (!dat_file->data)
       goto error;
@@ -155,6 +139,12 @@ void logiqx_dat_free(logiqx_dat_t *dat_file)
 
    dat_file->current_node = NULL;
 
+   if (dat_file->index)
+   {
+      free(dat_file->index);
+      dat_file->index = NULL;
+   }
+
    if (dat_file->data)
    {
       rxml_free_document(dat_file->data);
@@ -416,6 +406,84 @@ bool logiqx_dat_get_next(
 /* Fetches information for the specified game.
  * Returns false if game does not exist, or arguments
  * are invalid. */
+/* qsort comparator for the search index: by name, then by document
+ * position, so equal names keep document order and the leftmost
+ * match is the first occurrence. */
+static int logiqx_dat_index_compare(const void *a_, const void *b_)
+{
+   const logiqx_dat_index_entry_t *a = (const logiqx_dat_index_entry_t*)a_;
+   const logiqx_dat_index_entry_t *b = (const logiqx_dat_index_entry_t*)b_;
+   int cmp = strcmp(a->name, b->name);
+   if (cmp)
+      return cmp;
+   if (a->seq < b->seq)
+      return -1;
+   if (a->seq > b->seq)
+      return 1;
+   return 0;
+}
+
+/* Build the sorted name index over every game node that carries a
+ * non-empty 'name' attribute.  Nodes without one are exactly the
+ * nodes the linear scan could never match (the matcher rejects
+ * them), so leaving them out changes no answer. */
+static void logiqx_dat_index_build(logiqx_dat_t *dat_file)
+{
+   rxml_node_t *root_node = NULL;
+   rxml_node_t *game_node = NULL;
+   size_t count           = 0;
+
+   dat_file->index_failed = true;   /* cleared on success */
+
+   if (!(root_node = rxml_root_node(dat_file->data)))
+      return;
+
+   for (game_node = root_node->children; game_node; game_node = game_node->next)
+   {
+      if (logiqx_dat_is_game_node(game_node))
+      {
+         const char *name = rxml_node_attrib(game_node, "name");
+         if (name && *name)
+            count++;
+      }
+   }
+
+   if (!count)
+   {
+      /* An empty index is a valid one: every search misses, which
+       * is what the linear scan would conclude too. */
+      dat_file->index_size   = 0;
+      dat_file->index_failed = false;
+      return;
+   }
+
+   if (!(dat_file->index = (logiqx_dat_index_entry_t*)
+         malloc(count * sizeof(*dat_file->index))))
+      return;
+
+   count = 0;
+   for (game_node = root_node->children; game_node; game_node = game_node->next)
+   {
+      if (logiqx_dat_is_game_node(game_node))
+      {
+         const char *name = rxml_node_attrib(game_node, "name");
+         if (name && *name)
+         {
+            dat_file->index[count].name = name;
+            dat_file->index[count].node = game_node;
+            dat_file->index[count].seq  = count;
+            count++;
+         }
+      }
+   }
+
+   qsort(dat_file->index, count, sizeof(*dat_file->index),
+         logiqx_dat_index_compare);
+
+   dat_file->index_size   = count;
+   dat_file->index_failed = false;
+}
+
 bool logiqx_dat_search(
       logiqx_dat_t *dat_file, const char *game_name,
       logiqx_dat_game_info_t *game_info)
@@ -429,6 +497,34 @@ bool logiqx_dat_search(
    if (!dat_file->data)
       return false;
 
+   /* Build the name index on first use; on failure fall through to
+    * the linear walk below, which remains authoritative. */
+   if (!dat_file->index && !dat_file->index_size && !dat_file->index_failed)
+      logiqx_dat_index_build(dat_file);
+
+   if (!dat_file->index_failed)
+   {
+      /* Binary search for the leftmost entry with this name: lower
+       * bound over (name, seq)-sorted entries. */
+      size_t lo = 0;
+      size_t hi = dat_file->index_size;
+
+      while (lo < hi)
+      {
+         size_t mid = lo + ((hi - lo) >> 1);
+         if (strcmp(dat_file->index[mid].name, game_name) < 0)
+            lo = mid + 1;
+         else
+            hi = mid;
+      }
+
+      if (   lo < dat_file->index_size
+          && string_is_equal(dat_file->index[lo].name, game_name))
+         return logiqx_dat_parse_game_node(dat_file->index[lo].node,
+               game_info);
+      return false;
+   }
+
    /* Get root node */
    root_node = rxml_root_node(dat_file->data);
 

libretro-common/formats/xml/rxml.c

@@ -1688,6 +1688,16 @@ rxml_document_t *rxml_load_document_string_opts(const char *str,
    return rxml_parse_owned(buf, len, opts, err);
 }
 
+rxml_document_t *rxml_load_document_owned(char *buf, size_t len)
+{
+   if (!buf)
+      return NULL;
+   /* The contract mirrors the tail of rxml_load_document: the caller
+    * hands over a NUL-terminated heap buffer and rxml_parse_owned
+    * either gives it to the document or frees it on failure. */
+   return rxml_parse_owned(buf, len, 0, NULL);
+}
+
 rxml_document_t *rxml_load_document_string(const char *str)
 {
    return rxml_load_document_string_opts(str, 0, NULL);

libretro-common/include/formats/logiqx_dat.h

@@ -58,25 +58,25 @@ typedef struct
    bool is_runnable;
 } logiqx_dat_game_info_t;
 
-/* Validation */
+/* Initialisation/de-initialisation */
 
-/* Performs rudimentary validation of the specified
- * Logiqx XML DAT file path (not rigorous - just
- * enough to prevent obvious errors).
- * Also provides access to file size (DAT files can
- * be very large, so it is useful to have this information
- * on hand - i.e. so we can check that the system has
- * enough free memory to load the file). */
-bool logiqx_dat_path_is_valid(const char *path, uint64_t *file_size);
-
-/* File initialisation/de-initialisation */
-
-/* Loads specified Logiqx XML DAT file from disk.
+/* Parses the specified Logiqx XML DAT document held
+ * in memory.  Takes ownership of @xml_data - a heap
+ * allocation of at least @len + 1 bytes with
+ * xml_data[len] == '\0' - which is kept for the
+ * lifetime of the returned object and released by
+ * logiqx_dat_free(); on failure it is freed here
+ * before NULL is returned, so the caller must not
+ * touch it again either way.
+ * Reading the document from disk is the caller's
+ * job (see e.g. the chunked, budgeted read in
+ * tasks/task_database.c); this module performs no
+ * file I/O.
  * Returned logiqx_dat_t object must be free'd using
  * logiqx_dat_free().
- * Returns NULL if file is invalid or a read error
- * occurs. */
-logiqx_dat_t *logiqx_dat_init(const char *path);
+ * Returns NULL if the document is not valid
+ * Logiqx/MAME XML. */
+logiqx_dat_t *logiqx_dat_init_owned(char *xml_data, size_t len);
 
 /* Frees specified DAT file */
 void logiqx_dat_free(logiqx_dat_t *dat_file);

libretro-common/include/formats/rxml.h

@@ -83,6 +83,17 @@ typedef struct rxml_parse_error
 rxml_document_t *rxml_load_document(const char *path);
 rxml_document_t *rxml_load_document_string(const char *str);
 
+/* As rxml_load_document_string, but takes ownership of @buf - a heap
+ * allocation of at least @len + 1 bytes with buf[len] == '\0' -
+ * instead of copying it.  The document keeps the buffer alive for its
+ * whole lifetime (the tree points into it) and releases it in
+ * rxml_free_document; on failure the buffer is freed here before NULL
+ * is returned, so the caller must not touch it again either way.
+ * For large inputs this halves peak memory against the copying string
+ * entry point, which is what callers that already hold the document
+ * bytes (e.g. a task that read the file itself) should care about. */
+rxml_document_t *rxml_load_document_owned(char *buf, size_t len);
+
 /* As rxml_load_document_string, with parse options; on failure, *err
  * (when non-NULL) receives the position the parser stopped at. */
 rxml_document_t *rxml_load_document_string_opts(const char *str,

libretro-common/include/lists/dir_list.h

@@ -64,6 +64,72 @@ bool dir_list_append(struct string_list *list, const char *dir, const char *ext,
 struct string_list *dir_list_new(const char *dir, const char *ext,
       bool include_dirs, bool include_hidden, bool include_compressed, bool recursive);
 
+/* Resumable directory walk.
+ *
+ * Produces, incrementally, exactly the listing dir_list_new() builds
+ * in one blocking call: same entries, same filters, same order,
+ * including the post-order append of directories when @include_dirs
+ * is set and the silent skip of unreadable subdirectories.  It exists
+ * so a task handler can spread a large recursive walk across gathers
+ * under a time budget instead of freezing the frame that starts it -
+ * and it is the only walk: entries land in the caller's list as they
+ * are read, nothing is enumerated twice.  See dir_list_iter_step()
+ * for the yielding contract. */
+typedef struct dir_list_iter dir_list_iter_t;
+
+/**
+ * dir_list_iter_new:
+ * @dir                : directory path.
+ * @ext                : allowed extensions of file directory entries to include.
+ * @include_dirs       : include directories as part of the finished directory listing?
+ * @include_hidden     : include hidden files and directories as part of the finished directory listing?
+ * @include_compressed : include compressed files, even when not part of ext.
+ * @recursive          : list directory contents recursively
+ * @list               : the string list entries are appended to.  Borrowed:
+ *                       the caller keeps ownership and must keep it alive
+ *                       until the iterator is freed.
+ *
+ * Begin a resumable directory walk equivalent to
+ * dir_list_append(list, dir, ...).
+ *
+ * @return iterator on success, NULL if @dir cannot be opened (the
+ * same condition under which dir_list_new() returns NULL) or on
+ * allocation failure.  Has to be freed with dir_list_iter_free().
+ **/
+dir_list_iter_t *dir_list_iter_new(const char *dir, const char *ext,
+      bool include_dirs, bool include_hidden, bool include_compressed,
+      bool recursive, struct string_list *list);
+
+/**
+ * dir_list_iter_step:
+ * @iter          : iterator from dir_list_iter_new().
+ * @within_budget : optional; consulted between entries, return false
+ *                  to make the step yield.  NULL runs to completion.
+ * @userdata      : passed to @within_budget.
+ *
+ * Advance the walk, appending entries to the list given at creation.
+ * At least one entry is always processed per call (the floor: a queue
+ * whose budget is already spent still makes progress), after which
+ * @within_budget is consulted before each further entry.
+ *
+ * @return 1 when the walk has completed (the list now equals what
+ * dir_list_new() would have produced, pre-sort), 0 when yielding with
+ * work remaining, -1 on allocation failure (the walk is abandoned;
+ * free the iterator).
+ **/
+int dir_list_iter_step(dir_list_iter_t *iter,
+      bool (*within_budget)(void *userdata), void *userdata);
+
+/**
+ * dir_list_iter_free:
+ * @iter : iterator, may be NULL.
+ *
+ * Release the iterator, closing any directory handles still open.
+ * Safe to call at any point of the walk; the entries already appended
+ * to the caller's list remain valid.
+ **/
+void dir_list_iter_free(dir_list_iter_t *iter);
+
 /**
  * dir_list_initialize:
  *

libretro-common/lists/dir_list.c

@@ -310,6 +310,304 @@ static int dir_list_read_ctx(size_t dir_len, struct dir_list_ctx *ctx)
    return 0;
 }
 
+/* Resumable directory walk ------------------------------------------
+ *
+ * dir_list_read_ctx() above is a depth-first recursion over a shared
+ * path-prefix buffer.  The iterator below is the same walk with the
+ * recursion made explicit - a frame per open directory - so it can
+ * stop between any two entries and resume later.  Every filtering
+ * decision is a transliteration of the recursive body, and the
+ * equivalence is pinned by the parity lane of
+ * samples/tasks/database/scan_begin_budget_test.c, which diffs the
+ * iterator's output against dir_list_new() on the same tree.
+ *
+ * Two behaviours of the recursion that read like accidents are
+ * contract here and preserved deliberately:
+ *  - the return value of the recursive call is ignored, so an
+ *    unreadable subdirectory is skipped silently while the walk
+ *    continues, and
+ *  - a directory itself is appended (under include_dirs) after its
+ *    contents, i.e. post-order, because the append sits after the
+ *    recursive call. */
+
+struct dir_list_iter_frame
+{
+   struct RDIR *entry;    /* open directory handle for this level    */
+   size_t       dir_len;  /* prefix length children append at        */
+   size_t       pend_len; /* path length of the dir itself, for the  *
+                           * post-order append on pop (root: unused) */
+   bool         pend_append; /* append this dir on pop?              */
+};
+
+struct dir_list_iter
+{
+   struct string_list *list;             /* borrowed                 */
+   struct string_list  ext_list;         /* owned storage...         */
+   struct string_list *ext_list_ptr;     /* ...NULL when no filter   */
+   char               *path;             /* shared prefix buffer     */
+   struct dir_list_iter_frame *stack;
+   size_t              depth;            /* frames in use            */
+   size_t              cap;              /* frames allocated         */
+   bool                include_dirs;
+   bool                include_hidden;
+   bool                include_compressed;
+   bool                recursive;
+   bool                done;
+   bool                failed;
+};
+
+static bool dir_list_iter_push(dir_list_iter_t *iter,
+      struct RDIR *entry, size_t dir_len, size_t pend_len,
+      bool pend_append)
+{
+   struct dir_list_iter_frame *frame;
+   if (iter->depth == iter->cap)
+   {
+      size_t new_cap = iter->cap ? (iter->cap * 2) : 8;
+      struct dir_list_iter_frame *new_stack =
+            (struct dir_list_iter_frame*)realloc(iter->stack,
+                  new_cap * sizeof(*new_stack));
+      if (!new_stack)
+         return false;
+      iter->stack = new_stack;
+      iter->cap   = new_cap;
+   }
+   frame              = &iter->stack[iter->depth++];
+   frame->entry       = entry;
+   frame->dir_len     = dir_len;
+   frame->pend_len    = pend_len;
+   frame->pend_append = pend_append;
+   return true;
+}
+
+dir_list_iter_t *dir_list_iter_new(const char *dir, const char *ext,
+      bool include_dirs, bool include_hidden, bool include_compressed,
+      bool recursive, struct string_list *list)
+{
+   dir_list_iter_t *iter;
+   struct RDIR *entry;
+   size_t dir_len;
+
+   if (!dir || !list)
+      return NULL;
+
+   if (!(iter = (dir_list_iter_t*)calloc(1, sizeof(*iter))))
+      return NULL;
+
+   if (!(iter->path = (char*)malloc(PATH_MAX_LENGTH)))
+   {
+      free(iter);
+      return NULL;
+   }
+
+   if (ext)
+   {
+      string_list_initialize(&iter->ext_list);
+      string_split_noalloc(&iter->ext_list, ext, "|");
+      iter->ext_list_ptr    = &iter->ext_list;
+   }
+
+   iter->list               = list;
+   iter->include_dirs       = include_dirs;
+   iter->include_hidden     = include_hidden;
+   iter->include_compressed = include_compressed;
+   iter->recursive          = recursive;
+
+   dir_len = strlcpy(iter->path, dir, PATH_MAX_LENGTH);
+
+   /* A root that cannot be opened is the condition under which
+    * dir_list_new() fails outright - unlike a child, which is
+    * skipped. */
+   entry   = retro_opendir_include_hidden(iter->path, include_hidden);
+   if (!entry)
+      goto error;
+   if (retro_dirent_error(entry))
+   {
+      retro_closedir(entry);
+      goto error;
+   }
+   if (!dir_list_iter_push(iter, entry, dir_len, 0, false))
+   {
+      retro_closedir(entry);
+      goto error;
+   }
+   return iter;
+
+error:
+   dir_list_iter_free(iter);
+   return NULL;
+}
+
+int dir_list_iter_step(dir_list_iter_t *iter,
+      bool (*within_budget)(void *userdata), void *userdata)
+{
+   char *path;
+   bool first = true;
+
+   if (!iter || iter->failed)
+      return -1;
+   if (iter->done)
+      return 1;
+
+   path = iter->path;
+
+   while (iter->depth)
+   {
+      struct dir_list_iter_frame *frame = &iter->stack[iter->depth - 1];
+      struct RDIR *entry                = frame->entry;
+      size_t dir_len                    = frame->dir_len;
+      const char *name;
+      size_t _len;
+      union string_list_elem_attr attr;
+
+      /* The floor: one entry per call regardless of the budget, so a
+       * spent window still makes progress.  After it, yield between
+       * any two entries. */
+      if (!first && within_budget && !within_budget(userdata))
+         return 0;
+      first = false;
+
+      if (!retro_readdir(entry))
+      {
+         /* Level exhausted: close, pop, and perform the parent's
+          * pending post-order append of the directory itself (the
+          * code after the recursive call in dir_list_read_ctx). */
+         bool pend_append = frame->pend_append;
+         size_t pend_len  = frame->pend_len;
+
+         retro_closedir(entry);
+         iter->depth--;
+
+         if (pend_append)
+         {
+            path[pend_len] = '\0';
+            attr.i         = RARCH_DIRECTORY;
+            if (!string_list_append(iter->list, path, attr))
+               goto fail;
+         }
+         continue;
+      }
+
+      name = retro_dirent_get_name(entry);
+
+      if (name[0] == '.' || name[0] == '$')
+      {
+         /* Do not include hidden files and directories */
+         if (!iter->include_hidden)
+            continue;
+         /* char-wise comparisons to avoid string comparison */
+         /* Do not include current dir */
+         if (name[1] == '\0')
+            continue;
+         /* Do not include parent dir */
+         if (name[1] == '.' && name[2] == '\0')
+            continue;
+      }
+
+      /* Append @name to the prefix instead of rebuilding the whole
+       * path: the prefix is already in place from the parent level. */
+      _len = dir_len;
+      if (_len && path[_len - 1] != '/' && path[_len - 1] != '\\')
+         path[_len++] = '/';
+      _len += strlcpy(path + _len, name, PATH_MAX_LENGTH - _len);
+
+      if (retro_dirent_is_dir(entry, NULL))
+      {
+         /* Exclude this frequent hidden dir on platforms which can not handle hidden attribute */
+         if (!iter->include_hidden && strcmp(name, "System Volume Information") == 0)
+            continue;
+
+#if defined(IOS) || defined(OSX)
+         {
+            size_t name_len = strlen(name);
+            if (name_len >= 10
+                  && !memcmp(name + name_len - 10, ".framework", 10))
+            {
+               attr.i = RARCH_PLAIN_FILE;
+               if (!string_list_append(iter->list, path, attr))
+                  goto fail;
+               continue;
+            }
+         }
+#endif
+         if (iter->recursive)
+         {
+            /* Descend.  A child that cannot be opened is skipped
+             * silently - the recursion ignored the child call's
+             * return value - but the post-order append it owed
+             * still happens, so perform it here directly. */
+            struct RDIR *child = retro_opendir_include_hidden(path,
+                  iter->include_hidden);
+
+            if (child && retro_dirent_error(child))
+            {
+               retro_closedir(child);
+               child = NULL;
+            }
+
+            if (child)
+            {
+               if (!dir_list_iter_push(iter, child, _len, _len,
+                        iter->include_dirs))
+               {
+                  retro_closedir(child);
+                  goto fail;
+               }
+               continue;
+            }
+
+            path[_len] = '\0';
+         }
+
+         if (!iter->include_dirs)
+            continue;
+         attr.i = RARCH_DIRECTORY;
+      }
+      else
+      {
+         const char *file_ext    = path_get_extension(name);
+
+         attr.i                  = RARCH_FILETYPE_UNSET;
+
+         if (string_list_find_elem_prefix(iter->ext_list_ptr, ".", file_ext))
+            attr.i            = RARCH_PLAIN_FILE;
+         else
+         {
+            bool is_compressed_file;
+            if ((is_compressed_file = path_is_compressed_file(path)))
+               attr.i               = RARCH_COMPRESSED_ARCHIVE;
+
+            if (iter->ext_list_ptr &&
+                  (!is_compressed_file || !iter->include_compressed))
+               continue;
+         }
+      }
+
+      if (!string_list_append(iter->list, path, attr))
+         goto fail;
+   }
+
+   iter->done = true;
+   return 1;
+
+fail:
+   iter->failed = true;
+   return -1;
+}
+
+void dir_list_iter_free(dir_list_iter_t *iter)
+{
+   size_t i;
+   if (!iter)
+      return;
+   for (i = 0; i < iter->depth; i++)
+      retro_closedir(iter->stack[i].entry);
+   free(iter->stack);
+   free(iter->path);
+   string_list_deinitialize(&iter->ext_list);
+   free(iter);
+}
+
 /**
  * dir_list_append:
  * @list               : existing list to append to.

manual_content_scan.c

@@ -262,6 +262,51 @@ void manual_content_scan_scrub_file_exts_custom(void)
    manual_content_scan_scrub_file_exts(scan_file_exts_custom);
 }
 
+/* Performs rudimentary validation of the specified
+ * Logiqx XML DAT file path (not rigorous - just enough
+ * to prevent obvious errors), optionally returning the
+ * file size (DAT files can be very large, so callers
+ * use it for free-memory checks).
+ * The parser (logiqx_dat.c) is path-agnostic - bytes
+ * in, structs out - so both the extension whitelist
+ * and the file system checks live here, with the scan
+ * configuration that owns the path. */
+bool manual_content_scan_dat_path_is_valid(
+      const char *path, uint64_t *file_size)
+{
+   const char *file_ext = NULL;
+   int64_t file_size_int;
+
+   if (!path || !*path)
+      return false;
+
+   /* Check file extension (.dat / .xml, the extensions
+    * a Logiqx XML DAT may legitimately carry) */
+   file_ext = path_get_extension(path);
+
+   if (!file_ext || !*file_ext)
+      return false;
+
+   if (   !string_is_equal_noncase(file_ext, "dat")
+       && !string_is_equal_noncase(file_ext, "xml"))
+      return false;
+
+   /* Ensure file exists */
+   if (!path_is_valid(path))
+      return false;
+
+   /* Get file size */
+   file_size_int = path_get_size(path);
+
+   if (file_size_int <= 0)
+      return false;
+
+   if (file_size)
+      *file_size = (uint64_t)file_size_int;
+
+   return true;
+}
+
 /* Checks 'dat_file_path' string and resets it
  * if invalid */
 enum manual_content_scan_dat_file_path_status
@@ -276,7 +321,7 @@ enum manual_content_scan_dat_file_path_status
       uint64_t file_size;
 
       /* Check if path itself is valid */
-      if (logiqx_dat_path_is_valid(scan_dat_file_path, &file_size))
+      if (manual_content_scan_dat_path_is_valid(scan_dat_file_path, &file_size))
       {
          uint64_t free_memory = mem_stats_free();
          dat_file_path_status = MANUAL_CONTENT_SCAN_DAT_FILE_OK;
@@ -1525,7 +1570,7 @@ bool manual_content_scan_get_task_config(
        || scan_settings.db_usage == MANUAL_CONTENT_SCAN_USE_DB_DAT_LOOSE)
        && *scan_dat_file_path)
    {
-      if (!logiqx_dat_path_is_valid(scan_dat_file_path, NULL))
+      if (!manual_content_scan_dat_path_is_valid(scan_dat_file_path, NULL))
          return false;
       strlcpy(
             task_config->dat_file_path,
@@ -1556,6 +1601,91 @@ bool manual_content_scan_get_task_config(
  * content directory
  * > Returns NULL in the event of failure
  * > Returned string list must be free()'d */
+/* Shared policy for both content-list builders below:
+ * whether extension filtering applies and whether
+ * compressed files must be listed. */
+static void manual_content_scan_content_list_policy(
+      manual_content_scan_task_config_t *task_config,
+      bool *filter_exts, bool *include_compressed)
+{
+   /* Check whether files should be filtered by
+    * extension */
+   *filter_exts = *task_config->file_exts;
+
+   /* Check whether compressed files should be
+    * included in the directory list
+    * > If compressed files are already listed in
+    *   the 'file_exts' string, they will be included
+    *   automatically
+    * > If we don't have a 'file_exts' list, then all
+    *   files must be included regardless of type
+    * > If user has enabled 'search inside archives',
+    *   then compressed files must of course be included */
+   *include_compressed = (!*filter_exts || task_config->search_archives);
+}
+
+bool manual_content_scan_content_list_iter_new(
+      manual_content_scan_task_config_t *task_config,
+      struct string_list **list, dir_list_iter_t **iter)
+{
+   bool filter_exts;
+   bool include_compressed;
+
+   if (list)
+      *list = NULL;
+   if (iter)
+      *iter = NULL;
+
+   /* Sanity check */
+   if (!task_config || !*task_config->content_dir || !list || !iter)
+      return false;
+
+   manual_content_scan_content_list_policy(task_config,
+         &filter_exts, &include_compressed);
+
+   if (path_is_directory(task_config->content_dir))
+   {
+      if (!(*list = string_list_new()))
+         return false;
+
+      /* Same parameters as the blocking getter's
+       * dir_list_new() call: exclude directories and
+       * hidden files */
+      *iter = dir_list_iter_new(
+            task_config->content_dir,
+            filter_exts ? task_config->file_exts : NULL,
+            false, /* include_dirs */
+            false, /* include_hidden */ /* todo: obey global settings? */
+            include_compressed,
+            task_config->search_recursively,
+            *list);
+
+      if (!*iter)
+      {
+         string_list_free(*list);
+         *list = NULL;
+         return false;
+      }
+      return true;
+   }
+
+   /* A plain-file content dir needs no walk: the list
+    * is complete here and *iter stays NULL. */
+   if (!(*list = string_list_new()))
+      return false;
+   {
+      union string_list_elem_attr attr;
+      attr.i = 0;
+      if (!string_list_append(*list, task_config->content_dir, attr))
+      {
+         string_list_free(*list);
+         *list = NULL;
+         return false;
+      }
+   }
+   return true;
+}
+
 struct string_list *manual_content_scan_get_content_list(
       manual_content_scan_task_config_t *task_config)
 {
@@ -1567,20 +1697,8 @@ struct string_list *manual_content_scan_get_content_list(
    if (!task_config || !*task_config->content_dir)
       return NULL;
 
-   /* Check whether files should be filtered by
-    * extension */
-   filter_exts = *task_config->file_exts;
-
-   /* Check whether compressed files should be
-    * included in the directory list
-    * > If compressed files are already listed in
-    *   the 'file_exts' string, they will be included
-    *   automatically
-    * > If we don't have a 'file_exts' list, then all
-    *   files must be included regardless of type
-    * > If user has enabled 'search inside archives',
-    *   then compressed files must of course be included */
-   include_compressed = (!filter_exts || task_config->search_archives);
+   manual_content_scan_content_list_policy(task_config,
+         &filter_exts, &include_compressed);
 
    if (path_is_directory(task_config->content_dir))
    {

manual_content_scan.h

@@ -29,6 +29,7 @@
 #include <boolean.h>
 
 #include <lists/string_list.h>
+#include <lists/dir_list.h>
 #include <formats/logiqx_dat.h>
 
 #include "playlist.h"
@@ -321,6 +322,16 @@ bool manual_content_scan_get_task_config(
       const char *path_dir_playlist
       );
 
+/* Performs rudimentary validation of a Logiqx XML DAT
+ * file path: extension whitelist (.dat / .xml,
+ * case-insensitive), existence, size > 0, with the
+ * size optionally returned for free-memory checks.
+ * The parser itself (logiqx_dat.c) is path-agnostic
+ * and performs no validation or I/O; every consumer
+ * of a DAT path goes through this one predicate. */
+bool manual_content_scan_dat_path_is_valid(const char *path,
+      uint64_t *file_size);
+
 /* Creates a list of all valid content in the specified
  * content directory
  * > Returns NULL in the event of failure
@@ -328,6 +339,24 @@ bool manual_content_scan_get_task_config(
 struct string_list *manual_content_scan_get_content_list(
       manual_content_scan_task_config_t *task_config);
 
+/* Resumable counterpart of
+ * manual_content_scan_get_content_list(): creates
+ * @*list and, when the content dir is a directory, a
+ * dir_list_iter_t in @*iter that fills it under the
+ * caller's budget via dir_list_iter_step().  The same
+ * extension / compressed-file policy is applied.  A
+ * content dir that is a plain file fills the list here
+ * and leaves *iter NULL.  When the walk completes the
+ * caller owes the list the same treatment the blocking
+ * getter applied: reject it when empty, then
+ * dir_list_sort(list, true).
+ * > Returns false (leaving *list / *iter NULL) if
+ *   config is invalid or the directory cannot be
+ *   opened */
+bool manual_content_scan_content_list_iter_new(
+      manual_content_scan_task_config_t *task_config,
+      struct string_list **list, dir_list_iter_t **iter);
+
 /* Adds specified content to playlist, if not already
  * present */
 void manual_content_scan_add_content_to_playlist(

samples/tasks/database/Makefile

@@ -91,6 +91,7 @@ SOURCES_C := \
 	$(CORE_DIR)/samples/tasks/database/main.c \
 	$(LIBRETRO_COMM_DIR)/memory/mem_stats.c \
 	$(CORE_DIR)/tasks/task_database.c \
+	$(CORE_DIR)/tasks/task_nbio_slice.c \
 	$(CORE_DIR)/tasks/task_database_cue.c \
 	$(CORE_DIR)/database_info.c \
 	$(CORE_DIR)/core_info.c \
@@ -187,6 +188,11 @@ INDEX_TEST_MAIN   := $(CORE_DIR)/samples/tasks/database/database_index_test.o
 INDEX_TEST_OBJECTS = $(filter-out $(CORE_DIR)/samples/tasks/database/main.o,$(OBJECTS)) \
 	$(INDEX_TEST_MAIN)
 
+BEGIN_TEST        := scan_begin_budget_test
+BEGIN_TEST_MAIN   := $(CORE_DIR)/samples/tasks/database/scan_begin_budget_test.o
+BEGIN_TEST_OBJECTS = $(filter-out $(CORE_DIR)/samples/tasks/database/main.o,$(OBJECTS)) \
+	$(BEGIN_TEST_MAIN)
+
 OBJOUT   = -o
 LINKOUT  = -o
 
@@ -202,7 +208,7 @@ else
 	LD = $(CC)
 endif
 
-all: $(TARGET)$(EXE_EXT)
+all: $(TARGET)$(EXE_EXT) $(BEGIN_TEST)$(EXE_EXT)
 $(TARGET)$(EXE_EXT): $(OBJECTS)
 	$(LD)  $(LINKOUT)$@ $(SHARED) $(OBJECTS) $(LDFLAGS) $(LIBS)
 
@@ -213,9 +219,13 @@ $(INDEX_TEST)$(EXE_EXT): $(INDEX_TEST_OBJECTS)
 $(SCAN_TEST)$(EXE_EXT): $(SCAN_TEST_OBJECTS)
 	$(LD)  $(LINKOUT)$@ $(SHARED) $(SCAN_TEST_OBJECTS) $(LDFLAGS) $(LIBS)
 
-check: $(INDEX_TEST)$(EXE_EXT) $(SCAN_TEST)$(EXE_EXT)
+$(BEGIN_TEST)$(EXE_EXT): $(BEGIN_TEST_OBJECTS)
+	$(LD)  $(LINKOUT)$@ $(SHARED) $(BEGIN_TEST_OBJECTS) $(LDFLAGS) $(LIBS)
+
+check: $(INDEX_TEST)$(EXE_EXT) $(SCAN_TEST)$(EXE_EXT) $(BEGIN_TEST)$(EXE_EXT)
 	./$(INDEX_TEST)$(EXE_EXT)
 	./$(SCAN_TEST)$(EXE_EXT)
+	./$(BEGIN_TEST)$(EXE_EXT)
 
 %.o: %.c
 	$(CC) $(INCFLAGS) $(CFLAGS) -c $(OBJOUT)$@ $<

samples/tasks/database/scan_begin_budget_test.c

@@ -0,0 +1,714 @@
+/* Pacing + throughput oracle for the manual content scanner's BEGIN
+ * path in tasks/task_database.c, compiled from the tree so it
+ * exercises the shipping translation unit.
+ *
+ * Two properties are asserted, because either one alone can be
+ * satisfied by a scanner that is worse:
+ *
+ *  1. Pacing.  Under the regular (non-threaded) task queue, no single
+ *     task_queue_check() attributable to the scanner may exceed the
+ *     shared I/O window by more than bounded slack.  Before the
+ *     BEGIN split, the first gather of a large scan performed the
+ *     entire recursive directory walk, the whole DAT load and the
+ *     playlist setup in one handler invocation - a multi-hundred-
+ *     millisecond freeze that this lane fails on the pre-split tree.
+ *
+ *  2. Throughput.  The total wall-clock time for the same fixed scan
+ *     must stay within a small factor of the blocking dir_list_new()
+ *     walk plus the DAT parse it replaces.  A scanner that no longer
+ *     freezes the UI but takes twice as long is a regression, so the
+ *     budget lane cannot pass on pacing alone.
+ *
+ * The tree is built here - CONTENT_FILES content files spread over
+ * CONTENT_DIRS subdirectories, plus a Logiqx DAT naming each entry -
+ * so the test needs nothing but a writable directory.
+ *
+ * Run with: make scan_begin_budget_test SANITIZER=address,undefined
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <stdint.h>
+#include <time.h>
+
+#include <queues/task_queue.h>
+#include <lists/dir_list.h>
+#include <lists/string_list.h>
+#include <retro_timers.h>
+#include <streams/file_stream.h>
+#include <file/file_path.h>
+#include <features/features_cpu.h>
+#include <formats/logiqx_dat.h>
+
+#include "../../../core_info.h"
+#include "../../../list_special.h"
+#include "../../../tasks/tasks_internal.h"
+#include "../../../manual_content_scan.h"
+#include "../../../playlist.h"
+#include "../../../configuration.h"
+#include "../../../verbosity.h"
+
+#define SCAN_TIMEOUT_SECONDS 300
+
+#define CONTENT_DIRS  50
+#define CONTENT_FILES 5000
+
+/* Per-gather cost for the pacing lane is measured in thread CPU
+ * time, not wall time.  The property under test is how much work the
+ * scanner chooses to do in one gather; on a shared CI runner a cheap
+ * gather can be preempted mid-flight and charged tens of wall-clock
+ * milliseconds it never spent, which fails the lane spuriously,
+ * while the regression the lane exists for - the pre-split tree
+ * doing the whole walk+load+flush in one invocation - burns its
+ * >160 ms as CPU and is caught either way.  Wall time remains the
+ * clock for the total/throughput and first-feedback lanes, where
+ * elapsed time is the property. */
+static retro_time_t thread_cpu_usec(void)
+{
+#if defined(CLOCK_THREAD_CPUTIME_ID)
+   struct timespec ts;
+   if (clock_gettime(CLOCK_THREAD_CPUTIME_ID, &ts) == 0)
+      return (retro_time_t)ts.tv_sec * 1000000
+           + (retro_time_t)(ts.tv_nsec / 1000);
+#endif
+   return cpu_features_get_time_usec();
+}
+
+/* Slack multiplier over the shared window for the pacing assertion.
+ * The window bounds the work the scanner *chooses* to start in a
+ * gather; a single readdir()/stat() that straddles the deadline, plus
+ * qsort of the final list, plus task-queue bookkeeping, land on top.
+ * 4x the 4ms window is still an order of magnitude below the
+ * pre-split freeze (>200ms on the same tree, debug build), so the
+ * discriminator stays sharp while absorbing CI jitter. */
+#define PACING_SLACK_MULTIPLIER 4
+
+/* Throughput guard: incremental scan total time must stay within
+ * this factor of (blocking walk + DAT parse + per-entry floor).
+ * Measured locally the ratio is ~1.0; the margin absorbs shared-
+ * runner noise without admitting a 2x regression. */
+#define THROUGHPUT_FACTOR 1.60
+
+/* ------------------------------------------------------------------ */
+/* Stubs (same set as database_scan_test.c - see rationale there)     */
+/* ------------------------------------------------------------------ */
+
+int msg_hash_get_help_us_enum(enum msg_hash_enums msg, char *s, size_t len)
+{
+   (void)msg;
+   if (s && len)
+      s[0] = '\0';
+   return 0;
+}
+
+const char *msg_hash_to_str_us(enum msg_hash_enums msg)
+{
+   (void)msg;
+   return "";
+}
+
+void msg_hash_us_index_init(void)
+{
+}
+
+settings_t *config_get_ptr(void)
+{
+   static settings_t settings;
+   return &settings;
+}
+
+void runloop_msg_queue_push(const char *msg, size_t len,
+      unsigned prio, unsigned duration,
+      bool flush, char *title, unsigned icon, unsigned category)
+{
+   (void)msg; (void)len; (void)prio; (void)duration;
+   (void)flush; (void)title; (void)icon; (void)category;
+}
+
+bool retroarch_override_setting_is_set(unsigned enum_idx, void *data)
+{
+   (void)enum_idx; (void)data;
+   return false;
+}
+
+struct string_list *dir_list_new_special(const char *input_dir,
+      enum dir_list_type type, const char *filter, bool show_hidden_files)
+{
+   (void)filter;
+   if (type == DIR_LIST_DATABASES)
+      return dir_list_new(input_dir, "rdb", false, show_hidden_files,
+            false, false);
+   return NULL;
+}
+
+static void msgq_push(retro_task_t *task, const char *msg,
+      unsigned prio, unsigned duration, bool flush)
+{
+   (void)task; (void)msg; (void)prio; (void)duration; (void)flush;
+}
+
+void ui_companion_driver_notify_refresh(void)
+{
+}
+
+void task_window_progress_cb(retro_task_t *task)
+{
+   (void)task;
+}
+
+/* ------------------------------------------------------------------ */
+/* Fixture                                                            */
+/* ------------------------------------------------------------------ */
+
+static char fixture_root[4096];
+
+static void rm_rf(const char *path)
+{
+   char cmd[4352];
+   snprintf(cmd, sizeof(cmd), "rm -rf '%s'", path);
+   if (system(cmd) != 0) { /* best effort */ }
+}
+
+static bool write_text_file(const char *path, const char *text)
+{
+   RFILE *f = filestream_open(path,
+         RETRO_VFS_FILE_ACCESS_WRITE, RETRO_VFS_FILE_ACCESS_HINT_NONE);
+   int64_t len;
+   if (!f)
+      return false;
+   len = (int64_t)strlen(text);
+   if (filestream_write(f, text, len) != len)
+   {
+      filestream_close(f);
+      return false;
+   }
+   filestream_close(f);
+   return true;
+}
+
+/* Build CONTENT_FILES small .bin files across CONTENT_DIRS subdirs,
+ * and a Logiqx DAT that names each one, into root (created here). */
+static bool build_fixture(const char *root, char *dat_path, size_t dat_path_len,
+      char *content_dir, size_t content_dir_len)
+{
+   size_t i;
+   FILE *dat;
+   char path[4096];
+
+   snprintf(content_dir, content_dir_len, "%s/content", root);
+   snprintf(dat_path, dat_path_len, "%s/games.dat", root);
+
+   if (!path_mkdir(root) || !path_mkdir(content_dir))
+      return false;
+
+   for (i = 0; i < CONTENT_DIRS; i++)
+   {
+      snprintf(path, sizeof(path), "%s/d%02u", content_dir, (unsigned)i);
+      if (!path_mkdir(path))
+         return false;
+   }
+
+   dat = fopen(dat_path, "w");
+   if (!dat)
+      return false;
+   fputs("<?xml version=\"1.0\"?>\n<datafile>\n"
+         "<header><name>ScanBench</name></header>\n", dat);
+
+   for (i = 0; i < CONTENT_FILES; i++)
+   {
+      char body[64];
+      snprintf(path, sizeof(path), "%s/d%02u/game%04u.bin",
+            content_dir, (unsigned)(i % CONTENT_DIRS), (unsigned)i);
+      snprintf(body, sizeof(body), "content-%04u", (unsigned)i);
+      if (!write_text_file(path, body))
+      {
+         fclose(dat);
+         return false;
+      }
+      fprintf(dat,
+            "<game name=\"game%04u\">\n"
+            "  <description>Game %04u</description>\n"
+            "  <year>2026</year>\n"
+            "  <manufacturer>Bench &amp; Co.</manufacturer>\n"
+            "</game>\n",
+            (unsigned)i, (unsigned)i);
+   }
+   fputs("</datafile>\n", dat);
+   fclose(dat);
+   return true;
+}
+
+/* ------------------------------------------------------------------ */
+/* Scan driver                                                        */
+/* ------------------------------------------------------------------ */
+
+static bool scan_done;
+static bool scan_err;
+
+static void scan_cb(retro_task_t *task,
+      void *task_data, void *user_data, const char *err)
+{
+   (void)task; (void)task_data; (void)user_data;
+   if (err && *err)
+   {
+      fprintf(stderr, "scan error: %s\n", err);
+      scan_err = true;
+   }
+   scan_done = true;
+}
+
+typedef struct
+{
+   retro_time_t total_usec;       /* push -> completion            */
+   retro_time_t max_gather_usec;  /* worst single task_queue_check */
+   retro_time_t first_feedback;   /* push -> first title/progress  */
+   unsigned     gathers;
+} scan_metrics_t;
+
+static retro_task_t *found_task;
+static bool find_any(retro_task_t *task, void *ud)
+{
+   (void)ud;
+   found_task = task;
+   return true;
+}
+
+/* Point the menu-backed scan settings at the fixture.  Shared by
+ * every lane below. */
+static bool configure_scan(const char *content_dir, const char *dat_path,
+      const char *playlist_dir)
+{
+   strlcpy(config_get_ptr()->paths.directory_playlist, playlist_dir,
+         sizeof(config_get_ptr()->paths.directory_playlist));
+
+   if (!manual_content_scan_set_menu_scan_method(
+            MANUAL_CONTENT_SCAN_METHOD_CUSTOM))
+      return false;
+   if (!manual_content_scan_set_menu_content_dir(content_dir))
+      return false;
+   if (!manual_content_scan_set_menu_system_name(
+            MANUAL_CONTENT_SCAN_SYSTEM_NAME_CUSTOM, "ScanBench"))
+      return false;
+   /* For the CUSTOM type the setter only records the type; the name
+    * itself lives in the menu-backed buffer. */
+   strlcpy(manual_content_scan_get_system_name_custom_ptr(), "ScanBench",
+         manual_content_scan_get_system_name_custom_size());
+   if (!manual_content_scan_set_menu_scan_use_db(
+            MANUAL_CONTENT_SCAN_USE_DB_DAT_LOOSE))
+      return false;
+
+   strlcpy(manual_content_scan_get_file_exts_custom_ptr(), "bin",
+         manual_content_scan_get_file_exts_custom_size());
+   strlcpy(manual_content_scan_get_dat_file_path_ptr(), dat_path,
+         manual_content_scan_get_dat_file_path_size());
+   if (manual_content_scan_validate_dat_file_path()
+         != MANUAL_CONTENT_SCAN_DAT_FILE_OK)
+   {
+      fprintf(stderr, "DAT fixture rejected\n");
+      return false;
+   }
+   *manual_content_scan_get_search_recursively_ptr() = true;
+   *manual_content_scan_get_overwrite_playlist_ptr() = true;
+   return true;
+}
+
+/* Configure and run one DAT-based manual scan of the fixture under
+ * the regular (non-threaded) queue, recording pacing metrics. */
+static bool run_scan(const char *content_dir, const char *dat_path,
+      const char *playlist_dir, scan_metrics_t *m)
+{
+   retro_time_t t0;
+   time_t started;
+   task_finder_data_t finder;
+
+   memset(m, 0, sizeof(*m));
+   scan_done = false;
+   scan_err  = false;
+
+   if (!configure_scan(content_dir, dat_path, playlist_dir))
+      return false;
+
+   t0 = cpu_features_get_time_usec();
+   if (!task_push_manual_content_scan(false, scan_cb))
+   {
+      fprintf(stderr, "task_push_manual_content_scan refused\n");
+      return false;
+   }
+
+   finder.func     = find_any;
+   finder.userdata = NULL;
+
+   started = time(NULL);
+   while (!scan_done)
+   {
+      retro_time_t g0, g1;
+
+      if (!m->first_feedback)
+      {
+         found_task = NULL;
+         if (task_queue_find(&finder) && found_task)
+         {
+            const char *title = found_task->title;
+            if ((title && *title) || found_task->progress > 0)
+               m->first_feedback =
+                     cpu_features_get_time_usec() - t0;
+         }
+      }
+
+      g0 = thread_cpu_usec();
+      task_queue_check();
+      g1 = thread_cpu_usec();
+
+      m->gathers++;
+      if (g1 - g0 > m->max_gather_usec)
+         m->max_gather_usec = g1 - g0;
+
+      if (difftime(time(NULL), started) > SCAN_TIMEOUT_SECONDS)
+      {
+         fprintf(stderr, "scan did not finish within %d seconds\n",
+               SCAN_TIMEOUT_SECONDS);
+         return false;
+      }
+   }
+   m->total_usec = cpu_features_get_time_usec() - t0;
+   return !scan_err;
+}
+
+/* Threaded lane: the same scan under the threaded task queue, where
+ * the handler runs on the worker thread while push/find/cancel come
+ * from this one.  Pacing is not asserted here - there is no frame to
+ * protect - only completion and playlist correctness; under TSan
+ * this is the lane that answers whether the BEGIN/END split kept any
+ * hidden shared state (the shared window deliberately opts threaded
+ * queues out of the static window, and this is where a mistake in
+ * that opt-out would surface). */
+static bool run_scan_threaded(const char *content_dir, const char *dat_path,
+      const char *playlist_dir)
+{
+   time_t started;
+
+   scan_done = false;
+   scan_err  = false;
+
+   if (!configure_scan(content_dir, dat_path, playlist_dir))
+      return false;
+
+   task_queue_init(true, msgq_push);
+
+   if (!task_push_manual_content_scan(false, scan_cb))
+   {
+      fprintf(stderr, "threaded: task_push_manual_content_scan refused\n");
+      task_queue_deinit();
+      return false;
+   }
+
+   started = time(NULL);
+   while (!scan_done)
+   {
+      task_queue_check();
+      retro_sleep(1);
+      if (difftime(time(NULL), started) > SCAN_TIMEOUT_SECONDS)
+      {
+         fprintf(stderr, "threaded scan did not finish in time\n");
+         task_queue_deinit();
+         return false;
+      }
+   }
+
+   task_queue_deinit();
+   return !scan_err;
+}
+
+/* Cancellation lane: push the scan, run a bounded number of gathers,
+ * then cancel and drain.  Sweeping cancel_after over a range lands
+ * the cancellation in every phase of the task - setup, mid-walk,
+ * mid-DAT-read, content iteration, mid-flush - which is exactly the
+ * set of states that now hold in-flight resources (an open directory
+ * iterator, an open DAT stream and half-filled buffer, an open flush
+ * playlist).  Under ASan/LSan a resource dropped by the teardown of
+ * any of those states is the failure; the lane itself only asserts
+ * clean termination. */
+static bool run_scan_cancelled(const char *content_dir, const char *dat_path,
+      const char *playlist_dir, unsigned cancel_after)
+{
+   retro_task_t *task = NULL;
+   task_finder_data_t finder;
+   unsigned i;
+   time_t started;
+
+   scan_done = false;
+   scan_err  = false;
+
+   if (!configure_scan(content_dir, dat_path, playlist_dir))
+      return false;
+
+   task_queue_init(false, msgq_push);
+
+   if (!task_push_manual_content_scan(false, scan_cb))
+   {
+      fprintf(stderr, "cancel lane: push refused\n");
+      task_queue_deinit();
+      return false;
+   }
+
+   for (i = 0; i < cancel_after && !scan_done; i++)
+      task_queue_check();
+
+   /* Cancel whatever is still running... */
+   finder.func     = find_any;
+   finder.userdata = NULL;
+   found_task      = NULL;
+   if (!scan_done && task_queue_find(&finder) && found_task)
+   {
+      task = found_task;
+      task_queue_cancel_task(task);
+   }
+
+   /* ...and drain to retirement. */
+   started = time(NULL);
+   while (!scan_done)
+   {
+      task_queue_check();
+      if (difftime(time(NULL), started) > SCAN_TIMEOUT_SECONDS)
+      {
+         fprintf(stderr, "cancelled scan (after %u) did not drain\n",
+               cancel_after);
+         task_queue_deinit();
+         return false;
+      }
+   }
+
+   task_queue_deinit();
+   /* A cancelled scan may or may not report an error; either way the
+    * teardown must have released everything, which the sanitizers
+    * decide. */
+   return true;
+}
+
+/* Playlist correctness: every fixture file present exactly once. */
+static bool check_playlist(const char *playlist_dir)
+{
+   char path[4096];
+   playlist_config_t cfg;
+   playlist_t *pl;
+   size_t n;
+
+   memset(&cfg, 0, sizeof(cfg));
+   snprintf(path, sizeof(path), "%s/ScanBench.lpl", playlist_dir);
+   playlist_config_set_path(&cfg, path);
+   cfg.capacity = COLLECTION_SIZE;
+
+   if (!(pl = playlist_init(&cfg)))
+      return false;
+   n = playlist_size(pl);
+   playlist_free(pl);
+
+   if (n != CONTENT_FILES)
+   {
+      fprintf(stderr, "playlist has %u entries, expected %u\n",
+            (unsigned)n, (unsigned)CONTENT_FILES);
+      return false;
+   }
+   return true;
+}
+
+int main(int argc, char *argv[])
+{
+   char content_dir[4096];
+   char dat_path[4096];
+   char playlist_dir[4352];
+   scan_metrics_t m;
+   retro_time_t walk_usec, dat_usec, floor_usec;
+   struct string_list *list;
+   int64_t dat_len = 0;
+   char *dat_buf   = NULL;
+   int rc = 1;
+   bool bench_only = (argc > 1 && strcmp(argv[1], "bench") == 0);
+   /* Sanitizer mode: run every lane - parity, threaded, the
+    * cancellation sweep - but skip the wall-clock assertions, which
+    * a sanitized build cannot honestly meet.  The sanitizers are the
+    * assertion in that mode. */
+   bool sanitize   = (argc > 1 && strcmp(argv[1], "sanitize") == 0);
+
+   (void)argc;
+   (void)argv;
+
+   snprintf(fixture_root, sizeof(fixture_root), "%s/scan_begin_fixture",
+         getenv("TMPDIR") ? getenv("TMPDIR") : "/tmp");
+   rm_rf(fixture_root);
+
+   /* Logging stays off for the measured lanes: the reference costs
+    * (blocking walk, DAT parse) perform no logging, so a measured
+    * scan writing ten thousand synchronous per-entry lines to stderr
+    * would be charged for the logging, not the scanning.  Failures
+    * print their own diagnostics below. */
+   retro_main_log_file_init(NULL, false);
+
+   if (!build_fixture(fixture_root, dat_path, sizeof(dat_path),
+         content_dir, sizeof(content_dir)))
+   {
+      fprintf(stderr, "could not build fixture under %s\n", fixture_root);
+      return 1;
+   }
+   snprintf(playlist_dir, sizeof(playlist_dir), "%s/playlists",
+         fixture_root);
+   if (!path_mkdir(playlist_dir))
+      return 1;
+
+   /* Reference cost: the blocking walk this change replaces, plus the
+    * DAT parse, plus a small per-entry floor for the matching the
+    * scan itself must do.  Taken as the min of 3 reps so a cold page
+    * cache does not inflate the budget. */
+   walk_usec = INT64_MAX;
+   {
+      int rep;
+      for (rep = 0; rep < 3; rep++)
+      {
+         retro_time_t w0 = cpu_features_get_time_usec();
+         retro_time_t w;
+         list = dir_list_new(content_dir, "bin", false, false, false, true);
+         if (!list || list->size != CONTENT_FILES)
+         {
+            fprintf(stderr, "reference walk produced %u entries\n",
+                  list ? (unsigned)list->size : 0u);
+            goto out;
+         }
+         dir_list_sort(list, true);
+         string_list_free(list);
+         w = cpu_features_get_time_usec() - w0;
+         if (w < walk_usec)
+            walk_usec = w;
+      }
+   }
+   {
+      /* Reference cost through the same I/O-free parser the task
+       * uses: one whole-file read plus one owned-buffer parse.
+       * logiqx_dat_init_owned() takes the buffer either way, so
+       * dat_buf must not be freed below. */
+      retro_time_t d0 = cpu_features_get_time_usec();
+      logiqx_dat_t *dat = NULL;
+      if (filestream_read_file(dat_path, (void**)&dat_buf, &dat_len) <= 0)
+         goto out;
+      dat     = logiqx_dat_init_owned(dat_buf, (size_t)dat_len);
+      dat_buf = NULL;
+      if (!dat)
+      {
+         fprintf(stderr, "reference DAT parse failed\n");
+         goto out;
+      }
+      logiqx_dat_free(dat);
+      dat_usec = cpu_features_get_time_usec() - d0;
+   }
+   floor_usec = (retro_time_t)CONTENT_FILES * 40; /* 40us/entry match floor */
+
+   task_queue_init(false /* regular queue: the case that freezes */,
+         msgq_push);
+
+   if (!run_scan(content_dir, dat_path, playlist_dir, &m))
+      goto out_queue;
+   if (!check_playlist(playlist_dir))
+      goto out_queue;
+
+   fprintf(stderr,
+         "[metrics] files=%u total=%.1fms max_gather_cpu=%.2fms "
+         "first_feedback=%.2fms gathers=%u walk_ref=%.1fms dat_ref=%.1fms\n",
+         (unsigned)CONTENT_FILES,
+         m.total_usec      / 1000.0,
+         m.max_gather_usec / 1000.0,
+         m.first_feedback  / 1000.0,
+         m.gathers,
+         walk_usec / 1000.0,
+         dat_usec  / 1000.0);
+
+   if (bench_only)
+   {
+      rc = 0;
+      goto out_queue;
+   }
+
+   /* 1. Pacing: the worst gather must be bounded by the shared window
+    *    plus slack - not by the size of the library. */
+   if (!sanitize)
+   {
+      retro_time_t limit = 4000 * PACING_SLACK_MULTIPLIER;
+      if (m.max_gather_usec > limit)
+      {
+         fprintf(stderr,
+               "FAIL pacing: max gather %.2fms CPU exceeds %.2fms budget\n",
+               m.max_gather_usec / 1000.0, limit / 1000.0);
+         goto out_queue;
+      }
+   }
+
+   /* 2. First feedback within ~2 frames of the push. */
+   if (!sanitize && m.first_feedback > 34000)
+   {
+      fprintf(stderr,
+            "FAIL feedback: first task feedback after %.2fms (limit 34ms)\n",
+            m.first_feedback / 1000.0);
+      goto out_queue;
+   }
+
+   /* 3. Throughput: total scan within factor of reference cost. */
+   if (!sanitize)
+   {
+      double ref = (double)(walk_usec + dat_usec + floor_usec);
+      if ((double)m.total_usec > ref * THROUGHPUT_FACTOR)
+      {
+         fprintf(stderr,
+               "FAIL throughput: total %.1fms vs reference %.1fms "
+               "(factor %.2f, limit %.2f)\n",
+               m.total_usec / 1000.0, ref / 1000.0,
+               (double)m.total_usec / ref, THROUGHPUT_FACTOR);
+         goto out_queue;
+      }
+   }
+
+   /* The regular-queue metrics lanes are done; tear that queue down
+    * before the threaded and cancellation lanes bring up their own. */
+   task_queue_deinit();
+
+   /* Threaded lane: correctness under the threaded queue (and the
+    * TSan target). */
+   {
+      char pl[4352];
+      snprintf(pl, sizeof(pl), "%s/ScanBench.lpl", playlist_dir);
+      filestream_delete(pl);
+   }
+   if (!run_scan_threaded(content_dir, dat_path, playlist_dir))
+      goto out;
+   if (!check_playlist(playlist_dir))
+      goto out;
+   fprintf(stderr, "[pass] threaded lane\n");
+
+   /* Cancellation sweep: land the cancel in every phase.  Gather
+    counts are exponential so the sweep covers setup (0-1), the
+    * budgeted walk and DAT read (single digits), content iteration
+    * (tens to thousands) and the flush (the tail), without running
+    * the full scan thousands of times. */
+   {
+      static const unsigned cancel_points[] =
+            { 0, 1, 2, 3, 4, 6, 8, 12, 16, 32, 64, 256, 1024, 4096 };
+      size_t ci;
+      for (ci = 0; ci < sizeof(cancel_points) / sizeof(cancel_points[0]); ci++)
+      {
+         if (!run_scan_cancelled(content_dir, dat_path, playlist_dir,
+               cancel_points[ci]))
+            goto out;
+      }
+      fprintf(stderr, "[pass] cancellation sweep (%u points)\n",
+            (unsigned)(sizeof(cancel_points) / sizeof(cancel_points[0])));
+   }
+
+   fprintf(stderr, "PASS scan_begin_budget_test\n");
+   rc = 0;
+   goto out;
+
+out_queue:
+   task_queue_deinit();
+out:
+   free(dat_buf);
+   rm_rf(fixture_root);
+   return rc;
+}

samples/tasks/file_transfer/Makefile

@@ -11,6 +11,7 @@ LIBRETRO_COMM_DIR := $(REPO_ROOT)/libretro-common
 
 SOURCES := task_file_transfer_test.c \
            $(REPO_ROOT)/tasks/task_file_transfer.c \
+           $(REPO_ROOT)/tasks/task_nbio_slice.c \
            $(LIBRETRO_COMM_DIR)/compat/compat_strl.c \
            $(LIBRETRO_COMM_DIR)/compat/compat_strcasestr.c \
            $(LIBRETRO_COMM_DIR)/compat/compat_strldup.c \

samples/tasks/image_decode_slice/Makefile

@@ -22,6 +22,7 @@ LIBRETRO_COMM_DIR := $(REPO_ROOT)/libretro-common
 SOURCES := image_decode_slice_test.c \
            $(REPO_ROOT)/tasks/task_image.c \
            $(REPO_ROOT)/tasks/task_file_transfer.c \
+           $(REPO_ROOT)/tasks/task_nbio_slice.c \
            $(LIBRETRO_COMM_DIR)/compat/compat_strl.c \
            $(LIBRETRO_COMM_DIR)/compat/compat_strldup.c \
            $(LIBRETRO_COMM_DIR)/compat/fopen_utf8.c \

tasks/task_file_transfer.c

@@ -80,102 +80,20 @@ void task_window_progress_cb(retro_task_t *task)
  * JPEG are avail-aware and decode against the partial buffer, but
  * they still need every byte to finish, so the time budget is theirs
  * too - task_image.c's is_prefix grouping answers a different
- * question from this one and the two are meant to differ.) */
-#define NBIO_XFER_TICK_USEC        4000
-
-/* ...and that budget is shared by every file-transfer task in a tick,
- * not handed to each one separately.
+ * question from this one and the two are meant to differ.)
  *
- * retro_task_regular_gather() calls every running task's handler once
- * per task_queue_check(), i.e. once per frame, so a per-task slice
- * multiplies: eight concurrent loads spent 30.65 ms in a single
- * frame, four already spent 15.07 ms, and it scaled linearly with
- * the task count.  A window is the fix - at most NBIO_XFER_TICK_USEC
- * of file I/O per NBIO_XFER_TICK_PERIOD_USEC of wall time, whatever
- * the number of tasks.
- *
- * A tick has no identity this file can see, so the window is a plain
- * fixed one rather than something keyed to the gather: consumed
- * microseconds accumulate until a period has elapsed, then reset.
- * It over-counts nothing (only time actually spent filling is
- * charged) and needs no cooperation from task_queue. */
-#define NBIO_XFER_TICK_PERIOD_USEC 16666
+ * The time budget's value itself lives with the window in
+ * task_nbio_slice.c (NBIO_XFER_TICK_USEC): the slice open/close
+ * calls below hand out exactly that allowance. */
 
-/* Every task gets one read regardless of the window, or a queue
- * whose budget is already spent would make no progress at all.  The
- * frame is then bounded by the window plus one chunk per task, which
- * is a few hundred microseconds for a queue of dozens, rather than
- * by the task count times the whole slice. */
-typedef struct
-{
-   retro_time_t start;       /* when this task's fill began       */
-   retro_time_t allowance;   /* usec this task may spend in it    */
-   uint8_t      floor;       /* the guaranteed first read         */
-} nbio_budget_t;
-
-/* Window state.  Only ever touched on the thread that runs handlers,
- * and only when that is the frame thread - see slice_open(). */
-static retro_time_t nbio_slice_start;
-static retro_time_t nbio_slice_used;
-
-/* The budget is handed to the fill rather than checked around it, so
- * it is consulted between the fill's own reads.  This used to be a
- * do/while around iterate() with a 256 KiB byte budget standing in
- * for the time slice, which meant the deadline could only be noticed
- * every 256 KiB - once per whole chunk, however long that chunk took
- * to arrive. */
-static bool task_file_transfer_within_budget(void *ud, size_t avail,
-      size_t len)
-{
-   nbio_budget_t *b = (nbio_budget_t*)ud;
-   (void)avail;
-   (void)len;
-   if (b->floor)
-   {
-      b->floor = 0;
-      return true;
-   }
-   return cpu_features_get_time_usec() - b->start < b->allowance;
-}
-
-/* Claim this task's share of the window.
- *
- * Declined entirely under a threaded task queue.  There the handler
- * runs on the worker thread, which rotates the task and re-enters
- * immediately; there is no frame to protect, slicing buys only
- * round-robin fairness between file tasks, and a shared static
- * across threads would be a race for no benefit.  Each task gets the
- * whole slice there, which is what it effectively had before. */
-static void task_file_transfer_slice_open(nbio_budget_t *b)
-{
-   retro_time_t now = cpu_features_get_time_usec();
-
-   b->start = now;
-   b->floor = 1;
-
-   if (task_queue_is_threaded())
-   {
-      b->allowance = NBIO_XFER_TICK_USEC;
-      return;
-   }
-   if (now - nbio_slice_start >= (retro_time_t)NBIO_XFER_TICK_PERIOD_USEC)
-   {
-      nbio_slice_start = now;
-      nbio_slice_used  = 0;
-   }
-   b->allowance = (nbio_slice_used < (retro_time_t)NBIO_XFER_TICK_USEC)
-      ? (retro_time_t)NBIO_XFER_TICK_USEC - nbio_slice_used
-      : 0;
-}
-
-/* Charge back what the fill actually spent, including the floor read
- * that ran outside the allowance. */
-static void task_file_transfer_slice_close(nbio_budget_t *b)
-{
-   if (task_queue_is_threaded())
-      return;
-   nbio_slice_used += cpu_features_get_time_usec() - b->start;
-}
+/* The shared per-frame I/O window itself (state + the three
+ * task_nbio_slice_* entry points) lives in task_nbio_slice.c: it is
+ * shared with other budgeted handlers (the manual content scanner's
+ * BEGIN spine), and a TU of its own keeps its dependencies to the
+ * clock and the task queue, so tests can link the real window
+ * without dragging the image/nbio spine in with it.  The design
+ * rationale - why a shared window rather than a per-task slice, why
+ * the floor - is documented there. */
 
 const uint8_t *nbio_xfer_ptr(nbio_handle_t *nbio, size_t *len)
 {
@@ -225,10 +143,10 @@ static int task_file_transfer_iterate_transfer(nbio_handle_t *nbio)
                         || nbio->type == NBIO_TYPE_WEBP)
          ? (size_t)NBIO_XFER_TICK_BYTES : 0;
 
-      task_file_transfer_slice_open(&b);
+      task_nbio_slice_open(&b);
       data_transfer_iterate_while(nbio->xfer, bytes,
-            task_file_transfer_within_budget, &b);
-      task_file_transfer_slice_close(&b);
+            task_nbio_slice_within_budget, &b);
+      task_nbio_slice_close(&b);
    }
    if (data_transfer_complete(nbio->xfer)
          || data_transfer_failed(nbio->xfer)
@@ -298,9 +216,9 @@ void task_file_load_handler(retro_task_t *task)
                 * much less. */
                {
                   nbio_budget_t b;
-                  task_file_transfer_slice_open(&b);
+                  task_nbio_slice_open(&b);
                   nbio->xfer = data_transfer_open_prefix(nbio->path, 0);
-                  task_file_transfer_slice_close(&b);
+                  task_nbio_slice_close(&b);
                }
                if (nbio->xfer)
                {

tasks/task_nbio_slice.c

@@ -0,0 +1,122 @@
+/* Copyright  (C) 2010-2026 The RetroArch team
+ *
+ * ---------------------------------------------------------------------------------------
+ * The following license statement only applies to this file (task_nbio_slice.c).
+ * ---------------------------------------------------------------------------------------
+ *
+ * Permission is hereby granted, free of charge,
+ * to any person obtaining a copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation the rights to
+ * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software,
+ * and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
+ * INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+ * IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+ * DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+ * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE
+ * OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+/* The shared per-frame I/O window used by budgeted task handlers:
+ * the file-transfer spine (task_file_transfer.c) and the manual
+ * content scanner's BEGIN states (task_database.c).  A TU of its
+ * own so its only dependencies are the clock and the task queue,
+ * which lets test harnesses link the real window rather than a
+ * reimplementation. */
+
+#include <features/features_cpu.h>
+#include <queues/task_queue.h>
+
+#include "tasks_internal.h"
+
+/* At most NBIO_XFER_TICK_USEC of budgeted work per
+ * NBIO_XFER_TICK_PERIOD_USEC of wall time, shared by every
+ * participating task in a tick rather than handed to each one
+ * separately.
+ *
+ * retro_task_regular_gather() calls every running task's handler once
+ * per task_queue_check(), i.e. once per frame, so a per-task slice
+ * multiplies: eight concurrent loads spent 30.65 ms in a single
+ * frame, four already spent 15.07 ms, and it scaled linearly with
+ * the task count.  A window is the fix - whatever the number of
+ * tasks, the frame pays the window once.
+ *
+ * A tick has no identity this file can see, so the window is a plain
+ * fixed one rather than something keyed to the gather: consumed
+ * microseconds accumulate until a period has elapsed, then reset.
+ * It over-counts nothing (only time actually spent working is
+ * charged) and needs no cooperation from task_queue.
+ *
+ * Every open grants a floor of one work item regardless of the
+ * window, or a queue whose budget is already spent would make no
+ * progress at all.  The frame is then bounded by the window plus one
+ * work item per task - a few hundred microseconds for a queue of
+ * dozens - rather than by the task count times the whole slice. */
+#define NBIO_XFER_TICK_USEC        4000
+#define NBIO_XFER_TICK_PERIOD_USEC 16666
+
+/* Window state.  Only ever touched on the thread that runs handlers,
+ * and only when that is the frame thread - see
+ * task_nbio_slice_open(). */
+static retro_time_t nbio_slice_start;
+static retro_time_t nbio_slice_used;
+
+/* The budget is handed to the work loop rather than checked around
+ * it, so the deadline is noticed between the loop's own items - not
+ * once per whole chunk, however long a chunk takes to arrive. */
+bool task_nbio_slice_within_budget(void *ud, size_t avail,
+      size_t len)
+{
+   nbio_budget_t *b = (nbio_budget_t*)ud;
+   (void)avail;
+   (void)len;
+   if (b->floor)
+   {
+      b->floor = 0;
+      return true;
+   }
+   return cpu_features_get_time_usec() - b->start < b->allowance;
+}
+
+/* Claim this task's share of the window.
+ *
+ * Declined entirely under a threaded task queue.  There the handler
+ * runs on the worker thread, which rotates the task and re-enters
+ * immediately; there is no frame to protect, slicing buys only
+ * round-robin fairness between tasks, and a shared static across
+ * threads would be a race for no benefit.  Each task gets the whole
+ * slice there, which is what it effectively had before. */
+void task_nbio_slice_open(nbio_budget_t *b)
+{
+   retro_time_t now = cpu_features_get_time_usec();
+
+   b->start = now;
+   b->floor = 1;
+
+   if (task_queue_is_threaded())
+   {
+      b->allowance = NBIO_XFER_TICK_USEC;
+      return;
+   }
+   if (now - nbio_slice_start >= (retro_time_t)NBIO_XFER_TICK_PERIOD_USEC)
+   {
+      nbio_slice_start = now;
+      nbio_slice_used  = 0;
+   }
+   b->allowance = (nbio_slice_used < (retro_time_t)NBIO_XFER_TICK_USEC)
+      ? (retro_time_t)NBIO_XFER_TICK_USEC - nbio_slice_used
+      : 0;
+}
+
+/* Charge back what the work actually spent, including the floor item
+ * that ran outside the allowance. */
+void task_nbio_slice_close(nbio_budget_t *b)
+{
+   if (task_queue_is_threaded())
+      return;
+   nbio_slice_used += cpu_features_get_time_usec() - b->start;
+}

tasks/tasks_internal.h

@@ -57,6 +57,32 @@ typedef struct nbio_buf
    unsigned bufsize;
 } nbio_buf_t;
 
+/* Shared per-frame I/O window (implementation and rationale in
+ * task_file_transfer.c).  A handler that performs bounded work per
+ * gather claims a share with task_nbio_slice_open(), consults
+ * task_nbio_slice_within_budget() between work items, and charges
+ * back what it actually spent with task_nbio_slice_close().  The
+ * window is shared by every participating task in a gather - file
+ * transfers, the content scanner - rather than handed to each one
+ * separately, because a per-task slice multiplies with the task
+ * count.  Every open grants a floor of one work item so a queue
+ * whose window is already spent still makes progress.  Under a
+ * threaded task queue each task simply gets a whole slice: there is
+ * no frame to protect there, and a shared static across threads
+ * would be a race for no benefit. */
+typedef struct
+{
+   retro_time_t start;       /* when this task's work began       */
+   retro_time_t allowance;   /* usec this task may spend in it    */
+   uint8_t      floor;       /* the guaranteed first work item    */
+} nbio_budget_t;
+
+void task_nbio_slice_open(nbio_budget_t *b);
+void task_nbio_slice_close(nbio_budget_t *b);
+/* Signature matches data_transfer's within-budget callback; the
+ * @avail / @len arguments are unused. */
+bool task_nbio_slice_within_budget(void *ud, size_t avail, size_t len);
+
 /* Generic progress_cb that forwards a task's progress (0-100) to the
  * platform's window/taskbar progress indicator (e.g. ITaskbarList3 on
  * Win32). Set this on any task whose progress should be reflected on