Skip to content

Modernize agent I/O with CompletableFuture and BufferedReader. #79

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
DomiKoPL wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

Modernize agent I/O with CompletableFuture and BufferedReader. #79

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 62 additions & 46 deletions runner/src/main/java/com/codingame/gameengine/runner/Agent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,16 @@
package com.codingame.gameengine.runner;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
Expand All @@ -15,34 +21,45 @@ abstract class Agent {
public static final int AGENT_MAX_BUFFER_SIZE = 10_000;
public static final int THRESHOLD_LIMIT_STDERR_SIZE = 4096 * 50;

/**
* Single dedicated thread for all blocking agent I/O — sequential game loop
* never needs more than one.
*/
private static final ExecutorService AGENT_IO_EXECUTOR = Executors.newSingleThreadExecutor(r -> {
Thread t = new Thread(r, "agent-reader");
t.setDaemon(true);
return t;
});

private static Log log = LogFactory.getLog(Agent.class);

private OutputStream processStdin;
private InputStream processStdout;
private BufferedReader processStdoutReader;
private InputStream processStderr;
private int totalStderrBytesSent = 0;
private int agentId;
private boolean lastAgentByteIsCarriageReturn = false;
private boolean failed = false;

private String nickname;
private String avatar;

public long lastExecutionTimeMs;

public Agent() {
}

abstract protected OutputStream getInputStream();

abstract protected InputStream getOutputStream();
abstract protected BufferedReader getOutputReader();

abstract protected InputStream getErrorStream();

/**
* Initialize an agent given global properties. A call to this function is needed before-all
* Initialize an agent given global properties. A call to this function is
* needed before-all
*
* @param conf
* Global configuration
* Global configuration
*/
public void initialize(Properties conf) {
this.lastExecutionTimeMs = 0;
Expand All @@ -54,7 +71,7 @@ public void initialize(Properties conf) {
public void execute() {
try {
this.processStdin = getInputStream();
this.processStdout = getOutputStream();
this.processStdoutReader = getOutputReader();
this.processStderr = getErrorStream();
runInputOutput();
} catch (Exception e) {
Expand All @@ -70,15 +87,15 @@ public void destroy() {
* Launch the agent. After the call, agent is ready to process input / output
*
* @throws Exception
* if an error occurs
* if an error occurs
*/
protected abstract void runInputOutput() throws Exception;

/**
* Write 'input' to standard input of agent
*
* @param input
* an input to write
* an input to write
*/
public void sendInput(String input) {
if (processStdin != null) {
Expand All @@ -98,57 +115,56 @@ public void sendInput(String input) {
* Get the output of an agent
*
* @param nbLine
* Number of lines wanted
* Number of lines wanted
* @param timeout
* Stop reading after timeout milliseconds
* @return the agent output
* Stop reading after timeout milliseconds
* @return the agent output, or null if timed out or reader closed
*/
public String getOutput(int nbLine, long timeout) {
if (processStdout == null) {
return getOutput(nbLine, timeout, AGENT_MAX_BUFFER_SIZE);
}

/**
* Read at most maxOutputSize bytes across nbLine lines.
*
* @param nbLine
* Number of lines wanted
* @param timeout
* Stop reading after timeout milliseconds
* @param maxOutputSize
* Maximum number of bytes to read
* @return the agent output, or null if timed out or reader closed
*/
protected final String getOutput(int nbLine, long timeout, int maxOutputSize) {
if (processStdoutReader == null) {
return null;
}

try {
byte[] tmp = new byte[AGENT_MAX_BUFFER_SIZE];
int offset = 0;
int nbOccurences = 0;

long t0 = System.nanoTime();

while ((offset < AGENT_MAX_BUFFER_SIZE) && (nbOccurences < nbLine)) {
long current = System.nanoTime();
if ((current - t0) > (timeout * 1_000_000l)) {
break;
}

if (processStdout.available() > 0) {
int nbRead = processStdout.read(tmp, offset, 1);
if (nbRead < 0) {
// Should not happen, just in case...
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
StringBuilder sb = new StringBuilder();
for (int i = 0; i < nbLine; i++) {
String line = processStdoutReader.readLine();

@sdhoine sdhoine Jul 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This removes the previous byte-level cap. readLine() can allocate an arbitrarily large line before the maxOutputSize check below runs, and the later check counts UTF-16 chars rather than UTF-8 bytes. A bot can now emit a huge unterminated line and bypass the intended output limits or cause memory pressure.

if (line == null) {
break;
}
byte curByte = tmp[offset];
if (!((curByte == '\n') && lastAgentByteIsCarriageReturn)) {
offset += nbRead;
if ((curByte == '\n') || (curByte == '\r')) {
++nbOccurences;
}
}
lastAgentByteIsCarriageReturn = curByte == '\r';
} else {
if ((offset < AGENT_MAX_BUFFER_SIZE) && (nbOccurences < nbLine)) {
Thread.sleep(1);
sb.append(line).append('\n');

@sdhoine sdhoine Jul 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This changes the validity contract for terminated lines. BufferedReader.readLine() can return the last partial line at EOF, and appending \n here makes checkOutput() count it as a complete line. Previously, output without an actual line terminator would remain too short.

if (sb.length() >= maxOutputSize) {
break;
}
}
return sb.toString();
} catch (IOException e) {
return null;
}
}, AGENT_IO_EXECUTOR);

return new String(tmp, 0, offset, UTF8);
} catch (IOException e1) {
processStdout = null;
} catch (InterruptedException e) {
// wtf
try {
return future.orTimeout(timeout, TimeUnit.MILLISECONDS).join();
} catch (CompletionException e) {
future.cancel(true);

@sdhoine sdhoine Jul 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

CompletableFuture.cancel(true) does not stop the supplier that is already blocked in BufferedReader.readLine(). Since all reads share the single agent-reader executor, one timed-out player that never emits a newline can permanently occupy that thread and make later referee/player reads timeout even when their output is ready. This needs a design where timed-out blocking reads cannot block subsequent reads.

return null;
}
return null;
}

/**
Expand Down
26 changes: 11 additions & 15 deletions runner/src/main/java/com/codingame/gameengine/runner/CommandLinePlayerAgent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,17 +1,19 @@
package com.codingame.gameengine.runner;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Properties;

/**
* This class is used in with the <code>GameRunner</code> to add an AI as a player.
* This class is used in with the <code>GameRunner</code> to add an AI as a
* player.
*/
public class CommandLinePlayerAgent extends Agent {

private OutputStream processStdin;
private InputStream processStdout;
private BufferedReader processInputReader;
private InputStream processStderr;
private String[] commandArray;
private Process process;
Expand All @@ -20,18 +22,18 @@ public class CommandLinePlayerAgent extends Agent {
* Creates an Agent for your game, will run the given commandLine at game start
*
* @param commandLine
* the command line to run
* the command line to run
*/
public CommandLinePlayerAgent(String commandLine) {
super();
this.commandArray = commandLine.split(" ");
}

/**
* Creates an Agent for your game, will run the given commandLine at game start
*
* @param commandArray
* the command array to run
* the command array to run
*/
public CommandLinePlayerAgent(String[] commandArray) {
super();
Expand All @@ -44,8 +46,8 @@ protected OutputStream getInputStream() {
}

@Override
protected InputStream getOutputStream() {
return processStdout;
protected BufferedReader getOutputReader() {
return processInputReader;
}

@Override
Expand All @@ -61,21 +63,15 @@ public void initialize(Properties conf) {
throw new RuntimeException("Failed to launch " + String.join(" ", commandArray), e);
}
processStdin = process.getOutputStream();
processStdout = process.getInputStream();
processInputReader = process.inputReader();

@sdhoine sdhoine Jul 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This uses the platform default charset, while the previous implementation decoded stdout with Agent.UTF8. Please preserve the existing behavior with process.inputReader(UTF8) or an explicit new InputStreamReader(process.getInputStream(), UTF8).

processStderr = process.getErrorStream();
}

@Override
public String getOutput(int nbLine, long timeout) {
String output = super.getOutput(nbLine, timeout);
return output;
}

/**
* Launch the agent. After the call, agent is ready to process input / output
*
* @throws Exception
* if an error occurs
* if an error occurs
*/

@Override
Expand Down
14 changes: 9 additions & 5 deletions runner/src/main/java/com/codingame/gameengine/runner/JavaPlayerAgent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
package com.codingame.gameengine.runner;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
Expand Down Expand Up @@ -34,11 +36,12 @@ public class JavaPlayerAgent extends Agent {

private OutputStream processStdin;
private InputStream processStdout;
private BufferedReader processStdoutReader;
private InputStream processStderr;

/**
* @param className
* The name of the class to use as a participating AI.
* The name of the class to use as a participating AI.
*/
public JavaPlayerAgent(String className) {
super();
Expand All @@ -48,6 +51,7 @@ public JavaPlayerAgent(String className) {
try {
processStdin = new PipedOutputStream(agentStdin);
processStdout = new PipedInputStream(agentStdout, 100_000);
processStdoutReader = new BufferedReader(new InputStreamReader(processStdout, UTF8));
processStderr = new PipedInputStream(agentStderr, 100_000);
} catch (IOException e) {
throw new RuntimeException("Cannot initialize Player Agent", e);
Expand All @@ -60,8 +64,8 @@ protected OutputStream getInputStream() {
}

@Override
protected InputStream getOutputStream() {
return processStdout;
protected BufferedReader getOutputReader() {
return processStdoutReader;
}

@Override
Expand All @@ -77,7 +81,7 @@ public void initialize(Properties conf) {
* Launch the agent. After the call, agent is ready to process input / output
*
* @throws Exception
* if an error occurs
* if an error occurs
*/
@Override
protected void runInputOutput() throws Exception {
Expand All @@ -99,7 +103,7 @@ public void destroy() {
if (javaRunnerThread.isAlive()) {
// TODO
javaRunnerThread.interrupt();
// javaRunnerThread.destroy();
// javaRunnerThread.destroy();
javaRunnerThread.stop();
}
}
Expand Down
Loading