Das deutsche Python-Forum

Hi,

folgendes Problem.

Ich habe einen Server der mehreren Clients sagen soll, wann sich ein interner State ändert.
Dazu benutze ich momentan eine angepasste threading._Event Klasse der Form:

Code: Alles auswählen

import threading
class StateEvent(threading._Event):
    def __init__(self):
        threading._Event.__init__(self)
        self.state = None
        self.lock = threading.Lock()
    def set_state(self, state):
        with self.lock:
            self.state = state
            self.set()
            self.clear()
    def wait_state(self, timeout=None)
        self.wait(timeout)
        return self.state

Meine Frage ist nun:
1. Ist das ganze wirklich Threadsafe?
2. Ist das die "schönste" Möglichkeit oder kann man das etwas eleganter machen?

Die Clients müssen nicht jeden State bekommen, nur möglichst schnell den letzten (das ganze ist eine Lichtsteuerung, ob ich 1000 den Schalter ein und aus mache ist egal solange das Programm schnellstmögliches das Licht an macht wenn er zum Schluss auf "On" steht.

Alternativ könnte ich auch das unterliegende Event umschreiben. Das würde dann Afaik etwa soa ussehen (ungetestet)

Code: Alles auswählen

from threading import Condition, Lock


class StateEvent(object):
	def __init__(self):
		self._condition = Condition(Lock())
		self._state = None
		self._flag = False

	def _reset_internal_locks(self):
		# private!  called by Thread._reset_internal_locks by _after_fork()
		self._condition.__init__()

	def send(self, state):
		with self._condition:
			self._state = state
			self._flag = True
			self._condition.notify_all()
			self._flag = False

	def get(self, timeout=None):
		with self._condition:
			if not self._flag:
				self._condition.wait(timeout)
			return self._state

Den Code vom Originalen Event bekomtm man hier: http://hg.python.org/cpython/file/2.7/Lib/threading.py

Beide Varianten funktionieren nicht.

Bei der ersten wird ein zweites Lock eingeführt. Das verwendest du dann völlig unabhängig von dem Lock der geerbten Condition. set_state() verwendet self.lock zum gegenseitigen Ausschluss und wait_state() verwendet das self.__cond.__lock der Parent-Klasse. Gegenseitiger Ausschluss bei gemeinsam genutzen Ressourcen (hier self.state) funktioniert aber nur, wenn immer dasselbe Ausschlusskriterium verwendet wird. Dein Programm hat nun aber zwei voneinander unabhängige Kriterien (self.lock und self.__cond der Parent-Klasse), es ist also nicht thread safe.

Die zweite Variante funktioniert ebenfalls nicht, weil Condition.notify_all() die Notifikation nicht sofort auslöst, sondern erst, wenn das Lock der Condition das nächste mal freigegeben wird. In deinem Fall knippst du in StateEvent.send() das Attribut self._flag an, benachrichtigst die Condition, dass nach der nächsten Freigabe ihres Locks alle wartenden Threads benachrichtigt werden sollen, dann knippst du self._flag wieder aus und dann erst gibst du das Lock von self._condition wieder frei. Alle schlafenden Threads werden jetzt benachrichtigt, dass es neue Daten gibt. Diejenigen dieser Threads, die auf StateEvent.get() warten, schlafen nun gleich wieder ein, weil in StateEvent.get() ja getestet wird, ob self._flag gesetzt ist. Das ist es aber nie, weil du es ja zuvor immer schon ausgeknippst hast.

Selbst, wenn beide Varianten tun würden, was du möchtest, würden nur diejenigen Clients die neue Information erhalten, die zufällig gerade auf einen Rückgabewert aus StateEvent.get() warten. Ist es wirklich das, was du möchtest? Soll nicht allen Clients diese Information zur Verfügung stehen, auch denen, die gerade was anderes tun oder auf etwas anderes warten?

Desweiteren frage ich mich, warum du die Kommunikation zwischen den Clients und dem Server überhaupt so kompliziert machen willst. Warum verwendest du nicht einfach das Observer Pattern? Beschreib doch mal genauer, was dein eigentliches Problem ist, dann kann man vielleicht eine einfachere Lösung finden.

Ich sollte es also eher so machen?
(hab das mit der Flag entfernt weil es eigentlich hier keinen Nutzen hat)

Code: Alles auswählen

from threading import Condition, Lock


class StateEvent(object):
	def __init__(self):
		self._condition = Condition(Lock())
		self._state = None

	def _reset_internal_locks(self):
		# private!  called by Thread._reset_internal_locks by _after_fork()
		self._condition.__init__()

	@property
	def state(self):
		return self._state

	def send(self, state):
		with self._condition:
			self._state = state
			self._condition.notify_all()

	def get(self, timeout=None):
		with self._condition:
			self._condition.wait(timeout)
			return self._state

Zum Verhalten des Codes: Doch, genau dieses Verhalten will ich. Zur Erklärung.
Wir haben hier alle Lampen etc. ansteuerbar gemacht und wollen nun "States" definieren die Beschreiben welche Lampen an sind und welche nicht. Als Beispiel sagen wir einfach alle Lampen an ist State 1, alle aus ist 0. Nun brauchen die Lampen aber eine Schaltzeit um von 0 -> 1 zu gehen während der sie auch nicht auf externe Kommandos reagieren. Sie sollen deshalb immer nur schnellstmöglicht den letzten gegebenen Zustand annehmen.
Beispiel:

Code: Alles auswählen

Schalter: 01010101010111100111100101011011001111001010011100000000000000
State   : 00----Verarbeitungszeit--->111----Verarbeitungszeit--->0000000
Lampen  : 00000000000000000000000000011111111111111111111111111110000000

Ob nun jemand wild den Schalter betätigt während die Lampe angeht ist mir egal da das anschalten der Lampen nicht unterbrechbar ist. Nach dem Anschalten vergleiche ich ob der momentane State der Lampe auch der letzte gesehene State ist. Wenn ja warte ich mit wait() auf einen neuen State, wenn nein versuche ich sofort den nun gesehenen State anzunehmen.

Fällt mir etwas schwer aus deinen Erklärungen zu verstehen, wie genau auf die Lampen zugegriffen werden soll. Ich gehe mal davon aus, dass du eine Anzahl von Lampen hast und möchtest dass mehrere Leute gleichzeitig auf eine Lampe zugreifen können. Wenn eine Lampe aktiviert wird und dies 10 Sekunden benötigt, sollen die parallel laufenden Threads(andere Leute) diesen Vorgang nicht unterbrechen können.

In diesem Fall solltest du Queue verwenden.

@Dami123

Ne, du hast es genau Falsch rum (glaube ich)
Eigentlich habe ich ein klassisches Obervers Model.

Ich habe einen Server (Schalter) der seinen State (an/aus) zu einer Menge von Clients (Lampen) bringen will.
Server und Clients sind jeweils in einzelnen Threads bzw. zum Teil auch noch übers Netzwerk Verteilt.
Die Clients sagen dem Server "Sag mir Bescheid wenn sich der State ändert" in dem sie ein StateEvent.get machen.
Nun braucht aber jeder Client zum verarbeiten des neuen States eine gewissen Zeit (zum Teil 10s weil einige z.B. über Funk angeschlossen sind und da halt mehrfach gesendet werden muss).
Nun entgeht dem Client zwar die state changes während des Ausführens, die will er aber eh nicht weiler ja immer nur den letzten state einstellen soll.

Das deutsche Python-Forum

threading.Event ->Wert zurückgeben

threading.Event ->Wert zurückgeben

Re: threading.Event ->Wert zurückgeben

Re: threading.Event ->Wert zurückgeben

Re: threading.Event ->Wert zurückgeben

Re: threading.Event ->Wert zurückgeben

Re: threading.Event ->Wert zurückgeben