I am using java IO streams to read commands from a Unix named pipe. The program has a thread that creates a named pipe with mkfifo and then listens to it, allowing the user to send commands to it using echo "command" > pipe.
It works fine most of the time, except when my program has to terminate without having received any command yet (on an unrecoverable exception for instance).
As described here: Not able to read from named pipe in Java the FileInputStream constructor blocks until some other process opens the pipe for writing. This leaves no chance for the thread implementation to handle the interruption and terminate properly.
I have produced a simple program that exhibits my issue. It listens for input commands for 3 seconds and then attempts (and fails) to stop itself.
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.function.Consumer;
/**
* Read the input named FIFO to dispatch commands to a command processor.
*/
public final class InputReaderThread {
private static final long LOOP_PERIOD_MS = 100;
private final Path fifoPath;
private final Consumer<String[]> processor;
private final Thread thread;
/**
* Constructor.
*
* @param fifoPath path to the named FIFO
* @param processor the command processor
*/
public InputReaderThread(final Path fifoPath, final Consumer<String[]> processor) {
this.fifoPath = fifoPath;
this.processor = processor;
this.thread = new Thread(this::run);
this.thread.setName(InputReaderThread.class.getSimpleName());
}
/**
* Start the input reader thread.<br>
* It is never legal to start a thread more than once. In particular, a thread may not be restarted once it has completed execution.
*
* @see Thread#start()
*/
public final void start() {
System.out.println("Requesting input reader thread start");
this.thread.start();
}
/**
* Request the input reader thread to stop.<br>
* This method does not wait for the thread to terminate, see {@link #join(long)}.
*/
public final void stop() {
System.out.println("Requesting input reader thread stop");
this.thread.interrupt();
}
/**
* Wait for the internal reader thread to terminate.<br>
* Throws an {@link InterruptedException} on timeout.
*
* @param timeout the maximum time to wait, in milliseconds
* @return {@code true} if the input thread terminated within timeout
* @throws InterruptedException the current thread was interrupted while waiting, or the timeout was reached
* @see Thread#join(long)
*/
public final boolean join(final long timeout) throws InterruptedException {
System.out.println("Awaiting input reader thread stop");
this.thread.join(timeout);
return !this.thread.isAlive();
}
private final void run() {
System.out.println("Input reader thread started");
try (final FileInputStream is = new FileInputStream(createFifoPipe(this.fifoPath))) {
final StringBuilder commandBuilder = new StringBuilder();
System.out.println("Listening to input FIFO");
while (!Thread.interrupted()) {
// Avoid reading when there is no data available
if (is.available() > 0) {
final int b = is.read();
if (b == '\n') {
// The command is complete: process it
final String command = commandBuilder.toString().trim();
System.out.println("Received command: " + command);
this.processor.accept(command.split(" "));
// Reset the command builder
commandBuilder.setLength(0);
} else {
// Append the character to the command
commandBuilder.append((char) b);
}
} else {
// Poll the input FIFO periodically
Thread.sleep(LOOP_PERIOD_MS);
}
}
} catch (final IOException e) {
throw new RuntimeException("An IO exception occurred on agent FIFO", e);
} catch (@SuppressWarnings("unused") final InterruptedException e) {
// Handle interruption by terminating the thread
}
System.out.println("Input reader thread terminated");
}
/**
* Helper method to create a Unix named FIFO.
*
* @param fifoPath the FIFO path
* @return the File handler to the created FIFO
* @throws IOException an I/O error occurs
* @throws InterruptedException the thread was interrupted while waiting for the FIFO creation
*/
private static final File createFifoPipe(final Path fifoPath) throws IOException, InterruptedException {
System.out.println("Creating fifo: " + fifoPath);
final File fifo = fifoPath.toFile();
if (fifo.exists()) {
System.err.println("Deleting existing fifo");
Files.delete(fifoPath);
}
final String[] command = new String[] { "mkfifo", fifo.getAbsolutePath() };
final Process process = new ProcessBuilder(command).start();
final int returnStatus = process.waitFor();
if (returnStatus != 0) {
throw new IOException("Failed to create fifo: " + returnStatus);
} else {
System.out.println("Created fifo: " + fifoPath);
}
return fifo;
}
}
NamedPipeTest.java:
import java.nio.file.Paths;
/**
* Dummy program for {@link InputReaderThread}.
*/
public final class NamedPipeTest {
/**
* Entry point.
*
* @param args one argument (path to named pipe)
*/
public static void main(final String[] args) {
final NamedPipeTest fifoTest = new NamedPipeTest(args[0]);
fifoTest.start();
try {
Thread.sleep(3000);
fifoTest.stop();
} catch (final InterruptedException e) {
System.err.println("Main thread interrupted");
e.printStackTrace(System.err);
}
}
private final InputReaderThread inputReader;
/**
* Constructor.
*
* @param fifoPath the named pipe path
*/
public NamedPipeTest(final String fifoPath) {
this.inputReader = new InputReaderThread(Paths.get(fifoPath), this::process);
}
/**
* Start the program.
*/
public void start() {
this.inputReader.start();
}
/**
* Stop the program.
*/
public void stop() {
try {
this.inputReader.stop();
if (!this.inputReader.join(5000)) {
System.err.println("Failed to terminate input reader thread");
System.exit(-1);
}
} catch (@SuppressWarnings("unused") final InterruptedException e) {
System.err.println("Interrupted while stopping");
System.exit(-1);
}
System.out.println("Stopped successfully");
}
/**
* Process a command.
*
* @param command the command
*/
public void process(final String[] command) {
if (command[0].equals("stop")) {
new Thread(this::stop).start();
}
}
}
Running this program writing "stop" in the named pipe shows the input reader thread terminates properly as it handles the interruption (unsure why the thread termination is logged twice, but that would be another question):
$ (java -jar NamedPipeTest.jar fifo &) && sleep 2s && echo "stop" > fifo
Requesting input reader thread start
Input reader thread started
Creating fifo: fifo
Deleting existing fifo
Created fifo: fifo
Listening to input FIFO
Received command: stop
Requesting input reader thread stop
Awaiting input reader thread stop
Input reader thread terminated
Stopped successfully
Requesting input reader thread stop
Awaiting input reader thread stop
Stopped successfully
However, running it without writing "stop" command exhibits the inability of FileInputStream to handle the interruption while it is opening the pipe:
$ java -jar NamedPipeTest.jar fifo &
Requesting input reader thread start
Input reader thread started
Creating fifo: fifo
Deleting existing fifo
Created fifo: fifo
Requesting input reader thread stop
Awaiting input reader thread stop
Failed to terminate input reader thread
Things I thought of:
- the thread should not be made daemon as it is performing IO (and it seems like a bad idea to mix those two)
- the Java NIO API relies on seek(), which seems not advisable with named pipes
I am at a loss as to how I could work around this, did I miss something? Thanks for the help!