マルチタスク処理

今日は同期性、並列性を学びました.
以前学んだことがあり、以前整理したものをリンクし直します.
同期vsパラレル

同期:同時実行のように見える

並列性:実際には複数のタスクを同時に処理

コメントリンク

同期vs非同期

モチベーション:タスクを別のスレッドに入れて待機します.

非同期:次のタスクの実行を待たずに別のスレッドにタスクを配置します.

コメントリンク

I/O Bound vs CPU Bound

I/Oフラクチャ:データ(ファイル)の入出力処理に時間がかかります.

CPUバインド:大量の計算(複雑な数式計算やグラフィック処理など)が必要です.

プロセスとスレッド

プロセス:オペレーティングシステムからリソースを割り当てるオペレーティングユニット、実行中のプログラム

スレッド:プロセスが指定したリソースを使用する実行単位

コメントリンク

プロファイル

profiling:アプリケーション内のリソースが最も集中している場所を特定するためのテクノロジー

Profiler:アプリケーションの実行と各関数の実行に要する時間を検索するプログラム

モジュール:profile、cProfile、line profiler

Scale Up vs Scale Out

Scale-Up:コンピュータのパフォーマンスを最適化する方法

Scale-Out:複数のコンピュータを1台のコンピュータとして使用

コメントリンク

実装-マルチスレッド
https://velog.io/@nameunzz/thread-threading

from threading import *
from time import sleep

Stopped = False

def worker(work, sleep_sec):  # 일꾼 스레드입니다.
    while not Stopped:        # 그만 하라고 할때까지
        print('do ', work)    # 시키는 일을 하고
        sleep(sleep_sec)      # 잠깐 쉽니다.
    print('retired..')        # 언젠가 이 굴레를 벗어나면, 은퇴할 때가 오겠지요?
        
t = Thread(target=worker, args=('Overwork', 3))    # 일꾼 스레드를 하나 생성합니다. 열심히 일하고 3초간 쉽니다.
t.start()    # 일꾼, 이제 일을 해야지?

# 이 코드 블럭을 실행하기 전까지는 일꾼 스레드는 종료하지 않습니다. 
Stopped = True    # 일꾼 일 그만하라고 세팅해 줍시다. 
t.join()          # 일꾼 스레드가 종료할때까지 기다립니다. 
print('worker is gone.')

インプリメンテーション-マルチプロセス

import multiprocessing as mp

def delivery():
    print('delivering...')

p = mp.Process(target=delivery, args=())
p.start()

p = mp.Process(target=delivery, args=())
p.start()      # 프로세스 시작
p.join()       # 실제 종료까지 기다림 (필요시에만 사용)
p.terminate()  # 프로세스 종료

スレッドプール、プロセスプール
スレッドプールを作成すると、タスクごとにスレッドが自動的に割り当てられ、終了します.
プールの作成方法

Queueリソース構造

concurrent.FuturesライブラリのThreadPollExector、ProcessPollExectorクラス

を使用
concurrent.futures
concurrent.先物公式文書

Executorオブジェクト

Futureオブジェクト

ThreadPoolExecutorオブジェクト

from concurrent.futures import ThreadPoolExecutor

class Delivery:
    def run(self):
        print("delivering")
w = Delivery()

with ThreadPoolExecutor() as executor:
    future = executor.submit(w.run)

multiprocessing.Pool.mapにより特定の関数を複数のプロセスにマッピングして並列処理

# multiprocessing.Pool
from multiprocessing import Pool
from os import getpid

def double(i):
    print("I'm process ", getpid())   # pool안에서 이 메소드가 실행될 때 pid확인
    return i * 2

with Pool() as pool:
      result = pool.map(double, [1, 2, 3, 4, 5])
      print(result)

ProcessPoolExecutorオブジェクト

import math, time
import concurrent

PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419]
    
def is_prime(n):
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    print("병렬처리 시작")
    start = time.time()
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))
    end = time.time()
    print("병렬처리 수행 시각", end-start, 's')

    start = time.time()
    for number, prime in zip(PRIMES, map(is_prime, PRIMES)):
        print('%d is prime: %s' % (number, prime))
    end = time.time()
    print("단일처리 수행 시각", end-start, 's')

main()

병렬처리 시작
112272535095293 is prime: True
112582705942171 is prime: True
112272535095293 is prime: True
115280095190773 is prime: True
115797848077099 is prime: True
1099726899285419 is prime: False
병렬처리 수행 시각 0.39751267433166504 s
112272535095293 is prime: True
112582705942171 is prime: True
112272535095293 is prime: True
115280095190773 is prime: True
115797848077099 is prime: True
1099726899285419 is prime: False
단일처리 수행 시각 1.9754359722137451 s