Running in parallel

Up until now we have running all our models in a single computational process. This is perfectly sufficient for simple models, or when your components can make use of Python's asyncio to avoid blocking.

As your models get larger and more computationally intensive you may benefit from running parts of the model in parallel. Plugboard integrates with the Ray framework, allowing you to split your computation across multiple CPU cores, or even across nodes in a Ray cluster.

Tip

Keep in mind that parallelising a model has a cost associated with it: the communication between the different components will be slower on Ray than it is locally.

For small models, or when a single component is the computational bottleneck then this overhead may not be worth it. However, when you have multiple computationally-intensive components in different branches of your Process then moving to Ray can give you a performance boost.

Before running this tutorial be sure to install Ray with pip, or install plugboard with its optional ray extra.

Parallelising a model

For demonstration purposes we're going to use a model with two branches that containing Sleep components to simulate computationally intensive activity. In real scenarios these might instead be calls to simulation software or machine-learning model inference.

graph LR;
    Iterator(input)-->Slow(slow-sleep);
    Iterator(input)-->VerySlow(very-slow-sleep);
    Slow(slow-sleep)-->Timestamper(timestamper);
    VerySlow(very-slow-sleep)-->Timestamper(timestamper);
    Timestamper(timestamper)-->FileWriter(save-results);

Defining the components

Let's define the various components that we need. The Timestamper component simply emits the current time in ISO format so that our output file will contain a record of how long each step of the model took. We can again use FileWriter to save the output to CSV.

class Iterator(Component):
    """Creates a sequence of numbers."""

    io = IO(outputs=["x"])

    def __init__(self, iters: int, **kwargs: _t.Unpack[ComponentArgsDict]) -> None:
        super().__init__(**kwargs)
        self._iters = iters

    async def init(self) -> None:
        self._seq = iter(range(self._iters))

    async def step(self) -> None:
        try:
            self.out_1 = next(self._seq)
        except StopIteration:
            await self.io.close()


class Sleep(Component):
    """Passes through input to output after a delay."""

    io = IO(inputs=["x"], outputs=["y"])

    def __init__(self, sleep_seconds: float, **kwargs: _t.Unpack[ComponentArgsDict]) -> None:
        super().__init__(**kwargs)
        self._duration = sleep_seconds

    async def step(self) -> None:
        time.sleep(self._duration)  # (1)!
        self.y = self.x


class Timestamper(Component):
    """Emits the current time when all inputs are ready."""

    io = IO(inputs=["x", "y"], outputs=["timestamp"])

    async def step(self) -> None:
        self.timestamp = datetime.datetime.now().isoformat()

We're using time.sleep here and not asyncio.sleep because we're deliberately blocking execution to simulate a computationally intensive component.

Running normally in a `LocalProcess`

First we can setup the

Running in parallel

Parallelising a model

Defining the components

Running normally in a `LocalProcess`