IntEnumValue – Ein multiprocessing-sicherer Enum-Wert für Python

Ankündigung der ersten Veröffentlichung des Moduls python-multiprocessing-intenum, das einen multiprocessing-sicheren Enum-Wert für die Python-Programmiersprache bereitstellt.

Hintergrund

Innerhalb eines großen Kundenprojekts mit mehreren Komponenten, die in einer Kubernetes-Umgebung laufen, standen wir vor einem spezifischen Problem, für das es keine generische Lösung gab, als wir eine Daemon-Worker-Komponente implementiert haben.
Da immer nur eine Berechnung gleichzeitig laufen kann, verfügt die Komponente über einen internen Zustand, um zu verfolgen, ob sie sich im Leerlauf befindet und bereit ist, neue Berechnungen anzunehmen, mit einer Berechnung beschäftigt ist oder darauf wartet, sich herunterzufahren, z. B. während eines Rolling Upgrade.
Neben dem eigentlichen Berechnungsprozess, der erzeugt wird, stellt die Komponente eine auf Django Rest Framework basierende API mittels mehrerer Webserver-Prozesse oder -Threads bereit, die es einer anderen Scheduler-Komponente ermöglicht, den aktuellen Zustand des Workers abzufragen und neue Berechnungsaufträge einzureichen. Daher muss auf den internen Zustand sicher von mehreren Prozessen und/oder Threads aus zugegriffen und dieser geändert werden können.

Der natürliche Datentyp, um einen Zustand in den meisten High-Level-Programmiersprachen einschließlich Python abzubilden, ist ein Enum-Typ, dem man symbolische Namen zuordnen kann, die den Zuständen entsprechen. Wenn es darum geht, Daten über mehrere Prozesse oder Threads hinweg in Python auszutauschen, kommt das Modul „ multiprocessing “ ins Spiel, das gemeinsam genutzte „ ctypes “ Objekte bereitstellt, d. h. grundlegende Datentypen wie Zeichen und Integer, sowie Locking-Primitive, um den Austausch größerer, komplexer Datenstrukturen zu handhaben.
Es bietet jedoch keine fertig nutzbare Möglichkeit, einen Enum-Wert auszutauschen.

Da das Problem ziemlich verbreitet ist, erforderte es eine generische und wiederverwendbare Lösung, und so wurde die Idee für „ IntEnumValue “ geboren, eine Implementierung eines multiprocessing-sicheren, gemeinsam genutzten Objekts für „ IntEnum “ Enum-Werte.

Funktionen

Für den Benutzer erscheint es und kann es wie ein Python „ IntEnum“ verwendet werden, das ein Enum ist, dessen benannte Werte jeweils einem eindeutigen Integer entsprechen.
Intern verwendet es dann ein „ multiprocessing.Value “ gemeinsam genutztes „ ctypes “ Integer-Objekt, um diesen Integer-Wert des Enum auf multiprocessing-sichere Weise zu speichern.

Um einen Programmierstil zu unterstützen, der proaktiv vor Programmierfehlern schützt, ist das Modul vollständig typisiert, was eine statische Prüfung mit „ mypy “ ermöglicht und als generische Klasse implementiert ist, die den Typ des zugrunde liegenden spezifischen Enum als Parameter entgegennimmt.

Das Modul python-multiprocessing-intenum wird mit Unit-Tests geliefert, die eine vollständige Abdeckung seines Codes gewährleisten.

Anwendungsbeispiele

Um die Verwendung zu veranschaulichen und die Funktionen besser zu demonstrieren, wird als Beispiel ein Multithread-Worker verwendet, der einen internen Zustand hat, der als Enum dargestellt wird und über seine Threads hinweg verfolgt werden soll.

Um ein multiprocessing-sicheres „ IntEnum“ zu verwenden, definieren Sie zuerst den eigentlichen zugrunde liegenden „ IntEnum “ Typ:

class WorkerStatusEnum(IntEnum):
   UNAVAILABLE = enum.auto()
   IDLE = enum.auto()
   CALCULATING = enum.auto()

IntEnumValue ist eine generische Klasse, die einen Typparameter für den Typ des zugrunde liegenden „ IntEnum“ akzeptiert.
Definieren Sie also den entsprechenden „ IntEnumValue “ Typ für diesen „ IntEnum wie folgt:

