fal.distributed

API reference for 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