Skip to content

Commit e8cb8a8

Browse files
committed
refactor: unify non-blocking sse client api
1 parent 60a3e48 commit e8cb8a8

18 files changed

Lines changed: 760 additions & 289 deletions

scripts/evaluate-sdk.zsh

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
#!/bin/zsh
2+
set -uo pipefail
3+
4+
repo_root=${0:a:h:h}
5+
package_root=src/main/java/io/github/easy4j/opencode
6+
result_dir=$repo_root/target/benchmark
7+
failed=0
8+
9+
for branch in feature/1.0.x feature/2.0.x feature/3.0.x; do
10+
print "STATIC opencode $branch"
11+
git -C "$repo_root" cat-file -e \
12+
"${branch}:${package_root}/api/OpenCodeSseClient.java" || failed=1
13+
git -C "$repo_root" cat-file -e \
14+
"${branch}:${package_root}/api/sse/SseSubscription.java" || failed=1
15+
if ! git -C "$repo_root" grep -q -E \
16+
'public[[:space:]]+OpenCodeSseClient[[:space:]]+sse\(\)' \
17+
"$branch" -- "$package_root/OpenCodeClient.java"; then
18+
print "FAIL missing OpenCodeClient.sse(): $branch"
19+
failed=1
20+
fi
21+
if git -C "$repo_root" grep -n -E \
22+
'eventClient|[[:space:]]events\(\)|public[[:space:]]+void[[:space:]]+stop\(\)|subscribeSessionStream\(|Thread\.sleep|\.execute\(\)|Netty event-loop' \
23+
"$branch" -- "$package_root/api/*HttpClient.java" \
24+
"$package_root/api/*ChatClient.java" "$package_root/api/*SseClient.java"; then
25+
print "FAIL legacy/blocking SSE implementation remains: $branch"
26+
failed=1
27+
fi
28+
done
29+
30+
if [[ -e "$repo_root/$package_root/api/model/Event.java" ]]; then
31+
print 'FAIL legacy api.model.Event remains'
32+
failed=1
33+
fi
34+
35+
if [[ "$(git -C "$repo_root" branch --show-current)" != feature/3.0.x ]]; then
36+
print 'FAIL feature/3.0.x must be checked out for latest verification'
37+
failed=1
38+
else
39+
JAVA_HOME=${JAVA_HOME:-$(/usr/libexec/java_home -v 21)}
40+
PATH="$JAVA_HOME/bin:$PATH" mvn -q -f "$repo_root/pom.xml" \
41+
-Dtest=OpenCodeSseApiShapeTest,OpenCodeNonBlockingConcurrencyTest test || failed=1
42+
fi
43+
44+
python3 "$repo_root/scripts/verify-benchmark-results.py" \
45+
--sdk opencode \
46+
--results "$result_dir/benchmark-results.csv" \
47+
--branches "$result_dir/branch-verification.tsv" || failed=1
48+
49+
if (( failed != 0 )); then
50+
print 'OPENCODE_SDK_EVALUATOR_FAIL'
51+
exit 1
52+
fi
53+
print 'OPENCODE_SDK_EVALUATOR_PASS'
Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
#!/bin/zsh
2+
set -euo pipefail
3+
4+
repo_root=${0:a:h:h}
5+
sdk=opencode
6+
workload=${1:-}
7+
concurrency=${2:-}
8+
result_file=${3:-$repo_root/target/benchmark/benchmark-results.csv}
9+
lock_dir=/tmp/easy4j-sdk-benchmark-exclusive.lock
10+
11+
if [[ "$workload" != http && "$workload" != sse ]]; then
12+
print -u2 'Usage: run-concurrency-benchmark.zsh <http|sse> <concurrency> [result.csv]'
13+
exit 64
14+
fi
15+
if [[ ! "$concurrency" =~ '^[1-9][0-9]*$' ]]; then
16+
print -u2 'concurrency must be a positive integer'
17+
exit 64
18+
fi
19+
if ! mkdir "$lock_dir" 2>/dev/null; then
20+
print -u2 "another SDK benchmark holds the global lock: $lock_dir"
21+
exit 75
22+
fi
23+
24+
benchmark_pid=
25+
cleanup() {
26+
if [[ -n "$benchmark_pid" ]] && kill -0 "$benchmark_pid" 2>/dev/null; then
27+
kill "$benchmark_pid" 2>/dev/null || true
28+
fi
29+
rmdir "$lock_dir" 2>/dev/null || true
30+
}
31+
trap cleanup EXIT INT TERM
32+
33+
mkdir -p "$repo_root/target/benchmark"
34+
classpath_file=$repo_root/target/benchmark/test-classpath.txt
35+
metrics_file=$repo_root/target/benchmark/${sdk}-${workload}-${concurrency}.metrics
36+
samples_file=$repo_root/target/benchmark/${sdk}-${workload}-${concurrency}.samples
37+
log_file=$repo_root/target/benchmark/${sdk}-${workload}-${concurrency}.log
38+
rm -f "$metrics_file" "$samples_file" "$log_file"
39+
40+
JAVA_HOME=${JAVA_HOME:-$(/usr/libexec/java_home -v 21)}
41+
PATH="$JAVA_HOME/bin:$PATH" mvn -q -f "$repo_root/pom.xml" -DskipTests test-compile \
42+
dependency:build-classpath -Dmdep.outputFile="$classpath_file"
43+
classpath="$repo_root/target/test-classes:$repo_root/target/classes:$(<"$classpath_file")"
44+
started_epoch_ms=$(python3 -c 'import time; print(time.time_ns() // 1000000)')
45+
"$JAVA_HOME/bin/java" -cp "$classpath" io.github.easy4j.opencode.OpenCodeConcurrencyBenchmark \
46+
"$workload" "$concurrency" "$metrics_file" >"$log_file" 2>&1 &
47+
benchmark_pid=$!
48+
49+
while kill -0 "$benchmark_pid" 2>/dev/null; do
50+
ps -o %cpu= -o rss= -p "$benchmark_pid" | awk 'NF == 2 { print $1, $2 }' >>"$samples_file"
51+
sleep 0.2
52+
done
53+
set +e
54+
wait "$benchmark_pid"
55+
benchmark_exit=$?
56+
set -e
57+
benchmark_pid=
58+
ended_epoch_ms=$(python3 -c 'import time; print(time.time_ns() // 1000000)')
59+
60+
if [[ ! -s "$metrics_file" ]]; then
61+
print -u2 "benchmark did not produce metrics; see $log_file"
62+
exit ${benchmark_exit:-1}
63+
fi
64+
operations=$(awk -F= '$1 == "operations" { print $2 }' "$metrics_file")
65+
errors=$(awk -F= '$1 == "errors" { print $2 }' "$metrics_file")
66+
duration_seconds=$(awk -F= '$1 == "duration_seconds" { print $2 }' "$metrics_file")
67+
throughput=$(awk -F= '$1 == "throughput_per_sec" { print $2 }' "$metrics_file")
68+
read avg_cpu peak_cpu peak_rss_mb <<<"$(awk '
69+
BEGIN { sum=0; count=0; peak_cpu=0; peak_rss=0 }
70+
NF == 2 { sum += $1; count++; if ($1 > peak_cpu) peak_cpu=$1; if ($2 > peak_rss) peak_rss=$2 }
71+
END { printf "%.3f %.3f %.3f", count ? sum/count : 0, peak_cpu, peak_rss/1024 }
72+
' "$samples_file")"
73+
benchmark_status=PASS
74+
if (( benchmark_exit != 0 || errors != 0 )) || [[ "$peak_cpu" == 0.000 ]]; then
75+
benchmark_status=FAIL
76+
fi
77+
git_sha=$(git -C "$repo_root" rev-parse HEAD)
78+
if [[ ! -s "$result_file" ]]; then
79+
print 'sdk,workload,concurrency,started_epoch_ms,ended_epoch_ms,duration_seconds,operations,errors,throughput_per_sec,avg_cpu_pct,peak_cpu_pct,peak_rss_mb,status,git_sha' >"$result_file"
80+
fi
81+
print "$sdk,$workload,$concurrency,$started_epoch_ms,$ended_epoch_ms,$duration_seconds,$operations,$errors,$throughput,$avg_cpu,$peak_cpu,$peak_rss_mb,$benchmark_status,$git_sha" >>"$result_file"
82+
print "$sdk $workload concurrency=$concurrency status=$benchmark_status cpu(avg/peak)=$avg_cpu/$peak_cpu rss_peak_mb=$peak_rss_mb throughput=$throughput"
83+
if [[ "$benchmark_status" != PASS ]]; then
84+
print -u2 "benchmark failed; see $log_file"
85+
exit 1
86+
fi
Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
#!/usr/bin/env python3
2+
"""校验单个 SDK 的分支测试和串行并发压测结果。"""
3+
4+
import argparse
5+
import csv
6+
from pathlib import Path
7+
8+
9+
CONCURRENCY_LEVELS = {"100", "300", "500", "800", "1000"}
10+
WORKLOADS = {"http", "sse"}
11+
BRANCHES = {"feature/1.0.x", "feature/2.0.x", "feature/3.0.x"}
12+
13+
14+
def parse_args():
15+
parser = argparse.ArgumentParser()
16+
parser.add_argument("--sdk", required=True)
17+
parser.add_argument("--results", required=True)
18+
parser.add_argument("--branches", required=True)
19+
return parser.parse_args()
20+
21+
22+
def read_rows(path, delimiter=","):
23+
target = Path(path)
24+
if not target.is_file():
25+
raise SystemExit(f"FAIL missing result file: {target}")
26+
with target.open(newline="", encoding="utf-8") as handle:
27+
return list(csv.DictReader(handle, delimiter=delimiter))
28+
29+
30+
def verify_branches(sdk, path):
31+
rows = read_rows(path, "\t")
32+
expected = {(sdk, branch) for branch in BRANCHES}
33+
actual = {
34+
(row.get("sdk"), row.get("branch"))
35+
for row in rows
36+
if row.get("status") == "PASS"
37+
}
38+
if len(rows) != 3 or actual != expected:
39+
raise SystemExit("FAIL branch verification must contain exactly 3 passing rows")
40+
41+
42+
def verify_benchmarks(sdk, path):
43+
rows = read_rows(path)
44+
expected = {
45+
(sdk, workload, concurrency)
46+
for workload in WORKLOADS
47+
for concurrency in CONCURRENCY_LEVELS
48+
}
49+
actual = {
50+
(row.get("sdk"), row.get("workload"), row.get("concurrency"))
51+
for row in rows
52+
}
53+
if len(rows) != 10 or actual != expected:
54+
raise SystemExit("FAIL benchmark results must contain exactly 10 unique runs")
55+
56+
intervals = []
57+
for row in rows:
58+
if row.get("status") != "PASS" or int(row.get("errors", "-1")) != 0:
59+
raise SystemExit(f"FAIL benchmark errors: {row}")
60+
if float(row.get("avg_cpu_pct", "-1")) < 0:
61+
raise SystemExit(f"FAIL invalid average CPU: {row}")
62+
if float(row.get("peak_cpu_pct", "0")) <= 0:
63+
raise SystemExit(f"FAIL invalid peak CPU: {row}")
64+
if float(row.get("peak_rss_mb", "0")) <= 0:
65+
raise SystemExit(f"FAIL invalid peak RSS: {row}")
66+
if float(row.get("throughput_per_sec", "0")) <= 0:
67+
raise SystemExit(f"FAIL invalid throughput: {row}")
68+
start = int(row["started_epoch_ms"])
69+
end = int(row["ended_epoch_ms"])
70+
if end <= start:
71+
raise SystemExit(f"FAIL invalid interval: {row}")
72+
intervals.append((start, end, row))
73+
74+
intervals.sort(key=lambda value: value[0])
75+
for previous, current in zip(intervals, intervals[1:]):
76+
if current[0] < previous[1]:
77+
raise SystemExit(
78+
f"FAIL overlapping benchmark runs: {previous[2]} and {current[2]}"
79+
)
80+
81+
82+
def main():
83+
args = parse_args()
84+
verify_branches(args.sdk, args.branches)
85+
verify_benchmarks(args.sdk, args.results)
86+
print(f"{args.sdk.upper()}_BENCHMARK_RESULTS_PASS")
87+
88+
89+
if __name__ == "__main__":
90+
main()

