<jupyter_start><jupyter_text>Enhancing performance using Parallel computing- Start by profiling a serial program to identify bottlenecks- Are there opportunities for parallelism?    - Loops    - Can data be split in parallel?    - Pipeline of stages?- Choose an approach and pattern- Map to the parallel environment    - Multicore    - GPU    - Multinode Embarrassingly parallel programsMany problems are embarrassingly parallel and can be easily decomposed into independent tasks or data sets. Here are several examples:- Monte Carlo integration- Bootstrap for calculating statistics- Fitting the same model on multiple data sets- Running simulation with different settingsThere are many parallel design patterns. The simplest way is to divide it into - Data parallelism means that the data is distributed across processes (e.g., MPI, Hadoop, Spark)- Task parallelism means that tasks (functions) are distributed across processes, and different units of work (data) are sent to each task (e.g., multithreading, multiprocessing, single GPU programming).  Using multiple cores with `multiprocessing`The standard implementation of Python uses a Global Interpreter Lock (GIL). This means that only one thread can be run at any one time, and multiple threads work by time-slicing. Hence multi-threaded code with lots of latency (waiting for the network to respond, I/O) can result in speed-ups, but multi-threaded code which is computationally intensive will not see any speed-up. For numerically intensive code, parallel code needs to be run in separate processes to see speed-ups.- Process    - Heavyweight    - Have a separate memory space    - Large cost for communications- Thread    - Lightweight    - Share the same memory space    - Small cost for communications First we see how to split the computation into pieces using a loop.<jupyter_code>from multiprocessing import (Pool, Process, cpu_count)
import multiprocessing as mp
import time
from numba import njit
from math import sqrt
import numpy as np
cpu_count()
[sqrt(i ** 2) for i in range(10)]
%%writefile defs.py
from math import sqrt

def sqrt_list(i):
    return sqrt(i**2)
import defs
with mp.Pool(processes=cpu_count()) as pool:
    res = pool.map(defs.sqrt_list, [i for i in range(10)])
res
res<jupyter_output><empty_output><jupyter_text>Functions with multiple arguments<jupyter_code>%%writefile defs2.py

def f(a, b, c):
    return a + b + c
