@@ -20,15 +20,17 @@ package resolver
2020
2121import (
2222 "context"
23+ "errors"
2324 "regexp"
2425 "testing"
2526 "time"
2627
2728 xxhash "github.com/cespare/xxhash/v2"
28- "github.com/google/go-cmp/cmp "
29+ "google.golang.org/grpc/internal/grpcsync "
2930 "google.golang.org/grpc/internal/grpctest"
3031 "google.golang.org/grpc/internal/grpcutil"
3132 iresolver "google.golang.org/grpc/internal/resolver"
33+ "google.golang.org/grpc/internal/testutils"
3234 _ "google.golang.org/grpc/internal/xds/balancer/cdsbalancer" // To parse LB config
3335 "google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
3436 "google.golang.org/grpc/metadata"
@@ -44,50 +46,113 @@ func Test(t *testing.T) {
4446 grpctest .RunSubTests (t , s {})
4547}
4648
47- func (s ) TestPruneActiveClusters (t * testing.T ) {
48- newClusterInfo := func (ref int32 , unsubscribe func ()) * clusterInfo {
49- ci := & clusterInfo {unsubscribe : unsubscribe }
50- ci .refCount .Store (ref )
51- return ci
52- }
49+ // newResolverForActiveEntryTests returns a resolver with just enough state to
50+ // exercise acquireActiveClusterInfo and the cleanup it registers. The current
51+ // config selector is an erroring one so that pushing a new service config does
52+ // not require an xDS config to be present.
53+ func newResolverForActiveEntryTests (t * testing.T ) * xdsResolver {
54+ t .Helper ()
55+
56+ ctx , cancel := context .WithCancel (context .Background ())
57+ t .Cleanup (cancel )
5358 r := & xdsResolver {
54- activeClusters : map [string ]* clusterInfo {
55- "zero" : newClusterInfo (0 , func () {}),
56- "one" : newClusterInfo (1 , func () {}),
57- "two" : newClusterInfo (2 , func () {}),
58- "anotherzero" : newClusterInfo (0 , func () {}),
59- },
60- activePlugins : map [string ]* clusterInfo {
61- "zero" : newClusterInfo (0 , nil ),
62- "one" : newClusterInfo (1 , nil ),
63- "two" : newClusterInfo (2 , nil ),
64- "anotherzero" : newClusterInfo (0 , nil ),
65- },
59+ cc : & testutils.ResolverClientConn {Logger : t },
60+ activeClusters : make (map [string ]* grpcsync.RefCounted [* clusterInfo ]),
61+ activePlugins : make (map [string ]* grpcsync.RefCounted [* clusterInfo ]),
62+ serializer : grpcsync .NewCallbackSerializer (ctx ),
63+ serializerCancel : cancel ,
64+ curConfigSelector : newErroringConfigSelector (errors .New ("test" ), "" ),
6665 }
67- wantActiveClusters := map [string ]int32 {
68- "one" : 1 ,
69- "two" : 2 ,
66+ r .logger = prefixLogger (r )
67+ return r
68+ }
69+
70+ // runOnSerializer runs f in the context of a serializer callback, which is the
71+ // only place the resolver's active cluster and plugin maps may be touched, and
72+ // blocks until it has run. Any callback queued by an earlier call to this
73+ // helper, including a removal queued when a reference count reached zero, is
74+ // guaranteed to have run by the time f is invoked.
75+ func runOnSerializer (ctx context.Context , t * testing.T , r * xdsResolver , f func ()) {
76+ t .Helper ()
77+
78+ done := make (chan struct {})
79+ r .serializer .TrySchedule (func (context.Context ) {
80+ defer close (done )
81+ f ()
82+ })
83+ select {
84+ case <- done :
85+ case <- ctx .Done ():
86+ t .Fatal ("Timeout waiting for serializer callback to run" )
7087 }
71- wantActivePlugins := map [string ]int32 {
72- "one" : 1 ,
73- "two" : 2 ,
88+ }
89+
90+ // TestActivePluginRefCounting verifies that repeated acquisitions of a cluster
91+ // specifier plugin share a single entry, and that the entry is removed from
92+ // activePlugins only once the last reference is released.
93+ func (s ) TestActivePluginRefCounting (t * testing.T ) {
94+ ctx , cancel := context .WithTimeout (context .Background (), defaultTestTimeout )
95+ defer cancel ()
96+
97+ r := newResolverForActiveEntryTests (t )
98+ const key = "cluster_specifier_plugin:test-plugin"
99+
100+ var first , second * grpcsync.RefCounted [* clusterInfo ]
101+ runOnSerializer (ctx , t , r , func () {
102+ first = r .acquireActiveClusterInfo (key , "" )
103+ second = r .acquireActiveClusterInfo (key , "" )
104+ })
105+ if first != second {
106+ t .Fatalf ("acquireActiveClusterInfo(%q) returned a new entry; want the existing one to be reused" , key )
74107 }
75- r .pruneActiveClustersAndPlugins ()
76108
77- getRefCounts := func (m map [string ]* clusterInfo ) map [string ]int32 {
78- res := make (map [string ]int32 )
79- for k , v := range m {
80- res [k ] = v .refCount .Load ()
109+ // Two references are outstanding, so releasing one must keep the entry.
110+ runOnSerializer (ctx , t , r , func () { first .Decrement () })
111+ runOnSerializer (ctx , t , r , func () {
112+ if _ , ok := r .activePlugins [key ]; ! ok {
113+ t .Errorf ("activePlugins[%q] was removed while a reference is still held" , key )
81114 }
82- return res
83- }
115+ })
84116
85- if d := cmp .Diff (getRefCounts (r .activeClusters ), wantActiveClusters ); d != "" {
86- t .Fatalf ("r.activeClusters refCounts mismatch (-got +want):\n %s" , d )
87- }
88- if d := cmp .Diff (getRefCounts (r .activePlugins ), wantActivePlugins ); d != "" {
89- t .Fatalf ("r.activePlugins refCounts mismatch (-got +want):\n %s" , d )
117+ // Releasing the last reference must remove the entry.
118+ runOnSerializer (ctx , t , r , func () { second .Decrement () })
119+ runOnSerializer (ctx , t , r , func () {
120+ if _ , ok := r .activePlugins [key ]; ok {
121+ t .Errorf ("activePlugins[%q] still present after the last reference was released" , key )
122+ }
123+ })
124+ }
125+
126+ // TestActivePluginNotRevivedAfterRelease verifies that an entry whose reference
127+ // count has already dropped to zero is replaced by a fresh entry rather than
128+ // resurrected, and that the pending removal of the dead entry does not delete
129+ // its replacement.
130+ func (s ) TestActivePluginNotRevivedAfterRelease (t * testing.T ) {
131+ ctx , cancel := context .WithTimeout (context .Background (), defaultTestTimeout )
132+ defer cancel ()
133+
134+ r := newResolverForActiveEntryTests (t )
135+ const key = "cluster_specifier_plugin:test-plugin"
136+
137+ var dead , live * grpcsync.RefCounted [* clusterInfo ]
138+ runOnSerializer (ctx , t , r , func () {
139+ // Release the only reference. The entry is now dead, but its removal is
140+ // queued behind this callback and so has not run yet. Acquiring again
141+ // must therefore hand back a fresh entry rather than revive this one.
142+ dead = r .acquireActiveClusterInfo (key , "" )
143+ dead .Decrement ()
144+ live = r .acquireActiveClusterInfo (key , "" )
145+ })
146+ if live == dead {
147+ t .Fatal ("acquireActiveClusterInfo() returned an entry whose refcount had already reached zero; want a new entry" )
90148 }
149+
150+ // The dead entry's queued removal must not evict the replacement.
151+ runOnSerializer (ctx , t , r , func () {
152+ if got := r .activePlugins [key ]; got != live {
153+ t .Errorf ("activePlugins[%q] = %p, want the newly created entry %p" , key , got , live )
154+ }
155+ })
91156}
92157
93158func (s ) TestGenerateRequestHash (t * testing.T ) {
0 commit comments