src/main/java/io/github/easy4j/opencode/OpenCodeClient.java

Lines changed: 14 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@ public class OpenCodeClient implements AutoCloseable {
6060
private final OpenCodeChatClient chatClient;
6161
private final OpenCodeSseClient sseClient;
6262
private final OpenCodeCli cli;
63+
private final OkHttpClient ownedHttpClient;
6364

6465
// ============================================================
6566
// 构造器
@@ -114,13 +115,19 @@ public OpenCodeClient(OpenCodeHttpClientConfig httpConfig, OpenCodeCliConfig cli
114115

115116
// HTTP 子系统初始化
116117
if (httpEnabled) {
117-
this.chatClient = new OpenCodeChatClient(httpConfig, objectMapper, httpClient);
118+
OkHttpClient sharedHttpClient = Objects.nonNull(httpClient)
119+
? httpClient : OpenCodeOkHttpClientFactory.create(httpConfig);
120+
this.ownedHttpClient = Objects.isNull(httpClient) ? sharedHttpClient : null;
121+
this.sseClient = new OpenCodeSseClient(
122+
httpConfig, objectMapper, sharedHttpClient);
123+
this.chatClient = new OpenCodeChatClient(
124+
httpConfig, objectMapper, sharedHttpClient, sseClient);
118125
this.httpClient = this.chatClient;
119-
this.sseClient = this.chatClient.events();
120126
} else {
121127
this.httpClient = null;
122128
this.chatClient = null;
123129
this.sseClient = null;
130+
this.ownedHttpClient = null;
124131
}
125132

126133
// CLI 子系统初始化
@@ -160,6 +167,7 @@ public OpenCodeClient(OpenCodeClientConfig config,
160167
this.chatClient = httpClient instanceof OpenCodeChatClient ? (OpenCodeChatClient) httpClient : null;
161168
this.sseClient = sseClient;
162169
this.cli = cli;
170+
this.ownedHttpClient = null;
163171
}
164172

165173
// ============================================================
@@ -431,25 +439,11 @@ public OpenCodeChatClient chat() {
431439
return chatClient;
432440
}
433441

434-
public OpenCodeSseClient eventStream() {
442+
/** 获取统一的 OpenCode SSE 场景客户端。 */
443+
public OpenCodeSseClient sse() {
435444
return sseClient;
436445
}
437446

438-
public okhttp3.sse.EventSource onSessionEvent(String sessionId,
439-
io.github.easy4j.opencode.api.event.EventHandler handler) {
440-
return sseClient.subscribeHandler(sessionId, handler);
441-
}
442-
443-
public okhttp3.sse.EventSource onEvent(
444-
io.github.easy4j.opencode.api.event.EventHandler handler) {
445-
return sseClient.subscribeHandler(null, handler);
446-
}
447-
448-
public okhttp3.sse.EventSource onEventTypes(java.util.Set<String> types,
449-
java.util.function.Consumer<Event> consumer) {
450-
return sseClient.subscribeEventTypes(types, consumer);
451-
}
452-
453447
// ============================================================
454448
// CLI
455449
// ============================================================
@@ -877,7 +871,8 @@ public io.github.easy4j.opencode.cli.OpenCodeCliResult cliPr(int number) {
877871

878872
@Override
879873
public void close() {
880-
if (httpClient != null) httpClient.close();
881874
if (sseClient != null) sseClient.close();
875+
if (httpClient != null) httpClient.close();
876+
OpenCodeOkHttpClientFactory.shutdown(ownedHttpClient);
882877
}
883878
}

0 commit comments

Comments
 (0)