File tree Expand file tree Collapse file tree 3 files changed +21
-2
lines changed
xds/internal/xdsclient/transport Expand file tree Collapse file tree 3 files changed +21
-2
lines changed Original file line number Diff line number Diff line change @@ -62,6 +62,11 @@ func (t *Transport) lrsStartStream() {
62
62
63
63
ctx , cancel := context .WithCancel (context .Background ())
64
64
t .lrsCancelStream = cancel
65
+
66
+ // Create a new done channel everytime a new stream is created. This ensures
67
+ // that we don't close the same channel multiple times (from lrsRunner()
68
+ // goroutine) when multiple streams are created and closed.
69
+ t .lrsRunnerDoneCh = make (chan struct {})
65
70
go t .lrsRunner (ctx )
66
71
}
67
72
@@ -78,6 +83,9 @@ func (t *Transport) lrsStopStream() {
78
83
79
84
t .lrsCancelStream ()
80
85
t .logger .Infof ("Stopping LRS stream" )
86
+
87
+ // Wait for the runner goroutine to exit. The done channel will be
88
+ // recreated when a new stream is created.
81
89
<- t .lrsRunnerDoneCh
82
90
}
83
91
Original file line number Diff line number Diff line change @@ -54,7 +54,7 @@ func (s) TestReportLoad(t *testing.T) {
54
54
NodeProto : nodeProto ,
55
55
}
56
56
57
- // Create a transport to the fake server.
57
+ // Create a transport to the fake management server.
58
58
tr , err := transport .New (transport.Options {
59
59
ServerCfg : serverCfg ,
60
60
UpdateHandler : func (transport.ResourceUpdate ) error { return nil }, // No ADS validation.
@@ -190,4 +190,16 @@ func (s) TestReportLoad(t *testing.T) {
190
190
if _ , err := mgmtServer .LRSStreamCloseChan .Receive (ctx ); err != nil {
191
191
t .Fatal ("Timeout waiting for LRS stream to close" )
192
192
}
193
+
194
+ // Calling the load reporting API again should result in the creation of a
195
+ // new LRS stream. This ensures that creating and closing multiple streams
196
+ // works smoothly.
197
+ _ , cancelLRS3 := tr .ReportLoad ()
198
+ if err != nil {
199
+ t .Fatalf ("Failed to start LRS load reporting: %v" , err )
200
+ }
201
+ if _ , err := mgmtServer .LRSStreamOpenChan .Receive (ctx ); err != nil {
202
+ t .Fatalf ("Timeout when waiting for LRS stream to be created: %v" , err )
203
+ }
204
+ cancelLRS3 ()
193
205
}
Original file line number Diff line number Diff line change @@ -202,7 +202,6 @@ func New(opts Options) (*Transport, error) {
202
202
versions : make (map [string ]string ),
203
203
nonces : make (map [string ]string ),
204
204
adsRunnerDoneCh : make (chan struct {}),
205
- lrsRunnerDoneCh : make (chan struct {}),
206
205
}
207
206
208
207
// This context is used for sending and receiving RPC requests and
You can’t perform that action at this time.
0 commit comments