Skip to content

Commit 97535ff

Browse files
committed
wip: execute subscription updates in parallel
1 parent 5086005 commit 97535ff

3 files changed

Lines changed: 5 additions & 4 deletions

File tree

v2/go.mod

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@ require (
3232
github.com/vektah/gqlparser/v2 v2.5.30
3333
github.com/wundergraph/astjson v1.1.0
3434
github.com/wundergraph/go-arena v1.1.0
35-
go.uber.org/atomic v1.11.0
3635
go.uber.org/goleak v1.3.0
3736
golang.org/x/sync v0.17.0
3837
golang.org/x/sys v0.37.0

v2/go.sum

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -162,8 +162,6 @@ go.opentelemetry.io/otel/sdk/metric v1.36.0/go.mod h1:qTNOhFDfKRwX0yXOqJYegL5WRa
162162
go.opentelemetry.io/otel/trace v1.36.0 h1:ahxWNuqZjpdiFAyrIoQ4GIiAIhxAunQR6MUoKrsNd4w=
163163
go.opentelemetry.io/otel/trace v1.36.0/go.mod h1:gQ+OnDZzrybY4k4seLzPAWNwVBBVlF2szhehOBB/tGA=
164164
go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
165-
go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
166-
go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
167165
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
168166
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
169167
go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4=

v2/pkg/engine/resolve/resolve.go

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1011,12 +1011,16 @@ func (r *Resolver) handleTriggerUpdate(id uint64, data []byte) {
10111011
}
10121012
trig.mu.Unlock()
10131013

1014+
var wg sync.WaitGroup
10141015
for _, pw := range pending {
10151016
if pw.sub.removed.Load() {
10161017
continue
10171018
}
1018-
r.executeSubscriptionUpdate(pw.sub.ctx, pw.sub, data)
1019+
wg.Go(func() {
1020+
r.executeSubscriptionUpdate(pw.sub.ctx, pw.sub, data)
1021+
})
10191022
}
1023+
wg.Wait()
10201024
}
10211025

10221026
// handleUpdateSubscription sends data to a single subscription using snapshot-and-release.

0 commit comments

Comments
 (0)