import defs2
x = np.arange(24)
x_s = np.array_split(x, x.shape[0]//3)
x_s
with mp.Pool(processes=cpu_count()) as pool:
    res = pool.starmap(defs2.f, x_s)
res<jupyter_output><empty_output><jupyter_text>MoteCarlo<jupyter_code>%%writefile defs3.py
import numpy as np

def monte_carlo_pi(n):
    x = np.random.uniform(-1, 1, (n,2))
    return 4*np.sum((x**2).sum(1) < 1)/n
import defs3
%%timeit  
global res
res = [defs3.monte_carlo_pi(int(1e7)) for i in range(10)]
%%timeit
global re2
with mp.Pool(processes=cpu_count()) as pool:
    res2 = pool.map(defs3.monte_carlo_pi, [int(1e7) for i in range(10)])
%%timeit
global re2
with mp.Pool(processes=40) as pool:
    res2 = pool.map(defs3.monte_carlo_pi, [int(1e7) for i in range(10)])<jupyter_output>1 loop, best of 5: 4.59 s per loop<jupyter_text>- Check map_async vs map https://discuss.python.org/t/differences-between-pool-map-pool-apply-and-pool-apply-async/6575/2   Using `Threading` - Check threading vs multiprocessing https://blog.floydhub.com/multiprocessing-vs-threading-in-python-what-every-data-scientist-needs-to-know/<jupyter_code>from multiprocessing.dummy import Pool as ThreadPool
import requests
def func(number):
    url = 'http://example.com/'
    for i in range(number):
        response = requests.get(url)
        print(len(response.text))
with ThreadPool(processes=4) as pool:
    res2 = pool.map(func, [3,3,3,3])<jupyter_output>1256
1256
1256
1256
12561256

1256
1256
1256
1256
1256
1256<jupyter_text>Using `Joblib` `joblib` provides parallel processing using a comprehension syntax<jupyter_code>from joblib import Parallel, delayed
from functools import partial
from tqdm import tqdm
tqdm = partial(tqdm, position=0, leave=True)
Parallel(n_jobs=4)(delayed(sqrt)(i ** 2) for i in tqdm(range(10)))
x = np.arange(24)
x_s = np.array_split(x, x.shape[0]//3)
res = Parallel(n_jobs=4)(delayed(defs2.f)(x_s[i][0], x_s[i][1], x_s[i][2]) for i in tqdm(range(len(x_s))))
res<jupyter_output>100%|██████████| 8/8 [00:00<00:00, 1286.40it/s]<jupyter_text>Using thread<jupyter_code>res = Parallel(n_jobs=4, prefer="threads")(delayed(defs2.f)(x_s[i][0], x_s[i][1], x_s[i][2]) for i in tqdm(range(len(x_s))))
res<jupyter_output>100%|██████████| 8/8 [00:00<00:00, 985.68it/s]<jupyter_text>MonteCarlo<jupyter_code>%%timeit
res = Parallel(n_jobs=4, prefer="threads")(delayed(defs3.monte_carlo_pi)(int(1e7)) for i in tqdm(range(10)))
%%timeit
res = Parallel(n_jobs=40, prefer="threads")(delayed(defs3.monte_carlo_pi)(int(1e7)) for i in tqdm(range(10)))
%%timeit
res = Parallel(n_jobs=2)(delayed(defs3.monte_carlo_pi)(int(1e7)) for i in tqdm(range(10)))
%%timeit
res = Parallel(n_jobs=40)(delayed(defs3.monte_carlo_pi)(int(1e7)) for i in tqdm(range(10)))<jupyter_output>100%|██████████| 10/10 [00:04<00:00,  2.41it/s]
100%|██████████| 10/10 [00:00<00:00, 1759.72it/s]
100%|██████████| 10/10 [00:00<00:00, 3135.22it/s]
100%|██████████| 10/10 [00:00<00:00, 1396.70it/s]
100%|██████████| 10/10 [00:00<00:00, 2373.28it/s]
100%|██████████| 10/10 [00:00<00:00, 4135.17it/s]<jupyter_text>- Scientific Python libraries such as numpy, `scipy`, `pandas` and `scikit-learn` often release the GIL in performance-critical code paths. It is therefore advised to always measure the speed of thread-based parallelism and use it when the GIL does not limit the scalability.- The thread-based approach can also ease debugging- Writing to shared memory requires careful coordination of processes, and many control and communication concepts are implemented in the multiprocessing library for this purpose, including semaphores, locks, barriers, etc. - Check share memory and reduction at https://milliams.com/courses/parallel_python/- Check Numba parallel features  Laboratories<jupyter_code># Baseline
def cdist(xs, ys):
    """Returns pairwise distance between row vectors in xs and ys.
    
    xs has shape (m, p)
    ys has shape (n, p)
    
    Return value has shape (m, n)    
    """
    
    m, p = xs.shape
    n, p = ys.shape
    
    res = np.empty((m, n))
    for i in range(m):
        for j in range(n):
            res[i, j] = np.sqrt(np.sum((ys[j] - xs[i])**2))
    return res
xs = np.arange(6).reshape(3,2).astype('float')
ys = np.arange(4).reshape(2,2).astype('float')
zs = cdist(xs, ys)
cdist(xs, ys)
np.split(xs, 3, 0)
res = np.concatenate([cdist(x, ys) for x in np.split(xs, 3, 0)])
res
m = 1000
n = 1000
p = 100

X = np.random.random((m, p))
Y = np.random.random((n, p))
%%timeit
Z = cdist(X, Y)<jupyter_output>6.84 s ± 263 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)<jupyter_text>Using `multiprocessing`<jupyter_code>from multiprocessing import Pool
%%writefile defs4.py
import numpy as np

def cdist(xs, ys):
    """Returns pairwise distance between row vectors in xs and ys.
    
    xs has shape (m, p)
    ys has shape (n, p)
    
    Return value has shape (m, n)    
    """
    
    m, p = xs.shape
    n, p = ys.shape
    
    res = np.empty((m, n))
    for i in range(m):
        for j in range(n):
            res[i, j] = np.sqrt(np.sum((ys[j] - xs[i])**2))
    return res
import defs4
%%timeit
with Pool(processes=4) as p:
    Z1 = p.starmap(defs4.cdist, [(X_, Y) for X_ in np.split(X, 100, 0)])
    Z1 = np.concatenate(Z1)
Z = cdist(X, Y)

with Pool(processes=4) as p:
    Z1 = p.starmap(defs4.cdist, [(X_, Y) for X_ in np.split(X, 100, 0)])
    Z1 = np.concatenate(Z1)

np.allclose(Z, Z1)<jupyter_output><empty_output><jupyter_text>Using threadsNote that there is no gain with using multiple threads for computationally intensive tasks because of the GIL.<jupyter_code>%%timeit
with ThreadPool(processes=4) as pool:
    Z2 = list(pool.starmap(defs4.cdist, [(X_, Y) for X_ in np.split(X, 100, 0)]))
    Z2 = np.concatenate(Z2)<jupyter_output>7.76 s ± 221 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)<jupyter_text>Check<jupyter_code>with ThreadPool(processes=4) as pool:
    Z2 = list(pool.starmap(defs4.cdist, [(X_, Y) for X_ in np.split(X, 100, 0)]))
    Z2 = np.concatenate(Z2)

