파이썬으로 간편하게 Multiprocessing 의 시작하기!

CPU의 코어는 우리의 두뇌와 같은 역할을 하는데요, 요즘 웬만한 CPU는 멀티코어를 가지고 있습니다. 즉 동시에 처리할 수 있는 구조라고 볼 수 있는데요. CPU 의 코어가 만약 1개라면 멀티프로세싱 자체는 시분할로 동시에 처리되는 것 처럼 구현할 수도 있습니다. 이렇게 말씀드리는 이유는 뭐냐면, 코어의 성능자체 역시 천차 만별이기 때문인데요. 오늘 저는 파이썬으로 멀티 프로세싱을 할 수 있는 방법에 대해설 설명해보도록 하겠습니다.

Global Interpreter Lock

https://en.wikipedia.org/wiki/Global_interpreter_lock

Global interpreter lock - Wikipedia

Python은 GIL을 갖고 있는 언어이며, 이로 인해 단일 스레드로 작동시키던, 멀티스레드로 작동시키던 단 하나의 CPU 코어만 사용하여 온전한 멀티스레딩이 불가능합니다.

https://stackoverflow.com/questions/4496680/python-threads-all-executing-on-a-single-core

Python threads all executing on a single core

• Java, C#은 멀티스레드 사용시 자동적으로 멀티프로세싱을 지원하는 데에 반해, 파이썬은 자동적으로 지원하지 않는다는 뜻
• 단, CPU 코어 사용량을 보면 모든 코어를 활용하는 것처럼 보일 수 있는데, 이는 OS에서 프로세스를 job scheduling에 따라 왔다갔다 이동시키는 것일 뿐, 여전히 하나의 파이썬 프로세스는 동시간대에 단 하나의 코어에서만 작동한다.

Multiprocessing

자 그럼 이제 파이썬을 통해 멀티프로세싱을 시작해볼까요?

• Python의 multiprocessing 라이브러리는 threading과 동일한 API를 제공한다.
• 기본적으로는 아래와 같이 사용할 수 있다.

from multiprocessing import Process

# 프로세스 객체 생성
p1 = Process(target=함수명, args=(함수의 매개변수))

# 프로세스 시작
p1.start()

# 프로세스 종료
p1.join()

IPC(Inter Process Communication)

Python의 multiprocessing에는 크게 3가지 방법으로 IPC를 구현할 수 있고, 3가지 방법 모두 user space가 아닌 kernel space에서 이뤄진다.

• Pipe
• Queue
• Shared Memory

 

Pipe

• 양방향 (Full-duplex) 파이프 구조의 IPC 메커니즘
• 매개변수로 Pipe(duplex=bool)을 받는데, duplex=False로 설정해놓는다면 단방향이 된다. (기본값은 True)
• 양방향 통신이 가능하다는 것에 이점이 있지만, 메서드가 존재하지 않는 것이 단점
• 사용법

from multiprocessing import Pipe, Process

def test(pipe):
    pipe.send([1, 2, "hey"])
		pipe.close()

def main():
	parent_conn, chile_conn = Pipe()
	p1 = Process(target=test, args=(child_conn,))
	p1.start()

	val = parent_conn.recv()
	print(val)    # [1, 2, "hey"]

	p1.join()

Queue

• 단방향 FIFO 큐 구조의 IPC 매커니즘
• 파이프와 몇 개의 록/세마포어를 사용하여 구현된 프로세스 공유 큐를 반환한다.
• 사용법

from multiprocessing import Queue, Process

def test(q):
    val = q.get()
    print(val) # [1, 2, 'good']

def main():
    q = Queue()
    q.put([1, 2, 'good'])
    p1 = Process(target=test, args=(q))
    
    p1.start()
    p1.join()

Shared Memory

• 커널에 존재하고 있는 커널 메모리로, 데이터가 저장된 후에는 함수가 직접 접근하여 데이터를 가져와야 한다.
• Lock를 활용하여 상호배제를 해야 한다.
• 사용법

from multiprocessing import shared_memory, Lock
import numpy as np

lock = Lock()

a = np.array([1, 2, 3, 4, 5, 6])
shm = shared_memory.SharedMemory(create=True, size=a.nbytes)

