Skip to content

Fix input stream piping in Forker (Issue #882) #2855

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

Open
dorodb-web22 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Fix input stream piping in Forker (Issue #882) #2855

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
07b6df0
Fix input stream piping in Forker
dorodb-web22 Jan 22, 2026
3661f70
Add reproduction test for Issue #882
dorodb-web22 Jan 22, 2026
e60e0ce
Enhance tests: add readLine and large input checks
dorodb-web22 Jan 22, 2026
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
61 changes: 23 additions & 38 deletions frontend/src/main/scala/bloop/exec/Forker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,51 +69,36 @@ object Forker {
logger: Logger,
opts: CommonOptions
): Task[Int] = {
val consumeInput: Option[Cancelable] = None
@volatile var shutdownInput: Boolean = false

/* We need to gobble the input manually with a fixed delay because otherwise
* the remote process will not see it.
*
* The input gobble runs on a 50ms basis and it can process a maximum of 4096
* bytes at a time. The rest that is not read will be read in the next 50ms. */
def goobleInput(to: OutputStream): Cancelable = {
val duration = FiniteDuration(50, TimeUnit.MILLISECONDS)
ExecutionContext.ioScheduler.scheduleWithFixedDelay(duration, duration) {
val buffer = new Array[Byte](4096)
if (shutdownInput) {
consumeInput.foreach(_.cancel())
} else {
try {
if (opts.in.available() > 0) {
val read = opts.in.read(buffer, 0, buffer.length)
if (read == -1) ()
else {
to.write(buffer, 0, read)
to.flush()
}
}
} catch {
case t: IOException =>
logger.debug(s"Error from input gobbler: ${t.getMessage}")
logger.trace(t)
// Rethrow so that Monix cancels future scheduling of the same task
throw t
}
}
}
}

val runTask = run(
Some(cwd.underlying.toFile),
cmd,
logger,
opts.env.toMap,
writeToStdIn = outputStream => {
val mainCancellable = goobleInput(outputStream)
val thread = new Thread {
Copy link
Copy Markdown
Contributor

@tgodzik tgodzik Jan 22, 2026

Choose a reason for hiding this comment

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

Are you able to add some tests? I can confirm that this is fixing the issues at hand

Copy link
Copy Markdown
Author

@dorodb-web22 dorodb-web22 Jan 22, 2026

Choose a reason for hiding this comment

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

@tgodzik

I've added the tests as requested.
The test cases verify that the blocking reader thread approach fixes the stdin readline issue, covering:

  1. Input forwarding without data loss
  2. No indefinite waiting when reading from stdin
  3. Compatibility with forker-realib patterns
    4)Standard I/O patterns work correctly

Could you approve the workflow run so CI can execute the tests? Thanks!

override def run(): Unit = {
val buffer = new Array[Byte](4096)
try {
while (opts.in != null) {
val read = opts.in.read(buffer, 0, buffer.length)
if (read == -1) return
else {
outputStream.write(buffer, 0, read)
outputStream.flush()
}
}
} catch {
case t: IOException =>
logger.debug(s"Error from input reader: ${t.getMessage}")
logger.trace(t)
}
}
}

thread.start()

Cancelable { () =>
shutdownInput = true
mainCancellable.cancel()
thread.interrupt()
}
},
debugLog = msg => {
Expand Down