Thread.interrupt() and java.io.InterruptedIOException

Question

I'm running Java 1.5 on Solaris 10. My program is a standalone java program, using java concurrency package and log4j-1.2.12.jar to log certain information. primary logic is as below

ExecutorService executor = new AppThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(Integer.MAX_VALUE), new AppThreadFactory("BSRT", true), new ThreadPoolExecutor.CallerRunsPolicy());
        CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(executor);
        for (final Integer id : taskList) {
            Callable<Integer> c = new Callable<Integer>() {
                public Integer call() throws Exception {
                    int newId = DB operation(id);
                    return newId;
                }
            };
            completionService.submit(c);
        }
        logger.debug("Start retrievie result");
        for (Integer id : taskList) {
            try {
                Future<Integer> future = completionService.poll(1, TimeUnit.SECONDS);               
                Integer taskId=null;
                if (future != null) {
                    logger.debug("future is obtained.");
                    taskId = future.get();
                } else {
                    logger.error("wait too long and get nothing!");
                    break;
                }
                if (taskId != null) {
                    taskIdList.add(taskId);
                }
            } catch (ExecutionException ignore) {
                // log the cause and ignore this aborted task,coninue with
                // next available task.
                logger.warn(ignore.getCause());
            } catch (InterruptedException e) {
                logger.warn("interrupted...");
                // Re-assert the thread’s interrupted status
                Thread.currentThread().interrupt();
            }
        }executor.shutdown();

During the execution of my program, Sometimes (not always) I'm getting this error ...

executor.shutdown(); 

will not be able to interrupt AppThread after return from the call super.run(); because the woker is already removed from workers set used internally by ThreadPoolExecutor, executor does not have reference to AppThread from that point of time.

btw: the log file is accessible and size is big enough.

log4j:ERROR Failed to flush writer,
java.io.InterruptedIOException
       at java.io.FileOutputStream.writeBytes(Native Method)
       at java.io.FileOutputStream.write(FileOutputStream.java:260)
       at sun.nio.cs.StreamEncoder$CharsetSE.writeBytes(StreamEncoder.java:336)
       at sun.nio.cs.StreamEncoder$CharsetSE.implFlushBuffer(StreamEncoder.java:404)
       at sun.nio.cs.StreamEncoder$CharsetSE.implFlush(StreamEncoder.java:408)
       at sun.nio.cs.StreamEncoder.flush(StreamEncoder.java:152)
       at java.io.OutputStreamWriter.flush(OutputStreamWriter.java:213)
       at org.apache.log4j.helpers.QuietWriter.flush(QuietWriter.java:57)
       at org.apache.log4j.WriterAppender.subAppend(WriterAppender.java:315)
       at org.apache.log4j.DailyRollingFileAppender.subAppend(DailyRollingFileAppender.java:358)
       at org.apache.log4j.WriterAppender.append(WriterAppender.java:159)
       at org.apache.log4j.AppenderSkeleton.doAppend(AppenderSkeleton.java:230)
       at org.apache.log4j.helpers.AppenderAttachableImpl.appendLoopOnAppenders(AppenderAttachableImpl.java:65)
       at org.apache.log4j.Category.callAppenders(Category.java:203)
       at org.apache.log4j.Category.forcedLog(Category.java:388)
       at org.apache.log4j.Category.debug(Category.java:257)
       at AppThread.run( AppThread.java: 33)  

33 is the line: if (debug) logger.info("Exiting " + getName());

import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

public class AppThread extends Thread {
    public static final String DEFAULT_NAME = "MyAppThread";
    private static volatile boolean debugLifecycle = false;
    private static final AtomicInteger created = new AtomicInteger();
    private static final AtomicInteger alive = new AtomicInteger();
    private static final Logger logger = Logger.getLogger(AppThread.class);
    private boolean dump = false;

    public AppThread(Runnable r) {
        this(r, DEFAULT_NAME);
    }