lock.acquire()
b = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)
lock.release()

existing_shm = shared_memory.SharedMemory(name=shm.name)

lock.acquire()
c = np.ndarray((6,), dtype=np.int64, buffer=existing_shm.buf)
lock.release()

lock.acquire()
shm.close()
shm.unlink()
lock.release()

혹은 아래와 같이 세마포어를 활용하는 방법도 있습니다.

from multiprocessing import Process, shared_memory, Semaphore
import numpy as np
import time

def worker(id, number, shm, arr, sem):
    increased_number = 0

    for _ in range(number):
        increased_number += 1

    sem.acquire()
    new_shm = shared_memory.SharedMemory(name=shm)
    tmp_arr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=new_shm.buf)
    tmp_arr[0] += increased_number
    sem.release()

if __name__ == "__main__":
    start_time = time.time()

    arr = np.array([0])
    shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
    np_shm = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)

    sem = Semaphore()

    th1 = Process(target=worker, args=(1, 50000000, shm.name, np_shm, sem))
    th2 = Process(target=worker, args=(2, 50000000, shm.name, np_shm, sem))

    th1.start()
    th2.start()

    th1.join()
    th2.join()

    print("--- %s seconds ---" % (time.time() - start_time))
    print("total_number=",end=""), print(np_shm[0])
    print("end of main")

    shm.close()
    shm.unlink()

Multiprocessing 속도 테스트

그럼 멀티 프로세싱이 엉ㄹ마나 처리가 빠른지 테스트를 해볼 텐데요 속도 차이를 비교하기 위해 소수 판별 알고리즘을 연산하는 시간을 비교해보도록 하겠습니다.

# Simple stress test between uni-process mechanism & multiprocess mechanism
# Uses prime number algorithm for stress testing

import time
from multiprocessing import Process
from math import sqrt

nums = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419]

def is_prime_number(x):
    for i in range(2, int(sqrt(x))+1):
        if x % i == 0:
            print(str(x) + ": Not a prime number")
            return False
    print(str(x) + ": Prime number")
    return True

if __name__ == '__main__':
    # Test uni-process mechanism
    start = time.time()
    print("Uniprocess Mechanism")
    for i in range(len(nums)):
        is_prime_number(nums[i])
    end = time.time()
    uniprocess_time = end - start
    print("Time:", str(uniprocess_time) + " s")
    print("=============================================================================")
    
    # Test multiprocess mechanism
    start = time.time()
    print("Multiprocess Mechanism")
    p1 = Process(target=is_prime_number, args=(nums[0],))
    p2 = Process(target=is_prime_number, args=(nums[1],))
    p3 = Process(target=is_prime_number, args=(nums[2],))
    p4 = Process(target=is_prime_number, args=(nums[3],))
    p5 = Process(target=is_prime_number, args=(nums[4],))
    
    p1.start()
    p2.start()
    p3.start()
    p4.start()
    p5.start()
    
    p1.join()
    p2.join()
    p3.join()
    p4.join()
    p5.join()
    
    end = time.time()
    multiprocess_time = end - start
    print("Time:", str(multiprocess_time) + " s")
    print("=============================================================================")
    
    faster = round((uniprocess_time / multiprocess_time) * 100, 2)
    
    print("Multiprocessing is " + str(faster) + "%% faster than uniprocessing.")

결과는 어떻게 되었을까요?

• 5개의 숫자에 대한 소수 판별 알고리즘 [시간 복잡도 O(n^(1/2))] 에 대하여:
  • 단일 프로세스: 3.88988초
  • 멀티 프로세싱: 0.85337초
• 멀티프로세싱이 단일프로세싱에 비해 약 5배 더 빠름

물론 멀티프로세싱을 했을 경우 프로세스간 통신을 해야하는 것과, CPU를 더 열렬히 사용하기 때문에 발열문제가 있을 수 있으나, 이렇게 빠르게 연산이 된다면 적절히 활용하는게 좋겠죠. 멀티 코어, 멀티 프로세싱이 당연한 세상인 지금 파이썬으로 프로세스를 멀티하게 돌려보시면 성능향상에 적절히 기여할 수 있을 것입니다.