파이썬 중급 - 9. 동시성과 병렬성 : Futures

 

개요

 

동시성, 병렬성과 관련한 부분은 실제 코딩이 어렵다기보다는 이론적인 배경을 정확하게 아는 것이 어렵고, 중요하다. 따라서 이번 강의 내용 정리는 개요 수준이고, 앞으로 더 많은 경험과 학습을 통해서 완전한 이해가 필요하다.

 

Futures를 이용하여 동시성 작업을 할 수 있다. 목적은 비동기 작업을 실행하여 지연 시간(Block)과 리소스 낭비를 방지하는 것이다. 보통 File, Network 등 I/O 관련 작업에서 동시성을 활용하는 것을 권장한다.

 

과거에는 threading, multiprocessing 등의 모듈을 import 해야했지만, 지금은 더 캡슐화되어 futures로 통합되었다. 특히 futures는 GIL(Global Interface Lock) 방식을 사용한다. 이것은 두 개 이상의 스레드가 동시에 하나의 자원에 접근할 때, 문제 방지를 위해서 리소스 전체에 lock을 거는 방식이다.

 

---

 

실습 : futures의 executor 사용하기

 

concurrent.futures에서 ThreadPoolExecutor 또는 ProcessPoolExecutor를 실행할 수 있다. Pycharm에서 futures에 들어가보면 두 가지 모듈을 모두 import하고 있는 것을 볼 수 있다. 멀티 스레드로해서 memory 자원을 사용할 것인지, 멀티 프로세스로 CPU 연산 자원을 사용할 것인지를 결정한다. futures 모듈 하나에서 두 가지 다른 방식을 간편하게 이름만 변경하여 사용할 수 있다는 것이 큰 장점이다.

import time
from concurrent import futures

# 동시에 4개의 스레드를 활용해서 1만, 10만, 100만, 1000만 까지의 숫자의 합을 동시에 구한다.
WORK_LIST = [10000, 100000, 1000000, 10000000]


def sum_generator(n):
    return sum(n for n in range(1, n + 1))


def main():
    worker = min(10, len(WORK_LIST))

    start_tm = time.time()  # 시작 시각

    #ThreadPoolExecutor | ProcessPoolExecutor
    with futures.ThreadPoolExecutor() as executor:
        # map : 작업 순서 유지, 즉시 실행
        result = executor.map(sum_generator, WORK_LIST)
    end_tm = time.time() - start_tm  # 종료 시각

    msg = '\n Result -> {} Time : {:.2f}s'
    print(msg.format(list(result), end_tm))


# 실행
if __name__ == '__main__':
    main()

 

 

 

 

---

 

실습 : executor.submit, futures.wait

 

실제 작업 목록은 위 처럼 순차적으로 작업 시간이 오래 걸릴 것이 예상되는 것이 아니라 얼마만큼의 시간이 걸릴지 모른다. 따라서 단순히 순차적으로 처리하는 것이 아니라 완료된 항목들부터 결과를 리스트에 받아서 담아놓으면 좋고, 그것이 executor.submit() 메서드를 통해서 가능하다. 결과는 Future 객체에 담긴다.

 

이 Future 객체에서 wait() 메서드를 사용하면 각 객체의 상태가 어떤지, 그리고 작업이 끝나야하는 기준 시간(timeout)을 설정할 수도 있다.

 

위 코드에서 나머지는 그대로 두고, main() 메서드 내 with 함수 부분만 변경해준다.

우선 submit() 메서드를 적용한다. 그러면 executor가 작업을 Future 객체로 저장하면서 메모리 주소, state, result 등의 결과를 갖게 된다. 그리고 이 Future 객체들을 리스트에 담는데, 아래 출력 결과를 보면 10000짜리 작업은 매우 빨리 끝나서 print문에서 state가 바로 finished로 처리된 것을 볼 수 있다.

with futures.ThreadPoolExecutor() as executor:
    for work in WORK_LIST:
        # 작업 완료 후 결과를 future에 저장 선언
        future = executor.submit(sum_generator, work)
        # 스케쥴링
        futures_list.append(future)
        print('Schedule for {} : {}'.format(work, future))

    results = wait(futures_list, timeout=5)
    # 성공
    print('Completed Tasks : ' + str(results.done))
    # 실패
    print('Pending ones after waiting for 5seconds : ' + str(results.not_done))
    # 성공 결과값
    print([future.result() for future in results.done])

end_tm = time.time() - start_tm  # 종료 시각
msg = '\n results -> {} Time : {:.2f}s'
print(msg.format(list(results.done), end_tm))

