Skip to content

python.md

Latest commit

 

History

History
146 lines (111 loc) · 5.06 KB

python.md

File metadata and controls

146 lines (111 loc) · 5.06 KB
Raw

Python client

A binding over Talon's Rust client core. Reads objects through the cache instead of from the origin, so repeated reads across a fleet are served from local NVMe.

Install

pip install talon-client

Wheels are abi3, so one artifact per platform covers CPython 3.8 and newer:

platform wheel
Linux x86_64 manylinux 2.28
Linux aarch64 manylinux 2.28
macOS x86_64 10.12+
macOS arm64 11.0+

Windows is not built, since the worker is Linux-only.

Wheels are built on release tags by the release wheels workflow; the per-PR job builds and tests a Linux x86_64 wheel as a regression check.

To build from a checkout:

pip install maturin
maturin build --release --manifest-path clients/python/Cargo.toml --out dist
pip install dist/*.whl

Reading

import talon

with talon.Client("coordinator-host:7000", block_size=8 << 20) as client:
    info = client.stat("az://container/datasets/train.parquet")
    print(info.size, info.version)

    chunk = client.read("az://container/datasets/train.parquet",
                        offset=0, length=1 << 20)

block_size must match the workers' configured block size. Placement is computed per block, so a mismatch addresses blocks that do not exist.

URIs use the same namespaces as the FUSE mount — s3://, gcs://, az:// — so a path addresses the same object through either.

Reading many ranges of one object

read resolves the object's size and version with a stat when both are omitted. That is one extra round trip per call, which is wasted work when reading many ranges of the same object: a version is stable for an object generation.

info = client.stat(uri)
for offset in range(0, info.size, chunk_size):
    part = client.read(uri, version=info.version, size=info.size,
                       offset=offset, length=chunk_size)

version and size must be passed together: the pair describes one source generation and lets the client clamp at end-of-file without another lookup. Passing only one raises ValueError instead of combining metadata from two different generations. Workers serve matching cached bytes or conditionally fetch that exact version from the origin. If it is unavailable, the read fails instead of returning bytes from a replacement object.

Threads

Blocking calls release the GIL, so a threaded loader is limited by the network rather than serialised on the interpreter:

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=8) as pool:
    parts = list(pool.map(
        lambda off: client.read(uri, version=info.version, size=info.size,
                                offset=off, length=chunk_size),
        offsets))

Short reads

A read at or past end-of-file returns an empty buffer, and a read overlapping the end is truncated to what exists — POSIX short-read semantics. Check the returned length rather than assuming the requested count:

data = client.read(uri, offset=offset, length=length)
if len(data) < length:
    ...  # reached the end of the object

Errors

Failures raise OSError with the worker's own message preserved, so "object does not exist" stays distinguishable from "every replica is down". Malformed URIs raise ValueError before any I/O happens.

Not yet available

list is implemented but returns an error until the backends gain a listing capability (#332). Writes go through the FUSE mount or the Rust client.

Prewarm

result = client.load("s3://bucket/file", version="etag-1", size=4096)
print(result.size, result.blocks)
results = client.batch_load([
    talon.LoadRequest("s3://bucket/a", "etag-a", 4096),
    talon.LoadRequest("s3://bucket/b", "etag-b", 8192),
])

Both methods release the GIL while waiting. The supplied version and size are required, and no HEAD is issued. batch_load sends up to 1024 block instructions per protocol frame and returns LoadResult objects in input order. Empty input succeeds without network I/O. Errors may leave completed fills cached; neither method pins cache residency. Transient S3 failures are retried by the worker. Both methods accept the optional trace_context keyword used by read. Invalid coordinates raise ValueError; availability and deadline failures use UnavailableError and TimeoutError. Worker refusals raise OSError.

On a submitted batch failure, catch talon.BatchLoadError and inspect error.failed_files: each LoadFailure has index (zero-based input index), uncertain, and error (diagnostic). Entries are unique and ordered; omitted inputs completed successfully. A confirmed failed block sets uncertain=False; a missing outcome sets it to True. Empty files are successful and duplicate inputs retain separate indices. Other files continue after individual failures. Invalid arguments rejected before dispatch still raise ValueError.

try:
    client.batch_load(requests)
except talon.BatchLoadError as error:
    retry_requests = [requests[f.index] for f in error.failed_files]