java.io.PipedInputStream doesn't reliably support multiple writers

Name: mc57594 Date: 01/27/97


below is a program to reproduce the problem-
// This program demonstrates how the PipedInputStream object is not
// thread safe.
//
// What *should* happen is that the consumer thread should run
// and print notification when a read() is completed from each of
// the 4 producer threads. What *does* happen is the consumer reads
// the data from the first producer, then is thrown a Pipe broken
IOException.
// Ignoring the IOException is not a valid option, as the pipe will
// continue to receive exceptions at a rate of about 1/second, instead
// of just blocking.
//
// PipedInputStream 'remembers' the thread ID of the
// writer when the receive() method of the object is called to write
// data into the pipe. When the read() method is called, and there
// is no data available to be read, PipedInputStream should block.
// Before it will block, it checks to see if the writer thread
// is still alive (presumably to see if there is hope of ever unblocking
// again). Unfortunately, if there is more than one thread that will
// be writing to the pipe, and the last one that visited the receive()
// method has terminated, the pipe will not block, and it will throw
// a broken pipe IOException instead.
//
// There should be 1) a way to specify several threads that are
'eligible'
// to write to the pipe, 2) a way to selectively disable the isAlive()
check
// in the read() method altogether, and/or 3) access to the writeSide
member
// to allow the user to focus the isAlive() check on a different thread.
// None of the 'useful' bits of PipedInputStream are public, so the only
// way to subclass it and alter this behavior is to write something that
// resides in the package java.io, which is not a 'clean' solution.
//
// Pat Mancuso
// Cabletron Systems, Inc.
// ###@###.###

import java.io.*;

/*
 * The producer class sleeps a specified time, then writes into the
pipe.
 */
class Producer extends Thread
{
  PipedOutputStream pos = null;
  int wait = 0;

  public Producer(PipedOutputStream p, int w)
  {
    pos = p;
    wait = w;
  }

  public void run()
  {
    byte b[] = new byte[1];
    b[0]='a';

    try {
      System.out.println("Producer sleeping "+wait+" seconds...");
      sleep(wait*1000);
      System.out.println("...slept "+wait+" seconds, writing");
    } catch (Exception e) {
      System.out.println("Producer sleep interrupted!");
      return;
    }

    try {
      pos.write(b, 0, 1);
    } catch (IOException e) {
      System.out.println("producer caught IOException: "+e.toString());
    }
  }
}

/*
 * The consumer class should read from the pipe forever
 */
class Consumer extends Thread
{
  PipedInputStream pis;

  public Consumer(PipedInputStream p)
  {
    pis = p;
  }

  public void run()
  {
    int i;

    while (true) {
      try {
        i = pis.read();
        System.out.println("consumer read()");
      } catch (IOException e) {
        System.out.println("consumer caught IOException:
"+e.toString());
        // comment out the 'return' below to retry after each exception.
        // this 'works', but wasteful, as the consumer is constantly
        // receiving exceptions at a rate of about 1/second.
        return;
      }
    }
  }
}

public class PipeTest
{
  static public PipedInputStream pis = null;
  static public PipedOutputStream pos = null;

  public static void main(String argv[]) {

    try {
      pis = new PipedInputStream();
      pos = new PipedOutputStream(pis);
    } catch (IOException e) {
      System.out.println("main() caught IOException: "+e.toString());
    }


    // set up consumer - it will block on the read() since
    // writeSide is null initially.
    Consumer consumer = new Consumer(pis);
    
    // set up several producers to wake up at various times
    Producer producer1 = new Producer(pos, 0);
    Producer producer2 = new Producer(pos, 5);
    Producer producer3 = new Producer(pos, 10);
    Producer producer4 = new Producer(pos, 15);

    // start everything off
    consumer.start();
    producer1.start();
    producer2.start();
    producer3.start();
    producer4.start();
    
  }
  
}
======================================================================