agentscope-ai / agentscope-ai/agentscope-java

[Bug]: JdkHttpTransport SSE streams exhaust the shared boundedElastic scheduler

Open
#2,991 4 comments 0 reactions 0 assignees View on GitHub
area/core bug
Dominant language
Java
Stars
5.6k
Forks
1.3k
Avg merge
4d 12h
Merged PRs (30d)
77

Description

**AgentScope-Java is an open-source project. To involve a broader community, we recommend asking your questions in English.**

**Describe the bug**
`JdkHttpTransport.stream(...)` obtains an `InputStream`, reads it through `BufferedReader.lines()`, and subscribes the reader on Reactor's shared `Schedulers.boundedElastic()` scheduler. A slow SSE connection therefore keeps one worker blocked inside `BufferedReader.readLine()` between chunks.

When the number of active streams reaches the boundedElastic size, additional JDK transport streams and unrelated work using the same shared scheduler are queued until an existing stream terminates.

**To Reproduce**
The opt-in reproduction test is available here: [JdkHttpTransportBoundedElasticConcurrencyTest.java](https://github.com/LeePui/agentscope-java/blob/codex/test-jdk-http-bounded-elastic/agentscope-core/src/test/java/io/agentscope/core/model/transport/JdkHttpTransportBoundedElasticConcurrencyTest.java).It is published in `LeePui/agentscope-java`, branch `codex/test-jdk-http-bounded-elastic`.

The experiment works as follows:

1. It starts an in-process JDK `HttpServer` on `127.0.0.1` with a random port. The mock server uses 48 worker threads for 40 requests, so the server is not the concurrency bottleneck.
2. It creates one real `JdkHttpTransport` and subscribes to 40 streaming requests with a `flatMap` concurrency of 40.
3. The server accepts all requests, then writes and flushes one SSE `data:` event per request every second. Each response contains 30 events followed by `data: [DONE]`, so each connection remains open for about 30 seconds.
4. The test runs in a fresh forked JVM with the platform-thread boundedElastic pool limited to 20 workers. Virtual-thread boundedElastic is explicitly disabled.
5. After the first 20 streams receive a chunk, the test schedules an unrelated canary task on the same boundedElastic scheduler. Between server emissions it also takes 10 JVM thread-stack samples, 50 ms apart, and counts boundedElastic workers inside `BufferedReader.readLine()`.
6. It records every stream's first-chunk latency, how many start within a 5-second early window, the canary queue delay, and the number of blocked reader threads. The expected non-blocking result is 40/40 streams in the early window, canary delay below 2 seconds, and zero blocked readers.

```bash
git clone --single-branch \
--branch codex/test-jdk-http-bounded-elastic \
https://github.com/LeePui/agentscope-java.git \
agentscope-java-jdk-http-repro
cd agentscope-java-jdk-http-repro

mvn -pl agentscope-core \
-Dtest=JdkHttpTransportBoundedElasticConcurrencyTest \
-Dagentscope.test.sse.concurrency.enabled=true \
-Dagentscope.test.sse.requests=40 \
-Dagentscope.test.sse.chunks=30 \
-Dagentscope.test.sse.chunkIntervalMillis=1000 \
-Dreactor.schedulers.defaultBoundedElasticSize=20 \
-Dreactor.schedulers.defaultBoundedElasticOnVirtualThreads=false \
-DforkCount=1 -DreuseForks=false test
```

The reproduction branch intentionally fails the test assertion after printing the measurements; that failure confirms the starvation condition was detected.

Observed with 20 workers / 40 requests:

```text
RESULT pool=20 requests=40 fastestFirstChunkMs=1091 slowestFirstChunkMs=30182
canaryDelayMs=29076 signalThreads=20 blockedReaders=20
```

Only 20 streams received their first chunk in the early window. Thread sampling found all 20
workers blocked in `BufferedReader.readLine()`. The other 20 streams received their first chunk
only after the first wave completed around 30 seconds later.

With 40 workers / 40 requests, all streams started around 1.1 seconds, but all 40 workers were
blocked and the canary still queued for about 29 seconds:

```text
RESULT pool=40 requests=40 fastestFirstChunkMs=1125 slowestFirstChunkMs=1147
canaryDelayMs=29026 signalThreads=40 blockedReaders=40
```

**Expected behavior**
Waiting for SSE network data should not retain a boundedElastic worker. All 40 streams should make progress with a pool size of 20, and unrelated scheduler work should execute promptly.

**Environment**

- AgentScope-Java: `2.0.3-SNAPSHOT`
- Java: Oracle JDK 25.0.3 (sources compiled with `--release 17`)
- OS: macOS arm64

**Additional context**

I have prepared a candidate fix and plan to submit a pull request for this issue.

The fix uses `BodyHandlers.ofPublisher()` and incrementally decodes lines with
`BodySubscribers.fromLineSubscriber(...)`, propagating Reactor demand and cancellation to the JDK
body subscription. This keeps the current public API and parsing behavior without blocking a
scheduler worker while the network is idle.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.