    public AppThread(Runnable runnable, String name) {
        super(runnable, name + "-" + created.incrementAndGet());
        logger.debug(name + "'s constructor running");
    }

    public void interrupt() {
        if (!dump) {
            super.interrupt();
        }
        if (dump) {
            logger.debug("interrupt : " + getName() + " <<<");
            Thread.dumpStack();
            logger.debug("interrupt : " + getName() + " >>>");
        }
    }

    public void run() {
        boolean debug = debugLifecycle;
        if (debug)
            logger.info("Created " + getName());
        try {
            alive.incrementAndGet();
            super.run();
            logger.debug("running!");
        } finally {
            alive.decrementAndGet();
            dump = true;
            try {
                Thread.sleep(100000);
            } catch (InterruptedException e) {
                logger.debug(e);
            }
            if (debug)
                logger.info("Exiting " + getName());
        }
    }

    public static int getThreadsCreated() {
        return created.get();
    }

    public static int getThreadsAlive() {
        return alive.get();
    }

    public static boolean getDebug() {
        return debugLifecycle;
    }

    public static void setDebug(boolean b) {
        debugLifecycle = b;
    }
}

Another problem is that in order to debug the cause of java.io.InterruptedIOException , I added

     try {
            Thread.sleep(100000);
        } catch (InterruptedException e) {
            logger.debug(e);
        }

in finally clause in the run method for AppThread. when InterruptedException is catched in the finally clause, the override interrupt() method is never called. so who interrupt AppThread? is the same guy cause java.io.InterruptedIOException?

Filipe

I modified the code and just call executor.shutdown(); but the problem still exist. – Filipe

Is there a reason to use a custom thread factory and custom executor? If you just want to know who calls Thread.interrupt(), put a breakpoint on that method. Also, did you try running with a stock executor and no special thread factory to rule out bugs in your AppThreadPoolExecutor/AppThread classes? – Calcutta

it never happened in windows environment. it only happened on the server which is running on Solaris 10. better logging and statistic info is the only reason to custom thread factory and executor. the bugs does not affect the correctness of the app. Just try to understand why it happens. – Filipe

Answer 1

Yes:

shutdownNow Attempts to stop all actively executing tasks, halts the processing of waiting tasks, and returns a list of the tasks that were awaiting execution.

There are no guarantees beyond best-effort attempts to stop processing actively executing tasks. For example, typical implementations will cancel via Thread.interrupt(), so any task that fails to respond to interrupts may never terminate.

JavaDoc.

Simply use shutdown() instead of shutdownNow(). When you are forcibly calling shutdownNow() this is what you should expect - JVM gracefully interrupts I/O and shuts down the thread as fast as possible.

However I would make sure that logging isn't the bottleneck in your application. Simply make few thread dumps during the execution of your program and see how often threads are writing or waiting for I/O. Poor man's profiling.

Cacka

I modified the code and just call executor.shutdown(); but the problem still exist. – Filipe

THANKS, plus nice explanations – Pastelki

Answer 2

Interrupting the worker threads is actually a feature of the Executor framework to allow worker threads to gracefully shut down when asked to do so through interrupt(). It's documented behavior for shutdownNow().

If you don't want this, call shutdown() -- it won't interrupt() your worker threads, the Executor will just stop accepting new tasks.

Calcutta

I modified the code and just call executor.shutdown(); but the problem still exist. – Filipe

Answer 3

I have similar problems. My research went so far that Thread.interrupt() sets the interrupt flag. This leads to an interrupted IO operation deep in the Java Stack. But the IO methods are typically not declared to throw an InterruptedException.

Instead an InterruptedIOException is thrown and the interrupted state of the Thread is cleared!. If you wrote a Worker that expects (catches) IOExceptions, you have to catch the InterruptedIOException separately and call Thead.currentThread().interrupt() in the catch clause.

Aronson