How to use multiprocessing in Python to run camera, detection, and robot loops in parallel

thijs · September 2, 2026, 12:17am

Based on contributions by Tijn.Francken.

Multiprocessing lets you run several parts of a Python program at the same time on separate CPU cores. Use it whenever one slow task (like image analysis) would otherwise freeze a fast task (like a robot’s heartbeat or a camera feed) if they ran in the same loop.

What you need

Python 3.8+ on a multi-core machine.
A task that naturally splits into independent parts (for example: read camera, run detection, control robot).
No extra packages: multiprocessing is part of the standard library.

Background: why not just use threading?

A normal Python script runs on a single CPU core. Threading lets you switch quickly between tasks, but Python’s Global Interpreter Lock (GIL) still only allows one thread to run Python code at a time — it looks parallel, but it isn’t.

Multiprocessing bypasses the GIL by starting separate operating-system processes. Each process has its own Python interpreter and its own memory, so they run truly in parallel on different cores.

The trade-off is that processes cannot share variables directly — if one process changes a variable, other processes never see that change. This is intentional (it prevents data corruption) but it means processes must communicate by explicitly passing messages through a shared Queue or similar object.

Why this matters in robotics: a typical vision-guided robot cell has three tasks running at very different speeds:

Camera – should run at 30–60 FPS for a smooth feed.
Detection – CPU-heavy image analysis, often only 5–10 FPS.
Robot control – needs a constant, fast “heartbeat” signal; movements themselves take seconds.

If you run all three in a single loop, the whole application is only as fast as the slowest part: the camera feed freezes while the robot moves, and a Universal Robot (or similar cobot) may read a delayed heartbeat as a lost connection and trigger a protective stop. Splitting each task into its own process means a slow step never blocks the others.

Steps

1. Define the architecture

Map out your independent tasks and the data that flows between them. A common 3-process pattern for a robot vision system:

camera_worker – reads frames from the camera as fast as possible and sends them to detection_worker.
detection_worker – receives frames, runs the (slow) analysis, and sends results (e.g. coordinates and angle) to robot_worker.
robot_worker – receives detection results and user commands, and manages the robot’s state and movements.

2. Create queues and a stop event

In your main script, set up the “mailboxes” the processes will use to talk to each other, and a shared flag to stop everything cleanly.

from multiprocessing import Process, Queue, Event
import queue

frame_q = Queue(maxsize=2)      # camera_worker -> detection_worker
detection_q = Queue(maxsize=5)  # detection_worker -> robot_worker
command_q = Queue(maxsize=5)    # main process -> robot_worker

stop_event = Event()

3. Write the worker functions

Each worker is a function that loops until stop_event is set.

def camera_worker(frame_q, stop_event):
    # cam = initialize_camera()
    while not stop_event.is_set():
        # frame = cam.read()
        try:
            frame_q.put_nowait(frame)   # send frame to detection_worker
        except queue.Full:
            pass  # detection is lagging behind, skip this frame


def detection_worker(frame_q, detection_q, stop_event):
    while not stop_event.is_set():
        try:
            frame = frame_q.get(timeout=1)  # wait for a new frame
            # result = heavy_analysis(frame)
            # result = {"x": 100, "y": 250, "angle": 0.5}
            detection_q.put(result)  # send result to robot_worker
        except queue.Empty:
            continue  # no new frame yet, loop again


def robot_worker(detection_q, command_q, stop_event):
    robot_state = "IDLE"
    # robot = initialize_robot()

    while not stop_event.is_set():
        # 1. check for user commands
        try:
            cmd = command_q.get_nowait()
            if cmd == "toggle_state":
                robot_state = "RUNNING" if robot_state == "IDLE" else "IDLE"
        except queue.Empty:
            pass

        # 2. run the main robot logic
        if robot_state == "RUNNING":
            try:
                result = detection_q.get_nowait()
                # tell the robot to move to result["x"], result["y"]
                # this can take several seconds - that's fine, the
                # camera and detection processes keep running meanwhile
            except queue.Empty:
                pass

4. Write the main orchestrator

The if __name__ == "__main__": block starts the workers and handles user input (for example key presses on an OpenCV display window). Guarding with __name__ == "__main__" is required on Windows and macOS, where multiprocessing re-imports the main module in each new process.

if __name__ == "__main__":
    workers = [
        Process(target=camera_worker, args=(frame_q, stop_event)),
        Process(target=detection_worker, args=(frame_q, detection_q, stop_event)),
        Process(target=robot_worker, args=(detection_q, command_q, stop_event)),
    ]

    print("Starting all workers...")
    for w in workers:
        w.start()

    try:
        while True:
            # display window / read a key, e.g. with OpenCV:
            # key = cv2.waitKey(1) & 0xFF

            if key == ord("q"):
                print("Shutdown signal sent.")
                stop_event.set()
                break

            if key == ord("g"):
                print("Toggling robot state.")
                command_q.put("toggle_state")
    finally:
        for w in workers:
            w.join()  # wait for each process to finish
        print("All processes have shut down.")

Common mistakes

Forgetting the if __name__ == "__main__": guard. Without it, starting a Process can re-run your whole script recursively on Windows and macOS.
Sharing variables directly between processes. This does not work — processes have separate memory. Always pass data through a Queue, Pipe, or a multiprocessing.Manager object.
Using a blocking queue.get() without a timeout in a loop that also needs to check stop_event — the worker will hang past the point where it should have stopped. Use get(timeout=...) or get_nowait() combined with a try/except queue.Empty.
Unbounded queues can silently use a lot of memory if a fast producer outpaces a slow consumer; the maxsize + put_nowait()/except queue.Full pattern above deliberately drops frames instead of piling them up.