Disk cache
Replay an expensive step without recomputing
.cached_task(fn, cache_dir=...) computes fn(x) and serializes the result to .pkl under a SHA-256 key derived from the input and the function. On the next run, if the key exists, the result is read from disk — the function is never called.
Signature
Pipeline.cached_task(task, cache_dir: str, count=1) -> Pipeline[CacheEnvelope[OutPacket]]
Pipeline.uncache() -> Pipeline[OutPacket]import tempfile
import time
from olympipe import Pipeline
def expensive_transform(x: int) -> dict:
time.sleep(0.5) # simulate heavy computation
return {"value": x ** 3, "meta": compute_stats(x)}
with tempfile.TemporaryDirectory() as cache_dir:
# 1st run: compute and persist to disk
results = (
Pipeline(range(1000))
.cached_task(expensive_transform, cache_dir=cache_dir)
.uncache()
.wait_for_result()
)
# ~500s
# 2nd run: disk read only
results_again = (
Pipeline(range(1000))
.cached_task(expensive_transform, cache_dir=cache_dir)
.uncache()
.wait_for_result()
)
# < 2s⚡ Performance
1 000 items, 0.5s/item computation
How it works
The key is the SHA-256 of (the cloudpickled input's hash + the serialized function's hash). Files are stored at cache_dir/<step_hash[:16]>/<key>.pkl, where step_hash is the SHA-256 of the serialized function. Works with multi-step pipelines: each step has its own cache subfolder.
Parallelising a cached step
Signature: .cached_task(task, cache_dir, count=1). As with .task, count starts several workers; they share the same cache folder. End with .uncache() to get the raw values back.
import tempfile
import time
from olympipe import Pipeline
def slow_square(x: int) -> int:
time.sleep(0.2)
return x * x
with tempfile.TemporaryDirectory() as cache_dir:
for run in (1, 2):
start = time.time()
res = (
Pipeline(range(20))
.cached_task(slow_square, cache_dir=cache_dir, count=4) # 4 workers
.uncache()
.wait_for_result()
)
print(f"run {run}: {sorted(res)[:5]}... in {time.time() - start:.1f}s")run 1: [0, 1, 4, 9, 16]... in 1.4s
run 2: [0, 1, 4, 9, 16]... in 0.4s