From 698c250501774668579b96b6b9fd3e6aea5d4d66 Mon Sep 17 00:00:00 2001 From: Jay Malhotra Date: Thu, 18 Jun 2026 20:07:08 +0100 Subject: [PATCH 1/3] Fix System.in.available() handling Closes #1452 The current implementation of `available()` on DaemonInputStream requests to read 1 byte and then attempts to calculate the available length from that. This seems to cause problems with shells though, and reading the 1 byte will cause `available()` to affect the standard input when it should be a no-op. This commit adds a new message type that instructs the client to report the number of bytes it has available, without actually reading any data. This is then added to the existing stdin data we already have buffered. It also adds an IT for this case, which failed before any code changes but now passes. Manual testing was performed with bash to confirm that the repro in #1452 now behaves identically on `mvnd` and `mvn`. --- .../org/mvndaemon/mvnd/common/Message.java | 74 +++++++++++++++++ .../common/logging/TerminalInputHandler.java | 34 +++++++- .../mvnd/common/logging/TerminalOutput.java | 10 +++ .../mvnd/daemon/DaemonInputStream.java | 80 +++++++++++++++---- .../org/mvndaemon/mvnd/daemon/Server.java | 6 +- .../mvnd/it/InputStreamNativeIT.java | 33 ++++++++ .../org/apache/maven/its/CountBytesMojo.java | 58 ++++++++++++++ 7 files changed, 276 insertions(+), 19 deletions(-) create mode 100644 integration-tests/src/test/projects/input-stream/src/main/java/org/apache/maven/its/CountBytesMojo.java diff --git a/common/src/main/java/org/mvndaemon/mvnd/common/Message.java b/common/src/main/java/org/mvndaemon/mvnd/common/Message.java index 2878f3e30..790e45097 100644 --- a/common/src/main/java/org/mvndaemon/mvnd/common/Message.java +++ b/common/src/main/java/org/mvndaemon/mvnd/common/Message.java @@ -65,6 +65,8 @@ public abstract class Message { public static final int PRINT_ERR = 26; public static final int REQUEST_INPUT = 27; public static final int INPUT_DATA = 28; + public static final int REQUEST_INPUT_AVAILABLE = 29; + public static final int INPUT_AVAILABLE_DATA = 30; final int type; @@ -122,6 +124,10 @@ public static Message read(DataInputStream input) throws IOException { return RequestInput.read(input); case INPUT_DATA: return InputData.read(input); + case REQUEST_INPUT_AVAILABLE: + return RequestInputAvailable.read(input); + case INPUT_AVAILABLE_DATA: + return InputAvailableData.read(input); } throw new IllegalStateException("Unexpected message type: " + type); } @@ -1123,6 +1129,66 @@ public void write(DataOutputStream output) throws IOException { } } + public static class RequestInputAvailable extends Message { + + private final String projectId; + + public static RequestInputAvailable read(DataInputStream input) throws IOException { + String projectId = readUTF(input); + return new RequestInputAvailable(projectId); + } + + RequestInputAvailable(String projectId) { + super(REQUEST_INPUT_AVAILABLE); + this.projectId = projectId; + } + + public String getProjectId() { + return projectId; + } + + @Override + public String toString() { + return "RequestInputAvailable{" + "projectId='" + projectId + "'}"; + } + + @Override + public void write(DataOutputStream output) throws IOException { + super.write(output); + writeUTF(output, projectId); + } + } + + public static class InputAvailableData extends Message { + + private final int bytesAvailable; + + public static InputAvailableData read(DataInputStream input) throws IOException { + int bytesAvailable = input.readInt(); + return new InputAvailableData(bytesAvailable); + } + + InputAvailableData(int bytesAvailable) { + super(INPUT_AVAILABLE_DATA); + this.bytesAvailable = bytesAvailable; + } + + public int getBytesAvailable() { + return bytesAvailable; + } + + @Override + public String toString() { + return "InputAvailableData{" + "bytesAvailable='" + bytesAvailable + "'}"; + } + + @Override + public void write(DataOutputStream output) throws IOException { + super.write(output); + output.writeInt(bytesAvailable); + } + } + public int getType() { return type; } @@ -1143,6 +1209,14 @@ public static InputData inputEof() { return new InputData(null); } + public static RequestInputAvailable requestInputAvailable(String projectId) { + return new RequestInputAvailable(projectId); + } + + public static InputAvailableData inputAvailableData(int bytesAvailable) { + return new InputAvailableData(bytesAvailable); + } + public static StringMessage out(String message) { return new StringMessage(PRINT_OUT, message); } diff --git a/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java b/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java index c4071cbef..2d0f36ece 100644 --- a/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java +++ b/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java @@ -67,25 +67,36 @@ private static class InputRequest { final String projectId; // null for control keys final Message.Prompt prompt; // non-null only for prompt requests final boolean isControlKey; // true for control key listening + final boolean isAvailableLength; // true for requesting number of available bytes final int bytesToRead; // max number of bytes to read - private InputRequest(String projectId, Message.Prompt prompt, boolean isControlKey, int bytesToRead) { + private InputRequest( + String projectId, + Message.Prompt prompt, + boolean isControlKey, + boolean isAvailableLength, + int bytesToRead) { this.projectId = projectId; this.prompt = prompt; this.isControlKey = isControlKey; + this.isAvailableLength = isAvailableLength; this.bytesToRead = bytesToRead; } static InputRequest forProject(String projectId, int bytesToRead) { - return new InputRequest(projectId, null, false, bytesToRead); + return new InputRequest(projectId, null, false, false, bytesToRead); } static InputRequest forPrompt(Message.Prompt prompt) { - return new InputRequest(prompt.getProjectId(), prompt, false, 0); + return new InputRequest(prompt.getProjectId(), prompt, false, false, 0); } static InputRequest forControlKeys() { - return new InputRequest(null, null, true, 0); + return new InputRequest(null, null, true, false, 0); + } + + static InputRequest forAvailable(String projectId) { + return new InputRequest(projectId, null, false, true, 0); } } @@ -108,6 +119,8 @@ public TerminalInputHandler(Terminal terminal, boolean dumb) { } else if (request.prompt != null) { // Always handle prompts handlePrompt(request.prompt); + } else if (request.isAvailableLength) { + handleAvailableLength(); } else if (request.projectId != null) { // Always handle project input handleProjectInput(request.projectId, request.bytesToRead); @@ -199,6 +212,14 @@ private void handlePrompt(Message.Prompt prompt) throws IOException { } } + private void handleAvailableLength() throws IOException { + // terminal.reader().available() + int available = System.in.available(); + System.err.println("GOT AVAILABLE " + available + " from reader " + + terminal.reader().getClass().getName()); + daemonReceive.accept(Message.inputAvailableData(available)); + } + private boolean isControlKey(int c) { return c == TerminalOutput.KEY_PLUS || c == TerminalOutput.KEY_MINUS @@ -230,6 +251,11 @@ public void requestProjectInput(String projectId, int bytesToRead) { inputRequests.offer(InputRequest.forProject(projectId, bytesToRead)); } + public void requestProjectInputAvailable(String projectId) { + inputRequests.clear(); + inputRequests.offer(InputRequest.forAvailable(projectId)); + } + public void requestPrompt(Message.Prompt prompt) { inputRequests.clear(); // Clear any pending requests inputRequests.offer(InputRequest.forPrompt(prompt)); diff --git a/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalOutput.java b/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalOutput.java index ea4164982..1c0b9a28b 100644 --- a/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalOutput.java +++ b/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalOutput.java @@ -50,6 +50,7 @@ import org.mvndaemon.mvnd.common.Message.MojoStartedEvent; import org.mvndaemon.mvnd.common.Message.ProjectEvent; import org.mvndaemon.mvnd.common.Message.RequestInput; +import org.mvndaemon.mvnd.common.Message.RequestInputAvailable; import org.mvndaemon.mvnd.common.Message.StringMessage; import org.mvndaemon.mvnd.common.Message.TransferEvent; import org.mvndaemon.mvnd.common.OsUtils; @@ -414,6 +415,15 @@ private boolean doAccept(Message entry) { daemonDispatch.accept(entry); break; } + case Message.REQUEST_INPUT_AVAILABLE: { + RequestInputAvailable ri = (RequestInputAvailable) entry; + inputHandler.requestProjectInputAvailable(ri.getProjectId()); + break; + } + case Message.INPUT_AVAILABLE_DATA: { + daemonDispatch.accept(entry); + break; + } default: throw new IllegalStateException("Unexpected message " + entry); } diff --git a/daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java b/daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java index 6843a9106..f267c3989 100644 --- a/daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java +++ b/daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java @@ -24,7 +24,12 @@ import java.nio.charset.Charset; import java.util.LinkedList; import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.BiConsumer; +import java.util.function.Consumer; import org.apache.maven.logging.ProjectBuildLogAppender; @@ -45,13 +50,13 @@ * - EOF is signaled by calling addInputData with null * * The stream coordinates between multiple threads: - * - Reader thread(s): Calling read() methods to get input - * - Writer thread: Calling addInputData to provide input data + * - Reader thread(s): Calling methods to get input + * - Writer thread: Calling addInputData to provide input data, addInputAvailableData to provide the number of bytes available * * Synchronization: - * - All buffer access is synchronized on the datas collection - * - Readers wait when no data is available using datas.wait() - * - Writers notify readers when new data arrives using datas.notifyAll() + * - A lock controls access to datas / available as well as internal state + * - Readers wait when no data is available using datasReadyCondition.await() / availableReadyCondition.await() + * - Writers notify readers when new data arrives datasReadyCondition.signalAll() / availableReadyCondition.signalAll() * * This implementation is particularly important for: * 1. Handling piped input (e.g., cat file | mvnd ...) @@ -60,26 +65,57 @@ */ class DaemonInputStream extends InputStream { private final BiConsumer startReadingFromProject; + private final Consumer requestAvailable; + + private final Lock lock = new ReentrantLock(); + private final Condition datasReadyCondition = lock.newCondition(); + private final Condition availableReadyCondition = lock.newCondition(); + private final LinkedList datas = new LinkedList<>(); + private Optional available = Optional.empty(); + private final Charset charset; private int pos = -1; private String projectReading = null; private volatile boolean eof = false; - DaemonInputStream(BiConsumer startReadingFromProject) { + DaemonInputStream(BiConsumer startReadingFromProject, Consumer requestAvailable) { this.startReadingFromProject = startReadingFromProject; + this.requestAvailable = requestAvailable; this.charset = Charset.forName(System.getProperty("file.encoding")); } @Override public int available() throws IOException { - synchronized (datas) { + lock.lock(); + try { + int buffered = datas.stream().mapToInt(a -> a.length).sum() - Math.max(pos, 0); + + if (eof) { + return buffered; + } + String projectId = ProjectBuildLogAppender.getProjectId(); - if (!eof && !Objects.equals(projectId, projectReading)) { + + if (!Objects.equals(projectId, projectReading)) { projectReading = projectId; - startReadingFromProject.accept(projectId, 1); } - return datas.stream().mapToInt(a -> a.length).sum() - Math.max(pos, 0); + + requestAvailable.accept(projectId); + + available = Optional.empty(); + + try { + while (available.isEmpty()) { + availableReadyCondition.await(); + } + } catch (InterruptedException e) { + throw new InterruptedIOException("Interrupted"); + } + + return buffered + available.orElse(0); + } finally { + lock.unlock(); } } @@ -95,7 +131,8 @@ public int read() throws IOException { @Override public int read(byte[] b, int off, int len) throws IOException { - synchronized (datas) { + lock.lock(); + try { if (eof && datas.isEmpty()) { return -1; // Return EOF if we've reached the end and no more data } @@ -115,7 +152,7 @@ public int read(byte[] b, int off, int len) throws IOException { // Always notify we need input when waiting for data startReadingFromProject.accept(projectReading, len - read); try { - datas.wait(); + datasReadyCondition.await(); } catch (InterruptedException e) { throw new InterruptedIOException("Interrupted"); } @@ -134,17 +171,32 @@ public int read(byte[] b, int off, int len) throws IOException { b[off + read++] = curData[pos++]; } return read; + } finally { + lock.unlock(); } } public void addInputData(String data) { - synchronized (datas) { + lock.lock(); + try { if (data == null) { eof = true; } else { datas.add(data.getBytes(charset)); } - datas.notifyAll(); + datasReadyCondition.signalAll(); + } finally { + lock.unlock(); + } + } + + public void addInputAvailableData(int bytesAvailable) { + lock.lock(); + try { + available = Optional.of(bytesAvailable); + availableReadyCondition.signalAll(); + } finally { + lock.unlock(); } } } diff --git a/daemon/src/main/java/org/mvndaemon/mvnd/daemon/Server.java b/daemon/src/main/java/org/mvndaemon/mvnd/daemon/Server.java index 87e18a156..3e386692c 100644 --- a/daemon/src/main/java/org/mvndaemon/mvnd/daemon/Server.java +++ b/daemon/src/main/java/org/mvndaemon/mvnd/daemon/Server.java @@ -509,7 +509,8 @@ private void handle(DaemonConnection connection, BuildRequest buildRequest) { final BlockingQueue recvQueue = new LinkedBlockingDeque<>(); final BuildEventListener buildEventListener = new ClientDispatcher(sendQueue); final DaemonInputStream daemonInputStream = new DaemonInputStream( - (projectId, bytesToRead) -> sendQueue.add(Message.requestInput(projectId, bytesToRead))); + (projectId, bytesToRead) -> sendQueue.add(Message.requestInput(projectId, bytesToRead)), + (projectId) -> sendQueue.add(Message.requestInputAvailable(projectId))); InputStream in = System.in; try { System.setIn(daemonInputStream); @@ -561,6 +562,9 @@ private void handle(DaemonConnection connection, BuildRequest buildRequest) { return; } else if (message instanceof Message.InputData) { daemonInputStream.addInputData(((Message.InputData) message).getData()); + } else if (message instanceof Message.InputAvailableData) { + daemonInputStream.addInputAvailableData( + ((Message.InputAvailableData) message).getBytesAvailable()); } else { synchronized (recvQueue) { recvQueue.put(message); diff --git a/integration-tests/src/test/java/org/mvndaemon/mvnd/it/InputStreamNativeIT.java b/integration-tests/src/test/java/org/mvndaemon/mvnd/it/InputStreamNativeIT.java index 78df741eb..e3aa0c560 100644 --- a/integration-tests/src/test/java/org/mvndaemon/mvnd/it/InputStreamNativeIT.java +++ b/integration-tests/src/test/java/org/mvndaemon/mvnd/it/InputStreamNativeIT.java @@ -61,4 +61,37 @@ public void accept(Message message) { client.execute(output, "org.mvndaemon.mvnd.test.input-stream:echo-maven-plugin:echo"); } + + @Test + void installPluginAndCount() throws IOException, InterruptedException { + final TestClientOutput output = new TestClientOutput() { + boolean sentInput = false; + String data = "0123456789"; + + @Override + public void accept(Message message) { + if (message instanceof Message.RequestInput) { + if (!sentInput) { + daemonDispatch.accept(Message.inputResponse(data)); + sentInput = true; + } else { + daemonDispatch.accept(Message.inputEof()); + } + } else if (message instanceof Message.RequestInputAvailable) { + daemonDispatch.accept(Message.inputAvailableData(data.length())); + } + + if (!(message instanceof Message.TransferEvent)) { + super.accept(message); + } + } + }; + client.execute(output, "install").assertSuccess(); + + client.execute(output, "org.mvndaemon.mvnd.test.input-stream:echo-maven-plugin:countbytes") + .assertSuccess(); + + output.assertContainsMatchingSubsequence( + "Saw 10 bytes available from stdin. Actually read 10 bytes from stdin."); + } } diff --git a/integration-tests/src/test/projects/input-stream/src/main/java/org/apache/maven/its/CountBytesMojo.java b/integration-tests/src/test/projects/input-stream/src/main/java/org/apache/maven/its/CountBytesMojo.java new file mode 100644 index 000000000..648f14863 --- /dev/null +++ b/integration-tests/src/test/projects/input-stream/src/main/java/org/apache/maven/its/CountBytesMojo.java @@ -0,0 +1,58 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.maven.its; + +import org.apache.maven.plugin.AbstractMojo; +import org.apache.maven.plugin.MojoExecutionException; + +import org.apache.maven.plugins.annotations.Mojo; +import org.apache.maven.plugins.annotations.Parameter; + +import java.io.BufferedReader; +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.util.Scanner; + +/** + * Goal which counts the number of bytes available on stdin. + */ +@Mojo( name = "countbytes", requiresProject = false ) +public class CountBytesMojo + extends AbstractMojo +{ + + public void execute() + throws MojoExecutionException + { + getLog().info("Reading from standard input. Type 'exit' to stop."); + + int available = 0; + byte[] bytes = new byte[] {}; + + try { + available = System.in.available(); + bytes = System.in.readAllBytes(); + } catch (Exception ex) { + getLog().error("Failed to read from stdin", ex); + } + + System.out.println("Saw " + available + " bytes available from stdin. Actually read " + bytes.length + " bytes from stdin."); + } +} From 4d1dcc6dca60cbc9a8e5240b4983e0e615464e7c Mon Sep 17 00:00:00 2001 From: Jay Malhotra Date: Thu, 18 Jun 2026 22:34:22 +0100 Subject: [PATCH 2/3] Don't send stdin EOF before message data If TerminalInputHandler::handleProjectInput receives a small amount of input (e.g. 12 chars) then it will reach the c < 0 EOF branch in the loop. This sends an EOF message, then proceeds to send the data buffer. This does not mesh well with how DaemonInputStream reads input. If read() is suspended waiting for new data, receiving an EOF message first will lead to the eof field being set to true, which means when `read` wakes up it will end before it can actually read anything. This leads to `System.in.readAllBytes` returning an empty array. We can fix this by changing the order so that the data message is sent before the EOF message. --- .../mvnd/common/logging/TerminalInputHandler.java | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java b/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java index 2d0f36ece..c5f0e5850 100644 --- a/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java +++ b/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java @@ -151,8 +151,13 @@ private void handleProjectInput(String projectId, int bytesToRead) throws IOExce int c = terminal.reader().read(timeout); if (c < 0) { // End of stream reached + if (idx > 0) { + String data = String.valueOf(buf, 0, idx); + daemonReceive.accept(Message.inputResponse(data)); + } + daemonReceive.accept(Message.inputEof()); - break; + return; } buf[idx++] = (char) c; timeout = idx > 0 ? 1 : 10; // Shorter timeout after first char @@ -213,10 +218,10 @@ private void handlePrompt(Message.Prompt prompt) throws IOException { } private void handleAvailableLength() throws IOException { - // terminal.reader().available() + // terminal.reader().available() will attempt to return the number of available characters using some math, but + // this method powers System.in.available() when called from a mojo, which needs to return the number of + // available bytes. int available = System.in.available(); - System.err.println("GOT AVAILABLE " + available + " from reader " - + terminal.reader().getClass().getName()); daemonReceive.accept(Message.inputAvailableData(available)); } From 4418273e039da2fa80b7b69cadb7fb37fad6dd45 Mon Sep 17 00:00:00 2001 From: Jay Malhotra Date: Tue, 23 Jun 2026 20:22:56 +0100 Subject: [PATCH 3/3] Add timeout to DaemonInputStream.available() Some tests which call `exec` goals were hanging because the exec plugin behind the scenes may call `System.in.available()`, sending a `RequestInputAvailable` message, but the `TestClientOutput` does not respond to these, leading to an infinite deadlock. We could make `TestClientOutput` respond, but it seems like it would be a good idea to prevent a deadlock by imposing a maximum time limit, in case a message gets dropped or something else goes wrong. If this does occur, then `available` will be `Optional.empty()` and the input stream will return only the number of bytes currently in its buffer. This timeout has not been applied to `read()` because it feels more natural / expected that `System.in.read()` could block indefinitely if there is no standard input ready but the stream is open. --- .../java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java b/daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java index f267c3989..e4c14bb2f 100644 --- a/daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java +++ b/daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java @@ -25,6 +25,7 @@ import java.util.LinkedList; import java.util.Objects; import java.util.Optional; +import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -106,8 +107,9 @@ public int available() throws IOException { available = Optional.empty(); try { - while (available.isEmpty()) { - availableReadyCondition.await(); + long remaining = TimeUnit.SECONDS.toNanos(1); + while (available.isEmpty() && remaining > 0) { + remaining = availableReadyCondition.awaitNanos(remaining); } } catch (InterruptedException e) { throw new InterruptedIOException("Interrupted");