juturna.transport package¶

Subpackages¶

Module contents¶

class juturna.transport.BrowserTransport(queue_capacity: int = 8, max_meta_json_bytes: int = 4096, max_payload_bytes: int = 2000000, remote_timeout: float = 30.0)¶

Bases: object

Transport backend for a pipeline running in a browser (Pyodide)

attach_queue(handle) → _BrowserQueue¶: Attach to a queue created in another Worker, from its handle

capabilities: ClassVar[frozenset[str]] = frozenset({'page_channel'})¶

connect(node_name: str, handle) → None¶: Register the handle of the inbound queue of a node of another Worker, as returned by its queue_handle(). Writes to that node’s destination wait for it. Calling it again for the same node, with the handle of a rebuilt node, makes the destination write to the new queue.

is_current(handle: WorkerHandle) → bool¶

new_condition() → _BrowserCondition¶

new_event() → _BrowserEvent¶

new_lock() → _BrowserLock¶

new_queue(maxsize: int = 0, local: bool = False) → _BrowserQueue | _BrowserLocalQueue¶

node_scope(shared: bool)¶: Declare, for the queues built inside the block, whether the node they belong to is written to from another Worker. When it is not, a queue is local (see _BrowserLocalQueue); outside a block, or with shared true, it is a ring.

queue_handle(queue: _BrowserQueue)¶: What to post to another Worker so it can attach_queue().

remote_destination(node_name: str) → _BrowserRemoteDestination¶: The destination standing for a node of another Worker, to be linked as if it were the node itself (see Pipeline.warmup).

spawn(target: Callable[[], None], name: str, daemon: bool = True) → _BrowserWorker¶

class juturna.transport.Condition(*args, **kwargs)¶

Bases: Protocol

A lock with wait/notify semantics, used to track pending work.

notify_all() → None¶

wait_for(predicate: Callable[[], bool], timeout: float | None = None) → None¶

exception juturna.transport.Empty¶

Bases: Exception

Raised by Queue.get() when no item is available within the timeout.

class juturna.transport.Event(*args, **kwargs)¶

Bases: Protocol

A boolean flag shared across workers, used to signal stop conditions.

clear() → None¶

is_set() → bool¶

set() → None¶

wait(timeout: float | None = None) → bool¶

exception juturna.transport.Full¶

Bases: Exception

Raised by Queue.put() when no free slot is available in time.

class juturna.transport.Lock(*args, **kwargs)¶

Bases: Protocol

A mutual exclusion primitive, used as a context manager.

class juturna.transport.Queue(*args, **kwargs)¶

Bases: Protocol

A FIFO channel used to move messages between nodes and workers.

empty() → bool¶

full() → bool¶

get(timeout: float | None = None) → Any¶

get_nowait() → Any¶

put(item: Any, timeout: float | None = None) → None¶

put_nowait(item: Any) → None¶

qsize() → int¶

class juturna.transport.ThreadingTransport¶

Bases: object

Default transport backend, based on real OS threads.

This backend preserves the current concurrency model of Node: every worker runs on its own threading.Thread, and queues are backed by queue.Queue.

capabilities: ClassVar[frozenset[str]] = frozenset({'shared_fs'})¶

is_current(handle: WorkerHandle) → bool¶

new_condition() → _ThreadCondition¶

new_event() → _ThreadEvent¶

new_lock() → _ThreadLock¶

new_queue(maxsize: int = 0, local: bool = False) → _ThreadQueue¶

node_scope(shared: bool) → AbstractContextManager¶

remote_destination(node_name: str) → Any¶

spawn(target: Callable[[], None], name: str, daemon: bool = True) → _ThreadWorker¶

class juturna.transport.TransportBackend(*args, **kwargs)¶

Bases: Protocol

Factory of concurrency primitives used by nodes and buffers.

An implementation decides how messages are moved and how workers are executed (e.g. real OS threads, or a cooperative scheduler); nodes and buffers only depend on this interface, never on the concrete primitives.

capabilities: ClassVar[frozenset[str]]¶

is_current(handle: WorkerHandle) → bool¶

new_condition() → Condition¶

new_event() → Event¶

new_lock() → Lock¶

new_queue(maxsize: int = 0, local: bool = False) → Queue¶: Build a queue. local is a hint: the queue is only used within the execution context that creates it (the messages never go to another worker), so a backend may use a cheaper implementation. A backend whose queues are all local anyway ignores it.

node_scope(shared: bool) → AbstractContextManager¶: Return a context manager around the construction of a node. shared tells whether the node is written to from another worker, which a backend may use to build the queues of the node accordingly. A backend with a single kind of queue returns a no-op context manager.

remote_destination(node_name: str) → Any¶: Return the object that stands for a node of another worker: anything with a put(message) method, to be used as a destination of the local nodes. Raises ValueError if the backend cannot reach other workers.

spawn(target: Callable[[], None], name: str, daemon: bool = True) → WorkerHandle¶

class juturna.transport.WorkerHandle(*args, **kwargs)¶

Bases: Protocol

A handle to a unit of concurrent execution spawned by a backend.

is_alive() → bool¶

join(timeout: float | None = None) → None¶

name: str¶

start() → None¶

juturna.transport.get_transport(spec: str | dict | None = None) → TransportBackend¶

Resolve a transport backend from its configuration.

Parameters:: spec (str | dict | None) – Either the name of the backend, as registered in _TRANSPORTS, or a dictionary with its name and the options of the backend, that are passed to its constructor, like {'name': 'browser', 'max_payload_bytes': 4000000}. Defaults to the threading backend if not provided.
Returns:: A new instance of the requested backend.
Return type:: TransportBackend
Raises:: ValueError – If the name does not match any registered backend, or the backend does not accept one of the options.