The exception to IO is reading the dead end - what causes it in this example and how to fix it - a multi-threaded application in Java

Question

This is an extension of my question posted here, while it seems to fix one part of my problem, now I see IO-Exception read end dead exception. I am using an application multithreadedwhere thread-1 producesrandom numbers and others are used thread-2 consumesto calculate the average value. as soon as the average reaches the threshold value, I signal thread-1to stop issuing numbers. This is the basic code design.

I get it IO-Exception read end dead exception. I want to know why this is happening and how to fix it. Thanks.

Code below:

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;

//

class NumGen extends Thread {

    PipedOutputStream pos;
    DataOutputStream dos;
    AtomicBoolean isDone;

    public NumGen(PipedOutputStream pos,AtomicBoolean isDone){
        this.pos=pos;
        dos=new DataOutputStream(pos);
        this.isDone=isDone;
    }

    public void run(){
        while (!isDone.get()){
            Random rand = new Random();
            try {
                dos.writeDouble(rand.nextDouble()+100.0);
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
}

class RunningAvg extends Thread {

    PipedInputStream pis;
    DataInputStream dis;
    Double avg;
    int count;
    Double runningTotal;
    AtomicBoolean isDone;

    public RunningAvg(PipedInputStream pis,AtomicBoolean isDone){
        this.pis=pis;
        dis=new DataInputStream(pis);
        runningTotal=0.0;
        avg=0.0;
        this.isDone=isDone;
    }

    public void run(){
        try {
        while (dis.available()>0){
            count+=1;
            runningTotal+=dis.readDouble();
                avg=runningTotal/count;
                System.out.printf("The average in count no : %s is %s%n",count,avg);
                if (avg>1E5)
                  isDone.set(true);
        }
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }


public class InterThreadComm {

    public static void main(String[] args){


    try {
        PipedOutputStream pos= new PipedOutputStream();
        PipedInputStream pis = new PipedInputStream(pos);
        AtomicBoolean isDone = new AtomicBoolean(false);
        NumGen ng = new NumGen(pos,isDone);
        RunningAvg ra = new RunningAvg(pis,isDone);
        ng.start();
        ra.start();
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    }


}

EDIT: As per the answer below: I tried closing threads using try-with-resources. but i get java.io.IOException: Pipe closed

public class InterThreadComm {

    public static void main(String[] args){


    try(PipedOutputStream pos= new PipedOutputStream();PipedInputStream pis =new PipedInputStream(pos)){ 
        AtomicBoolean isDone = new AtomicBoolean(false);
        NumGen ng = new NumGen(pos,isDone);
        RunningAvg ra = new RunningAvg(pis,isDone);
        ng.start();
        ra.start();
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    }


}

Answer 1

:

while (dis.available()>0){

, . , .

, PipedInputStream/PipedOutputStream , . , checkStateForReceive, , , :

} else if (readSide != null && !readSide.isAlive()) {
    throw new IOException("Read end dead");
}

(readSide - .) ​​

, :

java.io.IOException: Read end dead
    at java.io.PipedInputStream.checkStateForReceive(PipedInputStream.java:246)
    at java.io.PipedInputStream.receive(PipedInputStream.java:210)
    at java.io.PipedOutputStream.write(PipedOutputStream.java:132)
    at java.io.DataOutputStream.writeLong(DataOutputStream.java:207)
    at java.io.DataOutputStream.writeDouble(DataOutputStream.java:242)

write receieve , checkStateForReceieve, , .

, :

while(!isDone.get()) {

, , :

if (avg>1E5)
    isDone.set(true);

100 101, 1e5. , , runningTotal > 1E5.

:

• , .
• IOException , . , : , . , , .

:

, join .

try(
    PipedOutputStream pos= new PipedOutputStream();
    PipedInputStream pis =new PipedInputStream(pos)
){ 
    AtomicBoolean isDone = new AtomicBoolean(false);
    NumGen ng = new NumGen(pos,isDone);
    RunningAvg ra = new RunningAvg(pis,isDone);

    ng.start();
    ra.start();

    try {
       ng.join(); // wait for ng and ra to complete
       ra.join(); //
    } catch(InterruptedException ie) {}
} catch (IOException e) {
    e.printStackTrace();
}