부록: 예제

튜플 이터레이터

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

이를 사용하여 스레드 안전 프로그램을 구성하는 작업은 독자에게 연습 문제로 남겨 둡니다.

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__()

카운터

이 예제는 경쟁 조건이 없는 카운터를 만드는 방법을 보여 줍니다. 이는 경쟁 조건 없는 연산에 뮤텍스를 사용하는 방법을 보여 주기 위한 것일 뿐입니다. 효율적인 공유 카운터에는 경합을 피하기 위한 추가 메커니즘이 필요합니다.

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 메서드는 get과 set 사이에 다른 스레드가 값을 수정할 수 있으므로 스레드 안전하지 않습니다.

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")