class WorkerStatus(IntEnumValue[WorkerStatusEnum]):
   pass

Was dies bewirkt, ist die Definition eines neuen WorkerStatus Typs, der ein „ IntEnumValue “ ist, der parametrisiert ist, um den spezifischen „ IntEnum “ Typ WorkerStatusEnum zu handhaben.
Er kann wie ein Enum verwendet werden, um den Status zu speichern:

>>> status = WorkerStatus(WorkerStatusEnum.IDLE)
>>> status.name
'IDLE'
>>> status.value
2
>>> with status.get_lock():
... status.set(WorkerStatusEnum.CALCULATING)
>>> status.name
'CALCULATING'
>>> status.value
3

Der Versuch, ihn auf einen anderen Typ von „ IntEnum “ zu setzen, wird jedoch als TypeError abgefangen:

>>> class FrutEnum(IntEnum):
... APPLE = enum.auto()
... ORANGE = enum.auto()
>>> status.set(FrutEnum.APPLE)
Traceback (most recent call last):
   File "", line 1, in 
   File "/home/elho/python-multiprocessing-intenum/src/multiprocessing_intenum/__init__.py", line 60, in set
     raise TypeError(message)
TypeError: Can not set '<enum 'WorkerStatusEnum'>' to value of type '<enum 'FrutEnum'>'

Dies hilft, Programmierfehler zu vermeiden, bei denen fälschlicherweise ein anderes Enum verwendet wird.

Neben der direkten Verwendung kann der erstellte WorkerStatus Typ natürlich auch in eine dedizierte Klasse verpackt werden, worauf die verbleibenden Beispiele näher eingehen werden:

class WorkerState:
    def __init__(self) -> None:
        self.status = WorkerStatus(WorkerStatusEnum.IDLE)

Bei Verwendung mehrerer „ multiprocessing.Value “ Instanzen (einschließlich „ IntEnumValue “), die sich ein Lock teilen sollen, um sicherzustellen, dass sie nur in einem konsistenten Zustand geändert werden können, übergeben Sie dieses gemeinsam genutzte Lock als Keyword-Argument bei der Instanziierung:

class WorkerState:
    def __init__(self) ->  None:
       self.lock = multiprocessing.RLock()
       self.status = WorkerStatus(WorkerStatusEnum.IDLE, lock=self.lock)
       self.job_id = multiprocessing.Value("i", -1, lock=self.lock)

    def get_lock(self) -> Lock | RLock:
       return self.lock

Um zu vermeiden, dass die set() Methode aufgerufen werden muss, um dem Attribut „ IntEnumValue “ einen Wert zuzuweisen, wird empfohlen, das eigentliche Attribut für die Klasse privat zu halten und Getter- und Setter-Methoden für eine öffentliche Eigenschaft zu implementieren, die dieses Implementierungsdetail verbirgt, z. B. wie folgt:

class WorkerState:
    def __init__(self) ->  None:
        self._status = WorkerStatus(WorkerStatusEnum.IDLE)

    @property
    def status(self) -> WorkerStatusEnum:
        return self._status # type: ignore[return-value]

    @status.setter
    def status(self, status: WorkerStatusEnum | str) ->  None:
        self._status.set(status)

Das Ergebnis kann eleganter verwendet werden, indem man Werte einfach dem Status-Attribut zuweist:

>>> state = WorkerState()
>>> state.status.name
'IDLE'
>>> with state.get_lock():
... state.status = WorkerStatusEnum.CALCULATING
>>>  state.status.name
'CALCULATING'

Der spezifische „ IntEnumValue “ Typ kann Methoden überschreiben, um weitere Funktionalitäten hinzuzufügen.

Ein häufiges Beispiel ist das Überschreiben der set() Methode, um Logging hinzuzufügen:

class WorkerStatus(IntEnumValue[WorkerStatusEnum]):
     def set(self, value: WorkerStatusEnum | str) ->  None:
         super().set(value)
         logger.info(f"WorkerStatus set to '{self.name}'")

Die gemeinsame Nutzung all dieser Funktionen und Anwendungsfälle ermöglicht die elegante, kohäsive und robuste Handhabung des internen Zustands von Multiprocessing-Workern.