fal.distributed

from fal.distributed import DistributedRunner, DistributedWorker

Classes

DistributedRunner

class fal.distributed.DistributedRunner

A class to launch and manage distributed workers.

Constructor Parameters

Name	Type	Default	Description
`worker_cls`	`type[DistributedWorker]`	`\<class 'fal.distributed.worker.DistributedWorker'\>`	-
`world_size`	`int`	`1`	-
`master_addr`	`str`	`'127.0.0.1'`	-
`master_port`	`int`	`29500`	-
`worker_addr`	`str`	`'127.0.0.1'`	-
`worker_port`	`int`	`54923`	-
`timeout`	`int`	`86400`	-
`keepalive_payload`	`dict[str, Any]`	`\{\}`	-
`keepalive_interval`	`int \| float \| NoneType`	`None`	-
`cwd`	`str \| Path \| NoneType`	`None`	-
`set_device`	`Optional[bool]`	`None`	-

Class Variables

Name	Type	Default	Description
`zmq_socket`	`Optional[Socket[Any]]`	-	-
`context`	`Optional[mp.ProcessContext]`	-	-
`keepalive_timer`	`Optional[KeepAliveTimer]`	-	-

Methods

close_zmq_socket

def close_zmq_socket(self) -> 'None'

Closes the ZeroMQ socket.Returns: NoneType

ensure_alive

def ensure_alive(self) -> 'None'

Ensures that the distributed worker processes are alive. If the processes are not alive, it raises an error.Returns: NoneType

gather_errors

def gather_errors(self) -> 'list[Exception]'

Gathers errors from the distributed worker processes. This method should be called to collect any errors that occurred during execution.Returns: list[Exception]

get_zmq_socket

def get_zmq_socket(self) -> 'Socket[Any]'

Returns a ZeroMQ socket of the specified type.Returns: A ZeroMQ socket.

invoke

async def invoke(self, payload: 'dict[str, Any]' = {}, timeout: 'Optional[int]' = None) -> 'Any'

Invokes the distributed worker with the given payload.

Parameter	Type	Default	Description
`payload`	`dict[str, Any]`	`\{\}`	The payload to send to the worker.
`timeout`	`Optional[int]`	`None`	The timeout for the overall operation.

Returns: Any

is_alive

def is_alive(self) -> 'bool'

Check if the distributed worker processes are alive.Returns: bool

keepalive

def keepalive(self, timeout: 'Optional[Union[int, float]]' = 60.0) -> 'None'

Sends the keepalive payload to the worker.

Parameter	Type	Default	Description
`timeout`	`int \| float \| NoneType`	`60.0`	-

Returns: NoneType

maybe_cancel_keepalive

def maybe_cancel_keepalive(self) -> 'None'

Cancels the keepalive timer if it is set.Returns: NoneType

maybe_reset_keepalive

def maybe_reset_keepalive(self) -> 'None'

Resets the keepalive timer if it is set.Returns: NoneType

maybe_start_keepalive

def maybe_start_keepalive(self) -> 'None'

Starts the keepalive timer if it is set.Returns: NoneType

run

def run(self, **kwargs: 'Any') -> 'None'

The main function to run the distributed worker. This function is called by each worker process spawned by torch.multiprocessing.spawn. This method must be synchronous.

Parameter	Type	Default	Description
`kwargs`	`Any`	-	The arguments to pass to the worker.

Returns: NoneType

start

async def start(self, timeout: 'int' = 1800, **kwargs: 'Any') -> 'None'

Starts the distributed worker processes.

Parameter	Type	Default	Description
`timeout`	`int`	`1800`	The timeout for the distributed processes.
`kwargs`	`Any`	-	-

Returns: NoneType

stop

async def stop(self, timeout: 'int' = 10) -> 'None'

Stops the distributed worker processes.

Parameter	Type	Default	Description
`timeout`	`int`	`10`	The timeout for the distributed processes to stop.

Returns: NoneType

stream

def stream(self, payload: 'dict[str, Any]' = {}, timeout: 'Optional[int]' = None, streaming_timeout: 'Optional[int]' = None, as_text_events: 'bool' = False) -> 'AsyncIterator[Any]'

Streams the result from the distributed worker.

Parameter	Type	Default	Description
`payload`	`dict[str, Any]`	`\{\}`	The payload to send to the worker.
`timeout`	`Optional[int]`	`None`	The timeout for the overall operation.
`streaming_timeout`	`Optional[int]`	`None`	The timeout in-between streamed results.
`as_text_events`	`bool`	`False`	Whether to yield results as text events.

Returns: AsyncIterator[Any]

