A process is usually the name given to a running program. To launch it, the operating system allocates a certain area of memory. Normally that area is isolated, so processes cannot affect each other's work by changing memory regions that don't belong to them.
In Python, processes are used to parallelize heavy computations: computing hashes, working with matrices, image processing and other similar operations. Creating processes is simple: it's enough to create a Process object and pass it a callable that should run in a separate process.
A basic example
Let's look at a basic example:
Pythonimport time
import os
from multiprocessing import Process, current_process
def foo(text: str) -> None:
time.sleep(3)
print('[foo] process name:', current_process().name)
print('[foo] process pid:', current_process().pid)
print('msg:', text)
if __name__ == '__main__':
print('Main process pid:', os.getpid())
p = Process(target=foo, args=('Hello world', ), name='foo-process')
process_pid = p.pid
process_is_alive = p.is_alive()
print(f'Process with pid: {process_pid} is alive: {process_is_alive}')
print('Process started')
p.start()
process_pid = p.pid
process_is_alive = p.is_alive()
print(f'Process with pid: {process_pid} is alive: {process_is_alive}')
print('Waiting for the process ends...')
p.join()
process_pid = p.pid
process_is_alive = p.is_alive()
print(f'Process with pid: {process_pid} is alive: {process_is_alive}')
print('End program')Console output:
TerminalMain process pid: 24569
Process with pid: None is alive: False
Process started
Process with pid: 24570 is alive: True
Waiting for the process ends...
[foo] process name: foo-process
[foo] process pid: 24570
msg: Hello world
Process with pid: 24570 is alive: False
End programAlso, while the script is running, run the command ps aux | grep <pid_number> in a second terminal window for each pid you obtained. The result will look roughly like this:
Terminal~> ps aux | grep 24569
darksto+ 24569 0.6 0.0 28316 11192 ? S 14:58 0:00 ~/PycharmProjects/edu/async-python-sprint-1/.venv/bin/python ~/.config/JetBrains/PyCharm2022.3/scratches/scratch.py
darksto+ 24578 0.0 0.0 17868 2280 pts/0 S+ 14:58 0:00 grep --color=auto 24569
~> ps aux | grep 24570
darksto+ 24570 0.0 0.0 28316 8968 ? S 14:58 0:00 ~/PycharmProjects/edu/async-python-sprint-1/.venv/bin/python ~/.config/JetBrains/PyCharm2022.3/scratches/scratch.py
darksto+ 24590 0.0 0.0 17868 2312 pts/0 S+ 14:58 0:00 grep --color=auto 24570We can see that at a single moment in time there are two processes with the corresponding pids.
Let's go through the example step by step. First we add the necessary imports and the foo function. Inside the function we print the name and pid of the current process the function runs in.
We start the program's main block. Using the os module we print the pid of the main process.
We create an instance of the Process class, passing it a callable (the foo function) via the target parameter, its arguments and the process name. The is_alive() method lets us determine whether the process is running or not. We start the process with the start() method, and adding the join() method to the main code lets us wait for the child process to finish.
If you need to interrupt a process from the main part of the program without waiting for it to complete, you can use the process's terminate() method.
For serializing and deserializing objects when interacting with processes, the pickle module is used; it only works with primitive types — numbers, strings, dictionaries, functions and so on.
Communication between processes
Since processes have no shared memory, it's very important to organize the delivery of parameters and results of various types for the program to work correctly.
There are various data-exchange models to meet this need. We'll focus on one of them — the queue.
A queue is a data structure of the FIFO type (First In, First Out) — first in, first out. Queues let groups of processes exchange messages.
The "producer-consumer" problem describes two processes: one is the supplier of data or tasks, that is, their producer, while the other receives and handles them, that is, consumes them and removes them from the queue. Together they use a shared buffer for exchanging messages — the queue.
Let's look at an example:
Pythonfrom multiprocessing import Process, get_context
from multiprocessing.queues import Queue
import time
class Producer(Process):
def __init__(self, queue: Queue):
super().__init__()
self.__queue = queue
def run(self):
for msg in range(5):
self.__queue.put(msg)
time.sleep(0.5)
class Consumer(Process):
def __init__(self, queue: Queue, queue_result: Queue):
super().__init__()
self.__queue = queue
self.__queue_result = queue_result
def run(self):
while True:
if self.__queue.empty():
print('Queue is empty. Exit.')
break
else:
item = self.__queue.get()
res = item ** 2
print(res)
self.__queue_result.put(res)
time.sleep(1)
if __name__ == '__main__':
data = []
context = get_context('spawn')
queue = Queue(ctx=context)
queue_result = Queue(ctx=context)
producer_process = Producer(queue)
consumer_process = Consumer(queue, queue_result)
producer_process.start()
consumer_process.start()
producer_process.join()
consumer_process.join()
while not queue_result.empty():
data.append(queue_result.get())
print(data)First we import the necessary libraries. Then we create a Producer class inherited from Process. In the __init__ method we store the queue and override the run method. By default this method is called in the child process after initialization. If it isn't defined, the function passed via the target parameter is called. In the current implementation we simply generate numbers from 0 to 4 and put them into the queue.
Next comes the Consumer class, also derived from the Process parent class. We pass two queues into the class constructor now. One from which we'll read data, and a second into which we'll put the result of the calculations. In the run method we square the value, move the result to the results queue, and also check whether the queue is empty. If it is empty, we stop the consumer. If we don't do this, the program won't exit and will run forever.
Then the main program. We create instances of the queues and of the producer and consumer. We start both processes with the start() method and specify that we wait for them to finish with the join() method.
After the data-processing processes finish, we read the values from the queue_result results queue in a while loop and print them to the screen.
The program's output is shown below.
Terminal0
1
4
9
16
Queue is empty. Exit.
[0, 1, 4, 9, 16]Let's also look separately at the get_context function. A context has to be specified for each queue. There are three options in total:
spawn (used by default) — the main process starts a new Python interpreter process. It is the slowest of the three, but available on all operating systems.
fork — uses the system fork command to create child processes. Faster than spawn, but not available on Windows.
forkserver — a dedicated server process is used, which the program's main process contacts to create a new process. Unnecessary resources are not copied. Available on all unix-based systems.
Process pools
To simplify working with several parallel processes, you can use the Pool class from the multiprocessing library.
The main methods of a Pool instance are:
apply — equivalent to an ordinary function call. It blocks until the result is received and runs in only one worker process of the pool. Syntax: apply(func, *args).
map — syntax: map(func, iter, [chunksize]); it splits all the iterable data iter into the specified number of chunksize fragments, which are supplied to the process pool as separate tasks. It blocks the pool until the final result is obtained.
imap — a lazy version of map. It lets you get results as they complete.
apply_async — a non-blocking version of apply. Syntax: apply_async(func, *args, [callback]). The results can be passed to a callback function for further processing, or you can get the result using the get() method.
Let's look at a few examples.
Pythonimport os
from multiprocessing import Pool, current_process
import time
import sys
import logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(stream=sys.stdout)
handler.setFormatter(logging.Formatter(fmt='%(asctime)s: %(message)s'))
logger.addHandler(handler)
def calc(value: int) -> int:
logger.info(f'calc started in process {current_process().name} with pid {current_process().pid}')
time.sleep(1)
return value**10
if __name__ == '__main__':
logger.info(f'main pid: {os.getpid()}')
with Pool(processes=2) as pool:
result1 = pool.apply(calc, (10,))
logger.info('result1: %s', result1)
result2 = pool.apply(calc, (10,))
logger.info('result2: %s', result2)
result3 = pool.apply(calc, (10,))
logger.info('result3: %s', result3)
result4 = pool.apply(calc, (10,))
logger.info('result4: %s', result4)
print('End program')Sample output:
Terminal2023-01-25 18:21:40,027: main pid: 34098
2023-01-25 18:21:40,041: calc started in process ForkPoolWorker-1 with pid 34099
2023-01-25 18:21:41,044: result1: 10000000000
2023-01-25 18:21:41,047: calc started in process ForkPoolWorker-2 with pid 34100
2023-01-25 18:21:42,050: result2: 10000000000
2023-01-25 18:21:42,050: calc started in process ForkPoolWorker-1 with pid 34099
2023-01-25 18:21:43,052: result3: 10000000000
2023-01-25 18:21:43,053: calc started in process ForkPoolWorker-2 with pid 34100
2023-01-25 18:21:44,056: result4: 10000000000
End programLet's look at the example with apply. At the start of the script we add the necessary imports and a basic logger setup to display timestamps. We define the base function calc, which raises the received number to the 10th power.
Then in the main block, using the with context manager, we define a process pool with a maximum size of two. And we start calling the calc function in a separate process.
In the console you can see that two child processes with pids 34100 and 34099 take part, but they don't run in parallel: the delay between each result printed is one second. There's no performance gain in this case.
Let's look at the example with the map function:
Pythonimport os
from multiprocessing import Pool, current_process
import time
import sys
import logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(stream=sys.stdout)
handler.setFormatter(logging.Formatter(fmt='%(asctime)s: %(message)s'))
logger.addHandler(handler)
def calc(value: int) -> int:
logger.info(f'calc started in process {current_process().name} with pid {current_process().pid}')
time.sleep(1)
return value**10
if __name__ == '__main__':
logger.info(f'main pid: {os.getpid()}')
l = [1, 2, 5, 10]
with Pool(processes=2) as pool:
results = pool.map(calc, l)
print(results)
print('End program')Console output:
Terminal2023-01-25 18:28:27,135: main pid: 34311
2023-01-25 18:28:27,148: calc started in process ForkPoolWorker-2 with pid 34313
2023-01-25 18:28:27,148: calc started in process ForkPoolWorker-1 with pid 34312
2023-01-25 18:28:28,150: calc started in process ForkPoolWorker-1 with pid 34312
2023-01-25 18:28:28,150: calc started in process ForkPoolWorker-2 with pid 34313
[1, 1024, 9765625, 10000000000]
End programIn the main program code we prepare in advance a list l with the numbers we'll compute for, and start calling the calc function with the map method.
In the console we see that two child processes started at the same time, then two more. The result is presented as a list of the calc function's output data. Note also that the list is ordered according to the input data in the list l.
Let's try the option using apply_async:
Pythonimport os
from multiprocessing import Pool, current_process
import time
import sys
import logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(stream=sys.stdout)
handler.setFormatter(logging.Formatter(fmt='%(asctime)s: %(message)s'))
logger.addHandler(handler)
def calc(value: int) -> int:
logger.info(f'calc started in process {current_process().name} with pid {current_process().pid}')
time.sleep(1)
return value**10
def callback(result):
logger.info(f'Callback result: {result}, process pid: {current_process().pid}')
if __name__ == '__main__':
logger.info(f'main pid: {os.getpid()}')
with Pool(processes=2) as pool:
result1 = pool.apply_async(calc, (1,), callback=callback)
logger.info('result1: %s', result1)
result2 = pool.apply_async(calc, (2,), callback=callback)
logger.info('result2: %s', result2)
result3 = pool.apply_async(calc, (3,), callback=callback)
logger.info('result3: %s', result3)
result4 = pool.apply_async(calc, (4,), callback=callback)
logger.info('result4: %s', result4)
pool.close()
pool.join()
print('End program')Compared with the previous examples, a callback function has been added that will be called when a process finishes. In the main code the numbers 1, 2, 3, 4 are passed sequentially to the calc function, and a callback is passed for each call as well. At the end of the context manager two statements are added — close() and join(). close() prevents new tasks from being submitted to the pool. Once all tasks are done, the worker processes will finish, and join() waits for the worker processes to complete. Without these commands, the main program would simply finish its work without waiting for the processes to run and the callback function to be called.
Console output:
Terminal2023-01-25 18:33:03,614: main pid: 34515
2023-01-25 18:33:03,625: result1: <multiprocessing.pool.ApplyResult object at 0x7fa55b754100>
2023-01-25 18:33:03,626: result2: <multiprocessing.pool.ApplyResult object at 0x7fa55b754130>
2023-01-25 18:33:03,626: result3: <multiprocessing.pool.ApplyResult object at 0x7fa55b754340>
2023-01-25 18:33:03,626: result4: <multiprocessing.pool.ApplyResult object at 0x7fa55b754460>
2023-01-25 18:33:03,626: calc started in process ForkPoolWorker-1 with pid 34516
2023-01-25 18:33:03,626: calc started in process ForkPoolWorker-2 with pid 34517
2023-01-25 18:33:04,628: calc started in process ForkPoolWorker-2 with pid 34517
2023-01-25 18:33:04,628: calc started in process ForkPoolWorker-1 with pid 34516
2023-01-25 18:33:04,628: Callback result: 1024, process pid: 34515
2023-01-25 18:33:04,628: Callback result: 1, process pid: 34515
2023-01-25 18:33:05,630: Callback result: 1048576, process pid: 34515
2023-01-25 18:33:05,630: Callback result: 59049, process pid: 34515
End programHere we can see that all the tasks for the processes were created at the same time. The apply_async function returned an object of the multiprocessing.pool.ApplyResult class. Then the first two processes with pids 34516 and 34517 started, and then the second two.
After that the callback function was called in the main process with pid 34515, but the results got mixed up. First the output for the statement that was second, then the output for the first statement, and so on. Unlike the map function, apply_async doesn't guarantee the correct order in which results are received.
Conclusion
Let's review once more the main advantages and disadvantages of processes in Python.
Advantages: the ability to use all available CPU cores; the GIL imposes no restrictions; each process has its own memory area; there's the ability to interrupt a process; suitable for heavy computations.
Disadvantages: a more complex interface and higher overhead; high memory consumption.
