Unlock the full potential of parallel computing in Python with FastFlow, a powerful C++ library now available in Python, that brings high-performance, scalable parallelism right to your fingertips.
- 🤩 Implement advanced parallel patterns and building blocks, like pipelines, farms, and all-to-all, with ease.
- 🚀🚀 Experience lightning-fast parallel execution with zero boilerplate code.
- 💡 It overcomes Python’s Global Interpreter Lock (GIL) limitations for you. You can leverage on multiple processes or subinterpreters.
Whether you’re processing massive datasets, building scalable applications, or designing high-throughput systems, FastFlow delivers the speed and efficiency you've been looking for.
To install fastflow, ensure you have the following dependencies: python3, python3-venv and python3-dev. For example on Ubuntu you can install them using apt install. Ensure you have updated submodules by running git submodule init && git submodule update.
- From the root directory, create virtual environment via
python3 -m venv .venv - From the root directory, activate the virtual environment by running
source .venv/bin/activate - Build and install by running
makefrom the root directory
The Farm pattern consists of an emitter, multiple workers, and a collector. It distributes tasks from the emitter to workers and collects the results in the collector. Both emitter and collector are optional.
from fastflow import FFFarm, EOS class source(): def __init__(self): self.counter = 1 def svc(self, *arg): if self.counter == 12: return EOS self.counter += 1 return list(['source']) class worker(): def __init__(self, id): self.id = id def svc(self, lis: list): lis.append(self.id) return lis class sink(): def svc(self, lis: list): print(lis) # create a farm, pass True as argument if you want to use subinterpreters farm = FFFarm() sourcenode = source() sinknode = sink() w_lis = [] for i in range(3): w = worker(f"worker{i+1}") w_lis.append(w) farm.add_emitter(sourcenode) farm.add_workers(w_lis) farm.add_collector(sinknode) # finally run the farm. Blocking call: will resume when the farm ends farm.run_and_wait_end()
The Pipeline pattern consists of multiple stages that process tasks in a linear sequence, where the output of one stage becomes the input of the next.
from fastflow import FFPipeline, EOS import sys class source(): def __init__(self): self.counter = 1 def svc(self): if self.counter > 5: return EOS self.counter += 1 return list(["source"]) class stage(): def __init__(self, id): self.id = id def svc(self, lis: list): lis.append(self.id) return lis class sink(): def svc(self, lis: list): lis.append("sink") print(lis) # create a pipeline, pass True if you want to use subinterpreters pipe = FFPipeline(True) # create the first stage sourcenode = source() # create the second stage st1 = stage('st1') # create last stage sinknode = sink() # add all the stages pipe.add_stage(sourcenode) pipe.add_stage(st1) pipe.add_stage(sinknode) # finally run the pipeline. Blocking call: will resume when the pipeline ends pipe.run_and_wait_end() # print how many milliseconds the pipeline took print(f"pipeline done in {farm.ffTime()}ms")
The All-to-All pattern connects multiple nodes to the left to other multiple nodes to the right, where every node to the left can send data to every node to the right.
from fastflow import FFAllToAll, EOS class leftnode(): def __init__(self, id): self.counter = 1 self.id = id def svc(self, *arg): if self.counter > 5: return EOS self.counter += 1 return list([self.id]) class rightnode(): def __init__(self, id): self.id = id def svc(self, lis: list): lis.append(f"rightnode{self.id}") print(lis) # All-to-All with 4 nodes to the left and 3 nodes to the right a2a = FFAllToAll() # Create and add left nodes a2a.add_firstset([leftnode(i+1) for i in range(4)]) # Create and add right nodes a2a.add_secondset([rightnode(i+1) for i in range(3)]) # Run All-to-All a2a.run_and_wait_end()In the All-to-All building block, two different scheduling policies can be used to distribute tasks from the left nodes to the right nodes:
- Round Robin (default): This policy distributes tasks evenly across the available right nodes in a cyclic manner. Each task is assigned to the next available right node in turn, regardless of the task's nature or the load on the right nodes. This method is efficient when tasks are uniform in size and processing time.
- On-Demand: In this policy, tasks are only sent to a right node when that node is ready to receive more tasks. It dynamically adapts to the load of each right node, making it more suited for scenarios where tasks have varying sizes or processing times. To enable on-demand scheduling, you can use the
ondemand=Trueoption in theadd_firstset()method.
The following is an example of On-Demand scheduling:
from fastflow import FFAllToAll, EOS class source(): def __init__(self, id): self.counter = 1 self.id = id def svc(self, *arg): if self.counter > 5: return EOS self.counter += 1 return [f"data from source {self.id}"] class sink(): def __init__(self, id): self.id = id def svc(self, lis: list): lis.append(f"processed by sink {self.id}") print(lis) # All-to-All with on-demand scheduling a2a = FFAllToAll() # First stage (sources) first_lis = [source(i+1) for i in range(3)] # Second stage (sinks) second_lis = [sink(i+1) for i in range(2)] # Add stages using on-demand scheduling a2a.add_firstset(first_lis, ondemand=True) a2a.add_secondset(second_lis) # Run All-to-All a2a.run_and_wait_end()You can specify svc_init and svc_end for Initialization and Cleanup purposes. This example shows how to use svc_init for initialization and svc_end for finalization logic in stages.
from fastflow import FFAllToAll, EOS class source(): def __init__(self, id): self.id = id self.counter = 1 def svc_init(self): print(f"Source {self.id} initialized") def svc(self, *arg): if self.counter > 5: return EOS self.counter += 1 return [self.id] def svc_end(self): print(f"Source {self.id} finished") class sink(): def __init__(self, id): self.id = id def svc_init(self): print(f"Sink {self.id} initialized") def svc(self, lis: list): lis.append(f"sink{self.id}") print(lis) def svc_end(self): print(f"Sink {self.id} finished") # All-to-All setup a2a = FFAllToAll() first_stage_size = 3 second_stage_size = 3 # Create and add first stages (sources) first_lis = [source(i+1) for i in range(first_stage_size)] a2a.add_firstset(first_lis) # Create and add second stages (sinks) second_lis = [sink(i+1) for i in range(second_stage_size)] a2a.add_secondset(second_lis) # Run All-to-All a2a.run_and_wait_end()You can send data multiple times or to specific nodes using ff_send_out.
from fastflow import FFAllToAll, EOS, ff_send_out class source(): def svc(self, *arg): # Send data items to sink1 and sink3 ff_send_out(["source-to-sink2"], 1) ff_send_out(["source-to-sink3"], 2) # Send multiple data items to any node ff_send_out(["source-to-any"]) ff_send_out(["source-to-any"]) return EOS class sink(): def __init__(self, id): self.id = id def svc(self, lis: list): lis.append(f"sink{self.id}") print(lis) # All-to-All setup a2a = FFAllToAll() first_stage_size = 4 second_stage_size = 3 # Create and add first stage (sources) first_lis = [source() for i in range(first_stage_size)] a2a.add_firstset(first_lis) # Create and add second stage (sinks) second_lis = [sink(i+1) for i in range(second_stage_size)] a2a.add_secondset(second_lis) # Run All-to-All a2a.run_and_wait_end()This example demonstrates combining All-to-All inside a Pipeline. The pipeline uses multiple stages where one stage is an All-to-All pattern.
from fastflow import FFPipeline, FFAllToAll, EOS class source(): def __init__(self): self.counter = 1 def svc(self, *arg): if self.counter > 5: return EOS self.counter += 1 return [f"data{self.counter}"] class stage(): def svc(self, lis: list): lis.append(f"stage") return lis class sink(): def svc(self, lis: list): lis.append("sink") print(lis) # Inner All-to-All setup a2a = FFAllToAll() first_lis = [stage() for _ in range(3)] second_lis = [stage() for _ in range(2)] a2a.add_firstset(first_lis) a2a.add_secondset(second_lis) # Pipeline setup pipeline = FFPipeline() pipeline.add_stage(source()) pipeline.add_stage(a2a) # All-to-All as a stage in the pipeline pipeline.add_stage(sink()) # Run pipeline pipeline.run_and_wait_end()This example shows Pipeline inside an All-to-All stage, combining the two patterns.
from fastflow import FFAllToAll, FFPipeline, EOS class source(): def __init__(self): self.counter = 1 def svc(self, *arg): if self.counter > 5: return EOS self.counter += 1 return [f"data{self.counter}"] class stage(): def svc(self, lis: list): lis.append(f"stage") return lis class sink(): def svc(self, lis: list): lis.append(f"sink") print(lis) # build a Pipeline to be used inside the second stage of All-to-All def build_pipeline(): pipeline = FFPipeline() pipeline.add_stage(stage()) pipeline.add_stage(stage()) pipeline.add_stage(sink()) return pipeline # All-to-All setup a2a = FFAllToAll() first_lis = [source() for _ in range(3)] # 3 first stages second_lis = [build_pipeline() for _ in range(2)] # 2 Pipelines in the second stage a2a.add_firstset(first_lis) a2a.add_secondset(second_lis) # Run All-to-All with embedded pipelines a2a.run_and_wait_end()The blocking_mode(boolean) function allows you to control whether the nodes block and wait for input or continuously check for data. Setting blocking_mode(True) enables blocking mode, reducing resource usage, while blocking_mode(False) disables it, making the nodes more responsive but more resource-hungry. By default, building blocks have blocking mode disabled. All the building blocks support it, this is an example using a Farm:
from fastflow import FFFarm, EOS class source(): def __init__(self): self.counter = 1 def svc(self, *arg): if self.counter > 5: return EOS self.counter += 1 return [f"data{self.counter}"] class worker(): def svc(self, lis: list): lis.append(f"processed") return lis class sink(): def svc(self, lis: list): print(lis) # Create a Farm with blocking mode enabled farm = FFFarm() farm.blocking_mode(True) # Enable blocking mode sourcenode = source() sinknode = sink() workers = [worker() for _ in range(3)] farm.add_emitter(sourcenode) farm.add_workers(workers) farm.add_collector(sinknode) farm.run_and_wait_end()By default, each nodes is set to run to one and only one core, improving performances. The no_mapping() function disables FastFlow's feature of pinning nodes to specific cores, allowing more flexibility in resource allocation but potentially increasing overhead due to movement between cores. All the building blocks support it, this is an example using a Pipeline:
from fastflow import FFPipeline, EOS class source(): def __init__(self): self.counter = 1 def svc(self): if self.counter > 5: return EOS self.counter += 1 return [f"data{self.counter}"] class stage(): def svc(self, lis: list): lis.append(f"processed") return lis class sink(): def svc(self, lis: list): lis.append("final") print(lis) # Create a Pipeline with no mapping pipeline = FFPipeline() pipeline.no_mapping() # Disable core pinning for nodes pipeline.add_stage(source()) pipeline.add_stage(stage()) pipeline.add_stage(sink()) pipeline.run_and_wait_end()To test the CI locally (optional), useful to test changes before actually change the whole ci, pip install cibuildwheel and then run cibuildwheel --platform linux or make ci.
Contributions are welcome! Please submit pull requests or open issues to help improve this project.