From 1ec9ec5a79bddd8a5d07455f101898d1dd3699f5 Mon Sep 17 00:00:00 2001 From: Eric Anderson Date: Wed, 3 Jun 2026 15:03:52 -0700 Subject: [PATCH] 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. --- .../java/io/grpc/xds/XdsServerBuilder.java | 3 + .../java/io/grpc/xds/XdsServerWrapper.java | 156 +----------------- .../XdsClientWrapperForServerSdsTestMisc.java | 4 +- .../io/grpc/xds/XdsServerWrapperTest.java | 20 ++- 4 files changed, 26 insertions(+), 157 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/XdsServerBuilder.java b/xds/src/main/java/io/grpc/xds/XdsServerBuilder.java index fb05f22499b..1d2dd81aac2 100644 --- a/xds/src/main/java/io/grpc/xds/XdsServerBuilder.java +++ b/xds/src/main/java/io/grpc/xds/XdsServerBuilder.java @@ -32,6 +32,8 @@ import io.grpc.Server; import io.grpc.ServerBuilder; import io.grpc.ServerCredentials; +import io.grpc.internal.GrpcUtil; +import io.grpc.internal.SharedResourcePool; import io.grpc.netty.InternalNettyServerBuilder; import io.grpc.netty.InternalNettyServerCredentials; import io.grpc.netty.InternalProtocolNegotiator; @@ -183,6 +185,7 @@ public Server build() { bootstrapOverride, ldsResourceNameResolver, filterRegistry, + SharedResourcePool.forResource(GrpcUtil.TIMER_SERVICE), this.channelConfigurator); } diff --git a/xds/src/main/java/io/grpc/xds/XdsServerWrapper.java b/xds/src/main/java/io/grpc/xds/XdsServerWrapper.java index eb6d5bda0fd..ad68dcb78c5 100644 --- a/xds/src/main/java/io/grpc/xds/XdsServerWrapper.java +++ b/xds/src/main/java/io/grpc/xds/XdsServerWrapper.java @@ -46,9 +46,8 @@ import io.grpc.StatusOr; import io.grpc.SynchronizationContext; import io.grpc.SynchronizationContext.ScheduledHandle; -import io.grpc.internal.GrpcUtil; +import io.grpc.internal.FixedObjectPool; import io.grpc.internal.ObjectPool; -import io.grpc.internal.SharedResourceHolder; import io.grpc.xds.EnvoyServerProtoData.FilterChain; import io.grpc.xds.Filter.FilterConfig; import io.grpc.xds.Filter.FilterContext; @@ -103,7 +102,7 @@ public void uncaughtException(Thread t, Throwable e) { static final long RETRY_DELAY_NANOS = TimeUnit.MINUTES.toNanos(1); private final String listenerAddress; private final ServerBuilder delegateBuilder; - private boolean sharedTimeService; + private final ObjectPool timeServicePool; private final ScheduledExecutorService timeService; private final FilterRegistry filterRegistry; private final ThreadSafeRandom random = ThreadSafeRandomImpl.instance; @@ -138,98 +137,6 @@ public void uncaughtException(Thread t, Throwable e) { private final ChannelConfigurator channelConfigurator; - XdsServerWrapper( - String listenerAddress, - ServerBuilder delegateBuilder, - XdsServingStatusListener listener, - FilterChainSelectorManager filterChainSelectorManager, - XdsClientPoolFactory xdsClientPoolFactory, - @Nullable Map bootstrapOverride, - @Nullable Function ldsResourceNameResolver, - FilterRegistry filterRegistry, - ChannelConfigurator channelConfigurator) { - this( - listenerAddress, - delegateBuilder, - listener, - filterChainSelectorManager, - xdsClientPoolFactory, - bootstrapOverride, - ldsResourceNameResolver, - filterRegistry, - SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE), - channelConfigurator); - sharedTimeService = true; - } - - XdsServerWrapper( - String listenerAddress, - ServerBuilder delegateBuilder, - XdsServingStatusListener listener, - FilterChainSelectorManager filterChainSelectorManager, - XdsClientPoolFactory xdsClientPoolFactory, - @Nullable Map bootstrapOverride, - FilterRegistry filterRegistry, - ChannelConfigurator channelConfigurator) { - this( - listenerAddress, - delegateBuilder, - listener, - filterChainSelectorManager, - xdsClientPoolFactory, - bootstrapOverride, - null, - filterRegistry, - SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE), - channelConfigurator); - sharedTimeService = true; - } - - XdsServerWrapper( - String listenerAddress, - ServerBuilder delegateBuilder, - XdsServingStatusListener listener, - FilterChainSelectorManager filterChainSelectorManager, - XdsClientPoolFactory xdsClientPoolFactory, - @Nullable Map bootstrapOverride, - FilterRegistry filterRegistry) { - this( - listenerAddress, - delegateBuilder, - listener, - filterChainSelectorManager, - xdsClientPoolFactory, - bootstrapOverride, - null, - filterRegistry, - SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE), - builder -> { }); - sharedTimeService = true; - } - - XdsServerWrapper( - String listenerAddress, - ServerBuilder delegateBuilder, - XdsServingStatusListener listener, - FilterChainSelectorManager filterChainSelectorManager, - XdsClientPoolFactory xdsClientPoolFactory, - @Nullable Map bootstrapOverride, - @Nullable Function ldsResourceNameResolver, - FilterRegistry filterRegistry) { - this( - listenerAddress, - delegateBuilder, - listener, - filterChainSelectorManager, - xdsClientPoolFactory, - bootstrapOverride, - ldsResourceNameResolver, - filterRegistry, - SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE), - builder -> { }); - sharedTimeService = true; - } - @VisibleForTesting XdsServerWrapper( String listenerAddress, @@ -249,58 +156,10 @@ public void uncaughtException(Thread t, Throwable e) { bootstrapOverride, null, filterRegistry, - timeService, - builder -> { }); - } - - @VisibleForTesting - XdsServerWrapper( - String listenerAddress, - ServerBuilder delegateBuilder, - XdsServingStatusListener listener, - FilterChainSelectorManager filterChainSelectorManager, - XdsClientPoolFactory xdsClientPoolFactory, - @Nullable Map bootstrapOverride, - @Nullable Function ldsResourceNameResolver, - FilterRegistry filterRegistry, - ScheduledExecutorService timeService) { - this( - listenerAddress, - delegateBuilder, - listener, - filterChainSelectorManager, - xdsClientPoolFactory, - bootstrapOverride, - ldsResourceNameResolver, - filterRegistry, - timeService, + new FixedObjectPool<>(timeService), builder -> { }); } - @VisibleForTesting - XdsServerWrapper( - String listenerAddress, - ServerBuilder delegateBuilder, - XdsServingStatusListener listener, - FilterChainSelectorManager filterChainSelectorManager, - XdsClientPoolFactory xdsClientPoolFactory, - @Nullable Map bootstrapOverride, - FilterRegistry filterRegistry, - ScheduledExecutorService timeService, - ChannelConfigurator channelConfigurator) { - this( - listenerAddress, - delegateBuilder, - listener, - filterChainSelectorManager, - xdsClientPoolFactory, - bootstrapOverride, - null, - filterRegistry, - timeService, - channelConfigurator); - } - @VisibleForTesting XdsServerWrapper( String listenerAddress, @@ -311,7 +170,7 @@ public void uncaughtException(Thread t, Throwable e) { @Nullable Map bootstrapOverride, @Nullable Function ldsResourceNameResolver, FilterRegistry filterRegistry, - ScheduledExecutorService timeService, + ObjectPool timeServicePool, ChannelConfigurator channelConfigurator) { this.listenerAddress = checkNotNull(listenerAddress, "listenerAddress"); this.delegateBuilder = checkNotNull(delegateBuilder, "delegateBuilder"); @@ -322,7 +181,8 @@ public void uncaughtException(Thread t, Throwable e) { this.xdsClientPoolFactory = checkNotNull(xdsClientPoolFactory, "xdsClientPoolFactory"); this.bootstrapOverride = bootstrapOverride; this.ldsResourceNameResolver = ldsResourceNameResolver; - this.timeService = checkNotNull(timeService, "timeService"); + this.timeServicePool = checkNotNull(timeServicePool, "timeServicePool"); + this.timeService = checkNotNull(timeServicePool.getObject(), "timeService"); this.filterRegistry = checkNotNull(filterRegistry,"filterRegistry"); this.delegate = delegateBuilder.build(); this.channelConfigurator = checkNotNull(channelConfigurator, "channelConfigurator"); @@ -439,9 +299,7 @@ private void internalShutdown() { if (restartTimer != null) { restartTimer.cancel(); } - if (sharedTimeService) { - SharedResourceHolder.release(GrpcUtil.TIMER_SERVICE, timeService); - } + timeServicePool.returnObject(timeService); isServing = false; internalTerminationLatch.countDown(); } diff --git a/xds/src/test/java/io/grpc/xds/XdsClientWrapperForServerSdsTestMisc.java b/xds/src/test/java/io/grpc/xds/XdsClientWrapperForServerSdsTestMisc.java index 81186d0639c..11e49db3e35 100644 --- a/xds/src/test/java/io/grpc/xds/XdsClientWrapperForServerSdsTestMisc.java +++ b/xds/src/test/java/io/grpc/xds/XdsClientWrapperForServerSdsTestMisc.java @@ -38,6 +38,7 @@ import io.grpc.Status; import io.grpc.StatusOr; import io.grpc.inprocess.InProcessSocketAddress; +import io.grpc.internal.FakeClock; import io.grpc.internal.TestUtils.NoopChannelLogger; import io.grpc.netty.GrpcHttp2ConnectionHandler; import io.grpc.netty.InternalProtocolNegotiationEvent; @@ -122,7 +123,8 @@ public void setUp() { when(mockServer.isShutdown()).thenReturn(false); xdsServerWrapper = new XdsServerWrapper("0.0.0.0:" + PORT, mockBuilder, listener, selectorManager, new FakeXdsClientPoolFactory(xdsClient), - XdsServerTestHelper.RAW_BOOTSTRAP, FilterRegistry.newRegistry()); + XdsServerTestHelper.RAW_BOOTSTRAP, FilterRegistry.newRegistry(), + new FakeClock().getScheduledExecutorService()); } @Test diff --git a/xds/src/test/java/io/grpc/xds/XdsServerWrapperTest.java b/xds/src/test/java/io/grpc/xds/XdsServerWrapperTest.java index 63200352c67..917c3e214cd 100644 --- a/xds/src/test/java/io/grpc/xds/XdsServerWrapperTest.java +++ b/xds/src/test/java/io/grpc/xds/XdsServerWrapperTest.java @@ -53,6 +53,7 @@ import io.grpc.StatusOr; import io.grpc.SynchronizationContext; import io.grpc.internal.FakeClock; +import io.grpc.internal.FixedObjectPool; import io.grpc.internal.ObjectPool; import io.grpc.testing.TestMethodDescriptors; import io.grpc.xds.EnvoyServerProtoData.CidrRange; @@ -163,7 +164,7 @@ public void testBootstrap() throws Exception { when(xdsClient.getBootstrapInfo()).thenReturn(b); xdsServerWrapper = new XdsServerWrapper("[::FFFF:129.144.52.38]:80", mockBuilder, listener, selectorManager, new FakeXdsClientPoolFactory(xdsClient), - XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry); + XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry, executor.getScheduledExecutorService()); Executors.newSingleThreadExecutor().execute(new Runnable() { @Override public void run() { @@ -204,7 +205,9 @@ public void testBootstrap_ldsResourceNameResolver() throws Exception { new FakeXdsClientPoolFactory(xdsClient), XdsServerTestHelper.RAW_BOOTSTRAP, addr -> "xdstp://resolved_name/" + addr, - filterRegistry); + filterRegistry, + new FixedObjectPool<>(executor.getScheduledExecutorService()), + builder -> { }); Executors.newSingleThreadExecutor() .execute( new Runnable() { @@ -241,7 +244,8 @@ private void verifyBootstrapFail(Bootstrapper.BootstrapInfo b) throws Exception when(xdsClient.getBootstrapInfo()).thenReturn(b); xdsServerWrapper = new XdsServerWrapper("0.0.0.0:1", mockBuilder, listener, selectorManager, new FakeXdsClientPoolFactory(xdsClient), - XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry); + XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry, + executor.getScheduledExecutorService()); final SettableFuture start = SettableFuture.create(); Executors.newSingleThreadExecutor().execute(new Runnable() { @Override @@ -281,7 +285,8 @@ public void testBootstrap_templateWithXdstp() throws Exception { when(xdsClient.getBootstrapInfo()).thenReturn(b); xdsServerWrapper = new XdsServerWrapper("[::FFFF:129.144.52.38]:80", mockBuilder, listener, selectorManager, new FakeXdsClientPoolFactory(xdsClient), - XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry); + XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry, + executor.getScheduledExecutorService()); Executors.newSingleThreadExecutor().execute(new Runnable() { @Override public void run() { @@ -2001,7 +2006,8 @@ private FilterRegistry filterStateTestFilterRegistry( private SettableFuture filterStateTestStartServer(FilterRegistry filterRegistry) { xdsServerWrapper = new XdsServerWrapper("0.0.0.0:1", mockBuilder, listener, selectorManager, new FakeXdsClientPoolFactory(xdsClient), - XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry); + XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry, + executor.getScheduledExecutorService()); SettableFuture serverStart = SettableFuture.create(); scheduleServerStart(xdsServerWrapper, serverStart); return serverStart; @@ -2184,8 +2190,8 @@ public void childChannelConfigurator_passedToXdsClientPool() { XdsServerWrapper serverWrapper = new XdsServerWrapper( "0.0.0.0:1", mockBuilder, listener, selectorManager, mockPoolFactory, - XdsServerTestHelper.RAW_BOOTSTRAP, filterRegistry, - executor.getScheduledExecutorService(), configurator); + XdsServerTestHelper.RAW_BOOTSTRAP, null, filterRegistry, + new FixedObjectPool<>(executor.getScheduledExecutorService()), configurator); Executors.newSingleThreadExecutor().execute(() -> { try {