출력 결과

Schedule for 10000 : <Future at 0x40022db0d0 state=finished returned int>
Schedule for 100000 : <Future at 0x40022dba30 state=pending>
Schedule for 1000000 : <Future at 0x40022ee1f0 state=running>
Schedule for 10000000 : <Future at 0x40022ee580 state=running>

Completed Tasks : {<Future at 0x40022dba30 state=finished returned int>, <Future at 0x40022db0d0 state=finished returned int>, <Future at 0x40022ee1f0 state=finished returned int>}
Pending ones after waiting for 5seconds : {<Future at 0x40022ee580 state=running>}
[5000050000, 50005000, 500000500000]

results -> [<Future at 0x40022dba30 state=finished returned int>, <Future at 0x40022db0d0 state=finished returned int>, <Future at 0x40022ee1f0 state=finished returned int>] Time : 6.34s

 

그리고 wait() 메서드로 Future의 처리 결과를 results라는 변수에 담고, timeout 옵션을 줄 수 있다. done은 timeout으로 설정한 5초 안에 처리된 작업만 가져온다. 그리고 not_done은 5초안에 처리되지 못한 작업을 가져온다. 성공 결과값을 future 객체로 가져와서 future.result() 메서드로 실행해보면 맨 마지막 작업은 완료되지 못해서 결과가 3개만 출력되는 것을 볼 수 있다. 맨 마지막 줄에 총 걸린 시간을 보면 6.34초가 걸렸기 때문이다.

 

 

wait와 as_completed의 차이

wait는 입력 순서대로 future를 반환하지만, as_completed는 먼저 처리된 작업부터 바로 future 객체로 넘겨줄 수 있다. timeout으로 전체 작업이 끝날 때까지 기다렸다가 결과를 갖고 처리할 수 있는 상황이면 wait, 각 객체 하나마다 완료 후 바로 실시간 처리가 필요하면 as_completed를 사용하는 것이 적합하다.

 

 

# ...중략

# as_completed
        for future in as_completed(futures_list):
            result = future.result()
            done = future.done()
            cancelled = future.cancelled

            print('Future Result : {}, Done : {}'.format(result, done))
            print('Future Cancelled : {}'.format(cancelled))
            
'''
Future Result : 50005000, Done : True
Future Cancelled : <bound method Future.cancelled of <Future at 0x400249e040 state=finished returned int>>
Future Result : 5000050000, Done : True
Future Cancelled : <bound method Future.cancelled of <Future at 0x400249e970 state=finished returned int>>
Future Result : 500000500000, Done : True
Future Cancelled : <bound method Future.cancelled of <Future at 0x40024ae130 state=finished returned int>>
Future Result : 50000005000000, Done : True
Future Cancelled : <bound method Future.cancelled of <Future at 0x40024ae460 state=finished returned int>>
'''

 

 

 

---

참고 :쓰레드와 코루틴 비교

 

쓰레드는 프로세스를 수행하기 위해 OS에서 관리하고, 실시간으로 시분할 비동기 작업이 가능하게 해준다(멀티쓰레드). 코루틴은 단일 스레드, 스택을 기반으로 동작하는 비동기 작업이라는 특징이 있다. 파이썬뿐 아니라 GoLang이나 유니티에서도 사용하는 개념이다.

 

쓰레드에서는 각 쓰레드간 자원이 공유되므로 코딩이 복잡하고 deadlock이나 race condition이 발생할 수도 있다. 그러나 코루틴에서는루틴 실행 중 작업을 중지하고 다른 작업을 수행한다. 또한 컨텍스트 스위칭 비용이 들지 않고 자원 소비 가능성이 적다는 장점이 있다. 컨텍스트 스위칭 비용은 예상보다 큰 경우가 많아서 멀티쓰레드보다 싱글쓰레드로 운용하는 것이 유리한 경우도 종종 있다.

 

 

---

 

참조

 

1. 인프런 강의 - 우리를 위한 프로그래밍 : 파이썬 중급 (Inflearn Original)

https://www.inflearn.com/course/%ED%94%84%EB%A1%9C%EA%B7%B8%EB%9E%98%EB%B0%8D-%ED%8C%8C%EC%9D%B4%EC%8D%AC-%EC%A4%91%EA%B8%89-%EC%9D%B8%ED%94%84%EB%9F%B0-%EC%98%A4%EB%A6%AC%EC%A7%80%EB%84%90/dashboard