Multiprocessing¶

In [4]:
%%writefile mp01.py

import multiprocessing as mp
import random
import os
from time import sleep

print(f'Processors: {mp.cpu_count()}')
print(f'Process name: {mp.current_process().name}')
print(f'PID: {mp.current_process().pid}')
print(f'PPID: {os.getppid()}')

def ftion(x):
    print(f'Process {mp.current_process().name} has started...')
    sleep(random.randrange(5, 10))
    return x**2

if __name__ == '__main__': #run only in main process
    p = mp.Process(target=ftion, args=(3,), name='PDS')
    p.start()
    p.join()

Overwriting mp01.py
In [5]:
!python mp01.py

Processors: 12
Process name: PDS
PID: 2064
PPID: 16064
Process PDS has started...
Processors: 12
Process name: MainProcess
PID: 16064
PPID: 17020
In [6]:
#not working directly in Jupyter Notebook
import multiprocessing as mp
import random
import os
from time import sleep

print(f'Processors: {mp.cpu_count()}')
print(f'Process name: {mp.current_process().name}')
print(f'PID: {mp.current_process().pid}')
print(f'PPID: {os.getppid()}')

def ftion(x):
    print(f'Process {mp.current_process().name} has started...')
    sleep(random.randrange(5, 10))
    return x**2

if __name__ == '__main__': #run only in main process
    p = mp.Process(target=ftion, args=(3,), name='PDS')
    p.start()
    p.join()

Processors: 12
Process name: MainProcess
PID: 4184
PPID: 13032

Run multiple processes¶

In [7]:
%%writefile mp02.py
#not working directly in Jupyter Notebook
import multiprocessing as mp
import random
import os
from time import sleep

print(f'Processors: {mp.cpu_count()}')
print(f'Process name: {mp.current_process().name}')
print(f'PID: {mp.current_process().pid}')
print(f'PPID: {os.getppid()}')

def ftion(x):
    print(f'Process {mp.current_process().name} has started...')
    sleep(random.randrange(5, 10))
    print(f'Process {mp.current_process().name} has finished...')
    return x**2

if __name__ == '__main__': #run only in main process
    ps = [mp.Process(target=ftion, args=(3,), name=f'PDS{index}') for index in range(5)]
    for p in ps:
        p.start()
    for p in ps:        
        p.join()

Writing mp02.py
In [8]:
!python mp02.py

Processors: 12
Process name: PDS0
PID: 18120
PPID: 10756
Process PDS0 has started...
Process PDS0 has finished...
Processors: 12
Process name: PDS3
PID: 3956
PPID: 10756
Process PDS3 has started...
Process PDS3 has finished...
Processors: 12
Process name: PDS1
PID: 10000
PPID: 10756
Process PDS1 has started...
Process PDS1 has finished...
Processors: 12
Process name: PDS4
PID: 17532
PPID: 10756
Process PDS4 has started...
Process PDS4 has finished...
Processors: 12
Process name: PDS2
PID: 10312
PPID: 10756
Process PDS2 has started...
Process PDS2 has finished...
Processors: 12
Process name: MainProcess
PID: 10756
PPID: 10348

In CLI starts with MainProcess output:

image.png

No global variable between processes ...

In [10]:
%%writefile mp03.py
#not working directly in Jupyter Notebook
import multiprocessing as mp
import random
import os
from time import sleep

print(f'Processors: {mp.cpu_count()}')
print(f'Process name: {mp.current_process().name}')
print(f'PID: {mp.current_process().pid}')
print(f'PPID: {os.getppid()}')

def ftion(x):
    global results
    print(f'Process {mp.current_process().name} has started...')
    sleep(random.randrange(5, 10))
    results[x] = x**2
    print(f'Process {mp.current_process().name} has finished with results: ', results)
    return x**2

results = {}

if __name__ == '__main__': #run only in main process
    ps = [mp.Process(target=ftion, args=(index+1,), name=f'PDS{index}') for index in range(5)]
    for p in ps:
        p.start()
    for p in ps:        
        p.join()

print(f'Process {mp.current_process().name} has stopped with results: ', results)

Overwriting mp03.py

image.png

Stopped before started ...

Pass data from MainProcess to all subprocesses ...¶

In [ ]:
q = mp.Queue() # create Queue
q.put(...)
q.get()

TASK: Create 5 subprocesses with parameter Queue, start all of them, pass values 2..6. Every subprocess get element and compute element**2 ...

