Part5. 제어 흐름

14. 반복형, 반복자, 제네레이터

 

파이썬은 반복형(Iterable)에서 반복자(Iterator)를 가져온다.

 

반복형

__iter__(): 반복자를 새로 생성

__iter__() 확인 -> __getitem__() 확인 (하위 버전과의 호환성) -> TypeError

* 반복형은 반복자로 작동하면 안된다. 다중 반복을 지원하려면 여러 독립적인 반복자를 가질 수 있어야하며, 각 반복자는 고유한 내부 상태를 유지해야 하기 때문.

 

반복자

__next__(): 다음 항복 반환, 없으면 StopIteration

__iter__(): 편의상 self 반환

 

제네레이터 함수

- 본체 안에 yield 키워드를 가진 함수, 제네레이터 팩토리

- 제네레이터 객체를 반환

- return은 오직 StopIteration 예외를 발생시키는 용도

class Sentence: #lazy loading
    def __init__(self, text):
        self.text = text
    
    def __repr__(self):
        wrapper = reprlib.Repr()
        return 'Sentence(%s)' % wrapper.repr(self.text)

    def __iter__(self):
        for match in RE_WORD.finditer(self.text):
            yield match.group()
#         return (match.group() for match in RE_WORD.finditer(self.text))

* 제네레이터 컴프리헨션은 편리하지만, 융통성과 가독성이 떨어짐

 

표준 라이브러리의 제네레이터 함수

1) 필터링 제네레이터

itertools.compress

itertools.takewhile

itertools.dropwhile

filter

itertools.filterfalse

itertools.islice

2) 매핑 제네레이터

itertools.accumulate

enumerate

map

itertools.starmap

3) 병합 제네레이터

itertools.chain

itertools.chain.from_iterable

itertools.product

zip

itertools.zip_longest

4) 확장 제네레이터

itertools.combinations

itertools.combinations_with_replacement

itertools.count

itertools.cycle

itertools.permutations

itertools.repeat

5) 재배치 제네레이터

itertools.groupby

reversed

itertools.tee

 

yield from

하위 제네레이터에 위임

def chain(*iterables):
	for i in iterables:
    	yield from i  # for루프를 한번 더 돌 필요없음

 

iter()

1) 어떤 객체 x를 반복할 때

2) 일반 함수나 콜러블 객체로부터 반복자를 생성할 때

iter(<인수없이 반복 호출되는 콜러블>, <이 값이 반환되면 StopIteration>)

예시) 1의 눈금이 절대 나오지 않는다.

import random
def d6():
    return random.randint(1, 6)

d6_iter = iter(d6, 1)
d6_iter

for roll in d6_iter:
    print(roll)

예시) 빈줄이 나올때까지 텍스트 라인별 전처리

with open('mydata.txt') as f:
    for line in iter(fp.readline, ''):
        process_line(line)

 

---

15. 콘텍스트 관리자와 else 블록

if문 이외에서의 else 절

- 예외, return break, continue문이 빠져나오게 만들면 else 블록이 실행되지 않는다.

1) EAFP(Easier to Ask for Forgiveness than Permission): 올바른 키 혹은 속성이 있다고 가정하고, 가정이 틀렸을 때 예외를 잡아서 처리하는 파이썬 코딩 스타일. try/except를 많이 사용

2) LBYL(Leap Before You Leap): 호출 혹은 조회 전 전제조건을 명시적으로 검사

 

with 블록

A;B;C 작업과 P;B;Q 작업이 있을 때 B작업을 공통요소로 만들면 서브루틴으로 만들 수 있다. 샌드위치 안의 내용물을 찍어내는 것과 같다. ... 그런데 동일한 빵에 내용물을 바꿔가면서 샌드위치를 만들기 위해 빵을 찍어내고 싶으면 어떻게 해야 할까? 그것이 바로 with문이 제공하는 것이다. 이는 서브루틴의 짝꿍이다.

- 레이몬드 헤팅거

• 공통적인 준비와 마무리 작업을 인수분해하는 도구
• try/finally와 같이 일정한 코드를 반드시 실행할 수 있게 보장

