java.io.IOException: Pipe broken

[Peter Chase]

I wanted to create a PrintWriter that I could give to JavaDB. The idea was that the lines printed to the PrintWriter by JavaDB would get logged to my logger. I thought PipedWriter and PipedReader sounded like good classes to help me with this.

I came up with the code below. It mostly worked, by sometimes suffered from "Pipe broken" IOException, which made it unusable, because the PrintWriter's println() method hung for ages when this happened.

Any ideas why that might have been happening? Any solutions?

This is on Windows XP at the moment, although it has to run on Mac and Linux too. Java is currently 1.4.1, thinking of going to 6 when it comes out properly on Mac.

  /** A PrintWriter that will log each line written to it to Logger. */
  public static final class LogPrintWriter extends PrintWriter
  {
    private final String _subject;
    private final int _level;
    private final PipedWriter _pipedWriter;
    private final PipedReader _pipedReader;
    private final Object _lock = new Object();
 
    private Thread _readerThread; // = null
 
    /** Constructs the writer, but it should not be written-to until
     *  {@link #start start()} has been called.
     */
    public LogPrintWriter(String subject, int level)
    {
      super(new PipedWriter(), true /* auto-flush at line end */);
      _pipedWriter = (PipedWriter) this.out;
 
      _subject = subject;
      _level = level;
 
      try
      {
        _pipedReader = new PipedReader(_pipedWriter);
      }
      catch (IOException ioe)
      {
        // No reason why this should fail during construction.
        throw logFailError(subject, ALARM_LEVEL_ERROR, ioe);
      }
    }
 
    /** Start daemon thread to consume the information written to the writer. */
    public void start()
    {
      Runnable runnable = new Runnable()
        {
          public void run()
          {
            synchronized(_lock) { _readerThread = Thread.currentThread(); }
            try
            {
              BufferedReader r = new BufferedReader(_pipedReader);
              String line;
              while ((! Thread.interrupted()) && (null != (line = r.readLine())))
                Logger.log(_subject, _level, line);
            }
            catch (IOException ioe)
            {
              Logger.log(_subject, ALARM_LEVEL_ERROR,
                            "LogPrintWriter failed due to: " + ioe);
            }
            finally
            {
              synchronized(_lock) { _readerThread = null; }
            }
          }
        };
 
      WorkerThreadPool.LONG_DAEMON_TASKS.schedule(runnable, _subject);
    }
 
    /** Request the associated thread to stop. May actually take quite a while
     *  to stop, but it is a daemon thread, so it will not prevent JVM shut-down.
     */
    public void stop()
    {
      synchronized(_lock)
      {
        if (null != _readerThread)
          _readerThread.interrupt();
      }
    }
  }

P.S. There's one or two bits of mildly disguised proprietory code in the above listing, but I hope it's pretty obvious what they do.
[ February 19, 2007: Message edited by: Peter Chase ]

[Jim Yingst]

Most likely this indicates that the reader thread either has completed, or has been unresponsive for an unacceptably long time. One possibility is that the reader thread has not yet started - perhaps all the threads in WorkerThreadPool.LONG_DAEMON_TASKS are busy, and the read task is still scheduled to begin whenever a thread becomes available? Which could take awhile if all the other tasks are long-running daemon tasks.

Another possibility is that the reader thread has completed due to an interrupt(). This could happen because someone has intentionally called stop() - but it could also happen because somewhere in the thread's past life some other component sent it an interrupt, and this may be the first opportunity the thread has had to handle it. Or another component may have merely retained a reference to the thread, and sent the interrupt much later. I would add a boolean flag (e.g. _isStopped) that can be set when stop() is called, and checked when an interrupt is detected. That way you can ignore interrupts sent for any other reason.

I'd also add a little debug-level logging to tell you when the reader runnable starts, and when it stops (for any reason). Does the broken pipe occur before the runnable starts, after it stops, or (unlikely) in between? I would also add a try/catch just inside run() to catch any unchecked throwables and make sure they're logged. Depending on how your logging is set up and what happens to System.err output, it's possible that uncaught exceptions may be overlooked. E.g. if you log all "normal" exceptions with a logger going to a log file, and don't normally have any reason to look at the standard error stuff, you may miss uncaught exceptions which simply bubble to the top of the stack, kill the thread, and print a stack trace to standard error.

[Peter Chase]

Thanks for the suggestions. I will stop using interrupts, and instead use some private protocol, which outside code can't get hold of. I'll also put in more tracing.

