Gevent 알아보기

gevent는 동시성과 네트워크 관련 작업들을 위한 다양한 API를 제공하는 동시성 라이브러리다. gevent에서는 Greenlet이라고하는 경량 코루틴을 사용한다. 한 번에 오직 하나의 greenlet만이 실행되기에, multiprocessing이나 threading을 이용한 병렬처리와는 다르다.

yield를 통해 컨텍스트 스위칭이 이루어지며, 네트워크, I/O bound 작업을 처리할때 그 힘이 발휘된다. gevent는 네트워크 라이브러리들이 컨텍스트 스위칭이 가능한 시점에 yield하도록 보장해준다.

 

Monkey patch

asyncio를 사용하는 것 보다 gevent가 나은 점 중에 하나는 바로 monkey patch로 gevent를 사용하지 않는 다른 라이브러리들도 동시처리를 가능하도록 만들어준다는 점이다.

Python의 표준 라이브러리들 (socket, time, requests..)를 사용한 기존 코드들을 그대로 유지하면서도 gevent의 협력적 멀티태스킹의 이점을 누릴 수 있다. 하지만 런타임에 기존 동작을 바꿔버리기때문에 디버깅하기 어렵다는 단점이 있다.

from gevent import monkey

monkey.patch_all()
# 다른 라이브러리들을 import 하기전에 monkey patch 해야만한다

 

Asynchronous Execution

gevent.spawn은 Greenlet 오브젝트를 생성하고, 인자로 넘어온 함수와 아규먼트를 즉시 스케쥴링한다. 이것이 가장 일반적이고 편리한 Greenlet 초기화 방법이다. 모든 작업이 끝나길 기다리려면 gevent.joinall 에 Greenlet 오브젝트들을 담아주면 된다.

import gevent

def foo():
    print('Running in foo')
    gevent.sleep(0)

jobs = [gevent.spawn(foo) for _ in range(10)]
gevent.joinall(jobs, timeout=5)

 

Greenlet State

Greenlet도 실패할 수 있다. 하지만 Greenlet에서 일어난 실패는 greenlet 안에서 잡혀버리기 때문에, 전파되지는 않는다.

def win():
    return "You win!"

def fail():
    raise Exception("You fail at failing.")

winner = gevent.spawn(win)
loser = gevent.spawn(fail)

try:
    gevent.joinall([winner, loser])
except Exception as e:
    print("this will never be printed")

Greenlet의 상태에 접근할 수 있는 프로퍼티와 함수들

• started - Boolean, Greenlet이 실행되었는지 여부를 나타낸다.
• ready() - Boolean, Greenlet이 정지되었는지 여부를 나타낸다.
• successful() - Boolean, Greenlet이 예외를 발생시키지 않고 정지되었는지 여부를 나타낸다.
• value - Any, Greenlet에 의해 리턴된 값이다.
• exception - Exception, Greenlet 내부에서 발생한 예외다.

print(winner.value, loser.value)  # You win! None
print(winner.ready(), loser.ready())  # True True
print(winner.successful(), loser.successful())  # True False
print(loser.exception) # You fail at failing.

 

Timeout

Timeout은 코드 블럭이나 Greenlet의 실행시간에 제한을 줄 수 있다. 실행시간을 초과할 경우 gevent.Timeout 예외를 던진다.

import gevent
from gevent import Timeout

seconds = 10

timeout = Timeout(seconds)
timeout.start()

def wait():
    gevent.sleep(10)

try:
    gevent.spawn(wait).join()
except Timeout:
    print('Could not complete')

 

Events & AsyncResult

Event는 Greenlet 간의 비동기 통신에 사용된다. Event을 set해서 같은 이벤트를 기다리고 있는 다른 Greenlet들을 깨울 수 있다.

def setter():
    print("A: Hey wait for me, I have to do something")
    gevent.sleep(3)
    print("OK, I'm done")
    evt.set()

def waiter():
    """After 3 seconds the get call will unblock"""
    print("I'll wait for you")
    evt.wait()  # blocking
    print("It's about time")

gevent.joinall(
    [
        gevent.spawn(setter)
        gevent.spawn(waiter),
        gevent.spawn(waiter),
    ]
)

A: Hey wait for me, I have to do something
I'll wait for you
I'll wait for you
OK, I'm done
It's about time
It's about tim

AsyncResult는 다른 언어의 동시성 라이브러리의 Deferred, Future와 비슷한데, Value 또는 Exception을 담을 수 있는 일회성 Event이다. 위에서 본 Event처럼 Greenlet간 통신에 사용할 수 있다. set 또는 set_exception이 호출되면 AsyncResult를 기다리는 다른 waiter들에게 값을 넘겨줄 수 있다.

import gevent
from gevent.event import AsyncResult

a = AsyncResult()