with open('untitled.ipynb') as fp:
    # 1. __enter__() 호출
    # open(): 이 경우 TextIOwrapper 객체, 컨텍스트 관리자
    # as ~: 컨텍스트 관리자 객체의 __enter__() 메서드 결과(이 경우 self)
    src = fp.read(60)
# 2. __exit__() 호출

• 컨텍스트 관리자 프로토콜: __enter__(), __exit__()

* __exit__() 메서드는 예외처리도 한다. True를 반환하면 예외를 억제하며, None를 반환하면 예외를 전파한다.

class LookingGlass:
    def __enter__(self):
        import sys
        
        self.original_write = sys.stdout.write
        sys.stdout.write = self.reverse_write
        return 'JABBERWOCKY'
    
    def reverse_write(self, text):
        self.original_write(text[::-1])
        
    def __exit__(self, exc_type, exc_value, traceback):
        import sys
        
        sys.stdout.write = self.original_write
        if exc_type is ZeroDivisionError:
            print('Please Do Not devide by Zero!')
            return True # 예외를 억제

 

contextlib

• ExitStack: 여러 콘텍스트 관리자를 입력할 수 있게 해주는 컨텍스트 관리자

from contextlib import ExitStack

filenames = ['untitled.ipynb', 'untitled1.ipynb']
with ExitStack() as stack:
    files = [stack.enter_context(open(f, encoding='utf-8')).read(60) for f in filenames]

• @contextmanager: 제네레이터 함수로 콘텍스트 관리자를 생성할 수 있게 해주는 데커레이터

• yield 문 앞의 코드: __enter__()를 호출할 때 실행
• yield 값: as절의 타겟 변수에 바인딩될 값
• yield 문 뒤의 코드: __exit__()가 호출될 때 실행

import contextlib

@contextlib.contextmanager
def looking_glass():
    import sys
    
    original_write = sys.stdout.write
    
    def reverse_write(text):
        original_write(text[::-1])
        
    sys.stdout.write = reverse_write
    msg = ''
    try:
        yield 'JABBERWOCKY' # 예외 시 gen.throw(exception) 실행
    except ZeroDivisionError:
        msg = 'Please Do Not divide by zero!'
    finally:
        sys.stdout.write = original_write
        if msg:
            print(msg)

* 제네레이터는 값을 반환하기 위해 예외를 이용해야 하므로, 예외를 전파하기 위해서는 함수 내에서 명시적으로 예외를 발생시켜야 한다.

 

---

16. 코루틴

-> 값을 주고받는 상호 연계 프로그램, 서브 루틴을 일시 정지하고 재개할 수 있는 구성 요소

-> send()를 호출하거나 yield from을 이용해서 데이터를 보내는 클라이언트(이벤트 기반)에 의해 실행되는 제네레이터

def simple_coro2(a):
    print('-> Started: a =', a)
    b = yield a
    print('-> Received: b =', b)
    c = yield a + b
    print('-> Received: c =', c)
    
my_coro2 = simple_coro2(14)
print(getgeneratorstate(my_coro2)) # GEN_CREATED
print(next(my_coro2)) # 14

print(getgeneratorstate(my_coro2)) # GEN_SUSPENDED
print(my_coro2.send(28)) # 42

try:
    print(my_coro2.send(99)) # 99
except:
    print(getgeneratorstate(my_coro2)) # GEN_CLOSED

• 할당문은 실제 값을 할당하기 전에 r-value부터 실행하기 때문에 b = yield a같은 코드에서는 호출자가 값을 보낸 후에야 변수 b가 설정된다.

상태

1. GEN_CREATED: 기동(priming)을 위해 대기
2. GEN_RUNNING: 인터프리터가 실행하고 있는 상태(다중 스레드에서만 확인 가능)
3. GEN_SUSPENDED: yield문에서 대기하고 있는 상태
4. GEN_CLOSED: 실행이 완료된 상태

 

API

send() *send(None) = next(self)

throw(exc_type, exc_value, traceback): yield 표현식에 예외 전달

close(): yield 표현식이 GeneratorExit 발생

 

코루틴을 기동하기 위한 데커레이터

from functools import wraps

def coroutine(func):
    @wraps(func) # func의 속성 래핑
    def primer(*args, **kwargs):
        gen = func(*args, **kwargs)
        next(gen)
        return gen
    return primer

 

