파이썬 동시성 프로그래밍 - (2) Condition & Semaphore & Event

연재 순서 

1. threading
2. Condition & Semaphore
3. Queue
4. multiprocessing
5. 비동기 (gevent) 
6. 분산 (celery)
7. GPGPU (PyCUDA)
8. 코루틴,asyncio,async/awiat
9. concurrent.future

2. Condition

파이썬의 Condidtion 은 쉽게 생각하면 Event + Mutex 쯤으로 보면된다.
다음예를보면 소비자 쓰레드들은 Condition 이 set 이 되길 기다리고 있다. 생산자 쓰레드는 이 Condition을 set 해줘서 다른 쓰레드들에게 진행해도 좋다고 고지한다. 기다리고 있던 쓰레드 모두가 통과할 수 는 없고 상호 배제되어 하나씩 통과된다 . 

import threading
import time
import logging

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-9s) %(message)s',)

def consumer(cv):
    logging.debug('Consumer thread started ...')
    with cv:
    	logging.debug('Consumer waiting ...')
        cv.wait()
        time.sleep(3)
        logging.debug('Consumer consumed the resource')

def producer(cv):
    logging.debug('Producer thread started ...')
    with cv:
        logging.debug('Making resource available')
        logging.debug('Notifying to all consumers')
        cv.notifyAll()

if __name__ == '__main__':
    condition = threading.Condition()
    cs1 = threading.Thread(name='consumer1', target=consumer, args=(condition,))
    cs2 = threading.Thread(name='consumer2', target=consumer, args=(condition,))
    pd = threading.Thread(name='producer', target=producer, args=(condition,))

    cs1.start()
    time.sleep(1)
    cs2.start()
    time.sleep(2)
    pd.start()

(consumer1) Consumer thread started ...

(consumer1) Consumer waiting ...

(consumer2) Consumer thread started ...

(consumer2) Consumer waiting ...

(producer ) Producer thread started ...

(producer ) Making resource available

(producer ) Notifying to all consumers

(consumer2) Consumer consumed the resource

(consumer1) Consumer consumed the resource

threading.Condition vs threading.Event

질문:

지금까지 Condition과 Event 에 대한 명확한 차이점에 대한 설명을 찾지 못해서 드리는 질문입니다. 각각의 사용처에 대해 알려 주실 수 있나요? 제가 지금까지 찾은 예제들은 모두 생산자-소비자모델이었는데요. queue.Queue가 더 좋은솔루션인듯요.

답변:

간단하게 말해서 쓰레드들이 다른곳에서 무엇인가 true 가 될 때까지 기다려야한다면 그리고 일단 그것이 true가 된다면 공유된 자원에 엑세스하는것이 배제적이 될때 Condition이 일반적이고, 반면에 Event 는 쓰레드들이 단지 무엇이 true가 되기만을 기다리는것에 흥미를 가지고 있으면 사용된다.

2-2 Semaphore

파이썬의 Semaphore 는 정해진 갯수의 쓰레드만 통과시켜준다. 예를들어 웹크롤링을 하는 쓰레드를 50개정도로 한정지어 놓는데 사용할 수 있다.  아래 예제에서는 10개의 쓰레드 중에서 3개만 일을 하도록 한다. 만약 세마포어를 통과한 3개의 쓰레드가 동일한 리소스를 사용하려고 할 때는 그들끼리 Lock 을 통해서 상호배제되어야 할 것이다.

import threading
import time
import logging

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-9s) %(message)s',)

class ThreadPool(object):
    def __init__(self):
        super(ThreadPool, self).__init__()
        self.active = []
        self.lock = threading.Lock()
    def makeActive(self, name):
        with self.lock:
            self.active.append(name)
            time.sleep(5)
            logging.debug('Running: %s', self.active)
    def makeInactive(self, name):
        with self.lock:
            self.active.remove(name)
            logging.debug('Running: %s', self.active)

def f(s, pool):
    logging.debug('Waiting to join the pool')
    with s:
        name = threading.currentThread().getName()
        pool.makeActive(name)
        time.sleep(1)
        pool.makeInactive(name)

if __name__ == '__main__':
    pool = ThreadPool()
    s = threading.Semaphore(3)
    for i in range(10):
        t = threading.Thread(target=f, name='thread_'+str(i), args=(s, pool))
        t.start()

2-3 Event

import threading
import time
import logging

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-9s) %(message)s',)
                    
def wait_for_event(e):
    logging.debug('wait_for_event starting')
    event_is_set = e.wait()
    logging.debug('event set: %s', event_is_set)

def wait_for_event_timeout(e, t):
    while not e.isSet():
        logging.debug('wait_for_event_timeout starting')
        event_is_set = e.wait(t)
        logging.debug('event set: %s', event_is_set)
        if event_is_set:
            logging.debug('processing event')
        else:
            logging.debug('doing other things')

if __name__ == '__main__':
    e = threading.Event()
    t1 = threading.Thread(name='blocking', 
                      target=wait_for_event,
                      args=(e,))
    t1.start()

    t2 = threading.Thread(name='non-blocking', 
                      target=wait_for_event_timeout, 
                      args=(e, 2))
    t2.start()

    logging.debug('Waiting before calling Event.set()')
    time.sleep(3)
    e.set()
    logging.debug('Event is set')

---

레퍼런스:

http://www.bogotobogo.com/python/Multithread/python_multithreading_Synchronization_Condition_Objects_Producer_Consumer.php