Remove Thread.interrupt() restriction. about jeromq HOT 16 CLOSED

Thank you for the comment.

"Close all the sockets properly otherwise Context.term() will wait forever" not actually means socket.close must be called before Context.term is called. Because Context.term() waits until all the ZMQ sockets are closed, a thread which creates the sockets has responsibility to close its sockets.

The pattern in the guide is also misleading. It was my boring translation of C examples.

while (true) { .. }

must have been correct guide. Sorry for the confusing.

Other example might be useful for clean exit is
https://github.com/zeromq/jeromq/blob/master/src/test/java/guide/interrupt.java

When the term is called, all the blocking call(usually recv) will raise an Exception, then you can do any cleanup including closing zmq socket.

from jeromq.

Today I stumbled upon this, which is very informative about Thread.interrupt(), Thread.interrupted, and InterruptedException in Java.

I looked through the codebase and found 2 places where we appear to be swallowing an InterruptedException -- I made a PR to fix that: #423.

I don't think that my PR solves this issue, but hopefully it is a step in the right direction.

At this point, I'm thinking the problem is that we aren't checking Thread.interrupted() often enough in our code, and there must be at least one place where we should be detecting that the thread is interrupted and taking the opportunity to shut down cleanly, but we aren't.

from jeromq.

Hi Miniway,

"Context.term() must be called before interrupting a thread by Thread.interrupt. Otherwise internal pipe which is used to signal ZMQ commands will be shutdown unexpectedly."

I have the same problem as the original question on this thread....The context.term restriction before calling interrupt is turning out to be a big issue for my use case. The thread that receives message can get block on some other call when processing the message so I need to interrupt that thread when stopping the application. I cannot interrupt it before terminating the context but if the thread is actually blocked on something outside ZMQ, context termination will hang until I interrupt because of the open sockets.....In the interrupt example that you mentioned above, just add the following line in run method and you will see what I am talking about.

     public void run() {
        ZMQ.Socket socket = context.socket(ZMQ.REP);
        socket.bind("tcp://*:5555");

        while (!Thread.currentThread().isInterrupted()) {
            try {
                socket.recv (0);
                blocking wait on something........
            } catch (ZMQException e) {
                if (e.getErrorCode () == ZMQ.Error.ETERM.getCode ()) {
                    break;
                }
            }
        }

        socket.close();
     }

Any help on how to cleanly handle this is appreciated. Thanks in advance for your help.

from jeromq.

I tried the suggested example but I get different exceptions. When I do context.term() I get zmq.ZError$IOException: java.nio.channels.ClosedByInterruptException in the run method and the same when I close the socket, then when I rerun the app I get org.zeromq.ZMQException: Errno 48 : Address already in use. I am using jeromq 0.3.4

from jeromq.

I had omitted the join thread after the interrupt, it now works as per the example, apologies

from jeromq.

I understand this restriction of termination is awkward and needs to be revised fundamentally.

There're some other (may be not that fancy) alternatives I've often used until then.

• Using shadow context and calls zcontext.destory in main thread. https://github.com/miniway/zper/blob/master/src/main/java/org/zper/ZPer.java
• Creating an inproc socket between main thread and worker threads, then send a control message to them. The worker threads need to poll on the inproc together. https://github.com/zeromq/jyre/blob/master/src/main/java/org/zyre/ZreInterface.java#L690
• recv with a timeout and check a shared control variable.

from jeromq.

+1 It would be great if jeromq was Thread.interrupt safe.

from jeromq.

+1 It would be great if jeromq was Thread.interrupt safe. :)

from jeromq.

+1 It would be great if jeromq was Thread.interrupt safe. :)

from jeromq.

+1
On Jul 6, 2016 7:49 PM, "Barbara Gomes" [email protected] wrote:

+1 It would be great if jeromq was Thread.interrupt safe. :)

—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub
#116 (comment), or mute
the thread
https://github.com/notifications/unsubscribe/ADhKBkp6beJ2acRHrBBOfLnuI0RtYr2zks5qS-qIgaJpZM4BPiOg
.

from jeromq.

I understand this restriction of termination is awkward and needs to be revised fundamentally.

@miniway It's been a while -- wondering if you've had any further insights on this? I would be interested in taking a stab at a PR to implement a fix for this issue, but I could use some direction on what the best approach might be.

