Skip to content

Commit 38db48e

Browse files
ejona86ejona86
committed
xds: Use ObjectPool to manage scheduler in XdsServerWrapper
Also, we should be passing the executor in all tests, as we don't want other threads to be running if we can avoid it.
1 parent 6d5848c commit 38db48e

4 files changed

Lines changed: 26 additions & 157 deletions

File tree

‎xds/src/main/java/io/grpc/xds/XdsServerBuilder.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@
3232
import io.grpc.Server;
3333
import io.grpc.ServerBuilder;
3434
import io.grpc.ServerCredentials;
35+
import io.grpc.internal.GrpcUtil;
36+
import io.grpc.internal.SharedResourcePool;
3537
import io.grpc.netty.InternalNettyServerBuilder;
3638
import io.grpc.netty.InternalNettyServerCredentials;
3739
import io.grpc.netty.InternalProtocolNegotiator;
@@ -183,6 +185,7 @@ public Server build() {
183185
bootstrapOverride,
184186
ldsResourceNameResolver,
185187
filterRegistry,
188+
SharedResourcePool.forResource(GrpcUtil.TIMER_SERVICE),
186189
this.channelConfigurator);
187190
}
188191

‎xds/src/main/java/io/grpc/xds/XdsServerWrapper.java‎

Lines changed: 7 additions & 149 deletions
Original file line numberDiff line numberDiff line change
@@ -46,9 +46,8 @@
4646
import io.grpc.StatusOr;
4747
import io.grpc.SynchronizationContext;
4848
import io.grpc.SynchronizationContext.ScheduledHandle;
49-
import io.grpc.internal.GrpcUtil;
49+
import io.grpc.internal.FixedObjectPool;
5050
import io.grpc.internal.ObjectPool;
51-
import io.grpc.internal.SharedResourceHolder;
5251
import io.grpc.xds.EnvoyServerProtoData.FilterChain;
5352
import io.grpc.xds.Filter.FilterConfig;
5453
import io.grpc.xds.Filter.FilterContext;
@@ -103,7 +102,7 @@ public void uncaughtException(Thread t, Throwable e) {
103102
static final long RETRY_DELAY_NANOS = TimeUnit.MINUTES.toNanos(1);
104103
private final String listenerAddress;
105104
private final ServerBuilder<?> delegateBuilder;
106-
private boolean sharedTimeService;
105+
private final ObjectPool<ScheduledExecutorService> timeServicePool;
107106
private final ScheduledExecutorService timeService;
108107
private final FilterRegistry filterRegistry;
109108
private final ThreadSafeRandom random = ThreadSafeRandomImpl.instance;
@@ -138,98 +137,6 @@ public void uncaughtException(Thread t, Throwable e) {
138137

139138
private final ChannelConfigurator channelConfigurator;
140139

141-
XdsServerWrapper(
142-
String listenerAddress,
143-
ServerBuilder<?> delegateBuilder,
144-
XdsServingStatusListener listener,
145-
FilterChainSelectorManager filterChainSelectorManager,
146-
XdsClientPoolFactory xdsClientPoolFactory,
147-
@Nullable Map<String, ?> bootstrapOverride,
148-
@Nullable Function<String, String> ldsResourceNameResolver,
149-
FilterRegistry filterRegistry,
150-
ChannelConfigurator channelConfigurator) {
151-
this(
152-
listenerAddress,
153-
delegateBuilder,
154-
listener,
155-
filterChainSelectorManager,
156-
xdsClientPoolFactory,
157-
bootstrapOverride,
158-
ldsResourceNameResolver,
159-
filterRegistry,
160-
SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE),
161-
channelConfigurator);
162-
sharedTimeService = true;
163-
}
164-
165-
XdsServerWrapper(
166-
String listenerAddress,
167-
ServerBuilder<?> delegateBuilder,
168-
XdsServingStatusListener listener,
169-
FilterChainSelectorManager filterChainSelectorManager,
170-
XdsClientPoolFactory xdsClientPoolFactory,
171-
@Nullable Map<String, ?> bootstrapOverride,
172-
FilterRegistry filterRegistry,
173-
ChannelConfigurator channelConfigurator) {
174-
this(
175-
listenerAddress,
176-
delegateBuilder,
177-
listener,
178-
filterChainSelectorManager,
179-
xdsClientPoolFactory,
180-
bootstrapOverride,
181-
null,
182-
filterRegistry,
183-
SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE),
184-
channelConfigurator);
185-
sharedTimeService = true;
186-
}
187-
188-
XdsServerWrapper(
189-
String listenerAddress,
190-
ServerBuilder<?> delegateBuilder,
191-
XdsServingStatusListener listener,
192-
FilterChainSelectorManager filterChainSelectorManager,
193-
XdsClientPoolFactory xdsClientPoolFactory,
194-
@Nullable Map<String, ?> bootstrapOverride,
195-
FilterRegistry filterRegistry) {
196-
this(
197-
listenerAddress,
198-
delegateBuilder,
199-
listener,
200-
filterChainSelectorManager,
201-
xdsClientPoolFactory,
202-
bootstrapOverride,
203-
null,
204-
filterRegistry,
205-
SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE),
206-
builder -> { });
207-
sharedTimeService = true;
208-
}
209-
210-
XdsServerWrapper(
211-
String listenerAddress,
212-
ServerBuilder<?> delegateBuilder,
213-
XdsServingStatusListener listener,
214-
FilterChainSelectorManager filterChainSelectorManager,
215-
XdsClientPoolFactory xdsClientPoolFactory,
216-
@Nullable Map<String, ?> bootstrapOverride,
217-
@Nullable Function<String, String> ldsResourceNameResolver,
218-
FilterRegistry filterRegistry) {
219-
this(
220-
listenerAddress,
221-
delegateBuilder,
222-
listener,
223-
filterChainSelectorManager,
224-
xdsClientPoolFactory,
225-
bootstrapOverride,
226-
ldsResourceNameResolver,
227-
filterRegistry,
228-
SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE),
229-
builder -> { });
230-
sharedTimeService = true;
231-
}
232-
233140
@VisibleForTesting
234141
XdsServerWrapper(
235142
String listenerAddress,
@@ -249,58 +156,10 @@ public void uncaughtException(Thread t, Throwable e) {
249156
bootstrapOverride,
250157
null,
251158
filterRegistry,
252-
timeService,
253-
builder -> { });
254-
}
255-
256-
@VisibleForTesting
257-
XdsServerWrapper(
258-
String listenerAddress,
259-
ServerBuilder<?> delegateBuilder,
260-
XdsServingStatusListener listener,
261-
FilterChainSelectorManager filterChainSelectorManager,
262-
XdsClientPoolFactory xdsClientPoolFactory,
263-
@Nullable Map<String, ?> bootstrapOverride,
264-
@Nullable Function<String, String> ldsResourceNameResolver,
265-
FilterRegistry filterRegistry,
266-
ScheduledExecutorService timeService) {
267-
this(
268-
listenerAddress,
269-
delegateBuilder,
270-
listener,
271-
filterChainSelectorManager,
272-
xdsClientPoolFactory,
273-
bootstrapOverride,
274-
ldsResourceNameResolver,
275-
filterRegistry,
276-
timeService,
159+
new FixedObjectPool<>(timeService),
277160
builder -> { });
278161
}
279162

