파이썬 concurrent.futures에서 Future 객체의 Wait 다루기

이번 예제는 ThreadPoolExecutor와 Future 객체를 활용해 작업을 스케줄링하고 완료 여부를 추적하는 방법을 보여줍니다.

 

1. 코드 구조

import os
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, wait, as_completed

WORK_LIST = [10000, 100000, 1000000, 10000000]


# 동시성 합계 계산 메인 함수
# 누적 합계 함수(제레네이터)
def sum_generator(n):
    return sum(n for n in range(1, n+1))

# wait
# as_completed
def main():
    # Worker Count
    worker = min(10, len(WORK_LIST))
    
    # 시작 시간
    start_tm = time.time()
    # Futures
    futures_list = []

    # 결과 건수
    # ProcessPoolExecutor
    with ThreadPoolExecutor() as excutor:
        for work in WORK_LIST:
            # future 반환
            future = excutor.submit(sum_generator, work)
            # 스케쥴링
            futures_list.append(future)
            # 스케쥴링 확인
            print('Scheduled for {} : {}'.format(work, future))
            # print()
        
        # wait 결과 출력
        # result = wait(futures_list, timeout=7)
        # # 성공
        # print('Completed Tasks : ' + str(result.done))
        # # 실패
        # print('Pending ones after waiting for 7seconds : ' + str(result.not_done))
        # # 결과 값 출력
        # print([future.result() for future in result.done])
                              
    # 종료 시간
    end_tm = time.time() - start_tm
    # 출력 포멧
    msg = '\n Time : {:.2f}s'
    # 최종 결과 출력
    print(msg.format(end_tm))



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

주요 개념

• executor.submit(func, arg)
  • 작업을 스케줄링하고 Future 객체를 반환.
  • Future는 아직 완료되지 않은 작업의 결과를 나중에 확인할 수 있는 핸들.
• wait(futures, timeout)
  • 모든 Future가 완료되거나 지정한 시간(timeout)이 지날 때까지 대기.
  • 완료된 Future와 미완료 Future를 구분해줌.
• future.result()
  • Future가 완료되면 실제 결과 값 반환.
  • 완료되지 않았다면 블로킹(기다림) 상태가 됨.

 

2. 실행 흐름 (ASCII 다이어그램)

Main Process]
|
+-- submit(sum_generator, 1000000실행 흐름 (ASCII 다이어그램)
) → Future-1
+-- submit(sum_generator, 100000) → Future-2
+-- submit(sum_generator, 1000000) → Future-3
+-- submit(sum_generator, 10000000)→ Future-4


wait(futures, timeout=2)
|
+-- done: {Future-1, Future-2, ...}
+-- not_done:{Future-4, ...}

출력 예시

Scheduled for 1000000 : <Future at 0x... state=running>
Scheduled for 100000 : <Future at 0x... state=running>
Scheduled for 1000000 : <Future at 0x... state=running>
Scheduled for 10000000 : <Future at 0x... state=running>


Completed Tasks : {<Future at 0x... state=finished returned int>, ...}
Pending Tasks : {<Future at 0x... state=running>}
[500000500000, 50005000, 500000500000]
Execution Time : 0.55s

 

 

 

정리하면

• submit()으로 개별 작업을 스케줄링하면 Future 객체가 반환됨.
• wait()를 사용하면 일정 시간 동안 완료된 작업과 미완료 작업을 구분할 수 있음.
• Future.result()를 통해 완료된 결과를 안전하게 가져올 수 있음.

이 방식은 작업을 개별적으로 제어하고 추적해야 할 때 유용합니다. 예를 들어, 긴 연산을 일부만 먼저 처리하거나, 타임아웃 기반으로 결과를 관리할 때 적합합니다.