Process

A process represents an instance of a computer program being executed, serving as the fundamental unit for resource allocation and scheduling within an operating system. In modern computing architectures, a process acts as a container for one or more threads.

The Process class models a process object. Creating a child process involves passing a target function and its arguments to instantiate a Process object. This mechanism allows for increased CPU utilization and improved efficiency.

Process Class Structure

Examination of the Process class reveals its core structure:

class Process(object):
    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
        self.name = ''
        self.daemon = False
        self.authkey = None
        self.exitcode = None
        self.ident = 0
        self.pid = 0
        self.sentinel = None

    def run(self):
        pass

    def start(self):
        pass

    def terminate(self):
        pass

    def join(self, timeout=None):
        pass

    def is_alive(self):
        return False

Beyond the __init__ constructor, the class provides five primary methods: run(), start(), terminate(), join(), and is_alive().

The run() Method

Consider this basic process creation example:

import os
from multiprocessing import Process

def task():
    print(f'Child Process ID: {os.getpid()}, Parent ID: {os.getppid()}')

if __name__ == '__main__':
    print(f'Main Process ID: {os.getpid()}, Parent ID: {os.getppid()}')
    worker = Process(target=task)
    worker.start()

Execution yields output similar to:

Main Process ID: 6864, Parent ID: 6100
Child Process ID: 5612, Parent ID: 6864

This demonstrates that a new child process is spawned to execute the task() function.

When employing object-oriented design, a custom class must inherit from Process and override the run() method, which serves as the entry point for the child process's execution.

import os
from multiprocessing import Process

class CustomProcess(Process):
    def run(self):
        print(f'Child Process ID: {os.getpid()}, Parent ID: {os.getppid()}')

if __name__ == '__main__':
    print(f'Main Process ID: {os.getpid()}, Parent ID: {os.getppid()}')
    worker = CustomProcess()
    worker.start()

Important Note for Windows: The if __name__ == '__main__': guard is essential on Windows systems. Windows spawns new processes by re-importing the module, which without this guard could lead to infinite recursion and a RuntimeError. Linux systems, in contrast, copy the existing memory space.

The start() Method

The start() method does not directly execute the program but instructs the operating system to create a new process. Internally, it invokes the run() method to begin execution in the child process.

The join() Method

join() blocks the calling process (typically the main process) until the target process completes its execution.

import os
from multiprocessing import Process

def worker_task(index):
    print(f'Process {index} starting')
    print(f'Process ID: {os.getpid()}')
    print(f'Process {index} completed')

if __name__ == '__main__':
    processes = []
    for i in range(5):
        p = Process(target=worker_task, args=(i,))
        p.start()
        processes.append(p)
    
    for p in processes:
        p.join()
    
    print('All processes have finished')

Note that the operating system schedules process execution independently, not necessarily in the order they were started.

The terminate() Method

terminate() immediately stops a child process and releases its resources.

The is_alive() Method

is_alive() returns True if the process is currently executing, otherwise False.

Daemon Processes

A daemon processs terminates automatically when the main process finishes, regardless of its own completion status. The daemon property defaults to False and must be set before starting the proces.

from time import sleep
from multiprocessing import Process

def background_task():
    print('Background task started')
    sleep(3)
    print('Background task completed')

def main_task():
    print('Main task started')
    sleep(1)
    print('Main task completed')

if __name__ == '__main__':
    Process(target=main_task).start()
    daemon_proc = Process(target=background_task)
    daemon_proc.daemon = True
    daemon_proc.start()
    sleep(2)
    print('Main program exiting')

In this example, the daemon process may not print its completion message if the main process exits first.

Lock

A Lock ensures safe access to shared resources by allowing only one process to hold it at a time.

class Lock(object):
    def acquire(self, blocking=True, timeout=-1):
        pass
    
    def release(self):
        pass

• acquire(): Obtains the lock.
• release(): Releases the lock.

Example implementing a ticket booking system:

import json
from multiprocessing import Process, Lock

class TicketAgent(Process):
    def __init__(self, lock):
        super().__init__()
        self.lock = lock
    
    def run(self):
        data = self.read_tickets()
        print(f'Tickets remaining: {data["count"]}')
        if data['count'] > 0:
            self.lock.acquire()
            self.purchase_ticket()
            self.lock.release()
        else:
            print('No tickets available')
    
    def read_tickets(self):
        with open('ticket_db.json', 'r') as f:
            return json.load(f)
    
    def purchase_ticket(self):
        data = self.read_tickets()
        if data['count'] > 0:
            data['count'] -= 1
            with open('ticket_db.json', 'w') as f:
                json.dump(data, f)
            print('Ticket purchased successfully')

if __name__ == '__main__':
    ticket_lock = Lock()
    for i in range(10):
        agent = TicketAgent(ticket_lock)
        agent.start()

Semaphore

A Semaphore is a lock with an internal counter, allowing multiple processes to access a resource up to a defined limit.

class Semaphore(object):
    def __init__(self, value=1):
        pass
    
    def acquire(self, blocking=True, timeout=None):
        pass
    
    def release(self):
        pass

• __init__(value=1): Initializes the semaphore with a maximum count.
• acquire(): Decrements the counter if positive, otherwise blocks.
• release(): Increments the counter.

Example simulating a karaoke room with limited capacity:

from time import sleep
from random import randint
from multiprocessing import Semaphore, Process

def karaoke_room(sem, person_id):
    sem.acquire()
    print(f'Person {person_id} entered the room')
    sleep(randint(1, 5))
    print(f'Person {person_id} left the room')
    sem.release()

if __name__ == '__main__':
    room_semaphore = Semaphore(4)
    for i in range(20):
        Process(target=karaoke_room, args=(room_semaphore, i)).start()

Event

An Event facilitates inter-process communication through signal-based synchronization.

class Event(object):
    def is_set(self):
        return False
    
    def set(self):
        pass
    
    def clear(self):
        pass
    
    def wait(self, timeout=None):
        pass

• is_set(): Returns the current state (default False).
• set(): Sets the flag to True.
• clear(): Resets the flag to False.
• wait(): Blocks until the flag becomes True, with an optional timeout.

Traffic light simulation example:

from time import sleep
from random import choice, randrange
from multiprocessing import Event, Process

def traffic_signal(evt):
    print('\033[1;31mRED light ON\033[0m')
    while True:
        sleep(2)
        if evt.is_set():
            print('\033[1;31mRED light ON\033[0m')
            evt.clear()
        else:
            print('\033[1;32mGREEN light ON\033[0m')
            evt.set()

def regular_vehicle(vehicle_id, evt):
    if not evt.is_set():
        print(f'Vehicle {vehicle_id} waiting')
        evt.wait()
    print(f'Vehicle {vehicle_id} passing')

def emergency_vehicle(vehicle_id, evt):
    if not evt.is_set():
        evt.wait(timeout=0.5)
    print(f'Emergency vehicle {vehicle_id} passing')

if __name__ == '__main__':
    signal_event = Event()
    Process(target=traffic_signal, args=(signal_event,)).start()
    vehicle_types = [regular_vehicle, emergency_vehicle]
    for i in range(15):
        Process(target=choice(vehicle_types), args=(i, signal_event)).start()
        sleep(randrange(0, 3, 2))