engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

MultiShardSidecarCorpus: parallel sidecar pre-download

Browse source 
Was serial — 4 × 745 MB = ~2 min cold for the 4-genesis-shard manifest.
Build a ThreadPoolExecutor with one worker per shard so the wall cost
is max(per-shard download), not sum. Idempotent: SidecarShardClient
re-reads cached files without re-downloading on subsequent runs.
This commit is contained in:
russell@unturf.com 2026-05-30 16:52:37 -04:00
parent b932784154
commit 0e7f599279
No known key found for this signature in database
1 changed files with 25 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

37
arborist/wallet/bucket.py
View file

@ -799,26 +799,39 @@ class MultiShardSidecarCorpus:
cache_bytes_per_shard: int = DEFAULT_CACHE_BYTES,
sidecar_cache_dir: str | None = None,
):
from concurrent.futures import ThreadPoolExecutor
self.manifest = manifest
self.shard_clients = []
self._has_sidecar: list[bool] = []
for sh in manifest.shards:
self._has_sidecar: list[bool] = [
bool(sh.get("sidecar_url")) for sh in manifest.shards
]
# Pre-download every sidecar in parallel so a fresh consumer
# pays max(per-sidecar download) instead of sum. SidecarShardClient
# is idempotent — if the cache file already exists, it just reads
# from disk (no re-download). Bucket-only shards (no sidecar)
# construct fast; they don't pre-pull the .db.
def _build(item):
i, sh = item
if sh.get("sidecar_url"):
client = SidecarShardClient(
return i, SidecarShardClient(
shard_url=sh["url"],
sidecar_url=sh["sidecar_url"],
blob_base=manifest.blob_base,
cache_bytes=cache_bytes_per_shard,
sidecar_cache_dir=sidecar_cache_dir,
)
self._has_sidecar.append(True)
else:
client = BucketClient(
BucketEndpoint(shard_url=sh["url"], blob_base=manifest.blob_base),
cache_bytes=cache_bytes_per_shard,
)
self._has_sidecar.append(False)
self.shard_clients.append(client)
return i, BucketClient(
BucketEndpoint(shard_url=sh["url"], blob_base=manifest.blob_base),
cache_bytes=cache_bytes_per_shard,
)
n = len(manifest.shards)
results: list = [None] * n
with ThreadPoolExecutor(max_workers=max(1, n)) as ex:
for i, client in ex.map(_build, enumerate(manifest.shards)):
results[i] = client
self.shard_clients = results
self._by_url = {sh["url"]: c for sh, c in zip(manifest.shards, self.shard_clients)}
def close(self):