np.allclose(Z, Z2)<jupyter_output><empty_output><jupyter_text>Exercise 3: - Calculate the pairwise euclidean distance between two matrices X and Y using `joblib` and report the speedup (or speed down) over baseline<jupyter_code>Z3 = Parallel(n_jobs=4)(delayed(cdist)(X_, Y) for X_ in tqdm(np.split(X, 100, 0)))
Z3 = np.concatenate(Z3)
np.allclose(Z, Z3)
%%timeit
## Solution here
Z3 = Parallel(n_jobs=4)(delayed(cdist)(X_, Y) for X_ in tqdm(np.split(X, 100, 0)))
Z3 = np.concatenate(Z3)<jupyter_output>100%|████████████████████████████████████████████████████████████████████████████████| 100/100 [00:04<00:00, 21.34it/s]
100%|████████████████████████████████████████████████████████████████████████████████| 100/100 [00:03<00:00, 30.60it/s]
100%|████████████████████████████████████████████████████████████████████████████████| 100/100 [00:03<00:00, 30.55it/s]
100%|████████████████████████████████████████████████████████████████████████████████| 100/100 [00:03<00:00, 31.16it/s]
100%|████████████████████████████████████████████████████████████████████████████████| 100/100 [00:03<00:00, 30.67it/s]
100%|████████████████████████████████████████████████████████████████████████████████| 100/100 [00:03<00:00, 30.93it/s]
100%|████████████████████████████████████████████████████████████████████████████████| 100/100 [00:03<00:00, 28.30it/s]
100%|████████████████████████████████████████████████████████████████████████████████| 100/100 [00:03<00:00, 30.71it/s]
{'fraction_non_alphanumeric': 0.19514178270149404, 'fraction_numerical': 0.058644171155605246, 'mean_word_length': 4.992691839220463, 'pattern_counts': {'":': 0, '<': 58, '<?xml version=': 0, '>': 41, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '227195', 'license': 'MIT', 'n_tokens_mistral': 4516, 'n_tokens_neox': 3780, 'n_words': 1002, 'path': 'static_files/presentations/MultiCore.ipynb', 'repo_name': 'phonchi/nsysu-math604', 'size': '9839'}