diff --git a/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/code_review_cloud_G08_0.log b/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/code_review_cloud_G08_0.log new file mode 100644 index 0000000..d60a443 --- /dev/null +++ b/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/code_review_cloud_G08_0.log @@ -0,0 +1,133 @@ + + +# Code Review Reference - REFACTOR + +> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** +> Do not implement until predecessor `10` has `complete.log`. + +## 개요 + +date=2026-06-06 +task=m-performance-hotspot-optimization/11+10_dart_isolate_transport_path, plan=0, tag=REFACTOR + +## Roadmap Targets + +- Milestone: `agent-roadmap/milestones/performance-hotspot-optimization.md` +- Task ids: + - `dart-isolate-real-path`: 실제 TCP/WS transport 수신 경로 hardening 연결 +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다. + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REFACTOR-1] TCP transport receive path | [x] | +| [REFACTOR-2] WS IO/web receive path | [x] | +| [REFACTOR-3] Performance row inclusion | [x] | + +## 구현 체크리스트 + +- [x] predecessor `10_dart_isolate_gateway_contract` complete.log 존재를 확인한다. (archive: `agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/complete.log` 존재 + git `9082c0c`에서 archive 확인. 의존성 충족.) +- [x] `ProtobufClient`가 decoded inline path 대신 hardened gateway-backed `onReceivedFrame`을 사용하게 한다. (`_parsing()`이 `onReceivedFrame(packetBytes)` 호출, 생성자에서 `IsolateInboundGateway` attach+start.) +- [x] `WsProtobufClient` IO/web이 같은 receive coordinator path를 사용하고 web은 `SyncInboundGateway` fallback을 유지한다. (IO는 `inbound_gateway_io.dart`의 `IsolateInboundGateway`, web은 `inbound_gateway_web.dart`의 `SyncInboundGateway` 기반 `IsolateInboundGateway`. 둘 다 `_dispatch`가 `onReceivedFrame(bytes)`만 수행.) +- [x] 내부 단일 수신 경로로 동작하며 public on/off 옵션을 추가하지 않는다. (gateway는 각 transport 생성자 내부에서만 attach. public API/생성자 시그니처 변경 없음.) +- [x] TCP/WS FIFO, nonce matching, close cleanup regression tests와 focused stress 결과를 기록한다. (`dart test` 67 통과, focused stress dart 12 rows violations=0. 아래 검증 결과 참조.) +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +- [x] 판정을 append한다. +- [x] PASS이면 complete/archive 절차와 런타임 완료 이벤트를 처리한다. + +## 계획 대비 변경 사항 + +- 계획 범위(REFACTOR-1/2)는 transport 3종을 `onReceivedFrame`으로 전환하는 것이었으나, 전환 후 기존 회귀 테스트 `TCP inbound backpressure pauses subscription on heavy load`가 실패했다. 원인은 transport swap 자체가 아니라 predecessor `10`의 gateway wiring이 **decode in-flight**(gateway capacity)로만 backpressure를 측정하는데, 실제 병목은 **dispatch**(stalled handler)라는 점이다. decode는 병목이 아니므로 handler가 막혀도 socket이 멈추지 않고 decoded frame이 unbounded로 쌓여 기존 64-bound 회귀가 발생했다. +- 이를 plan의 "심볼 참조"가 명시한 변경 후보(`Communicator.attachInboundGateway`, `onReceivedFrame`) 범위 안에서 `dart/lib/src/communicator.dart`를 추가 수정해 해결했다. 계획의 "수정 파일 요약" 표에는 communicator.dart가 없었으나, 같은 plan 본문이 두 심볼을 변경 후보로 적시했으므로 범위 내로 판단했고 사용자 리뷰 요청으로 올리지 않았다. +- web `IsolateInboundGateway`는 `SyncInboundGateway`를 상속한 conditional-export 구현이라, 계획 체크리스트의 "web `SyncInboundGateway` fallback 유지"를 그대로 만족한다(별도 분기 코드 불필요). + +## 주요 설계 결정 + +- **gateway 경로 전용 backlog 카운터로 dispatch backpressure를 transport read loop에 전파.** `_gatewayBacklog`는 "gateway에 submit했지만 아직 dispatch 완료되지 않은 frame 수"다. `onReceivedFrame`(gateway branch)이 submit 전에 `_gatewayBacklog >= _inboundQueueCapacity`(64)면 대기하고, 통과 시 `_gatewayBacklog++` 후 submit한다. backlog는 frame이 실제로 handle된 시점(데이터/요청은 dispatch 체인 `whenComplete`, 응답은 pending 완료 직후, error는 즉시)에만 `_releaseGatewayBacklog()`로 감소한다. 따라서 stalled handler 하에서 backlog가 64까지 차면 read loop가 멈춰 inline 경로와 동일한 outstanding-frame bound를 회복한다. +- **gateway 경로를 `onReceivedData`의 큐 회계와 분리.** `onReceivedData`(+`_inboundQueueLength`/`_awaitInboundCapacity`)는 inline 경로와 직접 호출 테스트가 계속 쓰므로 그대로 두고, gateway listener는 decoded frame을 `_onGatewayResult`에서 직접 dispatch 체인에 태운다. 이렇게 하면 큐 슬롯을 enqueue 시점이 아니라 submit 시점에 예약(backlog)하게 되어, submit→decode→enqueue 사이 비동기 지연으로 read loop가 앞서 달리던 문제를 제거한다(단일 카운터로 두 회계를 섞지 않는다). +- **응답 immediacy 보존.** gateway 경로에서도 `responseNonce > 0` frame은 dispatch 체인을 거치지 않고 pending request를 즉시 완료한다 — inline `onReceivedData`와 동일한 out-of-band 처리. +- **decode pipeline 깊이는 gateway capacity(1024)가, dispatch bound는 backlog(64)가 담당**하도록 두 backpressure 축을 분리해, 병렬 decode 이점과 메모리 bound를 동시에 유지한다. +- **gateway start()는 await하지 않는다.** async 함수는 첫 await 전까지 동기 실행되어 `_ready` completer가 생성자 동기 구간에 설정되므로, isolate spawn 완료 전 도착한 첫 frame도 `submit`이 readiness를 기다려 drop되지 않는다. +- **teardown 안전.** `close()`/`shutdown()`에 `_drainGatewayBacklog()`를 추가해 backlog를 0으로 리셋하고 대기 중인 `onReceivedFrame`을 깨워, gateway 경로가 read loop를 strand하지 않게 한다. + +## 사용자 리뷰 요청 + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- `10` predecessor가 실제 PASS인지 확인한다. +- TCP/WS/web 모두 inline decode bypass 없이 단일 receive coordinator를 쓰는지 확인한다. +- public gateway on/off 옵션이 생기지 않았는지 확인한다. + +## 검증 결과 + +### REFACTOR-1/2 중간 검증 +```text +$ cd dart && dart analyze lib/src/protobuf_client.dart lib/src/ws_protobuf_client_io.dart lib/src/ws_protobuf_client_web.dart lib/src/communicator.dart +No issues found! + +$ cd dart && dart test test/socket_test.dart test/communicator_test.dart -r expanded +00:15 +64: ... Request-Response (WS plain) TCP inbound backpressure pauses subscription on heavy load +00:16 +65: ... TCP fragmentation and large payload TCP receives fragmented large frame without reordering or truncation +00:16 +66: ... TCP fragmentation and large payload TCP 1MB payload roundtrip +00:17 +67: All tests passed! +``` +- FIFO/fragmentation/1MB/request-response/close/broadcast 회귀 및 gateway 단위 테스트 모두 통과(67). +- backpressure 회귀 테스트는 transport swap 직후 FAIL이었고, communicator backlog gate 수정 후 PASS로 회복(계획 대비 변경 사항 참조). + +### REFACTOR-3 중간 검증 (focused stress) +```text +$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_stress.sh --quick --lang dart --transport tcp,ws --profile roundtrip,payload +DONE dart: PASS (rows=12 skips=0 required_skips=0 violations=0) +| 언어 | 결과 | ROW | SKIP(전체) | SKIP(required) | stability_violations | 비고 | +| dart | PASS | 12 | 0 | 0 | 0 | | +전체 결과값: PASS +결과 기록 파일: agent-test/runs/20260606-042547-proto-socket-stress-quick.md +``` + +### 최종 검증 +```text +$ cd dart && dart test +00:17 +67: All tests passed! + +$ cd dart && dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js +Compiled 10,874,863 input bytes (5,876,091 characters source) to 11,692 characters JavaScript in 0.50 seconds + +$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_performance.sh --quick +| 구성 | 결과 | exit code | +| same-language | PASS | 0 | (Dart 18 rows, 모든 row stability_violations=0) +| cross-language | PASS | 0 | +| typescript-gateway | PASS | 0 | +전체 결과값: PASS +결과 기록 파일: agent-test/runs/20260606-042556-proto-socket-performance-quick.md +``` +- same-language run의 Dart 실제 수신 경로(gateway-backed) rows: roundtrip(c1/16/64/256), burst, sustained, parallel, payload(1KB/64KB) × tcp/ws 전부 PASS, stability_violations 0. + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - correctness: Pass + - completeness: Pass + - test coverage: Pass + - API contract: Pass + - code quality: Pass + - plan deviation: Pass + - verification trust: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS이므로 active plan/review를 `.log`로 아카이브하고 `complete.log` 작성 후 task 디렉터리를 archive로 이동한다. diff --git a/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/complete.log b/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/complete.log new file mode 100644 index 0000000..76975d9 --- /dev/null +++ b/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/complete.log @@ -0,0 +1,44 @@ +# Complete - m-performance-hotspot-optimization/11+10_dart_isolate_transport_path + +## 완료 일시 + +2026-06-06 + +## 요약 + +Dart TCP/WS transport 수신 경로를 gateway-backed receive coordinator로 연결한 구현을 1회 리뷰했고 최종 판정은 PASS다. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G08_0.log` | `code_review_cloud_G08_0.log` | PASS | TCP/WS IO/web 수신 경로 전환, communicator gateway backlog 보정, focused stress/performance 검증이 계획 범위를 충족했다. | + +## 구현/정리 내용 + +- `ProtobufClient` TCP 수신 path를 inline `PacketBase.fromBuffer` decode 대신 `onReceivedFrame` gateway-backed coordinator로 전환했다. +- `WsProtobufClient` IO/web 수신 path를 동일한 coordinator contract로 전환하고 web conditional export의 sync fallback을 유지했다. +- gateway decode와 dispatch 병목을 분리해 `_gatewayBacklog`로 stalled handler 상황의 transport backpressure를 보존했다. + +## 최종 검증 + +- `dart analyze lib/src/protobuf_client.dart lib/src/ws_protobuf_client_io.dart lib/src/ws_protobuf_client_web.dart lib/src/communicator.dart` - PASS; No issues found. +- `dart test test/socket_test.dart test/communicator_test.dart -r expanded` - PASS; 67 tests passed. +- `dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js` - PASS; JavaScript output generated. +- `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_stress.sh --quick --lang dart --transport tcp,ws --profile roundtrip,payload` - PASS; 12 rows, required skips 0, stability violations 0; result=`agent-test/runs/20260606-045133-proto-socket-stress-quick.md`. +- `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_performance.sh --quick` - PASS; same-language/cross-language/typescript-gateway all PASS, regression SKIPPED; result=`agent-test/runs/20260606-045148-proto-socket-performance-quick.md`. + +## Roadmap Completion + +- Milestone: `agent-roadmap/milestones/performance-hotspot-optimization.md` +- Completed task ids: + - `dart-isolate-real-path`: PASS; evidence=`agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/plan_cloud_G08_0.log`, `agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/code_review_cloud_G08_0.log`; verification=`dart test test/socket_test.dart test/communicator_test.dart -r expanded`, `dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js`, `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_stress.sh --quick --lang dart --transport tcp,ws --profile roundtrip,payload`, `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_performance.sh --quick` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/PLAN-cloud-G08.md b/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/plan_cloud_G08_0.log similarity index 100% rename from agent-task/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/PLAN-cloud-G08.md rename to agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/plan_cloud_G08_0.log diff --git a/agent-task/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/CODE_REVIEW-cloud-G08.md b/agent-task/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/CODE_REVIEW-cloud-G08.md deleted file mode 100644 index 44c2e3f..0000000 --- a/agent-task/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/CODE_REVIEW-cloud-G08.md +++ /dev/null @@ -1,92 +0,0 @@ - - -# Code Review Reference - REFACTOR - -> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** -> Do not implement until predecessor `10` has `complete.log`. - -## 개요 - -date=2026-06-06 -task=m-performance-hotspot-optimization/11+10_dart_isolate_transport_path, plan=0, tag=REFACTOR - -## Roadmap Targets - -- Milestone: `agent-roadmap/milestones/performance-hotspot-optimization.md` -- Task ids: - - `dart-isolate-real-path`: 실제 TCP/WS transport 수신 경로 hardening 연결 -- Completion mode: check-on-pass - -## 이 파일을 읽는 리뷰 에이전트에게 - -> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다. - -## 구현 항목별 완료 여부 - -| 항목 | 완료 여부 | -|------|---------| -| [REFACTOR-1] TCP transport receive path | [ ] | -| [REFACTOR-2] WS IO/web receive path | [ ] | -| [REFACTOR-3] Performance row inclusion | [ ] | - -## 구현 체크리스트 - -- [ ] predecessor `10_dart_isolate_gateway_contract` complete.log 존재를 확인한다. -- [ ] `ProtobufClient`가 decoded inline path 대신 hardened gateway-backed `onReceivedFrame`을 사용하게 한다. -- [ ] `WsProtobufClient` IO/web이 같은 receive coordinator path를 사용하고 web은 `SyncInboundGateway` fallback을 유지한다. -- [ ] 내부 단일 수신 경로로 동작하며 public on/off 옵션을 추가하지 않는다. -- [ ] TCP/WS FIFO, nonce matching, close cleanup regression tests와 focused stress 결과를 기록한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. - -## 코드리뷰 전용 체크리스트 - -- [ ] 판정을 append한다. -- [ ] PASS이면 complete/archive 절차와 런타임 완료 이벤트를 처리한다. - -## 계획 대비 변경 사항 - -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ - -## 주요 설계 결정 - -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ - -## 사용자 리뷰 요청 - -- 상태: 없음 -- 사유 유형: 없음 -- 결정 필요: 없음 -- 차단 근거: 없음 -- 실행한 검증/명령: 없음 -- 자동 후속 불가 이유: 없음 -- 재개 조건: 없음 - -## 리뷰어를 위한 체크포인트 - -- `10` predecessor가 실제 PASS인지 확인한다. -- TCP/WS/web 모두 inline decode bypass 없이 단일 receive coordinator를 쓰는지 확인한다. -- public gateway on/off 옵션이 생기지 않았는지 확인한다. - -## 검증 결과 - -### REFACTOR-1 중간 검증 -```text -$ cd dart && dart test test/socket_test.dart -r expanded -(output) -``` - -### REFACTOR-3 중간 검증 -```text -$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_stress.sh --quick --lang dart --transport tcp,ws --profile roundtrip,payload -(output) -``` - -### 최종 검증 -```text -$ cd dart && dart test test/socket_test.dart test/communicator_test.dart -r expanded -(output) -$ cd dart && dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js -(output) -$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_performance.sh --quick -(output) -``` diff --git a/dart/lib/src/communicator.dart b/dart/lib/src/communicator.dart index 9d2cba3..e6555c7 100644 --- a/dart/lib/src/communicator.dart +++ b/dart/lib/src/communicator.dart @@ -47,6 +47,16 @@ abstract class Communicator { StreamSubscription? _gatewaySubscription; int _frameSeq = 0; + /// Frames handed to the gateway but not yet accepted into the dispatch queue. + /// + /// Decode is never the bottleneck, so the gateway's own in-flight capacity + /// cannot throttle a stalled dispatch handler. Gating [onReceivedFrame] on + /// this backlog makes a full inbound queue propagate to the transport read + /// loop and keeps decoded frames from piling up unbounded behind a stuck + /// handler — the same bound the inline path gets for free. + int _gatewayBacklog = 0; + Completer? _gatewayBacklogWaiter; + /// Monotonically increasing nonce. Shared by send / sendRequest / response. int _nonce = 0; int get nonce => _nonce; @@ -233,24 +243,56 @@ abstract class Communicator { /// Attaches a worker [gateway] in front of the receive coordinator. /// - /// Decoded results are forwarded to [onReceivedData] in input [seq] order, so - /// the coordinator keeps its FIFO dispatch and sole ownership of stateful - /// handling. Attaching is opt-in; without it [onReceivedFrame] decodes inline. + /// Decoded results are dispatched in input [seq] order, so the coordinator + /// keeps its FIFO dispatch and sole ownership of stateful handling. Each + /// result releases the backlog slot its frame reserved in [onReceivedFrame], + /// only once the frame is actually handled. Attaching is opt-in; without it + /// [onReceivedFrame] decodes inline. @protected void attachInboundGateway(InboundGateway gateway) { _inboundGateway = gateway; - _gatewaySubscription = gateway.results.listen((frame) { - if (frame.isError) { - onGatewayDecodeError(frame.seq, frame.error!); - return; - } - onReceivedData( - frame.typeName, - frame.data, - incomingNonce: frame.incomingNonce, - responseNonce: frame.responseNonce, - ); - }); + _gatewaySubscription = gateway.results.listen(_onGatewayResult); + } + + /// Consumes one ordered gateway result and frees its reserved backlog slot + /// once handled — not merely decoded — so a stalled handler keeps the + /// transport read loop paused via [onReceivedFrame]. + void _onGatewayResult(DecodedFrame frame) { + if (frame.isError) { + onGatewayDecodeError(frame.seq, frame.error!); + _releaseGatewayBacklog(); + return; + } + if (frame.responseNonce > 0) { + // Responses complete out of band, exactly as the inline coordinator does. + final pending = _pendingRequests.remove(frame.responseNonce); + pending?.complete(frame.typeName, frame.data); + _releaseGatewayBacklog(); + return; + } + // Data/request frames serialize on the dispatch chain. The reserved backlog + // slot is held until dispatch completes, so the chain depth — and thus the + // outstanding frame bound — matches the inline coordinator's queue. + final item = _InboundItem( + typeName: frame.typeName, + data: frame.data, + incomingNonce: frame.incomingNonce, + responseNonce: frame.responseNonce, + ); + _inboundDispatch = _inboundDispatch + .then((_) => _dispatchInbound(item)) + .whenComplete(_releaseGatewayBacklog); + } + + /// Frees one backlog slot and wakes a waiting [onReceivedFrame] if the gateway + /// has fallen back under capacity. + void _releaseGatewayBacklog() { + if (_gatewayBacklog > 0) _gatewayBacklog--; + final waiter = _gatewayBacklogWaiter; + if (waiter != null && _gatewayBacklog < _inboundQueueCapacity) { + _gatewayBacklogWaiter = null; + waiter.complete(); + } } /// Handles an ordered decode failure surfaced by the gateway. @@ -279,6 +321,15 @@ abstract class Communicator { if (!isAlive || _isClosing) return; final gateway = _inboundGateway; if (gateway != null) { + // Throttle on the dispatch backlog before handing the frame off. This + // propagates a full inbound queue to the transport read loop even though + // the gateway's in-flight capacity (decode) is never the bottleneck. + while (isAlive && !_isClosing && _gatewayBacklog >= _inboundQueueCapacity) { + _gatewayBacklogWaiter ??= Completer(); + await _gatewayBacklogWaiter!.future; + } + if (!isAlive || _isClosing) return; + _gatewayBacklog++; // Await so gateway backpressure propagates to the transport read loop. await gateway.submit(InboundFrame(seq: _nextFrameSeq(), bytes: frame)); return; @@ -288,6 +339,15 @@ abstract class Communicator { incomingNonce: common.nonce, responseNonce: common.responseNonce); } + /// Blocks until the inbound dispatch queue has room. Shared by the inline and + /// gateway receive paths so a stalled handler throttles the transport equally. + Future _awaitInboundCapacity() async { + while (_inboundQueueLength >= _inboundQueueCapacity) { + _queueWaiter ??= Completer(); + await _queueWaiter!.future; + } + } + /// Enqueues an inbound packet for serial dispatch. /// /// The receive worker processes items one at a time, preserving FIFO order @@ -302,10 +362,7 @@ abstract class Communicator { } return; } - while (_inboundQueueLength >= _inboundQueueCapacity) { - _queueWaiter ??= Completer(); - await _queueWaiter!.future; - } + await _awaitInboundCapacity(); if (!isAlive || _isClosing) return; _inboundQueueLength++; final item = _InboundItem( @@ -337,6 +394,7 @@ abstract class Communicator { _queueWaiter!.complete(); _queueWaiter = null; } + _drainGatewayBacklog(); await _closeInboundGateway(); cancelPendingRequests(); } @@ -351,10 +409,22 @@ abstract class Communicator { _queueWaiter!.complete(); _queueWaiter = null; } + _drainGatewayBacklog(); unawaited(_closeInboundGateway()); cancelPendingRequests(); } + /// Releases any [onReceivedFrame] blocked on backlog and clears the counter so + /// teardown never strands the transport read loop. + void _drainGatewayBacklog() { + _gatewayBacklog = 0; + final waiter = _gatewayBacklogWaiter; + if (waiter != null) { + _gatewayBacklogWaiter = null; + waiter.complete(); + } + } + Future _closeInboundGateway() async { final subscription = _gatewaySubscription; final gateway = _inboundGateway; diff --git a/dart/lib/src/protobuf_client.dart b/dart/lib/src/protobuf_client.dart index 13868a4..c30d468 100644 --- a/dart/lib/src/protobuf_client.dart +++ b/dart/lib/src/protobuf_client.dart @@ -7,6 +7,7 @@ import 'dart:collection'; import 'package:protobuf/protobuf.dart'; import 'base_client.dart'; +import 'inbound_gateway_io.dart'; import 'packets/message_common.pb.dart'; import 'transport.dart'; @@ -52,6 +53,12 @@ abstract class ProtobufClient extends BaseClient { (HeartBeat).toString(): HeartBeat.fromBuffer, }); super.initialize(parserMap, transport: _transport); + // 제품 수신 경로를 단일 gateway-backed coordinator로 통일한다. start()는 + // await하지 않아도 _ready completer를 동기 구간에서 설정하므로, spawn 완료 + // 전에 도착한 첫 프레임도 submit에서 readiness를 기다려 drop되지 않는다. + final gateway = IsolateInboundGateway(); + attachInboundGateway(gateway); + unawaited(gateway.start()); _subscription = _socket.listen(onData, onError: onError); _subscription!.asFuture().then((_) { close(); @@ -108,9 +115,10 @@ abstract class ProtobufClient extends BaseClient { _frameBuffer.takeBytes(_headerSize); final packetBytes = _frameBuffer.takeBytes(_length!); _length = null; - final common = PacketBase.fromBuffer(packetBytes); - await onReceivedData(common.typeName, common.data, - incomingNonce: common.nonce, responseNonce: common.responseNonce); + // raw frame을 coordinator로 넘긴다. gateway가 envelope decode를 off-loop로 + // 처리하고 seq 순서로 onReceivedData에 dispatch한다. submit await는 gateway + // backpressure를 socket read loop로 전파한다. + await onReceivedFrame(packetBytes); sendHeartBeat(); } } diff --git a/dart/lib/src/ws_protobuf_client_io.dart b/dart/lib/src/ws_protobuf_client_io.dart index 22db749..4e4d2af 100644 --- a/dart/lib/src/ws_protobuf_client_io.dart +++ b/dart/lib/src/ws_protobuf_client_io.dart @@ -6,6 +6,7 @@ import 'dart:typed_data'; import 'package:protobuf/protobuf.dart'; import 'base_client.dart'; +import 'inbound_gateway_io.dart'; import 'packets/message_common.pb.dart'; import 'transport.dart'; @@ -44,6 +45,11 @@ abstract class WsProtobufClient extends BaseClient { isAlive = true; parserMap.addAll({(HeartBeat).toString(): HeartBeat.fromBuffer}); super.initialize(parserMap, transport: _transport); + // TCP 경로와 동일한 gateway-backed receive coordinator를 사용한다. start()는 + // _ready completer를 동기 구간에서 설정하므로 spawn 전 첫 프레임도 drop되지 않는다. + final gateway = IsolateInboundGateway(); + attachInboundGateway(gateway); + unawaited(gateway.start()); _subscription = _ws.listen(_onMessage, onError: onError, onDone: close); addListener(onHeartBeat); sendHeartBeat(); @@ -65,9 +71,7 @@ abstract class WsProtobufClient extends BaseClient { Future _dispatch(dynamic data) async { final bytes = data is List ? data : (data as Uint8List).toList(); - final common = PacketBase.fromBuffer(bytes); - await onReceivedData(common.typeName, common.data, - incomingNonce: common.nonce, responseNonce: common.responseNonce); + await onReceivedFrame(bytes); sendHeartBeat(); } diff --git a/dart/lib/src/ws_protobuf_client_web.dart b/dart/lib/src/ws_protobuf_client_web.dart index 3826a59..15b1f65 100644 --- a/dart/lib/src/ws_protobuf_client_web.dart +++ b/dart/lib/src/ws_protobuf_client_web.dart @@ -6,6 +6,7 @@ import 'dart:typed_data'; import 'package:protobuf/protobuf.dart'; import 'base_client.dart'; +import 'inbound_gateway_web.dart'; import 'packets/message_common.pb.dart'; import 'transport.dart'; @@ -67,6 +68,12 @@ abstract class WsProtobufClient extends BaseClient { isAlive = true; parserMap.addAll({(HeartBeat).toString(): HeartBeat.fromBuffer}); super.initialize(parserMap, transport: _transport); + // web에는 worker isolate가 없으므로 IsolateInboundGateway는 SyncInboundGateway + // fallback으로 in-process decode하되 동일한 seq-ordered receive coordinator + // contract를 유지한다. + final gateway = IsolateInboundGateway(); + attachInboundGateway(gateway); + unawaited(gateway.start()); _subscription = _ws.onMessage.listen(_onMessage, onError: onError); _ws.onClose.listen((_) => close()); addListener(onHeartBeat); @@ -104,9 +111,7 @@ abstract class WsProtobufClient extends BaseClient { } Future _dispatch(List bytes) async { - final common = PacketBase.fromBuffer(bytes); - await onReceivedData(common.typeName, common.data, - incomingNonce: common.nonce, responseNonce: common.responseNonce); + await onReceivedFrame(bytes); sendHeartBeat(); }