lib.multithreading Module

Multithreading/processing utils for faceswap

class lib.multithreading.BackgroundGenerator(generator: Callable, prefetch: int = 1, name: str | None = None, args: tuple | None = None, kwargs: dict[str, T.Any] | None = None)

Run a task in the background background and queue data for consumption

Parameters:

• generator (Callable) – The generator to run in the background

• prefetch (int) – The number of items to pre-fetch from the generator before blocking (see Notes). Default: 1

• name (str | None) – The thread name. if None a unique name is constructed of the form {generator.__name__}_N where N is an incrementing integer. Default: None

• args (tuple | None) – The argument tuple for generator invocation. Default: None.

• kwargs (dict[str, T.Any] | None) – keyword arguments for the generator invocation. Default: None.

Notes

Putting to the internal queue only blocks if put is called while queue has already reached max size. Therefore this means prefetch is actually 1 more than the parameter supplied (N in the queue, one waiting for insertion)

References

https://stackoverflow.com/questions/7323664/

iterator() → Generator

Iterate items out of the queue

Yields:

The items from the generator

Return type:

Generator

class lib.multithreading.ErrorState

An object for tracking error state across threads

The “set” method should be called from within a thread to set the thread error traceback

The “check_and_raise” method should be called from the main thread to check for and re-raise any errors

clear() → None

Clear any stored errors

Return type:

None

errors: list[tuple[type[BaseException], BaseException, TracebackType] | tuple[Any, Any, Any]]

list of errors that have been detected within threads

property has_error: bool

Check whether any running FSThread thread has an error.

Return type:

True if an FSThread has an error

re_raise() → None

Check if a thread error is stored and re-raise it if so. Should be called from main thread. Only the first error received is re-raised (in the event of multiple errors)

Return type:

None

set(exc_info: tuple[type[BaseException], BaseException, TracebackType] | tuple[Any, Any, Any]) → None

Set the error traceback information to the error state object. Errors are appended to the error list in the order that they are received

Parameters:

• set (The traceback error information to)

• exc_info (tuple[type[BaseException], BaseException, TracebackType] | tuple[Any, Any, Any])

Return type:

None

class lib.multithreading.FSThread(target: Callable | None = None, name: str | None = None, args: tuple = (), kwargs: dict[str, T.Any] | None = None, *, daemon: bool | None = None)

Subclass of thread that passes errors back to parent

Parameters:

• target (Callable | None) – The callable object to be invoked by the run() method. If None nothing is called. Default: None

• name (str | None) – The thread name. if None a unique name is constructed of the form “Thread-N” where N is a small decimal number. Default: None

• args (tuple) – The argument tuple for the target invocation. Default: ().

• kwargs (dict[str, T.Any] | None) – keyword arguments for the target invocation. Default: {}.

• daemon (bool | None)

check_and_raise_error() → None

Checks for errors in thread and raises them in caller.

Raises:

Error – Re-raised error from within the thread

Return type:

None

error_state = <lib.multithreading.ErrorState object>

Class attribute to track error state across multiple threads

run() → None

Runs the target, and captures any thread errors for re-raising in the caller.

Errors are also captured in a class attribute so that threads in any other running FSThreads can be captured

Return type:

None

class lib.multithreading.MultiThread(target: Callable, *args, thread_count: int = 1, name: str | None = None, **kwargs)

Threading for IO heavy ops. Catches errors in thread and rethrows to parent.

Parameters:

• target (Callable) – The callable object to be invoked by the run() method.

• args – The argument tuple for the target invocation. Default: ().

• thread_count (int) – The number of threads to use. Default: 1

• name (str | None) – The thread name. if None a unique name is constructed of the form {target.__name__}_N where N is an incrementing integer. Default: None

• kwargs – keyword arguments for the target invocation. Default: {}.

check_and_raise_error() → None

Checks for errors in thread and raises them in caller.

Raises:

Error – Re-raised error from within the thread

Return type:

None

completed() → bool

Check if all threads have completed

Return type:

True if all threads have completed otherwise False

property errors: list[tuple[type[BaseException], BaseException, TracebackType] | tuple[Any, Any, Any]]

List of thread error values

Type:

list

property has_error: bool

True if a thread has errored, otherwise False

is_alive() → bool

Check if any threads are still alive

Return type:

True if any threads are alive. False if no threads are alive

join() → None

Join the running threads, catching and re-raising any errors

Clear the list of threads for class instance re-use

Return type:

None

property name: str

The name of the thread

start() → None

Start all the threads for the given method, args and kwargs

Return type:

None

lib.multithreading.total_cpus() → int

Return total number of cpus

Return type:

int

Functions

total_cpus()

Return total number of cpus

Classes

BackgroundGenerator(generator[, prefetch, ...])	Run a task in the background background and queue data for consumption
ErrorState()	An object for tracking error state across threads
FSThread([target, name, args, kwargs, daemon])	Subclass of thread that passes errors back to parent
MultiThread(target, *args[, thread_count, name])	Threading for IO heavy ops.

Class Inheritance Diagram