Flaws with PipedInputStream/PipedOutputStream

Question

I've seen two answers on SO that claim that the PipedInputStream and PipedOutputStream classes provided by Java are flawed. But they did not elaborate on what was wrong with them. Are they really flawed, and if so in what way? I'm currently writing some code that uses them, so I'd like to know whether I'm taking a wrong turn.

One answer said:

PipedInputStream and PipedOutputStream are broken (with regards to threading). They assume each instance is bound to a particular thread. This is bizarre.

To me that seems neither bizarre nor broken. Perhaps the author also had some other flaws in mind?

Another answer said:

In practice they are best avoided. I've used them once in 13 years and I wish I hadn't.

But that author could not recall what the problem was.

---

As with all classes, and especially classes used in multiple threads, you will have problems if you misuse them. So I do not consider the unpredictable "write end dead" IOException that PipedInputStream can throw to be a flaw (failing to close() the connected PipedOutputStream is a bug; see the article Whats this? IOException: Write end dead, by Daniel Ferbers, for more information). What other claimed flaws are there?

Answer 1

They are not flawed.

As with all classes, and especially classes used in multiple threads, you will have problems if you misuse them. The unpredictable "write end dead" IOException that PipedInputStream can throw is not a flaw (failing to close() the connected PipedOutputStream is a bug; see the article Whats this? IOException: Write end dead, by Daniel Ferbers, for more information).

Answer 2

I have used them nicely in my project and they are invaluable for modifying streams on the fly and passing them around. The only drawback seemed to be that PipedInputStream had a short buffer (around 1024) and my outputstream was pumping in around 8KBs.

There is no defect with it and it works perfectly well.

-------- Example in groovy

public class Runner{


final PipedOutputStream source = new PipedOutputStream();
PipedInputStream sink = new PipedInputStream();

public static void main(String[] args) {
    new Runner().doit()
    println "Finished main thread"
}


public void doit() {

    sink.connect(source)

    (new Producer(source)).start()
    BufferedInputStream buffer = new BufferedInputStream(sink)
    (new Consumer(buffer)).start()
}
}

class Producer extends Thread {


OutputStream source
Producer(OutputStream source) {
    this.source=source
}

@Override
public void run() {

    byte[] data = new byte[1024];

    println "Running the Producer..."
    FileInputStream fout = new FileInputStream("/Users/ganesh/temp/www/README")

    int amount=0
    while((amount=fout.read(data))>0)
    {
        String s = new String(data, 0, amount);
        source.write(s.getBytes())
        synchronized (this) {
            wait(5);
        }
    }

    source.close()
}

}

class Consumer extends Thread{

InputStream ins

Consumer(InputStream ins)
{
    this.ins = ins
}

public void run()
{
    println "Consumer running"

    int amount;
    byte[] data = new byte[1024];
    while ((amount = ins.read(data)) >= 0) {
        String s = new String(data, 0, amount);
        println "< $s"
        synchronized (this) {
            wait(5);
        }
    }

}

}

Answer 3

One flaw might be that there is not clear way for the writer to indicate to the reader that it encountered a problem:

PipedOutputStream out = new PipedOutputStream();
PipedInputStream in = new PipedInputStream(out);

new Thread(() -> {
    try {
        writeToOut(out);
        out.close();
    }
    catch (SomeDataProviderException e) {
        // Have to notify the reading side, but how?
    }
}).start();

readFromIn(in);

The writer could close out, but maybe the reader misinterprets that as end of data. To handle this correctly additional logic is needed. It would be easier if functionality to manually break the pipe was provided.

There is now JDK-8222924 which requests a way to manually break the pipe.

Answer 4

From my point of view there is a flaw. More precisely there is a high risk of a deadlock if the Thread which should pump data into the PipedOutputStream dies prematurely before it actually writes a single byte into the stream. The problem in such a situation is that the implementation of the piped streams is not able to detect the broken pipe. Consequently the thread reading from PipedInputStream will wait forever (i.e. deadlock) in it's first call to read().

Broken pipe detection actually relies on the first call to write() as the implementation will than lazily initialize the write side thread and only from that point in time broken pipe detection will work.

The following code reproduces the situation:

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;

import org.junit.Test;

public class PipeTest
{
    @Test
    public void test() throws IOException
    {
        final PipedOutputStream pout = new PipedOutputStream();
        PipedInputStream pin = new PipedInputStream();

        pout.connect(pin);

        Thread t = new Thread(new Runnable()
        {
            public void run()
            {
                try
                {
                    if(true)
                    {
                        throw new IOException("asd");
                    }
                    pout.write(0); // first byte which never get's written
                    pout.close();
                }
                catch(IOException e)
                {
                    throw new RuntimeException(e);
                }
            }
        });
        t.start();

        pin.read(); // wait's forever, e.g. deadlocks
    }
}

Answer 5

The flaws that I see with the JDK implementation are:

1) No timeouts, reader or writer can block infinitely.

2) Suboptimal control over when data is transferred (should be done only with flush, or when circular buffer is full)

So I created my own to address the above, (timeout value passed via a ThreadLocal):

PipedOutputStream

How to use:

PiedOutputStreamTest

Hope it helps...