코드는 다음과 같다.
import time
from multiprocessing import Pool
def work_log(name: str, duration: int):
print(f"Process {name} waiting {duration} seconds")
time.sleep(duration)
print(f"Process {name} Finished.")
if __name__ == "__main__":
tasks = [('A', 5), ('B', 2), ('C', 1), ('D', 3)]
with Pool(2) as p:
p.starmap(work_log, tasks)
생각보다 간단하고, 멀티 쓰레드 모듈인 Threading이랑 매우 비슷해보인다. 하지만 멀티 쓰레드와 멀티 프로세스의 가장 큰 차이는 GIL에 영향을 받는지 여부이다. Global Interpreter Lock은 파이썬 프로그램 실행시 thread-safe 환경을 만들기 위해 한 번에 하나의 쓰레드만 바이트 코드를 실행할 수 있도록 Lock으로 강제한다.
때문에 multithreading을 사용하더라도 한 쓰레드가 코드를 실행중이면 다른 모든 쓰레드는 Lock이 풀릴 때까지 기다려야 한다. 그럼에도 쓰는 이유는 여러 번의 Input, Output이 있는 task의 경우, Input을 위해 대기한다던지 Output 작업을 하는 동안에 병렬로 처리를 할 수 있어 단일 쓰레드보다 더 효율적이기 때문이다. 또한 각 쓰레드들은 하나의 공유 자원을 사용하기 때문에 쓰레드간 통신에도 유리하다.
반면에 multiprocessing은 여러 프로세스를 쓰기에 독립적인 공간을 가지고, 각각의 GIL을 가진다. 따라서 병렬적으로 실행되는 것에 대한 제한이 없어 계산 작업과 같은 CPU가 많이 필요할 때 사용된다. 하지만 프로세스간 직접 통신은 불가능하기 때문에 작업에 관한 공유가 필요하면 따로 구현을 해주어야 한다.
여기서 사용한 Pool은 최대 subprocess의 개수를 제한하여 멀티프로세싱을 한다. (semaphore 개념)
코드는 다음과 같다.
from multiprocessing import Process
def print_continent(name = "Asia"):
print(f"The name of continent is : {name}")
if __name__ == "__main__":
processes = []
p = Process(target=print_continent)
p.start()
processes.append(p)
for continent in ['America', 'Europe', 'Africa']:
p = Process(target=print_continent, args = (continent,))
p.start()
processes.append(p)
for process in processes:
process.join()
default로 Asia를 출력하는 default process와 정해진 continent 이름을 출력하는 여러 프로세스들을 동시에 실행시켰다. 마지막으로 processes list를 만들고, join으로 모든 프로세스가 끝나야 메인 프로세스도 끝날 수 있도록 했다.
from multiprocessing import Queue
def print_color(number, color):
print(f"item no: {number} {color}")
if __name__ == "__main__":
colors = ['red', 'green', 'blue', 'black']
q = Queue()
print("pushing items to queue:")
for number in range(4):
print_color(number + 1, colors[number])
q.put((number + 1, colors[number]))
print("popping items from queue")
while not q.empty():
number, color = q.get()
print_color(number - 1, color)
요구사항을 잘 읽어보면 Multiprocessing module을 이용하라고 했지, multiprocessing을 쓰라는 말이 없었다. (민우님 감사합니다!) 그래서 단일 프로세스로 코딩하되, multiprocessing 모듈 내에 있는 Queue를 사용했다.
인덱스가 push때는 1-index를 쓰다가 pop때는 0-index를 사용해서 print_color 호출 시 index를 조정했다.
import time
from multiprocessing import Process, Queue
def do_task(p_num: int, q_todo: Queue, q_done: Queue):
try:
while not q_todo.empty():
current_task = q_todo.get_nowait()
print(current_task)
time.sleep(0.5)
q_done.put(f"{current_task} is done by Process-{p_num}")
except Exception as e:
# Error handling when no tasks are left
return
if __name__ == "__main__":
# Task Distribution
tasks_to_accomplish = Queue()
for i in range(10):
task = f"Task no {i}"
tasks_to_accomplish.put(task)
# Process Execution
tasks_that_are_done = Queue()
processes = []
for process_num in range(1, 5):
p = Process(target=do_task, args=(process_num, tasks_to_accomplish, tasks_that_are_done))
p.start()
processes.append(p)
# Task Completion
for process in processes:
process.join()
while not tasks_that_are_done.empty():
completion_message = tasks_that_are_done.get()
print(completion_message)
특이한 점은 Task Distribution때 Queue에 순서대로 pish하여 작업을 시작하는 순서는 보장된다. 하지만 Multiprocessing으로 동일한 Queue에서 가져오게 되면 작업이 끝나는 순서는 보장되지 않는다는 것이다. 아래 실제 출력 결과로 알 수 있다. (출력 결과는 처음 push시 Task no 0~9는 동일하지만, process execution 이후 done by 앞에 있는 번호는 실행할 때마다 조금씩 달라진다. 아주 가끔 print가 겹치면 이렇게 나오기도 한다)
Task no 0
Task no 1
Task no 2
Task no 3
Task no 4Task no 5
Task no 6
Task no 7
Task no 8
Task no 9
Task no 1 is done by Process-1
Task no 2 is done by Process-4
Task no 0 is done by Process-2
Task no 3 is done by Process-3
Task no 6 is done by Process-1
Task no 5 is done by Process-2
Task no 4 is done by Process-4
Task no 7 is done by Process-3
Task no 8 is done by Process-1
Task no 9 is done by Process-4
이는 OS의 Process scheduling 방식, context switching에 따라 먼저 시작한 task가 먼저 끝나지 않을 수 있기 때문이다.