from jeromq.

I'm sorry that I have to say I haven't had time to look into it. Also I haven't heard that someone started to work on this. So if you have an idea to fix this issue, PR might be the best approach and will be appreciated.

from jeromq.

OK -- I'll do some brainstorming and see what I can come up with.

Questions for everyone who has participated in this thread so far ( @inscriber @amartinwest @manishag @virtuald @zhivko @barbaragomes ), so I can better understand the problem:

• How do you want this work, ideally?

• Is interrupting the thread the preferred way to end a thread?

• How do you feel about using shared global state and doing something like this?:

  boolean running = true;
  
  while (running) {
    // receive and handle messages here
  }
  
  // shut down sockets and destroy context here

  You could run the while loop above in any number of worker threads, and then in the main thread when you want to shut the workers down, you could set running = false;.

  Would that solve this problem, or are there pitfalls I'm not aware of?

  If this is a good pattern, then perhaps no code change is needed, and we just need to update our examples to discourage the use of Thread.interrupt().

• Is there any advantage to using Thread.interrupt() instead of doing something like my example above?

• Is the shared global state idea compatible with adding shutdown hooks? In other words, could you add a shutdown hook that sets running = false; and somehow wait for all the threads to finish? Would we need some special handling that lets the main thread know when all the worker threads are done, or does this happen automatically?

Any insight on this would be really helpful -- I'm new to this project.

from jeromq.

I think the point I was missing in my last post is that the work loop can involve blocking calls when receiving, sending or polling. So even if running got set to false, the worker thread can still potentially be blocked, e.g. waiting for a message.

From what @miniway said, it sounds like the while (!Thread.currentThread().isInterrupted()) thing we see so often in the examples is not actually the correct way to respond to the context being terminated.

It seems like the correct way to do this is to wrap your blocking calls in a try/catch, capture ZMQExceptions and check if the error code is ETERM.

Quoting @miniway earlier in this issue:

Other example might be useful for clean exit is
https://github.com/zeromq/jeromq/blob/master/src/test/java/guide/interrupt.java

When the term is called, all the blocking call(usually recv) will raise an Exception, then you can do any cleanup including closing zmq socket.

So I think the work loops in the zguide examples ought to look more like this:

while (true) {
  try {
    // handle messages / do work here
  } catch (ZMQException e) {
    if (e.getErrorCode () == ZMQ.Error.ETERM.getCode ()) {
      break;
    }
  }
}

// if using ZMQ.Context
socket.setLinger(0);
socket.close();
context.term();

// else if using ZContext, which handles closing sockets for you
context.destroy();

This still doesn't address the fact that you can't call .interrupt() on a blocked worker thread, but unless I'm missing something, I don't see why you would ever need to do that, when it's probably better to do one of these two things instead:

• Terminate the context in the main thread, causing the ETERM exception to get caught and the worker thread to break out of the while loop and shut down.
• Send a "shutdown" message from the main thread and when the worker receives it, it can handle that type of message by shutting down. (This may be useful if you only want to shut down a particular worker thread, but you still want to keep your ZMQ application running and therefore don't want to terminate the context.)

from jeromq.

I re-read this thread and I think I fully understand the problem now.

Clearly it is possible to unblock a worker thread that is blocked on a ZMQ task like waiting to receive a message. You can either send it a shutdown message or terminate its context and let it handle the ETERM exception.

But it becomes problematic when the worker thread gets blocked on something outside of JeroMQ's control. We need some way of interrupting the thread so we don't have to wait for it to finish before the context can terminate. But, we supposedly can't call Thread.interrupt() before terminating the context, so where does that leave us?

I don't have an answer yet, but I at least thought I'd summarize the problem, in case it's helpful.

from jeromq.

Open question: what is the undesirable thing that happens when you call Thread.interrupt() before terminating a context?

The JeroMQ wiki says:

Context.term() must be called before interrupting a thread by Thread.interrupt. Otherwise internal pipe which is used to signal ZMQ commands will be shutdown unexpectedly.

But I have no idea what "internal pipe which is used to signal ZMQ commands will be shutdown unexpectedly" means. It would be useful to know where the problem is in the codebase so we can potentially fix it. Perhaps there is some way we can catch InterruptedException and shut down cleanly.

from jeromq.