위코드 Pre Course - 파이썬의 Process

파이썬에서 병렬 처리를 위해서 사용하는 개념인 프로세스에 대해서 정리한다.

 

전까지 내가 알기로는 프로세스와 스레드는 작업 단위로서 사용하는 말로 알고 있는편이었다.

 

보통 스레드보다 프로세스 단위가 더 크고 상위의 개념이라고 알고 있었는데, 파이썬에서는 병렬 처리를 위해 사용할 수 있는 독립된 개념인 것 같다.

 

Process란?

• 프로세스는 프로세스 별도의 메모리 영역을 가진다.
• IPC(Inter-Process Communication)을 지원해서 프로세스간 통신이 가능하다.
• 스레드보다는 프로세스를 사용해서 다중 코어 프로그래밍을 하는 것을 권장한다고 함

프로세스 예제

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)     #공유자원 q에 계산한 값을 저장함

    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()  # join() 메소드의 인자가 없을 때 해당 프로세스의 작업이 완료될 때 까지 대기
    th2.join()


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

    total = 0
    while True:
        tmp = q.get()
        if tmp == 'exit':   #q에 'exit' 문자열 확인되면 반복문 탈출
            break
        else:       #'exit' 문자열 확인되기 전까지는 각 요소들 누적
            total += tmp

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

• 스레드에서도 사용했던 각각 5천만의 값까지 1씩 누적하는 로직을 2개의 프로세스에 할당해서 실행시켜보면 결과는 다음과 같다.

#####################################
# 실행 결과 - 프로세스
--- 6.085736989974976 seconds ---
total_number=100000000
end of main

#####################################
# 실행 결과 - 스레드 (전에 사용한 코드)
number = 50000000
number = 50000000
--- 12.252983570098877 seconds ---
shared_number=53298096
end of main

• 스레드에 비교해보면 의미있는 실행 시간의 향상이 확인된다.

IPC(Inter-Process Communication)

• 예전에 사용해봤던 운영체제의 IPC와 각 개념들이 기능은 동일한 것으로 보인다.
• 종류
  • Queue(큐) - FIFO 특성을 가지지만 단방향 통신
  • Pipe(파이프) - FIFO 특성을 가지고, 양방향 통신
  • 공유 메모리(Shared Memory)

Assignment

# 공유 메모리 방식 사용한 방법으로 코드 수정해보기

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

def func(id, number, new_array, shm, sem):
    increased_number = 0
    for i in range(number):
        increased_number += 1
    sem.acquire()
    ex_shm = shared_memory.SharedMemory(name=shm)
    b = np.ndarray(new_array.shape, dtype=new_array.dtype, buffer=ex_shm.buf)
    b[0] += increased_number
    sem.release()
    

if __name__ == "__main__":

    sem = Semaphore(1)         # semaphore 객체 생성, 접근 가능 프로세스 수 1개
    new_array = np.array([0])       # 1차원 numpy 배열 생성
    shm = shared_memory.SharedMemory(create=True, size=new_array.nbytes)        #shared memory 생성
    c = np.ndarray(new_array.shape, dtype=new_array.dtype, buffer=shm.buf)      #shared memory에 버퍼용도의 numpy 어레이 연결

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

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

    print("total_number=",end=""), print(c[0])
    print("end of main")
    shm.close()
    shm.unlink()
    
##########################################################
# 실행 결과
total_number=100000000
end of main

 

• 세마포어는 스레드 포스팅에 정리하였으니 넘어간다.