Thread

지난 스레드 포스트에서는 파이썬에서 스레드가 어떻게 구현되고 동작하는지에 관하여 간단하게 정리하였습니다. 이번에는 파이썬 스레드의 동작을 구현하며 고려해야할 자원의 동기화에 관련된 Lock과 Condition에 대한 글을 작성하고자합니다.

Thread Lock

여러 스레드가 사용하는 자원이 있다면 무결성을 보장하기 위한 로직을 구현해주어야 합니다. 그렇지 않으면 서로 다른 스레드에서 동시에 자원에 접근하여 수정하게되는 경우 무결성이 훼손될 수 있기 때문입니다.

파이썬의 내장 자료 구조 중에는 이러한 무결성을 보장해주는 자료구조가 있습니다. 내장된 리스트나 사전자료형의 경우, 값을 설정할 때 atomicity하게 실행되므로 스레드에서도 무결성을 보장할 수 있습니다.

하지만 정수나 실수와 같은 자료형들은 무결성이 보장되지 않으므로, 프로그래밍을 통해 무결성을 보장하기 위한 로직을 구현해주어야 합니다.

Lock

파이썬의 threading 모듈에는 Lock 클래스가 구현되어 있습니다. 데이터를 수정하기 전에 자원의 접근 권한인 Lock을 먼저 획득하도록 구현하는데 필요한 여러 기능들을 제공합니다. 자원의 값을 수정하기 위해 자원의 권한을 획득하고, 수정한 후에는 권한을 반환하는 방법으로 동시에 여러 스레드에서 자원을 수정하는 것을 방지하는 로직을 구현하는데 사용될 수 있습니다.

import time
import logging
import threading

logging.basicConfig(level=logging.DEBUG, format="(%(threadName)s) %(message)s")

def blocking_lock(lock):
    logging.debug("Start blocking lock")

    while True:
        time.sleep(1)
        lock.acquire()						# lock을 획득할때까지 block된다.
        try:
            logging.debug("Acquired lock")
            time.sleep(0.5)
        finally:
            logging.debug("Release lock")
            lock.release()

def nonblocking_lock(lock):
    logging.debug("Start non-blocking lock")
    attempt, grab = 0, 0
    
    while grab < 3:
        time.sleep(1)
        logging.debug("Attempt to acquire lock")
        success = lock.acquire(False)		# lock을 획득하면 True, 그렇지 않으면 False

        try:
            attempt += 1
            if success:
                logging.debug("Acquired lock")
                grab += 1
        finally:
            if success:
                logging.debug("Release lock")
                lock.release()

        logging.debug("Attempt : %s, Grab : %s"%(attempt, grab))

def main():
    lock = threading.Lock()

    blocking = threading.Thread(target=blocking_lock,
                                name="Blocking",
                                args=(lock,))
    blocking.setDaemon(True)
    blocking.start()

    nonblocking = threading.Thread(target=nonblocking_lock,
                                   name="Non-blocking",
                                   args=(lock,))
    nonblocking.start()

if __name__=="__main__":
    main()

두개의 스레드를 구현하여 하나의 Lock을 획득하여 실행하도록 구현하였습니다. blocking_lock함수로 구현한 스레드는 데몬 스레드로 설정하여 동작하도록 설정하였고, nonblocking_lock함수로 구현한 스레드는 Lock을 3회 이상 획득하게 되면 종료하여 프로그램이 종료되도록 구현하였습니다.

두 스레드에 인자로 주어진 lock에서 acquire(),release()가 사용되었습니다.

• acquire(blocking=True, timeout=-1)
  lock을 획득하기 위한 메소드입니다. blocking 인자의 기본값은 True로, 락을 획득할때까지 메소드는 block되고, locl이 획득하면 True를 반환하게 됩니다. blocking 인자로 False가 주어지게 되면 메소드는 nonblock되어 lock을 획득하면 True를, 획득하지 못하면False를 즉시 반환하고 동작을 이어가게 됩니다.
  
• release()
  lock을 반환하는 메소드입니다. 반환된 lock은 다시 다른 스레드에서 얻게됩니다.

위의 코드의 동작을 정리하면 다음과 같습니다.

데몬 스레드로 실행한 blocking_lock 스레드는 while문을 돌며 lock을 획득한 경우에만 lock.acquire() 이후의 코드를 수행하며 로깅을 남기고 lock을 반환하는 동작을 반복합니다.

반면 nonblocking_lock 스레드는 lock.acquire(False) 문을 nonblocking으로 처리하여 lock을 획득한 경우에는 시도 횟수와 획득 횟수를 모두 1씩 증가시키고, 획득하지 못한 경우네는 시도 횟수만 1 증가하도록 구현하였습니다.

두 스레드는 락을 획득, 반환하는 과정을 반복하며 각 스레드에 구현된 lock의 획득/반환에 따라 각자 구현된 로직을 수행하며 프로그램이 실행됩니다.

