[분산 시스템] 1-4 Synchronization

gimseonjin616·2021년 12월 26일

study 분산시스템

분산시스템

목록 보기

5/14

Thread Synchronization

스레드는 Stack을 제외하고 공유된 자원을 사용한다. 이때 하나의 자원에 여러 개의 스레드가 동시에 접근할 경우 문제가 생긴다. 그 중 가장 대표적인 문제가 DeadLock이다.

DeadLock

데드락은 프로세스가 자원을 획득하지 못해 다음 처리를 하지 못하는 무한 대기 상황(교착 상태)를 뜻한다.

프로세스 A가 자원 1을 점유한 상태이고 프로세스 B가 자원 2를 점유한 상태이다. 이때 프로세스 A가 자원 2를 요구하고 프로세스 B가 자원 1을 요구하면 서로가 자원을 점유한 상태기 때문에 무한 대기 상태에 빠진다. 이를 데드락이라 한다.

세마포어와 뮤텍스

세마포어와 뮤텍스는 모두 병렬 프로그래밍 환경에서 공유 자원 접근의 상호 배제를 위해 사용된다. 세마포어는 프로세스간 공유된 자원에 접근 시 문제 발생 가능성을 막기 위해 단 한 개의 프로세스만 접근 가능하도록 한다.

뮤텍스는 공유된 자원의 데이터를 여러 스레드가 접근하는 것을 막는 것이지만 세마포어와의 가장 큰 차이점은 한 번에 여러 스레드가 접근 가능하다는 것이다. 즉 세마포어는 뮤텍스가 될 수 있지만 뮤텍스는 세마포어가 될 수 없다.

메인 스레드와 서브 스레드 & 공유 자원 without Synchronization

서브 스레드 & 공유 자원

class FakeDataStore:
    # 공유 변수(value)
    def __init__(self):
        self.value = 0

    # 변수 업데이트 함수
    def update(self, n):
        logging.info("Thread %s: starting update", n)

        # 뮤텍스 & Lock 등 동기화(Thread synchronization) 필요
        local_copy = self.value
        local_copy += 1
        time.sleep(0.1)
        self.value = local_copy

        logging.info("Thread %s: finishing update", n)

메인 스레드

if __name__ == "__main__":
    # Logging format 설정
    format = "%(asctime)s: %(message)s"
    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")

    # 클래스 인스턴스화
    store = FakeDataStore()

    logging.info("Testing update. Starting value is %d.", store.value)

    # With Context 시작
    with ThreadPoolExecutor(max_workers=2) as executor:
        for n in ['First', 'Second', 'Third']:
            executor.submit(store.update, n)

    logging.info("Testing update. Ending value is %d.", store.value)

실행 결과

실행 결과를 살펴보면 결과값이 2로 나오는 것을 확인할 수 있다. 서브 스레드는 공유 자원 값을 가져와서 1을 더하는 작업을 한다. 그렇기 때문에 결과값는 3이 나와야 한다.

하지만 공유 자원에 동기화 작업을 하지 않았기 때문에 Thread First와 Thread Second는 동시에 공유자원에 접근하였고 초기 공유 값인 0을 가져온 것이다.

서브 스레드 With Synchronization

class FakeDataStore:
    # 공유 변수(value)
    def __init__(self):
        self.value = 0
        # Lock 선언
        self._lock = threading.Lock()

    # 변수 업데이트 함수
    def update(self, n):
        logging.info("Thread %s: starting update", n)

        # 뮤텍스 & Lock 등 동기화(Thread synchronization) 필요
        
        # Lock 획득(방법1)
        self._lock.acquire()
        logging.info("Thread %s has lock", n)
        
        local_copy = self.value
        local_copy += 1
        time.sleep(0.1)
        self.value = local_copy

        logging.info("Thread %s about to release lock", n)

        # Lock 반환
        self._lock.release()

        logging.info("Thread %s: finishing update", n)