def setter():
    gevent.sleep(3)
    a.set("Hello!")

def waiter():
    print(a.get())

gevent.joinall(
    [
        gevent.spawn(setter),
        gevent.spawn(waiter),
    ]
)

# prints "Hello!"

 

Queue

gevent.Queue는 일반적인 Queue처럼 동작하지만 Greenlet 사이에서 안전하게 사용할 수 있다. 예컨대 한 Greenlet이 Queue에서 값 하나를 가져왔을때, 해당 값이 다른 Greenlet에 의해 동시에 접근되지 못하도록 보장된다.

import gevent
from gevent.queue import Queue

tasks = Queue()

def worker(n):
    while not tasks.empty():
        task = tasks.get()
        print(f"Worker {n} processing task {task}")
        gevent.sleep(0)
    print(f"Worker {n} done")

def boss():
    for i in range(1, 10):
        tasks.put_nowait(i)

gevent.spawn(boss).join()

gevent.joinall(
    [
        gevent.spawn(worker, "steve"),
        gevent.spawn(worker, "john"),
        gevent.spawn(worker, "nancy"),
    ]
)

Worker steve processing task 1
Worker john processing task 2
Worker nancy processing task 3
Worker steve processing task 4
Worker john processing task 5
Worker nancy processing task 6
Worker steve processing task 7
Worker john processing task 8
Worker nancy processing task 9
Worker steve done
Worker john done
Worker nancy done

Queue는 put 또는 get 연산시 블락되지만, non-blocking 연산이 필요할때는 예시 코드처럼 _nowait을 사용하면된다.

Queue는 연산 중 다음 예외를 발생시킬 수 있다:

• gevent.queue.Empty
• gevent.queue.Full

import gevent
from gevent.queue import Empty, Queue

tasks = Queue(maxsize=3)

def worker(name):
    try:
        while True:
            task = tasks.get(timeout=1)  # 1초 동안 작업이 없다면 퇴근!
            print("Worker %s got task %s" % (name, task))
            gevent.sleep(0)
    except Empty:
        print("Quitting time!")

def boss():
    for i in range(1, 5):
        tasks.put(i)  # Queue에 남은 공간이 있을떄까지 block
    print("Assigned all work in iteration 1")

    for i in range(5, 10):
        tasks.put(i)
    print("Assigned all work in iteration 2")

gevent.joinall(
    [
        gevent.spawn(boss),
        gevent.spawn(worker, "steve"),
        gevent.spawn(worker, "john"),
        gevent.spawn(worker, "bob"),
    ]
)

Worker steve got task 1
Worker john got task 2
Worker bob got task 3
Assigned all work in iteration 1
Worker steve got task 4
Worker john got task 5
Worker bob got task 6
Assigned all work in iteration 2
Worker steve got task 7
Worker john got task 8
Worker bob got task 9
Quitting time!
Quitting time!
Quitting time!

 

Pool

Pool은 동시에 실행되는 Greenlet의 개수를 제한할 수 있도록 해준다.

from gevent.pool import Pool

pool = Pool(2)

def hello_from(n):
    print("Size of pool %s" % len(pool))

pool.map(hello_from, range(3))

Size of pool 2
Size of pool 2
Size of pool 1

 

Lock & Semaphore

Semaphore는 Greenlet들이 동시에 특정 코드블록에 접근하는 것을 제한한다. Semaphore bound가 0에 도달하면 다른 Greenlet이 해당 Semaphore를 release할때까지 block된다.

import gevent
from gevent import sleep
from gevent.lock import BoundedSemaphore

sem = BoundedSemaphore(2)

def worker1(n):
    with sem:
        print("Worker %i acquired semaphore" % n)
        sleep(2)
    print("Worker %i released semaphore" % n)

def worker2(n):
    with sem:
        print("Worker %i acquired semaphore" % n)
        sleep(2)
    print("Worker %i released semaphore" % n)

jobs = [gevent.spawn(worker1, i) for i in range(5)]
gevent.joinall(jobs)

Worker 0 acquired semaphore
Worker 1 acquired semaphore
Worker 0 released semaphore
Worker 1 released semaphore
Worker 2 acquired semaphore
Worker 3 acquired semaphore
Worker 2 released semaphore
Worker 3 released semaphore
Worker 4 acquired semaphore
Worker 4 released semaphore

Semaphore bound가 1인 경우를 Lock이라고하며, 한번에 하나의 Greenlet에 의해서만 사용되는 것을 보장해야할때 사용된다.

RLock은 재진입 가능한 락(Re-entrant Lock)으로, 같은 Greenlet이 여러번 획득가능한 Lock이다.

참고 자료

https://leekchan.com/gevent-tutorial-ko/

https://f-lab.kr/blog/ways-to-improve-python-application-performance