CODEWITHSUNDEEP
javazeromqjeromq
Alessandro Polverini
Alessandro Polverini

Reputation: 2459

How to interrupt a call to .read() method on a JeroMQ socket?

In my code I've a thread waiting for events on a ZeroMQ endpoint, and its main loop looks similar to this:

while (externalCondition) {
  byte[] bytes = subscriber.recv(0);
  // Do things
}

The problem is that I don't know how to interrupt the call to .recv() method, if I want to exit the thread (for example because I want to shut down the app).

I tried to interrupt the thread but this seems not to work.

Upvotes: 2

Views: 444

Answers (1)

user3666197
user3666197

Reputation: 1

Welcome to ZeroMQ and distributed system architectures.

There is no cheap way to interrupt such call ( except a brutal one ), yet

there is an easy way to re-factor the idea, so as it can work in distributed system realms:

/* THIS CODE IS NOT A COPY+PASTE SYNDROME-PROOF, BUT HAS THE DESIGN IDEA CLEAR,
        CODE IS NOT A READY TO RUN, AS THE TARGET LANGUAGE
                AND THE ACTUAL ZeroMQ BINDING / VERSION
                                      WILL DECIDE ON FURTHER DETAILS TO CODE IN
   */
while ( True ) { // -[NEVER-BLOCK]----------- aSoftRealTIME-control-loop-<body>--

   bool ec = externalCondition;
   int  rc = subscriber.poll( someSoftRealTimeDELAY_in_ms );

   if ( ec
      & rc == 0
        ){
        //--------------------------------------------------
       // DO NOTHING, AS THERE IS NOTHING TO WORK ON
      //  or sleep, or
     //   DO ANY SoftRealTime-controlled / yet interruptible
    //           maintenance workpackage(s)
   //-------------------------------------------------------
   }
   if ( !ec ){
        break; // __________________________ J.I.T. / JMP / RET __^
   }
   if (  rc < 0 ){
               // ZeroMQ .poll() Error Analysis & Error Recovery
   }
   if (  rc > 0 ){

               // ZeroMQ has a thing to indeed .recv(), so read it:
         byte[] bytes = subscriber.recv( zmq.DONTWAIT );

         ...   // DO things ... on bytes ...
   }
}  // ---------------[NEVER-BLOCK]----------- aSoftRealTIME-control-loop-<body>--

Nota bene

while this remark might be a fairly well know subject for you, in case you have already got familiar with ZeroMQ internalities and various defaults, yet, let me mention it also here, as not all kind reader may have such deep ZeroMQ practice.

The SUB-side(s) of the Scalable Formal Communication Pattern used above has a dark-side. Unless one explicitly "subscribed" to some newspapers and/or magazine(s), there will nothing arrive into one's SUB-side mailbox, nothing gets delivered at all, as the "delivery service" has not received any subscription from you, telling 'em what you would like to ( receive ) and read

So, always do not forget to submit a call to such a

subscriber.setsockopt( aSubject_What_To_Subscribe_To_Indeed_Receive_STRING );

Not having done so, one will never .recv() a single byte.

Thanks. What would be the "brutal" one? –Alessandro Polverini 23 hours ago

Well, ZeroMQ native API, as of 2018-Q1 permits one, rather brutal way to achieve an enforced interruption of a blocking call to .recv() method - calling a <Context>.term() - so, it works more less like a "Demolition Man" when asking just for a few pieces of paper ... indeed that way, plus there are further pre-requisities not to hang your further code-execution in an uncontrolled zombie-alike state, if not infinitely ( if not having protected all your Socket-instances generated from native-API with an immediately configured .setsockopt( zmq.LINGER, 1 ) ( since native API 2.1+ ~ 4.(?) where a promised change of defaults was published ... respective language binding / wrappers differ in version adoption, so cannot refer in general and best check the right upon instantiations took place with published .getLinger()-exposed method to see this reactively per-incident :o) ).

Calling a WMD-alike annihilation, using .term()-method, will in those cases, where Socket-instances were not left waiting ( indeterminatly, if not infinitely ) by zmq.LINGER-instance attribute, cancel internal-FSA and exit even the still blocked-{ .recv() | .send() | poll() }-s and return the code-execution control-path to your "Demolition Man"-controls :o)

Doable, but a bit harsh, isn't it?

Upvotes: 1

Related Questions