Buckets:
| """Rank-independent work assignment for frozen-policy rollout and evaluation.""" | |
| import time | |
| import uuid | |
| from pathlib import Path | |
| import torch.distributed as dist | |
| def map_rank_tasks(tasks, function, *, directory, parallel, scheduling="dynamic"): | |
| """Return rank-local (task position, result) pairs; synchronize stage errors. | |
| No actor update occurs here. Task seeds must depend on task identity, never | |
| on the rank that happens to claim it. Wait in the store before collectives | |
| so uneven generation does not exhaust a collective timeout. | |
| """ | |
| if not parallel.enabled: | |
| return [(index, function(task)) for index, task in enumerate(tasks)] | |
| queue_name = parallel.broadcast_object(uuid.uuid4().hex if parallel.is_main else None) | |
| root = Path(directory) / ".queues" | |
| root.mkdir(parents=True, exist_ok=True) | |
| store = dist.FileStore(str(root / queue_name), -1) | |
| result = [] | |
| static_positions = iter(range(parallel.rank, len(tasks), parallel.world_size)) | |
| try: | |
| while not store.check(["failure"]): | |
| index = (store.add("next", 1) - 1 if scheduling == "dynamic" | |
| else next(static_positions, len(tasks))) | |
| if index >= len(tasks): | |
| break | |
| result.append((index, function(tasks[index]))) | |
| except Exception as error: | |
| store.compare_set("failure", "", f"rank {parallel.rank}: {type(error).__name__}: {error}") | |
| store.set(f"done/{parallel.rank}", "1") | |
| done = [f"done/{rank}" for rank in range(parallel.world_size)] | |
| while not store.check(done): | |
| if store.check(["failure"]): | |
| raise RuntimeError(store.get("failure").decode()) | |
| time.sleep(0.1) | |
| if store.check(["failure"]): | |
| raise RuntimeError(store.get("failure").decode()) | |
| return result | |
| def ordered_results(shards, expected): | |
| indexed = sorted((item for shard in shards for item in shard), key=lambda item: item[0]) | |
| if [index for index, _value in indexed] != list(range(expected)): | |
| raise RuntimeError("distributed workers produced missing or duplicate task results") | |
| return [value for _index, value in indexed] | |
Xet Storage Details
- Size:
- 2.15 kB
- Xet hash:
- 32898b64e542c41a197d78cb379ba2da1ebdc21632109339c08cf5d71aa22da1
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.