Tutorials

This page helps the user to get started with the SnoParallel package.

General guidelines

• Always declare the code within a main guard.
• Always give target parameters as keyword arguments.
• When a worker fails, the remaining workers will continue until the Manager.wait() method is called.
• Without defining a timeout, the code could hang endlessly if an underlying worker fails.
• Do not use the Manager class within references.

Perform a single work cycle

Add a few items to the queue and wait for the workers to finish processing them.

from snoparallel import Manager


def target(work_item: str) -> None:
    print(work_item)


def main() -> None:
    manager = Manager()
    manager.add_workers(workers=2)  # number of workers should not exceed CPU cores

    for work_item in range(3):
        manager.add_work(target=target, work_item=work_item)  # give target parameters as keyword arguments

    manager.wait(timeout=10)  # add timeout to prevent endless hang


if __name__ == "__main__":  # all code must be within a main guard
    main()

Perform multiple independent work cycles

Add a few items to the queue and wait for the workers to finish processing them, repeatedly.

from snoparallel import Manager


def target(work_cycle: str, work_item: str) -> None:
    print(work_cycle, work_item)


def main() -> None:
    manager = Manager()
    manager.add_workers(workers=2)

    for work_cycle in range(3):
        for work_item in range(3):
            manager.add_work(target=target, work_cycle=work_cycle, work_item=work_item)

        manager.wait(timeout=10)


if __name__ == "__main__":
    main()

Use of shared variables

Shared variables enable work cycles in which the workers are dependent on a previous state, but can work independently. Floats, integers, and numpy arrays are supported as shared variables. In the example below, each worker reads the variables, and writes a new variable value based on their total. Once the work cycle is done, the variables are updated with the new values.

from numpy import float32
from snoparallel import Manager, Variable


def target(variable: Variable, variables: list[Variable]) -> None:
    total = sum(var.read() for var in variables)
    value = variable.read()
    variable.write(value=value + total)


def main() -> None:
    manager = Manager()
    manager.add_workers(workers=2)
    variables = 4

    for i in range(variables):
        manager.request_variable(value=float32(i))

    for variable in manager.variables:
        manager.add_work(target=target, variable=variable, variables=manager.variables)

    manager.wait(timeout=10)

    print([variable.read() for variable in manager.variables])

    for variable in manager.variables:
        variable.update()

    print([variable.read() for variable in manager.variables])


if __name__ == "__main__":
    main()

Use of a class method(s) as a target

Methods from class instances can be targeted directly, but could lead to performance degradation. Be aware that changes made by the workers to the class instance, will only apply to the instance in the worker process, and not to the one in the main process. A code snippet targeting a class method directly is given below, showing what happens in each process.

from snoparallel import Manager


class Reference:

    def __init__(self) -> None:
        self.value = 0

    def target(self, new_value: int) -> None:
        self.value = new_value
        print(self.value)  # will print 10


def main() -> None:
    reference = Reference()
    target = reference.target

    manager = Manager()
    manager.add_workers(workers=2)

    manager.add_work(target=target, new_value=10)
    manager.wait(timeout=10)

    print(reference.value)  # will print 0


if __name__ == "__main__":
    main()

It is recommended to reference the class instance(s) to the workers for improved performance. When adding work, the name of the class method needs to be given, instead of the callable itself, and the index of the class instance. Do not use the manager within the reference class! This can lead to errors from the underlying multiprocessing module.

from snoparallel import Manager


class GoodReference:

    def __init__(self, value: int) -> None:
        self.value = value

    def target(self, value: int) -> None:
        print(self.value + value)  # will print 2


class BadReference:

    def __init__(self, manager: Manager) -> None:
        self.manager = manager  # manager is copied to the workers, leading to unexpected errors


def main() -> None:
    references = [GoodReference(value=i) for i in range(2)]
    target = GoodReference.target.__name__

    manager = Manager()
    manager.add_references(references=references)  # references need to be added before the workers
    manager.add_workers(workers=2)

    for i in range(2):
        manager.add_work(target=target, index=i, value=i)  # index of the reference to call

    manager.wait(timeout=10)


if __name__ == "__main__":
    main()