Fetch a run of missing blocks in one request in BlockCache

[itzzdev09]

Closes the second half of #1960. The duplicate download reported there is gone (#1984, #2150), but the other complaint in the thread still holds: a large read costs one serialized request per block.

cache = BlockCache(5 * 2**20, fetcher, size=1 << 30, maxblocks=32)
cache._fetch(0, 200 * 2**20)

That is 40 requests on master, one per 5 MiB block, each waiting for the last. With this change it is 2, for the same 200 MiB of data: the blocks the cache does not hold are fetched in contiguous runs.

The old loop had a note on it:

# Note: it'd be nice to combine these into one big request. However
# that doesn't play nicely with our LRU cache.

The obstacle was functools.lru_cache, which cannot say whether a block is held and cannot be given a value fetched elsewhere. BackgroundBlockCache already uses UpdatableLRU from this module, which has is_key_cached and add_key, so BlockCache now uses it too. _fetch_blocks walks the requested range, takes what is cached and groups the rest into runs; each run is one fetcher call whose blocks are then added to the LRU individually, so eviction stays per block, exactly as before.

A run is capped at maxblocks. Without the cap a single read could both ask a backend for an unbounded range and evict its own earlier blocks before they were read, which is the regression #1984 fixed.

Behaviour that does not change: block boundaries and sizes, what ends up cached, LRU eviction order, cache_info(), hit_count, miss_count, total_requested_bytes, and pickling. A cached or single-block read still goes through the ordinary path.

UpdatableLRU.add_key takes a new keyword-only count_miss=False, so a caller that fetched the value itself can still have it counted as a miss. BackgroundBlockCache is untouched.

Tests

Four new tests in test_caches.py: one request for a run of eight missing blocks; only the gaps fetched when some blocks are already held; a run split at maxblocks; and a run stopping at the end of the file. All four fail on master.

fsspec/tests/ plus test_cached.py: 856 passed, 235 skipped, 2 xfailed.

[martindurant]

I suggest that you reuse cat_ranges - this will join a number of requests based on heuristics for maximum read size, and submit them concurrently for backends that support it.

[itzzdev09]

Done in 3fbd8a6: BlockCache now takes an optional multi_fetcher (same as MMapCache) and passes it every run of missing blocks in one call. AbstractBufferedFile sets it to fs.cat_ranges for blockcache on async filesystems, so the runs go out concurrently. On sync filesystems it still calls fetcher once per run, because the default cat_ranges reopens the file for every range. Should that be enabled for sync filesystems too?