The WorkerThreadPool.LONG_DAEMON_TASKS does not have a limited number of threads available, so there should not be a long delay between scheduling and starting the task, although I realise that there are no guarantees with thread scheduling. FYI there are SHORT_*_TASKS pools (not used in this bit of code) that do have a limited number of threads; the reasoning is that if the tasks are short, it's not unreasonable to have to wait for some of them to finish, during busy periods.

[Peter Chase]

Unfortunately, none of the above suggestions worked. I still got "Pipe broken" IOExceptions. So I went to look in the JDK sources. Disappointment lurked there.

PipedReader expects to be used by exactly two threads. One thread should do all the reading and another thread should do all the writing. In my application, there is one thread doing all the reading, but writes come from all sorts of different threads. When the first write occurs, PipedReader stashes the Thread object. When that Thread dies, PipedReader says "Pipe broken". Game over.

Sun's own code contains the following revealing comment: -

    /* REMIND: identification of the read and write sides needs to be
       more sophisticated.  Either using thread groups (but what about
       pipes within a thread?) or using finalization (but it may be a
       long time until the next GC). */

I'll have to write my own similar classes that don't have this restriction.
[ February 21, 2007: Message edited by: Peter Chase ]

[Jim Yingst]

Ah, that's disappointing. Have you considered simply using an existing logging framework such as log4j instead? Might be quicker, and you'd get a lot more functionality. Unless you've got a lot of existing client code that expects to use a Writer. Or the Writer could just append Strings to the end of a LinkedList, while a reader thread pulls them off from the front. With synchronization as necessary of course. Well, I'm sure you can come up with something. Enjoy...

[Peter Chase]

I ended up with a simpler scheme. I now have a LogWriter, which is just a Writer, not a PrintWriter. It doesn't need to do fancy threading. Then, to get a PrintWriter, as required by JavaDB, I just construct a PrintWriter with a LogWriter as the wrapped Writer.

  /** A Writer that transfers data written to it to the HQNlogger, every time the
   *  {@link #flush flush()} method is called. Best used with an auto-flushing
   *  <code>PrintWriter</code>.
   */
  public static final class LogWriter extends Writer
  {
    private final String _subject;
    private final int _level;
 
    private StringBuffer _buf = new StringBuffer(INIT_BUF);
    private boolean _isClosed; // = false
 
    private static final int INIT_BUF = 80;
 
    /** Constructs the writer. */
    public LogWriter(String subject, int level)
    {
      super(new Object());
 
      _subject = subject;
      _level = level;
    }
 
    public void close() throws IOException
    {
      synchronized(this.lock)
      {
        if (! _isClosed)
        {
          flush();
          _isClosed = true;
        }
      }
    }
 
    public void flush() throws IOException
    {
      if (logging(_subject, _level))
      {
        String text;
        synchronized(this.lock)
        {
          if (_isClosed)
            throw new IOException("Already closed");
 
          text = _buf.toString();
 
          // Do not retain StringBuffer that has grown big, but reuse a small one
          if (_buf.capacity() > INIT_BUF)
            _buf = new StringBuffer(INIT_BUF);
          else
            _buf.setLength(0);
        }
 
        if (text.length() > 0)
          HQNlogger.log(_subject, _level, text);
      }
    }
 
    public void write(char[] cbuf, int off, int len) throws IOException
    {
      // May as well discard log data early, if this subject is not being
      // logged at sufficient level.
      if (logging(_subject, _level))
      {
        synchronized(this.lock)
        {
          if (_isClosed)
            throw new IOException("Already closed");
 
          _buf.append(cbuf, off, len);
        }
      }
    }
  }
 

... later ...
 

    // Log DriverManager output to HQNlogger. There is rather a lot of output, so
    // use INFO_VERBOSE level, so it is discarded by default.
    LogWriter logWriter
      = new LogWriter(getName(), HQNlogger.HQN_ALARM_LEVEL_INFO_VERBOSE);
    DriverManager.setLogWriter(new PrintWriter(logWriter));
 

Note that it's JavaDB that insists on having a PrintWriter. Would Log4J have given me a PrintWriter for free? I have a lot of code using my logger, so it would be hard to switch everything to another logger. Where using third-party code with a different log system, I write an adapter to mine; I have one for Log4J, JDK Logging, Apache Commons Logging etc.

[Jim Yingst]

Ah, I see. I overlooked the JavaDB part of your needs there, so didn't realize why you needed the PrintWriter. No, I don't think log4j has such an adapter built in. Looks like yours will work nicely though. One possible problem I see is that if you use a PrintWriter without autoflush, you quite possibly won't get any flushing at all until the writer is closed. That could use a lot more memory that you'd like. Might be useful to insert a check of the buffer size, and flush whenever it exceeds some maximum. Or just make sure you use autoflush. Anyway, glad you found a simpler solution.