Skip to content

Reduce websocket subscription dispatch and receive-loop overhead - #678

Open
maedula wants to merge 2 commits into
graphql-dotnet:masterfrom
maedula:perf/websocket-dispatch-async-receive
Open

Reduce websocket subscription dispatch and receive-loop overhead#678
maedula wants to merge 2 commits into
graphql-dotnet:masterfrom
maedula:perf/websocket-dispatch-async-receive

Conversation

@maedula

@maedula maedula commented Jul 16, 2026

Copy link
Copy Markdown

Summary

  • replace the per-operation Where subscriptions on the shared incoming stream with one ID-keyed dispatcher
  • replace the per-message Task -> Observable -> Repeat receive chain with an owned async receive pump
  • isolate operation callback failures so one malformed response does not terminate unrelated subscriptions
  • guard duplicate IDs, synchronous terminal delivery, stale source generations, reconnects, and receive-pump disposal
  • keep each receive pump bound to its own websocket and buffer and await the previous pump before reconnecting

This changes incoming message routing from broadcasting every message to every active operation to a single dictionary lookup, and removes the Rx task-bridging objects allocated for each received message.

Validation

  • dotnet test tests/GraphQL.Integration.Tests/GraphQL.Integration.Tests.csproj -c Release
  • 117 passed, 2 skipped
  • focused dispatcher regression tests: 7 passed
  • dotnet build src/GraphQL.Client/GraphQL.Client.csproj -c Release -f netstandard2.0

Roman Elias added 2 commits July 16, 2026 15:55
…comingMessageDispatcher

Every active GraphQL subscription independently subscribed to the shared
incoming websocket message stream and filtered it with its own
Where(response.Id == myId) predicate, so the per-message cost scaled with
the number of concurrent subscriptions instead of being constant.
IncomingMessageDispatcher keeps a single subscription to the shared stream
and routes each message directly to its owner via a dictionary lookup.
…sync pump

Observable.Defer(() => GetReceiveTask().ToObservable()).Repeat().Catch(...).Publish()
allocated a SlowTaskObservable (bridging via Task.ContinueWith) and re-entered
Repeat's TailRecursiveSink trampoline for every single incoming websocket message.
Profiling showed this Rx plumbing, not application logic, was the largest CPU and
allocation cost in the receive path. A plain async while loop pushing into a Subject
gives the same hot/multicast/run-until-cancelled semantics without the per-message
Rx bridging overhead.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant