Python Process

이전 포스팅은 Thread에 대해 다루었는데요! 2개의 스레드를 만들었지만 시간이 단축되지 않고 1억까지 증가되지 않았습니다. 하지만 병렬로 두개의 프로세스를 만들어 병렬의 프로그래밍 방식으로 구현하면 시간을 단축할 수 있는데요!

프로세스를 만들면 각 프로세스별로 별도의 메모리 영역을 가지며 queue나 pipe, shared memory를 이용한 IPC(Inter-process-communication_프로세스간 통신)과 같은 방법으로 객체들의 교환을 구현할 수 있습니다. '멀티 프로세스 프로그램'은 각각 별도의 메모리 영역을 가지고, 여러 작업을 동시에 나눠서 처리할 수 있습니다. 그리고 병렬로 작업을 처리하니 하나의 프로세스에서 작업하던 것을 두배 이상 빠르게 할 수 있습니다.

다음의 코드는 thread를 생성했던 것처럼 multiprocessing패키지에서 process 클래스로부터 객체를 만들어 각각의 프로세스에 50000000씩 증가킨 후 q에 삽입해서 최종으로 1억까지 증가시키는 코드입니다.

from multiprocessing import Process, Queue
import time

def worker(id, number, q):
    increased_number = 0

    for i in range(number):
        increased_number += 1
    
    q.put(increased_number)

    return


if __name__ == "__main__":

    start_time = time.time()
    q = Queue()

    th1 = Process(target=worker, args=(1, 50000000, q))
    th2 = Process(target=worker, args=(2, 50000000, q))

    th1.start()
    th2.start()
    th1.join()
    th2.join()


    print("--- %s seconds ---" % (time.time() - start_time))
    q.put('exit')

    total = 0
    while True:
        tmp = q.get()
        if tmp == 'exit':
            break
        else:
            total += tmp

    print("total_number=",end=""), print(total)
    print("end of main")

두개의 쓰레드를 사용해서 구현했을 때보다 두개의 프로세스로 병렬 프로그래밍을 구현하면 훨씬 더 빠르게 진행됨을 알 수 있습니다.

IPC

프로세스의 IPC 방법은 앞에서 설명했듯, queue, pipe, shared memory 등이 있는데요. 파이썬의 multiprocessing은 queue와 pipe 두 가지 유형의 IPC를 지원하며 각각 queue()나 pipe()로 호출하면 됩니다.

Queue

from multiprocessing import Process, Queue

def f(q):
    q.put([42, None, 'hello'])

if __name__ == '__main__':
    q = Queue() # queue
    p = Process(target=f, args=(q,))
    p.start()
    p.join()
    print(q.get())
	
# "[42, None, 'hello']"

pipe

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([42, None, 'hello'])
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    p.join()
    print(parent_conn.recv())   
   
# "[42, None, 'hello']"
    

shared_memory

스레드는 프로세스를 나눈 단위이며 공유 데이터를 사용할 때 제약이 없지만, 프로세스는 독집적인 메모리 공간을 가지기 때문에 shared_memory를 사용합니다. 공유 메모리를 사용하면 메모리의 일부 공간을 각각의 독립적인 프로세스에서 공유하고, 해당 메모리를 통해 데이터를 주고받을 수 있습니다.

semaphore

semaphore는 지정된 변수만큼의 프로세스 혹은 쓰레드가 자원에 접근할 수 있도록 하는 방법입니다. 공유 메모리에 여러 프로세스가 동시에 사용하기 시작하면 데이터가 손상될 수도 있습니다. 그래서 여러 프로세스 사이에 동작의 순서를 지정해주어야 하는데 이 때 semaphore가 사용됩니다.

Mutex(상호배제)는 오직 하나의 프로세스 혹은 쓰레드만 자원에 접근하도록 하지만 Semaphore는 여러 대상을 처리할 수 있고 공유 메모리라는 임계 영역을 사용할 수있도록 합니다. 그리고 Lock을 사용하지 않기에 현재 수행하지 않는 다른 프로세스가 semaphore를 해제할 수 있습니다.

다음의 코드는 queue를 통한 IPC를 구현한 코드를 shared_memory방식과 semaphore를 사용한 프로그램입니다.

from multiprocessing import Process, shared_memory, Semaphore
import numpy as np
import time
def worker(id, number, shm, arr, sem):
	#매개 변수 : 프로세스명, 숫자, 공유메모리, 공유메모리를 저장할 배열, 세마포어 객체
	increased_number = 0

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

    sem.acquire() 
    
    # 기존 공유 메모리 블록에 연결
    existing_shm = shared_memory.SharedMemory(name=shm)
    # numpy 배열 형태에 맞게 변환
    tmp_arr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=existing_shm.buf)
    # 각각의 프로세스에서 연산한 값을 합해서 numpy 배열에 저장
    tmp_arr[0] += increased_number
    
    sem.release() # mutex와 동일



if __name__ == "__main__":

    start_time = time.time()
    sem = Semaphore()
    # 숫자를 저장할 numpy 배열 생성
    arr = np.array([0])
    # 공유 메모리 생성
    shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
    # 공유 메모리의 버퍼를 numpy 배열로 변환
    np_shm = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
    
    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))
    #프로세스에서 계산된 값을 저장한 np_shm 배열 출력
    print("total_number=",end=""), print(np_shm[0]) 
    print("end of main")

    # 공유 메모리 사용 종료
    shm.close()
    # 공유 메모리 블록 삭제
    shm.unlink()