부록: 예제

튜플 이터레이터

이 예제는 protect 메커니즘을 사용하여 객체를 여러 ThreadGroup에서 사용할 수 있는 동기화된 객체처럼 보이게 만드는 방법을 보여 줍니다.

이를 사용하여 스레드 안전한 프로그램을 구성하는 것은 독자의 연습 과제로 남겨 둡니다.

from threading import Lock

class SynchronizedTupleIter:

    def __init__(self, iterable):
        self.mutex = Lock()
        with self.mutex:
            self._iterator = self.mutex.protect(iter(iterable))
        self.__freeze__()

    def __iter__(self):
        return self

    def __next__(self):
        with self.mutex:
            return self._iterator.__next__()

카운터

이 예제는 경쟁 상태가 없는 Counter를 만드는 방법을 보여 줍니다. 이는 단지 경쟁 상태 없는 연산을 위해 뮤텍스를 사용하는 방법을 보여 주기 위한 것입니다. 효율적인 공유 카운터라면 경합을 피하기 위해 추가적인 메커니즘을 사용해야 합니다.

class MutableInt:

    def __init__(self, value):
        self.value = value

class Counter:

    def __init__(self):
        self.mutex = Lock()
        with self.mutex:
           self.number = self.mutex.protect(MutableInt(0))
        self.__freeze__()

    def value(self):
        with self.mutex:
            return self.number.value

    def increment(self):
        with self.mutex:
            self.number.value += 1

안전하지 않은 카운터

보호는 스레드 안전성을 보장하지 않으며, 단지 잠금 규율을 강제할 뿐입니다. 이로 인해 실수로 스레드 안전하지 않은 코드를 작성하기는 더 어려워지지만, 불가능해지는 것은 아닙니다. 이 예제에서 increment 메서드는 값을 읽는 시점과 설정하는 시점 사이에 다른 스레드가 값을 수정할 수 있으므로 스레드 안전하지 않습니다.

class MutableInt:

    def __init__(self, value):
        self.value = value

class Counter:

    def __init__(self):
        self.mutex = Lock()
        with self.mutex:
           self.number = self.mutex.protect(MutableInt(0))
        self.__freeze__()

    def value(self):
        with self.mutex:
            return self.number.value

    def set_value(self, val):
        with self.mutex:
            self.number.value = val

    def increment(self):
        val = self.value()
        self.set_value(val+1)

경쟁 상태를 허용하는 대신 중단하기

특정 알고리즘에서는 공유 입력을 추가로 보호하는 것이 비실용적이거나 별다른 가치가 없을 수 있습니다. 이 PEP는 코드가 동시성을 처리하는 대신 연산을 중단할 수 있도록 허용합니다. 직렬화 라이브러리가 이러한 경우에 해당할 수 있습니다:

def dump(mapping: dict):
    if mapping.__shareable__ is SYNCHRONIZED:
        raise ValueError("cannot cope with data races.")
    # other states are fine:
    #   LOCAL -- no concurrent accesses
    #   PROTECTED -- mutual exclusion prevents races
    #   IMMUTABLE -- no concurrent modifications
    for key, value in mapping.items():
        dump_one(key, value)

파일에 대한 접근 직렬화하기

여러 스레드가 같은 파일에 동시에 쓰도록 허용하면 비결정적인 동작만 발생할 수 있습니다. 몇 가지 간단한 직렬화 메커니즘을 구현할 수 있습니다:

class ThreadSectionedFile:

    def __init__(self, f: file):
        self._lock = Lock()
        with self._lock:
            self._file = self._lock.protect(del f)
        self._sections: dict[Thread, list[bytes]] = dict().synchronized()

    def __enter__(self):
        self._sections[threading.current_thread()] = []
        # Note that the list is thread-local, no other thread may
        # inadvertently write into it.

    def write(self, data: bytes):
        me = threading.current_thread()
        if me not in self._sections:
            raise Exception("must call __enter__")
        self._sections[me].append(data)

    def __exit__(self, t, v, tb):
        me = threading.current_thread()
        data = self._sections[me]
        del self._sections[me]
        with self._lock:
            self._file.write(f"Thread {me.name} says:\n".encode())
            for d in data:
                self._file.write(d)
            self._file.write(b"\n")