Master Python Multiprocessing: Processes, Locks, Queues, and More

Python's threading does not provide true parallelism because of the GIL, so the multiprocessing package is used to run code on multiple CPU cores. It offers classes for creating processes and various synchronization and communication primitives.

Process

The Process class creates a new process. Its constructor signature is Process([group, target, name, args, kwargs]). Important methods include start(), join([timeout]), is_alive(), terminate(). Example:

import multiprocessing, time

def worker(interval):
    n = 5
    while n > 0:
        print("The time is {0}".format(time.ctime()))
        time.sleep(interval)
        n -= 1

if __name__ == "__main__":
    p = multiprocessing.Process(target=worker, args=(3,))
    p.start()
    print("p.pid:", p.pid)
    print("p.name:", p.name)
    print("p.is_alive:", p.is_alive())

Lock

A Lock prevents concurrent access to shared resources. Example using a lock with a file:

import multiprocessing, sys

def worker_with(lock, f):
    with lock:
        fs = open(f, 'a+')
        for _ in range(10):
            fs.write("Lock acquired via with
")
        fs.close()

def worker_no_with(lock, f):
    lock.acquire()
    try:
        fs = open(f, 'a+')
        for _ in range(10):
            fs.write("Lock acquired directly
")
        fs.close()
    finally:
        lock.release()

if __name__ == "__main__":
    lock = multiprocessing.Lock()
    f = "file.txt"
    w = multiprocessing.Process(target=worker_with, args=(lock, f))
    nw = multiprocessing.Process(target=worker_no_with, args=(lock, f))
    w.start()
    nw.start()
    print("end")

Semaphore

A Semaphore limits the number of processes that can access a resource simultaneously. Example:

import multiprocessing, time

def worker(s, i):
    s.acquire()
    print(multiprocessing.current_process().name + " acquire")
    time.sleep(i)
    print(multiprocessing.current_process().name + " release")
    s.release()

if __name__ == "__main__":
    s = multiprocessing.Semaphore(2)
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(s, i*2))
        p.start()

Event

An Event allows processes to wait for a flag to be set, enabling synchronization.

import multiprocessing, time

def wait_for_event(e):
    print("wait_for_event: starting")
    e.wait()
    print("wait_for_event: e.is_set() ->", e.is_set())

def wait_for_event_timeout(e, t):
    print("wait_for_event_timeout: starting")
    e.wait(t)
    print("wait_for_event_timeout: e.is_set() ->", e.is_set())

if __name__ == "__main__":
    e = multiprocessing.Event()
    w1 = multiprocessing.Process(name="block", target=wait_for_event, args=(e,))
    w2 = multiprocessing.Process(name="non-block", target=wait_for_event_timeout, args=(e, 2))
    w1.start()
    w2.start()
    time.sleep(3)
    e.set()
    print("main: event is set")

Queue

multiprocessing.Queue

provides a process‑safe FIFO queue for data exchange.

import multiprocessing

def writer_proc(q):
    try:
        q.put(1, block=False)
    except:
        pass

def reader_proc(q):
    try:
        print(q.get(block=False))
    except:
        pass

if __name__ == "__main__":
    q = multiprocessing.Queue()
    writer = multiprocessing.Process(target=writer_proc, args=(q,))
    reader = multiprocessing.Process(target=reader_proc, args=(q,))
    writer.start()
    reader.start()
    reader.join()
    writer.join()

Pipe

Pipe()

returns a pair of connection objects for two‑way communication. Example:

import multiprocessing, time

def proc1(pipe):
    for i in range(5):
        print("send:", i)
        pipe.send(i)
        time.sleep(1)

def proc2(pipe):
    while True:
        print("proc2 recv:", pipe.recv())
        time.sleep(1)

if __name__ == "__main__":
    conn1, conn2 = multiprocessing.Pipe()
    p1 = multiprocessing.Process(target=proc1, args=(conn1,))
    p2 = multiprocessing.Process(target=proc2, args=(conn2,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

Pool

The Pool class manages a pool of worker processes, simplifying parallel execution of many tasks.

# Non‑blocking example
import multiprocessing, time

def func(msg):
    print("msg:", msg)
    time.sleep(3)
    print("end")

if __name__ == "__main__":
    pool = multiprocessing.Pool(processes=3)
    for i in range(4):
        msg = "hello %d" % i
        pool.apply_async(func, (msg,))
    print("Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~")
    pool.close()
    pool.join()
    print("Sub‑process(es) done.")

Similar code can be written using apply for blocking execution or collecting results with apply_async(...).get().

Overall, the multiprocessing module provides a powerful set of tools for parallel programming in Python, enabling developers to fully utilize multi‑core CPUs through processes, synchronization primitives, inter‑process communication, and process pools.