terminate

def terminate(self, timeout: 'Union[int, float]' = 10) -> 'None'

Terminates the distributed worker processes. This method should be called to clean up the worker processes.

Parameter	Type	Default	Description
`timeout`	`int \| float`	`10`	-

Returns: NoneType

DistributedWorker

class fal.distributed.DistributedWorker

A base class for distributed workers.

Constructor Parameters

Name	Type	Default	Description
`rank`	`int`	`0`	-
`world_size`	`int`	`1`	-

Class Variables

Name	Type	Default	Description
`queue`	`queue.Queue[bytes]`	-	-
`loop`	`asyncio.AbstractEventLoop`	-	-
`thread`	`threading.Thread`	-	-

Properties

Name	Type	Description
`device`	-	The device for the current worker
`running`	`bool`	Whether the event loop is running

Methods

add_streaming_error

def add_streaming_error(self, error: 'Exception') -> 'None'

Add an error to the queue.

Parameter	Type	Default	Description
`error`	`Exception`	-	The error to add to the queue.

Returns: NoneType

add_streaming_result

def add_streaming_result(self, result: 'Any', image_format: 'str' = 'jpeg', as_text_event: 'bool' = False) -> 'None'

Add a streaming result to the queue.

Parameter	Type	Default	Description
`result`	`Any`	-	The result to add to the queue.
`image_format`	`str`	`'jpeg'`	-
`as_text_event`	`bool`	`False`	-

Returns: NoneType

initialize

def initialize(self, **kwargs: 'Any') -> 'None'

Initialize the worker.

Parameter	Type	Default	Description
`kwargs`	`Any`	-	-

Returns: NoneType

rank_print

def rank_print(self, message: 'str', debug: 'bool' = False) -> 'None'

Print a message with the rank of the current worker.

Parameter	Type	Default	Description
`message`	`str`	-	The message to print.
`debug`	`bool`	`False`	Whether to print the message as a debug message.

Returns: NoneType

run_in_worker

def run_in_worker(self, func: 'Callable[..., Any]', *args: 'Any', **kwargs: 'Any') -> 'Future[Any]'

Run a function in the worker.

Parameter	Type	Default	Description
`func`	`Callable[Ellipsis, Any]`	-	-
`args`	`Any`	-	-
`kwargs`	`Any`	-	-

Returns: Future[Any]

setup

def setup(self, **kwargs: 'Any') -> 'None'

Override this method to set up the worker. This method is called once per worker.

Parameter	Type	Default	Description
`kwargs`	`Any`	-	-

Returns: NoneType

shutdown

def shutdown(self, timeout: 'Optional[Union[int, float]]' = None) -> 'None'

Shutdown the event loop.

Parameter	Type	Default	Description
`timeout`	`int \| float \| NoneType`	`None`	The timeout for the shutdown.

Returns: NoneType

submit

def submit(self, coro: 'Coroutine[Any, Any, Any]') -> 'Future[Any]'

Submit a coroutine to the event loop.

Parameter	Type	Default	Description
`coro`	`Coroutine[Any, Any, Any]`	-	The coroutine to submit to the event loop.

Returns: Future[Any]

teardown

def teardown(self) -> 'None'

Override this method to tear down the worker. This method is called once per worker.Returns: NoneType

Client Libraries

CLI Reference

Python SDK

fal.distributed

Classes

DistributedRunner

close_zmq_socket

ensure_alive

gather_errors

get_zmq_socket

invoke

is_alive

keepalive

maybe_cancel_keepalive

maybe_reset_keepalive

maybe_start_keepalive

run

start

stop

stream

terminate

DistributedWorker

add_streaming_error

add_streaming_result

initialize

rank_print

run_in_worker

setup

shutdown

submit

teardown

​Classes

​DistributedRunner

​close_zmq_socket

​ensure_alive

​gather_errors

​get_zmq_socket

​invoke

​is_alive

​keepalive

​maybe_cancel_keepalive

​maybe_reset_keepalive

​maybe_start_keepalive

​run

​start

​stop

​stream

​terminate

​DistributedWorker

​add_streaming_error

​add_streaming_result

​initialize

​rank_print

​run_in_worker

​setup

​shutdown

​submit

​teardown

Classes

DistributedRunner

close_zmq_socket

ensure_alive

gather_errors

get_zmq_socket

invoke

is_alive

keepalive

maybe_cancel_keepalive

maybe_reset_keepalive

maybe_start_keepalive

run

start

stop

stream

terminate

DistributedWorker

add_streaming_error

add_streaming_result

initialize

rank_print

run_in_worker

setup

shutdown

submit

teardown