A Polite Interruption

We thought we were being clever when unit testing some IO code using PipedInputStreams instead of an implementation of InputStream that used system resources. Oops.

No less an expert than Brian Goetz is somewhat misleading when he says:

"The input and output stream classes may block waiting for an I/O to complete, but they do not throw InterruptedException, and they do not return early if they are interrupted." [1]

PipedInputStream behaves differently (from what I can tell) to all other implementations of InputStream when the thread blocking on its read method is interrupted. It returns immediately with an InterruptedIOException (basically, an InterruptedException that fits into the IOException class hierarchy).

From the JDK (1.6) source code:

    /**

     * The index of the position in the circular buffer at which the

     * next byte of data will be stored when received from the connected

     * piped output stream. in<0 implies the buffer is empty,

     * in==out implies the buffer is full

     * @since   JDK1.1

     */

    protected int in = -1;

.

.

    public synchronized int read()  throws IOException {

.
.

  while (in < 0) {

    if (closedByWriter) {

      /* closed by writer, return EOF */

      return -1;

    }

    if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) {

      throw new IOException("Pipe broken");

    }

            /* might be a writer waiting */

    notifyAll();

    try {

        wait(1000);

    } catch (InterruptedException ex) {

      throw new java.io.InterruptedIOException();

    }

  }

The read() method it just checking to see if anybody called interrupt() on the Thread running this code and if they had then the exception is thrown.

Although a call to Thread.interrupt() ultimately hits a native method, using strace I could not see any kernel calls being made (JDK build 1.7.0-ea-b125, Linux 2.6.32.9-70.fc12.i686). It appears that interrupt() just sets a flag. That is all.

So, how do you stop a thread that's blocking on reading from a stream? One way is to close the underlying resource (eg, by calling Socket.setSoTimeout() if the stream is associated with a socket).

But another way is to use Java NIO. If you look at the Thread.interrupt() method, you'll see this code:

.
.

    /* The object in which this thread is blocked in an interruptible I/O

     * operation, if any.  The blocker's interrupt method should be invoked

     * after setting this thread's interrupt status.

     */

    private volatile Interruptible blocker;

.

.

    public void interrupt() {

.

.

    Interruptible b = blocker;

    if (b != null) {

      interrupt0(); // Just to set the interrupt flag

      b.interrupt();

      return;

    }

The object on which Thread.interrupt in turn calls interrupt is an instance of sun.nio.ch.Interruptible. This instance is set in the package protected blockedOn method and the call tree of this looks like:

blockedOn(Interruptible) : void - java.lang.Thread

blockedOn(Thread, Interruptible) : void - java.lang.new JavaLangAccess() {...}

[constructor] new JavaLangAccess() {...} - java.lang 

[callers]

blockedOn(Interruptible) : void - java.nio.channels.spi.AbstractInterruptibleChannel

begin() : void - java.nio.channels.spi.AbstractInterruptibleChannel

begin() : void - java.nio.channels.spi.AbstractSelector

Looking at java.nio.channels.spi.AbstractInterruptibleChannel, for instance, we find that the Interruptible is being used as a callback:

    interruptor = new Interruptible() {

    public void interrupt() {

      synchronized (closeLock) {

        if (!open)

          return;

        interrupted = true;

        open = false;

        try {

          AbstractInterruptibleChannel.this.implCloseChannel();

        } catch (IOException x) { }

      }

    }};

Whereupon:

"An implementation of this method must arrange for any other thread that is blocked in an I/O operation upon this channel to return immediately, either by throwing an exception or by returning normally." [2]

And this is how NIO works: the Thread class needed to be changed to accommodate the ability to interrupt a thread blocking on input. It also closes the underlying resource as explained in Java NIO:

"[T]he blocked thread will be sent a ClosedByInterruptException.  Additionally, if a thread's interrupt status is set, and that thread attempts to access a channel, the channel will immediately be closed, and the same exception will be thrown... It may seem rather draconian to shut down a channel just because a thread sleeping on that channel was interrupted. But this is an explicit design decision made by the NIO architects. Experience has shown that it's impossible to reliably handle interrupted I/O operations consistently across all operating systems. The requirement to provide deterministic channel behavior on all platforms led to the design choice of always closing channels when I/O operations are interrupted. This was deemed acceptable, because a thread is most often interrupted so it can be told to shut down. The java.nio package mandates this behavior to avoid the quagmire of operating-system peculiarities, which is especially treacherous in this area. This is a classic trade-off of features for robustness." [3]

[1] http://www.ibm.com/developerworks/java/library/j-jtp05236/index.html
[2] http://docs.oracle.com/javase/6/docs/api/java/nio/channels/spi/AbstractInterruptibleChannel.html
[3] Java NIO (O'Reilly) - Ron Hitchens