magmalake Dista

Dista · API

dista.store

The node-local object store: large results in shared mappings.

A worker writes each large result into a file in a node-local directory (/dev/shm on Linux, $TMPDIR on macOS, or DISTA_STORE_DIR). A reader that can see that file maps it, through memory-region's SharedMapping, read-only and reads the bytes where they lie: no socket, no copy through the kernel's network stack. A reader that can't fetches the result over a socket from the worker, as before.

"Can see the file" is the test, not a matching host name: file names carry a random token per worker, so a file with the right name is the right file. That covers one machine, and Kubernetes pods that share a volume. (Host names turned out to be unreliable: on GitHub's macOS runners, processes on one machine reported different ones.)

DISTA_SHM=0 turns the store off; large results then stay in the worker's memory and are always fetched over sockets.

Spilling: a worker keeps at most DISTA_STORE_BYTES of results in the store (no limit when unset), and a result past that, or one the store can't take, goes to DISTA_SPILL_DIR on disk instead (default $TMPDIR, or /tmp). A spilled result is read the same way, mapped or fetched, only from disk. On Kubernetes /dev/shm is an in-memory volume whose size limit evicts the pod rather than failing a write, so the limit is what keeps it in bounds.

A location says where a result is: the worker's object server address, its host name, the file's path ("" when it's only in memory) and the exact size, since mapped files are padded.

Structs

Location

struct Location

Where a large result lives.

Implements Copyable, Movable

Fields
ip: Array[UInt8, Int(4)]
port: Int
node: String
path: String
size: Int

__init__

def __init__(out self, ip: Array[UInt8, Int(4)], port: Int, var node: String, var path: String, size: Int)

is_local

def is_local(self) -> Bool

Whether this process can map it instead of fetching it.

encode

def encode(self) -> List[UInt8]

decode

def decode(buf: Span[UInt8], mut pos: Int) -> Self

Functions

node_name

def node_name() -> String

This host's name, which identifies a shared /dev/shm, or DISTA_NODE if it's set: tests use it to run one host as several nodes.

worker_token

def worker_token() -> UInt64

A number naming one worker's files, so that no two workers, on this host or another, write the same file name: a hash of the host name, the process ID and a nanosecond clock.

store_dir

def store_dir() -> String

Where large results go, or "" when the shared store is off.

remove_file

def remove_file(path: String)

write_shared

def write_shared(path: String, data: Span[UInt8])

Writes data into a new file at path, for readers to map.

It's written under a temporary name and renamed into place, so a reader that maps path sees a whole result, old or new, never one mid-write. It's written with write(2), not through a mapping: a full file system then fails the write with an error, where a mapped store would kill the process with SIGBUS on a page it couldn't back.

spill_dir

def spill_dir() -> String

Where results that don't fit the store go: DISTA_SPILL_DIR, or the temporary directory, which in a container is on disk.

store_limit

def store_limit() -> Int

DISTA_STORE_BYTES: how many bytes of results a worker keeps in the store before it spills; 0, no limit, when unset.

read_shared

def read_shared(path: String, size: Int) -> List[UInt8]

The first size bytes of the shared mapping at path.