Zeromq: Reset REQ/REP Socket Status

Question

Wenn Sie das einfache ZeroMQ REQ/REP Muster verwenden, sind Sie auf eine feste send() -> recv()/recv() -> send() Sequenz angewiesen. Als this Artikel beschreibt Sie in Schwierigkeiten geraten, wenn ein Teilnehmer mitten in einer Anfrage trennt, weil dann Sie nicht einfach mit dem Empfang der nächsten Anfrage von einer anderen Verbindung neu beginnen, aber die State-Maschine würde Sie zwingen, eine Anfrage an die Verbindung zu senden ein.

Hat sich eine elegantere Lösung herauskristallisiert, seit der erwähnte Artikel geschrieben wurde?

eine erneute Verbindung der einzige Weg, dies zu lösen (abgesehen von nicht REQ/REP verwenden, aber verwenden Sie ein anderes Muster)

Answer 1

Die gute Nachricht ist, dass, wie von ZMQ 3.0 und höher (die Neuzeit) können Sie eine Zeitüberschreitung für einen Socket festlegen. Wie andere haben an anderer Stelle erwähnt, müssen Sie dies tun, nachdem Sie den Socket erstellt haben, aber bevor Sie verbinden es:

zmq_req_socket.setsockopt(zmq.RCVTIMEO, 500) # milliseconds

Dann, wenn Sie tatsächlich versuchen, die Antwort zu erhalten (nachdem Sie eine Nachricht gesendet haben den REP-Socket), können Sie den Fehler abfragen, der bei Überschreiten des Timeouts gemeldet wird:

Jedoch! Wenn dies passiert, wird Ihr Socket in einem lustigen Zustand sein, da er immer noch die Antwort erwartet. An dieser Stelle kann ich nichts finden, was zuverlässig funktioniert, außer einfach den Socket neu zu starten.Beachten Sie, dass, wenn Sie den Socket trennen() und dann erneut verbinden(), es immer noch in diesem schlechten Zustand ist. So müssen Sie

def reset_my_socket: 
    zmq_req_socket.close() 
    zmq_req_socket = zmq_context.socket(zmq.REQ) 
    zmq_req_socket.setsockopt(zmq.RCVTIMEO, 500) # milliseconds 
    zmq_req_socket.connect(zmq_endpoint) 

Sie werden auch feststellen, dass, weil ich in der Nähe() d die Buchse, die Timeout-Option erhalten wurde „verloren“, so ist es wichtig, festgelegt, dass auf dem neuen Sockel ist.

Ich hoffe, das hilft. Und ich hoffe, dass dies nicht die beste Antwort auf diese Frage ist. :)

Answer 2

eine Lösung gibt, um dies und das ist das Hinzufügen von Timeouts für alle Anrufe. Da ZeroMQ selbst keine einfache Timeout-Funktionalität bietet, empfehle ich eine Unterklasse des ZeroMQ-Sockets, die allen wichtigen Aufrufen einen Timeout-Parameter hinzufügt.

Also, anstatt s.recv() aufzurufen, würden Sie s.recv aufrufen (timeout = 5.0) und wenn eine Antwort nicht innerhalb dieses 5 Sekunden Fensters zurückkommt, gibt sie None zurück und hört auf zu blockieren. Ich hatte einen vergeblichen Versuch unternommen, als ich auf dieses Problem stieß.

Answer 3

Ich schaue mir das gerade an, weil ich ein Legacy-System nachrüste.

Ich stoße ständig auf Code, der über den Zustand der Verbindung "braucht". Die Sache ist jedoch, dass ich zum Paradigma der Nachrichtenübergabe übergehen möchte, das die Bibliothek fördert.

fand ich die folgende Funktion: zmq_socket_monitor

Was sie tut, ist die Steckdose, um es und erzeugen Ereignisse übergeben zu überwachen, die auf einen „inproc“ Endpunkt dann übergeben werden - an diesem Punkt Sie hinzufügen können, den Code verarbeitet tatsächlich zu tun etwas.

Es gibt auch ein Beispiel (tatsächlich Code-Test) hier: github

Ich habe keine spezifischen Code bekam zur Zeit zu geben (vielleicht am Ende der Woche), aber meine Absicht ist es, die reagieren verbinden und trennen, so dass ich tatsächlich jedes Zurücksetzen der Logik durchführen kann.

Hoffe, dass dies hilft, und trotz der 4.2 Dokumente zitieren, verwende ich 4.0.4, die die Funktionalität auch zu haben scheint.

Hinweis ich Sie über Python oben sprechen bemerken, aber die Frage ist C markiert ++ so das ist, wo meine Antwort herkommt ...

Answer 4

Da die angenommene Antwort so schrecklich traurig zu mir scheint, habe ich einige Nachforschungen gemacht und festgestellt, dass alles, was wir brauchen, tatsächlich in der Dokumentation war.

Die .setsockopt() mit dem richtigen Parameter können Sie Ihre Socket-Zustand-Maschine helfen, sie brutal zu zerstören, ohne und Wiederaufbau eine andere auf dem vorherigen Leiche zurückzusetzen.

(ja ich mag das Bild).

ZMQ_REQ_CORRELATE: match replies with requests
The default behavior of REQ sockets is to rely on the ordering of messages to match requests and responses and that is usually sufficient. When this option is set to 1 , the REQ socket will prefix outgoing messages with an extra frame containing a request id. That means the full message is (request id , identity , 0 , user frames…). The REQ socket will discard all incoming messages that don't begin with these two frames.
Option value type int
Option value unit 0 , 1
Default value 0
Applicable socket types ZMQ_REQ

ZMQ_REQ_RELAXED: relax strict alternation between request and reply
By default, a REQ socket does not allow initiating a new request with zmq_send(3) until the reply to the previous one has been received. When set to 1, sending another message is allowed and has the effect of disconnecting the underlying connection to the peer from which the reply was expected, triggering a reconnection attempt on transports that support it. The request-reply state machine is reset and a new request is sent to the next available peer.
If set to 1 , also enable ZMQ_REQ_CORRELATE to ensure correct matching of requests and replies. Otherwise a late reply to an aborted request can be reported as the reply to the superseding request.
Option value type int
Option value unit 0 , 1
Default value 0
Applicable socket types ZMQ_REQ

A complete documentation is here