코루틴의 값 반환

from collections import namedtuple

Result = namedtuple('Result', 'count average')

def averager():
    total = .0
    count = 0
    average = None
    while True:
        term = yield
        if term is None:
            break
        total += term
        count += 1
        average = total / count
        
    return Result(count, average)
    
coro_avg = averager()
next(coro_avg)
coro_avg.send(10)
coro_avg.send(30)
coro_avg.send(6.5)
try:
    coro_avg.send(None)
except StopIteration as exc: # StopIteration의 첫 번째 인수에 결과값이 밀반입됨
    result = exc.value

 

yield from

• 값을 생성하는 내포된 for루프를 대체
• 가장 바깥쪽 호출자와 가장 안쪽에 있는 하위 제네레이터 사이에 양뱡향 채널을 열어준다->코루틴 위임(coroutine delegation)
• 특징

1. 하위 제네레이터가 생성하는 값은 모두 대표 제네레이터의 호출자(클라이언트)에 전달된다.
2. send()를 통해 대표 제네레이터에 전달한 값은 모두 하위 제네레이터에 전달된다.
3. 제네레이터나 하위 제네레이터에서 return expr문을 실행하면 StopIteration(expr)가 발생한다. 이때 expr가 yield from 표현식의 값이다.
4. 대표 제네레이터에 던져진 예외는 GeneratorExit을 제외하고 하위 제네레이터 throw() 메서드에 전달된다. throw() 메서드를 호출해서 StopIteration이 발생하면 대표 제네레이터가 실행되고, 아니면 예외가 전파된다.
5. 대표 제네레이터의 close()메서드가 호출되면 하위 제네레이터의 close()메서드가 호출된다. 예외가 발생하면 대표 제네레이터에 전파되고, 아니면 대표 제네레이터에 GeneratorExit 예외가 발생한다.

 

코루틴 위임

1) 대표 제네레이터(delegating generator): 하위 제네레이터를 호출, 제어권을 클라이언트에게 넘김

2) 하위 제네레이터(subgenerator): 서브루틴

3) 호출자(caller), 클라이언트: 대표 제네레이터를 호출

Result = namedtuple('Result', 'count average')

def averager():
    total = .0
    count = 0
    average = None
    while True:
        term = yield
        if term is None:
            break
        total += term
        count += 1
        average = total / count
        
    return Result(count, average)

def grouper(results, key):
    while True:
        results[key] = yield from averager() # 하위 제네레이터의 종료를 await, 값을 전달 및 가져온다
        
def main(data):
    results = {}
    for key, values in data.items():
        group = grouper(results, key)
        next(group)
        for value in values:
            group.send(value)
        group.send(None) # 하위 제네레이터를 종료시키지 않으면 대표 제네레이터가 실행되지 않음

    report(results)
    
def report(results):
    for key, value in sorted(results.items()):
        group, unit = key.split(';')
        print('{:2} {:5} averaging {:.2f}{}'.format(result.count, group, result.average, unit))
        
data = {"girls;kg" : [40.9, 38.5, 44.3, 42.2, 45.2, 41.7, 44.5, 38.0, 40.6, 44.5],
       'girls;m' : [1.6, 1.51, 1.4, 1.3, 1.41, 1.39, 1.33, 1.46, 1.45, 1.43],
       'boys;kg' : [39., 40.8, 43.2, 40.8, 43.1, 38.6, 41.4, 40.6, 36.3],
       'boys;m' : [1.38, 1.6, 1.32, 1.25, 1.37, 1.48, 1.25, 1.49, 1.46],
       }

if __name__ == '__main__':
    main(data)

 

---

17. Future를 이용한 동시성

 

 concurrent.futures

ThreadPoolExecutor, ProcessPoolExecutor: 콜러블 객체를 서로 다른 스레드나 프로세스에서 실행할 수 있게 해주는 인터페이스를 구현

  1. map()을 이용한 구현: 호출한 순서대로 결과를 반환

from concurrent import futures

from flags import save_flag, get_flag, show, main

MAX_WORKERS = 20