280-
@VisibleForTesting
281-
XdsServerWrapper(
282-
String listenerAddress,
283-
ServerBuilder<?> delegateBuilder,
284-
XdsServingStatusListener listener,
285-
FilterChainSelectorManager filterChainSelectorManager,
286-
XdsClientPoolFactory xdsClientPoolFactory,
287-
@Nullable Map<String, ?> bootstrapOverride,
288-
FilterRegistry filterRegistry,
289-
ScheduledExecutorService timeService,
290-
ChannelConfigurator channelConfigurator) {
291-
this(
292-
listenerAddress,
293-
delegateBuilder,
294-
listener,
295-
filterChainSelectorManager,
296-
xdsClientPoolFactory,
297-
bootstrapOverride,
298-
null,
299-
filterRegistry,
300-
timeService,
301-
channelConfigurator);
302-
}
303-
304163
@VisibleForTesting
305164
XdsServerWrapper(
306165
String listenerAddress,
@@ -311,7 +170,7 @@ public void uncaughtException(Thread t, Throwable e) {
311170
@Nullable Map<String, ?> bootstrapOverride,
312171
@Nullable Function<String, String> ldsResourceNameResolver,
313172
FilterRegistry filterRegistry,
314-
ScheduledExecutorService timeService,
173+
ObjectPool<ScheduledExecutorService> timeServicePool,
315174
ChannelConfigurator channelConfigurator) {
316175
this.listenerAddress = checkNotNull(listenerAddress, "listenerAddress");
317176
this.delegateBuilder = checkNotNull(delegateBuilder, "delegateBuilder");
@@ -322,7 +181,8 @@ public void uncaughtException(Thread t, Throwable e) {
322181
this.xdsClientPoolFactory = checkNotNull(xdsClientPoolFactory, "xdsClientPoolFactory");
323182
this.bootstrapOverride = bootstrapOverride;
324183
this.ldsResourceNameResolver = ldsResourceNameResolver;
325-
this.timeService = checkNotNull(timeService, "timeService");
184+
this.timeServicePool = checkNotNull(timeServicePool, "timeServicePool");
185+
this.timeService = checkNotNull(timeServicePool.getObject(), "timeService");
326186
this.filterRegistry = checkNotNull(filterRegistry,"filterRegistry");
327187
this.delegate = delegateBuilder.build();
328188
this.channelConfigurator = checkNotNull(channelConfigurator, "channelConfigurator");
@@ -439,9 +299,7 @@ private void internalShutdown() {
439299
if (restartTimer != null) {
440300
restartTimer.cancel();
441301
}
442-
if (sharedTimeService) {
443-
SharedResourceHolder.release(GrpcUtil.TIMER_SERVICE, timeService);
444-
}
302+
timeServicePool.returnObject(timeService);
445303
isServing = false;
446304
internalTerminationLatch.countDown();
447305
}

‎xds/src/test/java/io/grpc/xds/XdsClientWrapperForServerSdsTestMisc.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@
3838
import io.grpc.Status;
3939
import io.grpc.StatusOr;
4040
import io.grpc.inprocess.InProcessSocketAddress;
41+
import io.grpc.internal.FakeClock;
4142
import io.grpc.internal.TestUtils.NoopChannelLogger;
4243
import io.grpc.netty.GrpcHttp2ConnectionHandler;
4344
import io.grpc.netty.InternalProtocolNegotiationEvent;
@@ -122,7 +123,8 @@ public void setUp() {
122123
when(mockServer.isShutdown()).thenReturn(false);
123124
xdsServerWrapper = new XdsServerWrapper("0.0.0.0:" + PORT, mockBuilder, listener,
124125
selectorManager, new FakeXdsClientPoolFactory(xdsClient),
125-
XdsServerTestHelper.RAW_BOOTSTRAP, FilterRegistry.newRegistry());
126+
XdsServerTestHelper.RAW_BOOTSTRAP, FilterRegistry.newRegistry(),
127+
new FakeClock().getScheduledExecutorService());
126128
}
127129

128130
@Test

‎xds/src/test/java/io/grpc/xds/XdsServerWrapperTest.java‎

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@
5353
import io.grpc.StatusOr;
5454
import io.grpc.SynchronizationContext;
5555
import io.grpc.internal.FakeClock;
56+
import io.grpc.internal.FixedObjectPool;
5657
import io.grpc.internal.ObjectPool;
5758
import io.grpc.testing.TestMethodDescriptors;
5859
import io.grpc.xds.EnvoyServerProtoData.CidrRange;
@@ -163,7 +164,7 @@ public void testBootstrap() throws Exception {
163164
when(xdsClient.getBootstrapInfo()).thenReturn(b);
164165
xdsServerWrapper = new XdsServerWrapper("[::FFFF:129.144.52.38]:80", mockBuilder, listener,
165166
selectorManager, new FakeXdsClientPoolFactory(xdsClient),
166-
XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry);
167+
XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry, executor.getScheduledExecutorService());
167168
Executors.newSingleThreadExecutor().execute(new Runnable() {
168169
@Override
169170
public void run() {
@@ -204,7 +205,9 @@ public void testBootstrap_ldsResourceNameResolver() throws Exception {
204205
new FakeXdsClientPoolFactory(xdsClient),
205206
XdsServerTestHelper.RAW_BOOTSTRAP,
206207
addr -> "xdstp://resolved_name/" + addr,
207-
filterRegistry);
208+
filterRegistry,
209+
new FixedObjectPool<>(executor.getScheduledExecutorService()),
210+
builder -> { });
208211
Executors.newSingleThreadExecutor()
209212
.execute(
210213
new Runnable() {
@@ -241,7 +244,8 @@ private void verifyBootstrapFail(Bootstrapper.BootstrapInfo b) throws Exception
241244
when(xdsClient.getBootstrapInfo()).thenReturn(b);
242245
xdsServerWrapper = new XdsServerWrapper("0.0.0.0:1", mockBuilder, listener,
243246
selectorManager, new FakeXdsClientPoolFactory(xdsClient),
244-
XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry);
247+
XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry,
248+
executor.getScheduledExecutorService());
245249
final SettableFuture<Server> start = SettableFuture.create();
246250
Executors.newSingleThreadExecutor().execute(new Runnable() {
247251
@Override
@@ -281,7 +285,8 @@ public void testBootstrap_templateWithXdstp() throws Exception {
281285
when(xdsClient.getBootstrapInfo()).thenReturn(b);
282286
xdsServerWrapper = new XdsServerWrapper("[::FFFF:129.144.52.38]:80", mockBuilder, listener,
283287
selectorManager, new FakeXdsClientPoolFactory(xdsClient),
284-
XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry);
288+
XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry,
289+
executor.getScheduledExecutorService());
285290
Executors.newSingleThreadExecutor().execute(new Runnable() {
286291
@Override
287292
public void run() {
@@ -2001,7 +2006,8 @@ private FilterRegistry filterStateTestFilterRegistry(
20012006
private SettableFuture<Server> filterStateTestStartServer(FilterRegistry filterRegistry) {
20022007
xdsServerWrapper = new XdsServerWrapper("0.0.0.0:1", mockBuilder, listener,
20032008
selectorManager, new FakeXdsClientPoolFactory(xdsClient),
2004-
XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry);
2009+
XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry,
2010+
executor.getScheduledExecutorService());
20052011
SettableFuture<Server> serverStart = SettableFuture.create();
20062012
scheduleServerStart(xdsServerWrapper, serverStart);
20072013
return serverStart;
@@ -2184,8 +2190,8 @@ public void childChannelConfigurator_passedToXdsClientPool() {
21842190

21852191
XdsServerWrapper serverWrapper = new XdsServerWrapper(
21862192
"0.0.0.0:1", mockBuilder, listener, selectorManager, mockPoolFactory,
2187-
XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry,
2188-
executor.getScheduledExecutorService(), configurator);
2193+
XdsServerTestHelper.RAW_BOOTSTRAP, null, filterRegistry,
2194+
new FixedObjectPool<>(executor.getScheduledExecutorService()), configurator);
21892195

21902196
Executors.newSingleThreadExecutor().execute(() -> {
21912197
try {

0 commit comments

Comments
 (0)