zeromq: reset stato socket REQ/REP

Question

Quando si utilizza il pattern ZeroMQ REQ/REP, si dipende da una sequenza send() -> recv()/recv() -> send() fissa. Come articolo this si descrive un problema quando un partecipante si disconnette nel mezzo di una richiesta perché non è possibile ricominciare da capo con la successiva richiesta da un'altra connessione, ma la macchina a stati ti obbliga a inviare una richiesta al disconnesso uno.

È emerso un modo più elegante per risolvere questo dato che l'articolo menzionato è stato scritto?

sta ricollegando l'unico modo per risolvere questo (oltre a non utilizzare REQ/REP ma usare un altro modello)

Answer 1

La buona notizia è che, come di ZMQ 3.0 e versioni successive (l'epoca moderna), è possibile impostare un timeout su un socket. Come altri hanno notato altrove, è necessario farlo dopo aver creato il socket, ma prima di collegarlo:

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

Poi, quando in realtà tenta di ricevere la risposta (dopo aver inviato un messaggio al la presa REP), è possibile rilevare l'errore che verrà affermato se il timeout è superato:

try: 
    send(message, 0) 
    send_failed = False 

except zmq.Again: 
    logging.warning("Image send failed.") 
    send_failed = True 

Tuttavia! Quando ciò accade, come osservato altrove, il tuo socket si troverà in uno stato divertente, perché sarà comunque in attesa della risposta. A questo punto, non riesco a trovare nulla che funzioni in modo affidabile diverso dal semplice riavvio del socket.Notare che se si disconnette() il socket e quindi si riconnette() esso, sarà ancora in questo stato negativo. Quindi è necessario

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) 

Noterete anche che, poiché chiudo() D la presa, l'opzione di ricezione di timeout è stato "perso", quindi è importante insieme che sul nuovo socket.

Spero che questo aiuti. E spero che questa non sia la migliore risposta a questa domanda. :)

Answer 2

C'è una soluzione a questo e che è l'aggiunta di timeout per tutte le chiamate. Dato che ZeroMQ di per sé non fornisce una funzionalità di timeout semplice, ti consiglio di utilizzare una sottoclasse del socket ZeroMQ che aggiunge un parametro di timeout a tutte le chiamate importanti.

Quindi, invece di chiamare s.recv() chiamereste s.recv (timeout = 5.0) e se una risposta non ritorna in quella finestra di 5 secondi restituirà None e interromperà il blocco. Ho fatto un tentativo inutile a questo quando ho incontrato questo problema.

Answer 3

Attualmente sto esaminando questo in questo momento, perché sono in stile retrò un sistema legacy.

Sto incrociando il codice costantemente che "ha bisogno" di sapere dello stato della connessione. Comunque la cosa è che voglio passare al paradigma di passaggio dei messaggi che la biblioteca promuove.

ho trovato la seguente funzione: zmq_socket_monitor

Ciò che fa è monitorare la presa passata ad esso e generare eventi che vengono poi passati a un endpoint "InProc" - a quel punto è possibile aggiungere il codice di gestione a che fare in realtà qualcosa.

C'è anche un esempio (in realtà il codice di prova) qui: github

non ho alcun codice specifico per dare al momento (forse alla fine della settimana), ma la mia intenzione è quella di rispondere alla connettersi e disconnettersi in modo tale da poter effettivamente effettuare il reset della logica richiesta.

Spero che questo aiuti, e nonostante la citazione di 4.2 documenti, sto usando 4.0.4 che sembra avere la funzionalità pure.

Nota noto si parla di pitone sopra, ma la questione è aggiunto C++ così che è dove la mia risposta è venuta da ...

Answer 4

Poiché la risposta accettata mi sembra così terribilmente triste, ho fatto qualche ricerca e ho scoperto che tutto ciò di cui abbiamo bisogno era effettivamente nella documentazione.

Il .setsockopt() con il parametro corretto può aiutare a reimpostare la presa state-macchina senza brutalmente distruggere e ricostruire un'altra sulla parte superiore del precedente corpo morto.

(sì, mi piace l'immagine).

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