def download_one(cc):
    image = get_flag(cc)
    show(cc)
    save_flag(image, cc.lower() + '.gif')
    return cc

def download_many(cc_list):
    workers = min(MAX_WORKERS, len(cc_list))
    with futures.ThreadPoolExecutor(workers) as executor:
        res = executor.map(download_one, sorted(cc_list))

    return len(list(res))

  2. submit()/futures.as_completed() : 완료되는 대로 결과를 가져옴

from concurrent import futures

from flags import save_flag, get_flag, show, main

MAX_WORKERS = 20

def download_one(cc):
    image = get_flag(cc)
    show(cc)
    save_flag(image, cc.lower() + '.gif')
    return cc

def download_many(cc_list):
    workers = min(MAX_WORKERS, len(cc_list))
    with futures.ThreadPoolExecutor(workers) as executor:
        to_do = [] # future 객체를 담을 리스트
        for cc in sorted(cc_list):
            future = executor.submit(download_one, cc)
            to_do.append(future)
            msg = 'Schduled for {}: {}'
            print(msg.format(cc, future))

        results = []
        for future in futures.as_completed(to_do): # future가 담긴 반복형을 받아 완료처리
            res = future.result() # 모두 완료처리됐으므로 블로킹되지 않음
            msg = '{} result: {!r}'
            print(msg.format(future, res))
            results.append(res)

    return len(results)

 

Future

• concurrent.futures와 asyncio의 핵심 컴포넌트
• 완료되었을 수도 있고 아닐 수도 있는 지연된 계산을 표현하기 위해 사용
• 직접 객체 생성 X (앞으로 일어날 일을 나타내기 때문에, 오직 프레임워크만 미래를 알 수 있음)
• 상태 변경 X (프레임워크가 연산 완료 시 상태 변경)

* API

done() : 실행 완료 여부

result() : 완료 시 결과 혹은 예외 반환, 미완료시 블로킹될 수도 (concurrent.futures는 블로킹, asyncio는 논블로킹)

 

GIL (Global Interpreter Lock)

• 하나의 쓰레드가 자원을 독점 (파이썬이 느린 이유)
• C로 작성된 확장은 GIL 해제 가능
• 블로킹 입출력(I/O 동안 작업 중단)을 실행하는 모든 표준 라이브러리 함수는 GIL 해제
• 멀티프로세싱은 GIL을 우회

따라서 결론,

• 입출력 App : 스레딩 (GIL 해제)
• 계산 App : 멀티프로세싱 (GIL 우회)

 

* threading과 multiprocessing: 상위 수준의 API에 잘 맞지 않는 동시성 작업 처리

 

---

18.  asyncio를 이용한 동시성

1. 동시성(Concurrency): dealing with a lot of things at once -> asyncio
2. 병렬성(Parallelism): doing a lot of things at once -> threading, multiprocessing

 

키워드

async: 이벤트 루프에 의해 스케쥴링(큐에 삽입)되는 비동기 함수

await: 이벤트 루프에 제어권 양도 

import asyncio
import itertools
import sys

async def spin(msg):
    write, flush = sys.stdout.write, sys.stdout.flush
    for char in itertools.cycle('|/-\\'):
        status = char + ' ' + msg
        write(status)
        flush()
        write('\x08' * len(status))
        try:
            await asyncio.sleep(.1) # 이벤트루프 논블로킹
        except asyncio.CancelledError:
            break
    write(' ' * len(status) + '\x08' * len(status))

async def slow_function(): # 일반 함수로 구현할 경우 제어권을 가져가버려 스피너가 안 나옴
    await asyncio.sleep(3) # 이벤트루프 논블로킹
    return 42

async def supervisor():
    spinner = asyncio.create_task(spin('thinking!')) # Task: 이벤트 루프 내 비동기 실행
    print('spinner object:', spinner)
    result = await slow_function()
    spinner.cancel()
    return result

def main():
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(supervisor())
    loop.close()
    print('Answer:', result)

if __name__ == '__main__':
    main()

 

Future 객체

concurrent.futures.Future: 코루틴의 실행이 완료되지 않았으면 result() 블로킹

asyncio.Future: : 실행이 완료되지 않았으면 예외 발생 -> yield from(await)사용하여 논블로킹 보장