Skip to content

Commit 92111dc

Browse files
easwarseaswars
authored
xds: keep ads flow control local to xdsclient/transport package (#7578)
1 parent 535bdce commit 92111dc

30 files changed

+306
-337
lines changed

Diff for: ‎xds/csds/csds_e2e_test.go

+33-33
Original file line numberDiff line numberDiff line change
@@ -70,50 +70,50 @@ func Test(t *testing.T) {
7070

7171
type nopListenerWatcher struct{}
7272

73-
func (nopListenerWatcher) OnUpdate(_ *xdsresource.ListenerResourceData, onDone xdsresource.DoneNotifier) {
74-
onDone.OnDone()
73+
func (nopListenerWatcher) OnUpdate(_ *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
74+
onDone()
7575
}
76-
func (nopListenerWatcher) OnError(_ error, onDone xdsresource.DoneNotifier) {
77-
onDone.OnDone()
76+
func (nopListenerWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
77+
onDone()
7878
}
79-
func (nopListenerWatcher) OnResourceDoesNotExist(onDone xdsresource.DoneNotifier) {
80-
onDone.OnDone()
79+
func (nopListenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
80+
onDone()
8181
}
8282

8383
type nopRouteConfigWatcher struct{}
8484

85-
func (nopRouteConfigWatcher) OnUpdate(_ *xdsresource.RouteConfigResourceData, onDone xdsresource.DoneNotifier) {
86-
onDone.OnDone()
85+
func (nopRouteConfigWatcher) OnUpdate(_ *xdsresource.RouteConfigResourceData, onDone xdsresource.OnDoneFunc) {
86+
onDone()
8787
}
88-
func (nopRouteConfigWatcher) OnError(_ error, onDone xdsresource.DoneNotifier) {
89-
onDone.OnDone()
88+
func (nopRouteConfigWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
89+
onDone()
9090
}
91-
func (nopRouteConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.DoneNotifier) {
92-
onDone.OnDone()
91+
func (nopRouteConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
92+
onDone()
9393
}
9494

9595
type nopClusterWatcher struct{}
9696

97-
func (nopClusterWatcher) OnUpdate(_ *xdsresource.ClusterResourceData, onDone xdsresource.DoneNotifier) {
98-
onDone.OnDone()
97+
func (nopClusterWatcher) OnUpdate(_ *xdsresource.ClusterResourceData, onDone xdsresource.OnDoneFunc) {
98+
onDone()
9999
}
100-
func (nopClusterWatcher) OnError(_ error, onDone xdsresource.DoneNotifier) {
101-
onDone.OnDone()
100+
func (nopClusterWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
101+
onDone()
102102
}
103-
func (nopClusterWatcher) OnResourceDoesNotExist(onDone xdsresource.DoneNotifier) {
104-
onDone.OnDone()
103+
func (nopClusterWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
104+
onDone()
105105
}
106106

107107
type nopEndpointsWatcher struct{}
108108

109-
func (nopEndpointsWatcher) OnUpdate(_ *xdsresource.EndpointsResourceData, onDone xdsresource.DoneNotifier) {
110-
onDone.OnDone()
109+
func (nopEndpointsWatcher) OnUpdate(_ *xdsresource.EndpointsResourceData, onDone xdsresource.OnDoneFunc) {
110+
onDone()
111111
}
112-
func (nopEndpointsWatcher) OnError(_ error, onDone xdsresource.DoneNotifier) {
113-
onDone.OnDone()
112+
func (nopEndpointsWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
113+
onDone()
114114
}
115-
func (nopEndpointsWatcher) OnResourceDoesNotExist(onDone xdsresource.DoneNotifier) {
116-
onDone.OnDone()
115+
func (nopEndpointsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
116+
onDone()
117117
}
118118

119119
// This watcher writes the onDone callback on to a channel for the test to
@@ -126,31 +126,31 @@ func (nopEndpointsWatcher) OnResourceDoesNotExist(onDone xdsresource.DoneNotifie
126126
// for ADS stream level flow control), and was causing CSDS to not receive any
127127
// updates from the xDS client.
128128
type blockingListenerWatcher struct {
129-
testCtxDone <-chan struct{} // Closed when the test is done.
130-
onDoneCh chan xdsresource.DoneNotifier // Channel to write the onDone callback to.
129+
testCtxDone <-chan struct{} // Closed when the test is done.
130+
onDoneCh chan xdsresource.OnDoneFunc // Channel to write the onDone callback to.
131131
}
132132

133133
func newBlockingListenerWatcher(testCtxDone <-chan struct{}) *blockingListenerWatcher {
134134
return &blockingListenerWatcher{
135135
testCtxDone: testCtxDone,
136-
onDoneCh: make(chan xdsresource.DoneNotifier, 1),
136+
onDoneCh: make(chan xdsresource.OnDoneFunc, 1),
137137
}
138138
}
139139

140-
func (w *blockingListenerWatcher) OnUpdate(_ *xdsresource.ListenerResourceData, onDone xdsresource.DoneNotifier) {
140+
func (w *blockingListenerWatcher) OnUpdate(_ *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
141141
writeOnDone(w.testCtxDone, w.onDoneCh, onDone)
142142
}
143-
func (w *blockingListenerWatcher) OnError(_ error, onDone xdsresource.DoneNotifier) {
143+
func (w *blockingListenerWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
144144
writeOnDone(w.testCtxDone, w.onDoneCh, onDone)
145145
}
146-
func (w *blockingListenerWatcher) OnResourceDoesNotExist(onDone xdsresource.DoneNotifier) {
146+
func (w *blockingListenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
147147
writeOnDone(w.testCtxDone, w.onDoneCh, onDone)
148148
}
149149

150150
// writeOnDone attempts to writes the onDone callback on the onDone channel. It
151151
// returns when it can successfully write to the channel or when the test is
152152
// done, which is signalled by testCtxDone being closed.
153-
func writeOnDone(testCtxDone <-chan struct{}, onDoneCh chan xdsresource.DoneNotifier, onDone xdsresource.DoneNotifier) {
153+
func writeOnDone(testCtxDone <-chan struct{}, onDoneCh chan xdsresource.OnDoneFunc, onDone xdsresource.OnDoneFunc) {
154154
select {
155155
case <-testCtxDone:
156156
case onDoneCh <- onDone:
@@ -545,7 +545,7 @@ func (s) TestCSDS_NACK(t *testing.T) {
545545
case <-ctx.Done():
546546
t.Fatal("Timed out waiting for watch callback")
547547
case onDone := <-watcher2.onDoneCh:
548-
onDone.OnDone()
548+
onDone()
549549
}
550550

551551
// Update the second resource with an empty ApiListener field which is
@@ -564,7 +564,7 @@ func (s) TestCSDS_NACK(t *testing.T) {
564564
case <-ctx.Done():
565565
t.Fatal("Timed out waiting for watch callback")
566566
case onDone := <-watcher2.onDoneCh:
567-
onDone.OnDone()
567+
onDone()
568568
}
569569

570570
// Verify that the xDS client reports the first listener resource as being

Diff for: ‎xds/internal/balancer/cdsbalancer/cluster_watcher.go

+9-9
Original file line numberDiff line numberDiff line change
@@ -32,19 +32,19 @@ type clusterWatcher struct {
3232
parent *cdsBalancer
3333
}
3434

35-
func (cw *clusterWatcher) OnUpdate(u *xdsresource.ClusterResourceData, onDone xdsresource.DoneNotifier) {
36-
handleUpdate := func(context.Context) { cw.parent.onClusterUpdate(cw.name, u.Resource); onDone.OnDone() }
37-
cw.parent.serializer.ScheduleOr(handleUpdate, onDone.OnDone)
35+
func (cw *clusterWatcher) OnUpdate(u *xdsresource.ClusterResourceData, onDone xdsresource.OnDoneFunc) {
36+
handleUpdate := func(context.Context) { cw.parent.onClusterUpdate(cw.name, u.Resource); onDone() }
37+
cw.parent.serializer.ScheduleOr(handleUpdate, onDone)
3838
}
3939

40-
func (cw *clusterWatcher) OnError(err error, onDone xdsresource.DoneNotifier) {
41-
handleError := func(context.Context) { cw.parent.onClusterError(cw.name, err); onDone.OnDone() }
42-
cw.parent.serializer.ScheduleOr(handleError, onDone.OnDone)
40+
func (cw *clusterWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
41+
handleError := func(context.Context) { cw.parent.onClusterError(cw.name, err); onDone() }
42+
cw.parent.serializer.ScheduleOr(handleError, onDone)
4343
}
4444

45-
func (cw *clusterWatcher) OnResourceDoesNotExist(onDone xdsresource.DoneNotifier) {
46-
handleNotFound := func(context.Context) { cw.parent.onClusterResourceNotFound(cw.name); onDone.OnDone() }
47-
cw.parent.serializer.ScheduleOr(handleNotFound, onDone.OnDone)
45+
func (cw *clusterWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
46+
handleNotFound := func(context.Context) { cw.parent.onClusterResourceNotFound(cw.name); onDone() }
47+
cw.parent.serializer.ScheduleOr(handleNotFound, onDone)
4848
}
4949

5050
// watcherState groups the state associated with a clusterWatcher.

Diff for: ‎xds/internal/balancer/clusterresolver/clusterresolver.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -216,7 +216,7 @@ func (b *clusterResolverBalancer) handleResourceUpdate(update *resourceUpdate) {
216216
b.updateChildConfig()
217217

218218
if update.onDone != nil {
219-
update.onDone.OnDone()
219+
update.onDone()
220220
}
221221
}
222222

Diff for: ‎xds/internal/balancer/clusterresolver/resource_resolver.go

+9-9
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ type resourceUpdate struct {
3737
priorities []priorityConfig
3838
// To be invoked once the update is completely processed, or is dropped in
3939
// favor of a newer update.
40-
onDone xdsresource.DoneNotifier
40+
onDone xdsresource.OnDoneFunc
4141
}
4242

4343
// topLevelResolver is used by concrete endpointsResolver implementations for
@@ -49,7 +49,7 @@ type topLevelResolver interface {
4949
// endpointsResolver implementation. The onDone callback is to be invoked
5050
// once the update is completely processed, or is dropped in favor of a
5151
// newer update.
52-
onUpdate(onDone xdsresource.DoneNotifier)
52+
onUpdate(onDone xdsresource.OnDoneFunc)
5353
}
5454

5555
// endpointsResolver wraps the functionality to resolve a given resource name to
@@ -215,7 +215,7 @@ func (rr *resourceResolver) updateMechanisms(mechanisms []DiscoveryMechanism) {
215215
}
216216
// Regenerate even if there's no change in discovery mechanism, in case
217217
// priority order changed.
218-
rr.generateLocked(xdsresource.NopDoneNotifier{})
218+
rr.generateLocked(func() {})
219219
}
220220

221221
// resolveNow is typically called to trigger re-resolve of DNS. The EDS
@@ -264,7 +264,7 @@ func (rr *resourceResolver) stop(closing bool) {
264264
select {
265265
case ru := <-rr.updateChannel:
266266
if ru.onDone != nil {
267-
ru.onDone.OnDone()
267+
ru.onDone()
268268
}
269269
default:
270270
}
@@ -281,14 +281,14 @@ func (rr *resourceResolver) stop(closing bool) {
281281
// clusterresolver LB policy.
282282
//
283283
// Caller must hold rr.mu.
284-
func (rr *resourceResolver) generateLocked(onDone xdsresource.DoneNotifier) {
284+
func (rr *resourceResolver) generateLocked(onDone xdsresource.OnDoneFunc) {
285285
var ret []priorityConfig
286286
for _, rDM := range rr.children {
287287
u, ok := rDM.r.lastUpdate()
288288
if !ok {
289289
// Don't send updates to parent until all resolvers have update to
290290
// send.
291-
onDone.OnDone()
291+
onDone()
292292
return
293293
}
294294
switch uu := u.(type) {
@@ -304,18 +304,18 @@ func (rr *resourceResolver) generateLocked(onDone xdsresource.DoneNotifier) {
304304
// receive path.
305305
case ru := <-rr.updateChannel:
306306
if ru.onDone != nil {
307-
ru.onDone.OnDone()
307+
ru.onDone()
308308
}
309309
default:
310310
}
311311
rr.updateChannel <- &resourceUpdate{priorities: ret, onDone: onDone}
312312
}
313313

314-
func (rr *resourceResolver) onUpdate(onDone xdsresource.DoneNotifier) {
314+
func (rr *resourceResolver) onUpdate(onDone xdsresource.OnDoneFunc) {
315315
handleUpdate := func(context.Context) {
316316
rr.mu.Lock()
317317
rr.generateLocked(onDone)
318318
rr.mu.Unlock()
319319
}
320-
rr.serializer.ScheduleOr(handleUpdate, func() { onDone.OnDone() })
320+
rr.serializer.ScheduleOr(handleUpdate, func() { onDone() })
321321
}

Diff for: ‎xds/internal/balancer/clusterresolver/resource_resolver_dns.go

+4-5
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@ import (
2727
"google.golang.org/grpc/internal/pretty"
2828
"google.golang.org/grpc/resolver"
2929
"google.golang.org/grpc/serviceconfig"
30-
"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
3130
)
3231

3332
var (
@@ -80,7 +79,7 @@ func newDNSResolver(target string, topLevelResolver topLevelResolver, logger *gr
8079
ret.logger.Infof("Failed to parse dns hostname %q in clusterresolver LB policy", target)
8180
}
8281
ret.updateReceived = true
83-
ret.topLevelResolver.onUpdate(xdsresource.NopDoneNotifier{})
82+
ret.topLevelResolver.onUpdate(func() {})
8483
return ret
8584
}
8685

@@ -90,7 +89,7 @@ func newDNSResolver(target string, topLevelResolver topLevelResolver, logger *gr
9089
ret.logger.Infof("Failed to build DNS resolver for target %q: %v", target, err)
9190
}
9291
ret.updateReceived = true
93-
ret.topLevelResolver.onUpdate(xdsresource.NopDoneNotifier{})
92+
ret.topLevelResolver.onUpdate(func() {})
9493
return ret
9594
}
9695
ret.dnsR = r
@@ -154,7 +153,7 @@ func (dr *dnsDiscoveryMechanism) UpdateState(state resolver.State) error {
154153
dr.updateReceived = true
155154
dr.mu.Unlock()
156155

157-
dr.topLevelResolver.onUpdate(xdsresource.NopDoneNotifier{})
156+
dr.topLevelResolver.onUpdate(func() {})
158157
return nil
159158
}
160159

@@ -177,7 +176,7 @@ func (dr *dnsDiscoveryMechanism) ReportError(err error) {
177176
dr.updateReceived = true
178177
dr.mu.Unlock()
179178

180-
dr.topLevelResolver.onUpdate(xdsresource.NopDoneNotifier{})
179+
dr.topLevelResolver.onUpdate(func() {})
181180
}
182181

183182
func (dr *dnsDiscoveryMechanism) NewAddress(addresses []resolver.Address) {

Diff for: ‎xds/internal/balancer/clusterresolver/resource_resolver_eds.go

+7-7
Original file line numberDiff line numberDiff line change
@@ -76,9 +76,9 @@ func newEDSResolver(nameToWatch string, producer xdsresource.Producer, topLevelR
7676
}
7777

7878
// OnUpdate is invoked to report an update for the resource being watched.
79-
func (er *edsDiscoveryMechanism) OnUpdate(update *xdsresource.EndpointsResourceData, onDone xdsresource.DoneNotifier) {
79+
func (er *edsDiscoveryMechanism) OnUpdate(update *xdsresource.EndpointsResourceData, onDone xdsresource.OnDoneFunc) {
8080
if er.stopped.HasFired() {
81-
onDone.OnDone()
81+
onDone()
8282
return
8383
}
8484

@@ -89,9 +89,9 @@ func (er *edsDiscoveryMechanism) OnUpdate(update *xdsresource.EndpointsResourceD
8989
er.topLevelResolver.onUpdate(onDone)
9090
}
9191

92-
func (er *edsDiscoveryMechanism) OnError(err error, onDone xdsresource.DoneNotifier) {
92+
func (er *edsDiscoveryMechanism) OnError(err error, onDone xdsresource.OnDoneFunc) {
9393
if er.stopped.HasFired() {
94-
onDone.OnDone()
94+
onDone()
9595
return
9696
}
9797

@@ -104,7 +104,7 @@ func (er *edsDiscoveryMechanism) OnError(err error, onDone xdsresource.DoneNotif
104104
// Continue using a previously received good configuration if one
105105
// exists.
106106
er.mu.Unlock()
107-
onDone.OnDone()
107+
onDone()
108108
return
109109
}
110110

@@ -120,9 +120,9 @@ func (er *edsDiscoveryMechanism) OnError(err error, onDone xdsresource.DoneNotif
120120
er.topLevelResolver.onUpdate(onDone)
121121
}
122122

123-
func (er *edsDiscoveryMechanism) OnResourceDoesNotExist(onDone xdsresource.DoneNotifier) {
123+
func (er *edsDiscoveryMechanism) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
124124
if er.stopped.HasFired() {
125-
onDone.OnDone()
125+
onDone()
126126
return
127127
}
128128

Diff for: ‎xds/internal/resolver/watch_service.go

+18-18
Original file line numberDiff line numberDiff line change
@@ -36,19 +36,19 @@ func newListenerWatcher(resourceName string, parent *xdsResolver) *listenerWatch
3636
return lw
3737
}
3838

39-
func (l *listenerWatcher) OnUpdate(update *xdsresource.ListenerResourceData, onDone xdsresource.DoneNotifier) {
40-
handleUpdate := func(context.Context) { l.parent.onListenerResourceUpdate(update.Resource); onDone.OnDone() }
41-
l.parent.serializer.ScheduleOr(handleUpdate, onDone.OnDone)
39+
func (l *listenerWatcher) OnUpdate(update *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
40+
handleUpdate := func(context.Context) { l.parent.onListenerResourceUpdate(update.Resource); onDone() }
41+
l.parent.serializer.ScheduleOr(handleUpdate, onDone)
4242
}
4343

44-
func (l *listenerWatcher) OnError(err error, onDone xdsresource.DoneNotifier) {
45-
handleError := func(context.Context) { l.parent.onListenerResourceError(err); onDone.OnDone() }
46-
l.parent.serializer.ScheduleOr(handleError, onDone.OnDone)
44+
func (l *listenerWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
45+
handleError := func(context.Context) { l.parent.onListenerResourceError(err); onDone() }
46+
l.parent.serializer.ScheduleOr(handleError, onDone)
4747
}
4848

49-
func (l *listenerWatcher) OnResourceDoesNotExist(onDone xdsresource.DoneNotifier) {
50-
handleNotFound := func(context.Context) { l.parent.onListenerResourceNotFound(); onDone.OnDone() }
51-
l.parent.serializer.ScheduleOr(handleNotFound, onDone.OnDone)
49+
func (l *listenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
50+
handleNotFound := func(context.Context) { l.parent.onListenerResourceNotFound(); onDone() }
51+
l.parent.serializer.ScheduleOr(handleNotFound, onDone)
5252
}
5353

5454
func (l *listenerWatcher) stop() {
@@ -68,22 +68,22 @@ func newRouteConfigWatcher(resourceName string, parent *xdsResolver) *routeConfi
6868
return rw
6969
}
7070

71-
func (r *routeConfigWatcher) OnUpdate(u *xdsresource.RouteConfigResourceData, onDone xdsresource.DoneNotifier) {
71+
func (r *routeConfigWatcher) OnUpdate(u *xdsresource.RouteConfigResourceData, onDone xdsresource.OnDoneFunc) {
7272
handleUpdate := func(context.Context) {
7373
r.parent.onRouteConfigResourceUpdate(r.resourceName, u.Resource)
74-
onDone.OnDone()
74+
onDone()
7575
}
76-
r.parent.serializer.ScheduleOr(handleUpdate, onDone.OnDone)
76+
r.parent.serializer.ScheduleOr(handleUpdate, onDone)
7777
}
7878

79-
func (r *routeConfigWatcher) OnError(err error, onDone xdsresource.DoneNotifier) {
80-
handleError := func(context.Context) { r.parent.onRouteConfigResourceError(r.resourceName, err); onDone.OnDone() }
81-
r.parent.serializer.ScheduleOr(handleError, onDone.OnDone)
79+
func (r *routeConfigWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
80+
handleError := func(context.Context) { r.parent.onRouteConfigResourceError(r.resourceName, err); onDone() }
81+
r.parent.serializer.ScheduleOr(handleError, onDone)
8282
}
8383

84-
func (r *routeConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.DoneNotifier) {
85-
handleNotFound := func(context.Context) { r.parent.onRouteConfigResourceNotFound(r.resourceName); onDone.OnDone() }
86-
r.parent.serializer.ScheduleOr(handleNotFound, onDone.OnDone)
84+
func (r *routeConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
85+
handleNotFound := func(context.Context) { r.parent.onRouteConfigResourceNotFound(r.resourceName); onDone() }
86+
r.parent.serializer.ScheduleOr(handleNotFound, onDone)
8787
}
8888

8989
func (r *routeConfigWatcher) stop() {

0 commit comments

Comments
 (0)