agentscope-ai / agentscope-ai/agentscope-java

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

Aperta
#2,991 4 commenti 0 reazioni 0 assegnatari Vedi su GitHub
area/core bug
Lingua principale
Java
Stelle
5.6k
Fork
1.3k
Merge medio
4g 12h
PR unite (30g)
77

Descrizione

**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.

Guida per i contributori

Apri la guida per i contributori

Valutazione

Questa issue non è ancora stata valutata.

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.