HPDS-03: Pendahuluan Pemrograman Parallel di Python

"Always code as if the person who ends up maintaining your code will be a violent psychopath who knows where you live." --unknown

• Watch The Video Here: https://youtu.be/Fp73A_2XNuA
• The code is available here

High Performance Data Science (HPDS)

https://taudata.blogspot.com

HPDS-03: Pendahuluan Pemrograman Parallel di Python

https://taudata.blogspot.com/2022/04/hpds-03.html

(C) Taufik Sutanto

Concurrent vs Parallel¶

• Pada dasarnya mirip. Namun bayangkan concurrent sebagai thread dan parallel sebagai pekerjaan yang dilakukan bersamaan di beberapa processor.

• image source: https://medium.com/@sanju.skm/parallel-programming-vs-concurrent-programming-f993d3f9ceea

Review¶

image source: https://code.tutsplus.com/articles/introduction-to-parallel-and-concurrent-programming-in-python--cms-28612

Berikut ini contoh yang sangat baik untuk membedakan ke-2-nya:¶

Jangan lupa seperti biasa jalankan di Terminal (Bukan Jupyter Notebook)¶

In [ ]:

# -*- coding: utf-8 -*-
import os
import time
import threading
import multiprocessing
 
NUM_WORKERS = 4
 
def only_sleep():
    """ Do nothing, wait for a timer to expire """
    print("PID: %s, Process Name: %s, Thread Name: %s" % (
        os.getpid(),
        multiprocessing.current_process().name,
        threading.current_thread().name)
    )
    time.sleep(1)
 
 
def crunch_numbers():
    """ Do some computations """
    print("PID: %s, Process Name: %s, Thread Name: %s" % (
        os.getpid(),
        multiprocessing.current_process().name,
        threading.current_thread().name)
    )
    x = 0
    while x < 10000000:
        x += 1

if __name__ == '__main__':
        
    ## Run tasks serially
    start_time = time.time()
    for _ in range(NUM_WORKERS):
        only_sleep()
    end_time = time.time()
     
    print("Serial time=", end_time - start_time)
     
    # Run tasks using threads
    start_time = time.time()
    threads = [threading.Thread(target=only_sleep) for _ in range(NUM_WORKERS)]
    [thread.start() for thread in threads]
    [thread.join() for thread in threads]
    end_time = time.time()
     
    print("Threads time=", end_time - start_time)
     
    # Run tasks using processes
    start_time = time.time()
    processes = [multiprocessing.Process(target=only_sleep()) for _ in range(NUM_WORKERS)]
    [process.start() for process in processes]
    [process.join() for process in processes]
    end_time = time.time()
     
    print("Parallel time=", end_time - start_time)

Pool Class¶

• Pengerjaan urut input

Contoh dibawah ini (not necessarily) lebih cepat dari serial code. Mengapa?¶

In [ ]:

import multiprocessing as mp

def f(x):
    return x*x

if __name__ == '__main__':
    print('Number of currently available processor = ', mp.cpu_count())
    input_ = [1, 2, 3, 4, 5, 7, 9, 10]
    print('input = ', input_)
    with mp.Pool(5) as p:
        print(p.map(f, input_))

Contoh Pool yang lebih cocok.¶

In [ ]:

import multiprocessing
import numpy as np
import time

def pungsi(N):
    s = 0.0
    for i in range(1,N):
        s += np.log(i)
    return s

if __name__ == '__main__':
    inputs = [10**6] * 20
    print('Serial/Sequential Programming biasa:')
    mulai =  time.time()
    outputs = [pungsi(x) for x in inputs]
    akhir  = time.time()
    print("Rata-rata Output: {}".format(np.mean(outputs)))
    print("Waktu Serial: {}".format(akhir-mulai))
    
    print('Parallel Programming:')
    mulai =  time.time()
    pool = multiprocessing.Pool(processes=8)
    outputs = pool.map(pungsi, inputs)
    akhir  = time.time()
    #print("Input: {}".format(inputs))
    print("Rata-rata Output: {}".format(np.mean(outputs)))
    print("Waktu parallel: {}".format(akhir-mulai))

map Async¶

In [ ]:

import multiprocessing as mp

def square(x):
    return x * x

if __name__ == '__main__':
    inputs = [0,1,2,3,4,5,6,7,8]
    
    print('Sync Parallel Processing')
    pool = mp.Pool()
    outputs = pool.map(square, inputs)
    print("Input: {}".format(inputs))
    print("Output: {} \n".format(outputs))
    pool.close(); del pool
    
    print('Async Parallel Processing')
    pool = mp.Pool()
    outputs_async = pool.map_async(square, inputs)
    outputs = outputs_async.get()
    print("Input: {}".format(inputs))
    print("Output: {}".format(outputs))

Ingat kita bisa assign fungsi apapun ke masing-masing processor secara manual jika dibutuhkan¶

In [ ]:

import multiprocessing
import os
import time
import threading

class ProsesA(multiprocessing.Process):
    def __init__(self, id):
        super(ProsesA, self).__init__()
        self.id = id

    def run(self):
        time.sleep(1)
        print("PID: %s, Process ID: %s, Process Name: %s, Thread Name: %s" % (
        os.getpid(), self.id,
        multiprocessing.current_process().name,
        threading.current_thread().name))
        
class ProsesB(multiprocessing.Process):
    def __init__(self, id):
        super(ProsesB, self).__init__()
        self.id = id

    def run(self):
        time.sleep(1)
        print("PID: %s, Process ID: %s, Process Name: %s, Thread Name: %s" % (
        os.getpid(), self.id,
        multiprocessing.current_process().name,
        threading.current_thread().name))

if __name__ == '__main__':
    p1 = ProsesA(0)
    p1.start()
    p2 = ProsesB(1)
    p2.start()
    p1.join(); p2.join()

Parallel Programming pada Fungsi Multivariate: StarMap¶

In [ ]:

import multiprocessing as mp

def f_sum(a, b):
    return a + b

if __name__ == '__main__':
    process_pool = mp.Pool(4)
    data = [(1, 1), (2, 1), (3, 1), (6, 9)]
    output = process_pool.starmap(f_sum, data)
    print("input = ", data)
    print("output = ", output)

End of Module¶

---