5252# few unwanted bytes for one less request, which at object-store latencies is
5353# always the right way round.
5454BLOCK_GAP = 4096
55+ # How many partly filled chunks keep their blocks in memory as well as in the
56+ # cache. Adding a block to a chunk rewrites that chunk, and the blocks already
57+ # in it have to come from somewhere: from here, or read back out of the cache and
58+ # taken apart again. Keeping the last few saves half the copying for the pattern
59+ # that costs the most -- one chunk filled a slice at a time -- and 8 of them is
60+ # the memory a fetch of 8 whole chunks already peaks at.
61+ BLOCK_HOT_CHUNKS = 8
5562
5663# `jit` kwargs that tune *how* an expression is evaluated, not what container the
5764# result is stored in. Unlike storage kwargs (`cparams`, `chunks`, `urlpath`, ...),
@@ -354,6 +361,9 @@ def __init__(
354361 for name in ("wants_blocks" , "chunk_layout" , "block_plan" , "read_range" )
355362 )
356363 self ._blocks_per_chunk = self .src .blocks_per_chunk if serves_blocks else 1
364+ # Blocks of the last few partly filled chunks, so a rewrite need not read
365+ # the chunk back out of the cache to find out what is already in it
366+ self ._hot_payloads = {}
357367 nchunks = self ._schunk_cache .nchunks
358368 nbits = nchunks * self ._blocks_per_chunk
359369 self ._fetched = bytearray ((nbits + 7 ) // 8 ) if fresh else self ._load_fetched (nchunks )
@@ -582,8 +592,7 @@ def fetch(
582592 missing = self ._missing_chunks (item )
583593 try :
584594 for nchunk , chunk in self ._get_chunks (missing , max_concurrency ):
585- self ._schunk_cache .update_chunk (nchunk , chunk )
586- self ._mark_fetched (nchunk )
595+ self ._store_chunk (nchunk , chunk )
587596 finally :
588597 if missing :
589598 self ._save_fetched ()
@@ -645,8 +654,7 @@ def fetch_one(task):
645654 try :
646655 for (nchunk , run ), data in zip (tasks , self ._run (fetch_one , tasks , max_concurrency ), strict = True ):
647656 if run is None :
648- self ._schunk_cache .update_chunk (nchunk , data )
649- self ._mark_fetched (nchunk )
657+ self ._store_chunk (nchunk , data )
650658 continue
651659 payloads = pending .setdefault (nchunk , {})
652660 for nblock , offset , size in run [2 ]:
@@ -661,22 +669,43 @@ def fetch_one(task):
661669 return self ._cache
662670
663671 def _write_blocks (self , nchunk : int , payloads : dict [int , bytes ], header : bytes ) -> None :
664- """Put the blocks just fetched into the cache, keeping those already there."""
672+ """Put the blocks just fetched into the cache, keeping those already there.
673+
674+ Every fetch that adds blocks to a chunk rewrites that chunk, so filling
675+ one a slice at a time is quadratic in bytes moved. It stays cheap --
676+ compressed bytes, and a rewrite never exceeds what fetching that chunk
677+ whole would have downloaded once -- and the write itself cannot be
678+ deferred, since the cache container is what the read comes out of. What
679+ can be avoided is the other half: `_hot_payloads` keeps the blocks of the
680+ last few partly filled chunks, so they need not be read back out of the
681+ cache and taken apart on every rewrite.
682+ """
665683 nblocks = self ._blocks_per_chunk
666- held = [n for n in range (nblocks ) if self ._is_fetched (nchunk , n )]
667- # ponytail: every fetch that adds blocks to a chunk rewrites it, so N
668- # separate fetches into the same chunk copy O(N^2) compressed bytes. Cheap
669- # -- memcpy, never more per rewrite than downloading that chunk once --
670- # until something pokes one big chunk many times over; then hold the
671- # payloads per chunk for the session and splice on close instead.
672- if held :
673- # The chunk in the cache was spliced by an earlier fetch, so its blocks
674- # come back out the way they went in, compressed and without a copy of
675- # anything else
676- payloads = _chunk_payloads (self ._schunk_cache .get_chunk (nchunk ), nblocks , held ) | payloads
677- self ._schunk_cache .update_chunk (nchunk , _splice_chunk (header , nblocks , payloads ))
684+ kept = self ._hot_payloads .pop (nchunk , None )
685+ if kept is None :
686+ # Not held any more (or not by this run): the chunk in the cache was
687+ # spliced by an earlier fetch, so its blocks come back out the way
688+ # they went in, compressed and without a copy of anything else
689+ held = [n for n in range (nblocks ) if self ._is_fetched (nchunk , n )]
690+ kept = _chunk_payloads (self ._schunk_cache .get_chunk (nchunk ), nblocks , held ) if held else {}
691+ kept .update (payloads )
692+ # ponytail: the write half stays quadratic -- a rewrite hands the cache
693+ # the whole chunk, and it cannot be deferred or batched further, since the
694+ # cache is what the next read comes out of. Removing it needs the cache to
695+ # hold blocks apart from their chunk, which is a different container.
696+ self ._schunk_cache .update_chunk (nchunk , _splice_chunk (header , nblocks , kept ))
678697 for nblock in payloads :
679698 self ._mark_fetched (nchunk , nblock )
699+ if len (kept ) < nblocks : # a chunk that is now complete will never be rewritten
700+ self ._hot_payloads [nchunk ] = kept
701+ while len (self ._hot_payloads ) > BLOCK_HOT_CHUNKS :
702+ self ._hot_payloads .pop (next (iter (self ._hot_payloads )))
703+
704+ def _store_chunk (self , nchunk : int , chunk : bytes ) -> None :
705+ """Put a whole chunk in the cache, dropping anything held about its blocks."""
706+ self ._schunk_cache .update_chunk (nchunk , chunk )
707+ self ._mark_fetched (nchunk )
708+ self ._hot_payloads .pop (nchunk , None )
680709
681710 async def afetch (
682711 self , item : slice | list [slice ] | None = (), max_concurrency : int | None = None
@@ -782,8 +811,7 @@ async def _fetch_one(nchunk):
782811 async with semaphore :
783812 chunk = await self .src .aget_chunk (nchunk )
784813 # Runs to completion between awaits, so concurrent writers can't interleave.
785- self ._schunk_cache .update_chunk (nchunk , chunk )
786- self ._mark_fetched (nchunk )
814+ self ._store_chunk (nchunk , chunk )
787815
788816 if to_fetch :
789817 try :
0 commit comments