Signal thread error to main thread

Question

I have a thread in Java that is connecting to a socket and sending information to another thread, which is processing that information.

Now, if the "producer" thread fails for any reason, I want the whole program to stop, so some sort of notification must happen.

Here's my program (very simplified):

public class Main {
  private Queue<String> q = new ConcurrentLinkedQueue();

  public static void main(String[] args) throws IOException {
    new Thread(new Producer(q)).start();
    new Thread(new Consumer(q)).start();

    // Catch any error in the producer side, then stop both consumer and producer, and do some extra work to notify me that there's an error...
  }
}

Main code just creates a shared queue, and starts both producer and consumer. So far, I guess it's ok? Now the Producer code is like this:

public class Producer implements Runnable {
  private Queue<String> q;

  public Producer(Queue<String> q) {
    this.q = q;
  }

  public void run() {
    try {
      connectToSocket();
      while(true) {
        String data = readFromSocket()
        q.offer(data);
      }
    } catch (Exception e) {
      // Something really bad happened, notify the parent thread so he stops the program...
    }
  }
}

Producer connects to socket, reads and sends to queue the string data... The consumer:

public class Consumer implements Runnable {
  private Queue<String> q;

  public Consumer(Queue<String> q) {
    this.q = q;
  }

  public void run() {
    while(true) {
      String dataFromSocket = q.poll();
      saveData(dataFromSocket);
    }
  }
}

My code does a lot more than that, but I think it's now self-explanatory what I'm trying to do. I've read about wait() and notify() but I think that wouldn't work, because I don't want to wait my thread for an exception, I want to deal with it in a better way. What are the alternatives?

In general, does my code look reasonable? Would using ExecutorService help here at all?

Thanks a lot!

Answer 1

you can use Thread's UncaughtExceptionHandler

Thread.setDefaultExceptionHandler(
new UncaughtExceptionHandler() {
    public void unchaughtException(Thread th, Throwable exception) {
        System.out.println("Exception from Thread" + th + ". Exception:" + exception);
    }
});

Java docs http://docs.oracle.com/javase/7/docs/api/java/lang/Thread.UncaughtExceptionHandler.html

Answer 2

The simplest solution given your current code would be to wait for the producer thread to finish and then interrupt the consumer:

Thread producerThread = new Thread(new Producer(q));
producerThread.start();     
Thread consumerThread = new Thread(new Consumer(q));
consumerThread.start();
try {
    producerThread.join();
} finally {
    consumerThread.interrupt();
}

As you mention, an executor would give you a more general purpose way to shut down everything when you need to exit (for example, when a interrupted in the terminal with ctrl-c).

ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);
Producer producer = new Producer(q);
Consumer consumer = new Consumer(q);
executor.submit(producer::run);
executor.submit(consumer::run);
Runtime.getRuntime().addShutdownHook(new Thread(executor::shutdownNow));

Note that your cleanup would have to be more comprehensive than just shutting down the executor. You would have to close the socket beforehand to allow the threads to be interrupted.

Here is a more complete example that handles shutdown from both sides. You can test it by starting a test server with nc -l 1234. Killing either process (nc or the java client) will result in a clean exit of the other.

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.Socket;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

public class Main {
    private ExecutorService executor;
    private Socket socket;
    private AtomicBoolean running = new AtomicBoolean(true);

    public static void main(String[] args) throws IOException {
        Main main = new Main();
        main.run();
    }

    private Main() throws IOException {
        executor = Executors.newCachedThreadPool();
        socket = new Socket("localhost", 1234);
    }

    private void run() throws IOException {
        BlockingQueue<String> q = new SynchronousQueue<>();
        Producer producer = new Producer(socket, q);
        Consumer consumer = new Consumer(q);

        // Start the producer. When it ends call stop
        CompletableFuture.runAsync(producer, executor).whenComplete((status, ex) -> stop());
        // Start the consumer.
        CompletableFuture.runAsync(consumer, executor);
        // Add a shutdown hook to stop everything on break
        Runtime.getRuntime().addShutdownHook(new Thread(this::stop));
    }

    private void stop() {
        if (running.compareAndSet(true, false)) { // only once
            // Close the socket to unblock the producer
            try {
                socket.close();
            } catch (IOException e) {
                // ignore
            }

            // Interrupt tasks
            executor.shutdownNow();
            try {
                // Give tasks some time to clean up
                executor.awaitTermination(1, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                // ignore
            }
        }
    }

    static class Producer implements Runnable {
        private BufferedReader in;
        private BlockingQueue<String> q;

        public Producer(Socket socket, BlockingQueue<String> q) throws IOException {
            this.in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            this.q = q;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    String data = in.readLine();
                    if (data == null) {
                        break;
                    }
                    q.put(data);
                }
            } catch (InterruptedException | IOException e) {
                // Fall through
            }
            System.err.println("Producer done");
        }
    }

    static class Consumer implements Runnable {
        private BlockingQueue<String> q;

        public Consumer(BlockingQueue<String> q) {
            this.q = q;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    System.out.println(q.take());
                }
            } catch (InterruptedException e) {
                // done
            }
            System.err.println("Client done");
        }
    }
}

Answer 3

Start consumer thread as 'daemon' thread

Mark the consumer thread as 'daemon' and let the main thread end too:

From the Java API doc for Thread.setDaemon(boolean):

Marks this thread as either a daemon thread or a user thread. The Java Virtual Machine exits when the only threads running are all daemon threads.

public class Main {
    private Queue<String> q = new ConcurrentLinkedQueue();

    public static void main(String[] args) throws IOException {
        Thread producerThread = new Thread(new Producer(q));
        // producerThread.setDefaultUncaughtExceptionHandler(...);
        producerThread.start();
        Thread consumerThread = new Thread(new Consumer(q));
        consumerThread.setDeamon(true);
        consumerThread.start();
    }

}

This way, your application automatically stops, when the main thread and the producer-thread have terminated (sucessfully or by exception).

You could combine this with the UncaughtExceptionHandler as @Manish suggested, if the main thread needs to know about the producerThread failing...

Answer 4

How about volatile?

public class Main {
  volatile boolean isStopMain = false;
  private Queue<String> q = new ConcurrentLinkedQueue();

  public static void main(String[] args) throws IOException {
    new Thread(new Producer(q)).start(); 
    new Thread(new Consumer(q)).start();
    // Catch any error in the producer side, then stop both consumer and producer, and do some extra work to notify me that there's an error...
    while (true) {
       if(isStopMain){
      System.exit(0); //or other operation to stop the main thread.
     }
    }
  }
}

And In Producer:

public class Producer implements Runnable {
  private Queue<String> q;

  public Producer(Queue<String> q) {
    this.q = q;
  }

  public void run() {
    try {
      connectToSocket();
      while(true) {
        String data = readFromSocket()
        q.offer(data);
      }
    } catch (Exception e) {
  // Something really bad happened, notify the parent thread so he stops the program...
      Main.isStopMain = true;
    }
  }
}

And I wonder if you are trying to kill the parent thread in child thread? If yes, here is something you may need to know:How do I terminate parent thread from child?