In [13]:
%%writefile mp04.py
#not working directly in Jupyter Notebook
import multiprocessing as mp
import random
import os
from time import sleep

print(f'Processors: {mp.cpu_count()}')
print(f'Process name: {mp.current_process().name}')
print(f'PID: {mp.current_process().pid}')
print(f'PPID: {os.getppid()}')

def ftion(que):
    print(f'Process {mp.current_process().name} has started...')
    sleep(random.randrange(5, 10))
    x = que.get()
    print(f'Process {mp.current_process().name} computed {x**2} for {x}.')
    print(f'Process {mp.current_process().name} has stopped.')

if __name__ == '__main__': #run only in main process
    q = mp.Queue() # create Queue
    ps = [mp.Process(target=ftion, args=(q,), name=f'PDS{index}') for index in range(5)]
    for p in ps:
        p.start()
    for index in range(2, 7):
        q.put(index)
    for p in ps:        
        p.join()

Overwriting mp04.py

image.png

TASK2: Consume several elements in ftion (for loop)

count as parameter for ftion

In [14]:
%%writefile mp05.py
#not working directly in Jupyter Notebook
import multiprocessing as mp
import random
import os
from time import sleep

print(f'Processors: {mp.cpu_count()}')
print(f'Process name: {mp.current_process().name}')
print(f'PID: {mp.current_process().pid}')
print(f'PPID: {os.getppid()}')

def ftion(que, count):
    print(f'Process {mp.current_process().name} has started...')
    for _ in range(count):
        sleep(random.randrange(5, 10))
        x = que.get()
        print(f'Process {mp.current_process().name} computed {x**2} for {x}.')
    print(f'Process {mp.current_process().name} has stopped.')

if __name__ == '__main__': #run only in main process
    q = mp.Queue() # create Queue
    count = 7
    ps = [mp.Process(target=ftion, args=(q, count), name=f'PDS{index}') for index in range(5)]
    for p in ps:
        p.start()
    for index in range(count*len(ps)):
        q.put(10+index)
    for p in ps:        
        p.join()

Writing mp05.py

image.png

In [15]:
!python mp05.py

Processors: 12
Process name: PDS2
PID: 10516
PPID: 12636
Process PDS2 has started...
Process PDS2 computed 100 for 10.
Process PDS2 computed 256 for 16.
Process PDS2 computed 441 for 21.
Process PDS2 computed 576 for 24.
Process PDS2 computed 900 for 30.
Process PDS2 computed 1089 for 33.
Process PDS2 computed 1369 for 37.
Process PDS2 has stopped.
Processors: 12
Process name: PDS3
PID: 4144
PPID: 12636
Process PDS3 has started...
Process PDS3 computed 121 for 11.
Process PDS3 computed 225 for 15.
Process PDS3 computed 400 for 20.
Process PDS3 computed 676 for 26.
Process PDS3 computed 961 for 31.
Process PDS3 computed 1156 for 34.
Process PDS3 computed 1521 for 39.
Process PDS3 has stopped.
Processors: 12
Process name: PDS0
PID: 2840
PPID: 12636
Process PDS0 has started...
Process PDS0 computed 169 for 13.
Process PDS0 computed 361 for 19.
Process PDS0 computed 625 for 25.
Process PDS0 computed 784 for 28.
Process PDS0 computed 1024 for 32.
Process PDS0 computed 1444 for 38.
Process PDS0 computed 1764 for 42.
Process PDS0 has stopped.
Processors: 12
Process name: PDS1
PID: 3752
PPID: 12636
Process PDS1 has started...
Process PDS1 computed 196 for 14.
Process PDS1 computed 324 for 18.
Process PDS1 computed 484 for 22.
Process PDS1 computed 729 for 27.
Process PDS1 computed 1225 for 35.
Process PDS1 computed 1600 for 40.
Process PDS1 computed 1849 for 43.
Process PDS1 has stopped.
Processors: 12
Process name: PDS4
PID: 4056
PPID: 12636
Process PDS4 has started...
Process PDS4 computed 144 for 12.
Process PDS4 computed 289 for 17.
Process PDS4 computed 529 for 23.
Process PDS4 computed 841 for 29.
Process PDS4 computed 1296 for 36.
Process PDS4 computed 1681 for 41.
Process PDS4 computed 1936 for 44.
Process PDS4 has stopped.
Processors: 12
Process name: MainProcess
PID: 12636
PPID: 8116

Pool¶

https://docs.python.org/3/library/multiprocessing.html#module-multiprocessing.pool