(Blocking) Start blocking lock
(Non-blocking) Start non-blocking lock
(Blocking) Acquired lock
(Non-blocking) Attempt to acquire lock
(Non-blocking) Attempt : 1, Grab : 0
(Blocking) Release lock
(Non-blocking) Attempt to acquire lock
(Non-blocking) Acquired lock
(Non-blocking) Release lock
(Non-blocking) Attempt : 2, Grab : 1
(Blocking) Acquired lock
(Blocking) Release lock
(Non-blocking) Attempt to acquire lock
(Non-blocking) Acquired lock
(Non-blocking) Release lock
(Non-blocking) Attempt : 3, Grab : 2
(Blocking) Acquired lock
(Non-blocking) Attempt to acquire lock
(Non-blocking) Attempt : 4, Grab : 2
(Blocking) Release lock
(Non-blocking) Attempt to acquire lock
(Non-blocking) Acquired lock
(Non-blocking) Release lock
(Non-blocking) Attempt : 5, Grab : 3

with 문

위의 예제에서는 lock을 잡을 때 오류에 대비하여 try구문을 사용하였습니다. 이와 같은 로직을 구현하는데 with 구문을 사용하여 구현할 수 있습니다. 위의 blocking_lock 함수를 with 구문을 사용하여 구현해보도록 하겠습니다.

def blocking_lock(lock):
    logging.debug("Start blocking lock")

	while True:
    	time.sleep(1)
        with lock:
        	logging.debug("Acquired lock")
            time.sleep(0.5)
            logging.debug("Release lock")

같은 로직을 구현한 코드입니다. try 구문보다 코드가 훨씬 간단하며, 가독성 또한 좋습니다. 이처럼 파일을 열 때 with 구문을 사용하면 자동으로 close가 되는 것처럼 with 구문이 끝나면 자동으로 lock을 반환하여 훨씬 편리하게 코드를 작성할 수 있습니다.
다만 이와 같은 with 구문은 lock을 획득하기 전까지 block되는 blocking lock에서만 사용할 수 있습니다. nonblocking lock에서 사용하기 위해선 따로 상속을 받아 구현해서 사용해야 합니다.

RLock(Reentrant Lock)

파이썬의 Lock을 사용하면 데이터의 무결성 문제를 해결하기 위한 로직을 구현하여 처리할 수 있습니다. 하지만 한 스레드가 lock을 가진 상태에서 다른 스레드가 해당 lock을 잡아야하는 경우 데드락 상태가 발생할 수 있습니다. 이와 같은 상황을 해결하기 위해 파이썬에서는 다른 스레드가 소유하고 있는 lock에 재진입하는 것이 가능한 RLock(Reentratn Lock)을 제공하고 있습니다.

import time
import logging
import threading

logging.basicConfig(level=logging.DEBUG, format="(%(threadName)s %(message)s)")

RESOURCE = 0

def reverse(lock):
    global RESOURCE
    logging.debug("Start Batch")

    with lock:
        logging.debug("Acquired Lock")
        set_reverse(lock, True)				# lock을 획득한 상태에서 같은 lock을 필요로하는 set_reverse 호출

    logging.debug("RESOURCE is reversed")

def set_reverse(lock, end=False):
    global RESOURCE
    logging.debug("Start set Resource to %s"%((RESOURCE+1)%2))

    while True:
        with lock:							# reverse 함수에서 호출한 경우, lock에 재진입하여 해당 구문을 실행할 수 있음
            logging.debug("Acquired lock and set resource to %s"%((RESOURCE+1)%2))
            RESOURCE = (RESOURCE+1)%2
            time.sleep(0.5)
        time.sleep(1)

        if end:
            break

def main():
    lock = threading.RLock()

    zero_one = threading.Thread(target=set_reverse, name="inner thread", args=(lock,))
    zero_one.setDaemon(True)
    zero_one.start()

    _reverse = threading.Thread(target=reverse, name="reverse", args=(lock,))
    _reverse.start()

if __name__=="__main__":
    main()

메인 스레드를 포함하여 3개의 스레드를 구현하였습니다.

reverse(lock) 함수를 사용하여 구현한 스레드는 lock을 획득하여 RESOURCE의 값을 0 또는 1로 설정하고 종료되는 스레드입니다. set_reverse(lock, end=False) 함수로 구현한 스레드는 락을 획득하여 현재 RESOURCE값을 반대 값으로 설정하고,end 인자에 따라 반복하거나 함수를 종료합니다.

여기서 reverse 스레드와 set_reverse 스레드는 모두 같은 lock을 사용하고 있습니다. 따라서 reverse 스레드에서 lock을 획득한 상태에서 set_reverse 함수를 호출하게 된다면 호출된 set_reverse 함수는 lock을 획득하기 위해 무한정 기다리게 되는 교착 상태에 빠지게 됩니다. 이러한 경우 RLock을 사용하여 재진입이 가능한 락을 사용할 수 있습니다.

(inner thread) Start set Resource to 1
(reverse) Start Batch
(inner thread) Acquired lock and set resource to 1
(reverse) Acquired Lock
(reverse) Start set Resource to 0				# reverse에서 set_reverse를 호출하여 재진입한 로그
(reverse) Acquired lock and set resource to 0	# set_reverse에서 RESOURCE 값 변경
(reverse) RESOURCE is reversed

reference

https://docs.python.org/ko/3/library/threading.html