[Python]Futures 동시성

파이썬의 futures 라이브러리 사용방법에 대해 알아보자.

# Futures 동시성
# 비동기 작업실행에 적합한 경우 성능이 크게 향상됨
# 비동기 실행을 위한 API 개발이 쉬워짐
# GIL(Global Interpreter Lock) : 두 개 이상의 스레드가 동시에 실행 될 때, 동시에 하나의 자원을 엑세스 하려는 경우
# -> 하나의 스레드에만 모든 자원을 허용하고 락을 건다. -> Thread context switching (문맥교환) 비용이 발생한다.


import os
import time
from concurrent import futures

WORK_LIST = [100_000, 1_000_000, 10_000, 10_000_000]


# 동시성 합계 계산 메인함수
# 누적 합계함수(제네레이터)

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


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

    # 시작시간
    start_time = time.time()

    # 결과 건수
    # with futures.ProcessPoolExecutor(max_workers=worker) as executor:
    with futures.ThreadPoolExecutor(max_workers=worker) as executor:
        # map : 작업순서 유지, 즉시실행, 모든 작업이 처리될 때까지 기다린다.
        result = executor.map(sum_generator, WORK_LIST)

    # 종료시간
    end_time = time.time() - start_time

    msg = '\nResult -> {} time : {:.2f}s'
    print(msg.format(list(result), end_time))


if __name__ == '__main__':
    main()

"""
Result -> [5000050000, 500000500000, 50005000, 50000005000000] time : 3.29s
"""

# Futures 동시성 : wait 사용예제(timeout 시간까지 모두처리 된 이후에 결과를 한번에 반환한다.)

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

WORK_LIST = [1_000, 100_000, 1_000_000, 100_000_000]


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


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

    # 시작시간
    start_time = time.time()

    # futures
    futures_list = []

    # 결과 건수
    # with ProcessPoolExecutor() as executor:
    with ThreadPoolExecutor() as executor:
        for work in WORK_LIST:
            future = executor.submit(sum_generator, work)

            # 스케줄링
            futures_list.append(future)

            # 스케줄링 확인
            print(f'Scheduled {work} for {future}')

        # wait 결과 출력
        result = wait(futures_list, timeout=2) # 2초 이내에 완료되어야 done 처리된다.
        # 성공
        print('성공한 작업 : ' + str(result.done))

        # 실패
        print('실패한 작업 : ' + str(result.not_done))

        # 결과값 출력
        print([future.result() for future in result.done])


    # 종료시간
    end_time = time.time() - start_time

    msg = '\n실행시간 : {:.2f}s'
    print(msg.format(end_time))


if __name__ == '__main__':
    main()

"""
Scheduled 1000 for <Future at 0x10265b610 state=finished returned int>
Scheduled 100000 for <Future at 0x10265b880 state=pending>
Scheduled 1000000 for <Future at 0x102888e50 state=pending>
Scheduled 100000000 for <Future at 0x102888f10 state=running>
성공한 작업 : {<Future at 0x10265b880 state=finished returned int>, <Future at 0x10265b610 state=finished returned int>, <Future at 0x102888e50 state=finished returned int>}
실패한 작업 : {<Future at 0x102888f10 state=running>}
[5000050000, 500500, 500000500000]

실행시간 : 2.57s
"""

# Futures 동시성 : as_completed 사용예제(먼저 처리된 것부터 결과를 먼저 반환한다.)

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

WORK_LIST = [1_000, 100_000, 1_000_000, 100_000_000]


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


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

    # 시작시간
    start_time = time.time()

    # futures
    futures_list = []

    # 결과 건수
    # with ProcessPoolExecutor() as executor:
    with ThreadPoolExecutor() as executor:
        for work in WORK_LIST:
            future = executor.submit(sum_generator, work)

            # 스케줄링
            futures_list.append(future)

            # 스케줄링 확인
            print(f'Scheduled {work} for {future}')

        # as_completed 결과 출력
        for future in as_completed(futures_list):
            result = future.result()
            done = future.done()
            cancelled = future.cancelled()

            # future 결과확인
            print(f'Future 결과값 : {result}, 성공여부 : {done}')
            print(f'Future 취소여부 : {cancelled}')

    # 종료시간
    end_time = time.time() - start_time

    msg = '\n실행시간 : {:.2f}s'
    print(msg.format(end_time))


if __name__ == '__main__':
    main()

"""
Scheduled 1000 for <Future at 0x10192d550 state=pending>
Scheduled 100000 for <Future at 0x101903a90 state=pending>
Scheduled 1000000 for <Future at 0x10199bdc0 state=pending>
Scheduled 100000000 for <Future at 0x1019a33a0 state=pending>
Future 결과값 : 500500, 성공여부 : True
Future 취소여부 : False
Future 결과값 : 5000050000, 성공여부 : True
Future 취소여부 : False
Future 결과값 : 500000500000, 성공여부 : True
Future 취소여부 : False
Future 결과값 : 5000000050000000, 성공여부 : True
Future 취소여부 : False

실행시간 : 29.48s
"""