In [18]:
%%writefile mp06.py

import multiprocessing as mp
import os
from time import sleep
from random import randrange

print(f'Process name: {mp.current_process().name}')

def ftion(x):
    print(f'Process {mp.current_process().name} started for {x}.')
    sleep(randrange(1, 5))
    return x**2

if __name__ == "__main__": #only in main process
    with mp.Pool(5) as p: #execute in parallel, None = all CPUs
        print(p.map(ftion, [x for x in range(15)])) #map ftion in parallel

Overwriting mp06.py
In [19]:
!python mp06.py

Process name: SpawnPoolWorker-3
Process SpawnPoolWorker-3 started for 4.
Process SpawnPoolWorker-3 started for 6.
Process SpawnPoolWorker-3 started for 7.
Process name: SpawnPoolWorker-1
Process SpawnPoolWorker-1 started for 0.
Process SpawnPoolWorker-1 started for 8.
Process SpawnPoolWorker-1 started for 13.
Process name: SpawnPoolWorker-5
Process SpawnPoolWorker-5 started for 2.
Process SpawnPoolWorker-5 started for 5.
Process SpawnPoolWorker-5 started for 11.
Process name: SpawnPoolWorker-4
Process SpawnPoolWorker-4 started for 3.
Process SpawnPoolWorker-4 started for 10.
Process SpawnPoolWorker-4 started for 12.
Process name: SpawnPoolWorker-2
Process SpawnPoolWorker-2 started for 1.
Process SpawnPoolWorker-2 started for 9.
Process SpawnPoolWorker-2 started for 14.
Process name: MainProcess
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196]

image.png

SharedMemory¶

https://docs.python.org/3/library/multiprocessing.shared_memory.html

In [21]:
from multiprocessing import shared_memory
shm = shared_memory.SharedMemory(create=True, size=10)
In [22]:
shm
Out[22]:
SharedMemory('wnsm_f3ec549e', size=10)
In [26]:
list(shm.buf[0:3])
Out[26]:
[0, 0, 0]
In [27]:
shm.buf[1] = 7
In [28]:
list(shm.buf[0:3])
Out[28]:
[0, 7, 0]

image.png

In [29]:
list(shm.buf[0:3])
Out[29]:
[0, 5, 0]

HW: Compare execution with and without SharedMemory

use 100_000 rows with 3 columns
count not a numbers
In [30]:
import pandas as pd
import numpy as np

from datetime import datetime

data = [
    [datetime.today(), 1, 'PDS1'],
    [datetime.today(), np.nan, 'NUMPY']
]*1_000_000

df = pd.DataFrame(data, columns=['date', 'val', 'description'])

df
Out[30]:
date val description
0 2026-02-25 15:57:35.320453 1.0 PDS1
1 2026-02-25 15:57:35.320453 NaN NUMPY
2 2026-02-25 15:57:35.320453 1.0 PDS1
3 2026-02-25 15:57:35.320453 NaN NUMPY
4 2026-02-25 15:57:35.320453 1.0 PDS1
... ... ... ...
1999995 2026-02-25 15:57:35.320453 NaN NUMPY
1999996 2026-02-25 15:57:35.320453 1.0 PDS1
1999997 2026-02-25 15:57:35.320453 NaN NUMPY
1999998 2026-02-25 15:57:35.320453 1.0 PDS1
1999999 2026-02-25 15:57:35.320453 NaN NUMPY

2000000 rows × 3 columns

In [31]:
np_array = df.to_records(index=False)
In [32]:
np_array.nbytes
Out[32]:
48000000

Run several processes which computes nansum:

copying data as argument
using shared memory

compare execution time (BONUS: memory usage)

In [33]:
np.nansum(np_array.val)
Out[33]:
1000000.0
In [34]:
import time

start = time.time()
sleep(3)
time.time()-start #in seconds
Out[34]:
3.001086473464966

BONUS - MEMORY¶

In [35]:
#Memory

import tracemalloc

tracemalloc.start()

print(tracemalloc.get_traced_memory()) #current, peak
tracemalloc.stop()

(930, 11635)
In [38]:
#Memory

import tracemalloc

tracemalloc.start()
data2 = [
    [datetime.today(), 1, 'PDS1'],
    [datetime.today(), np.nan, 'NUMPY']
]*1_000_000
print(tracemalloc.get_traced_memory()) #current, peak
tracemalloc.stop()

(16001058, 16012170)
In [ ]: