Skip to content

xds: Use ObjectPool to manage scheduler in XdsServerWrapper #12841

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ejona86 merged 1 commit into from
Oct 9, 2026
Merged

xds: Use ObjectPool to manage scheduler in XdsServerWrapper #12841

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions xds/src/main/java/io/grpc/xds/XdsServerBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -183,6 +185,7 @@ public Server build() {
bootstrapOverride,
ldsResourceNameResolver,
filterRegistry,
SharedResourcePool.forResource(GrpcUtil.TIMER_SERVICE),
this.channelConfigurator);
}

Expand Down
156 changes: 7 additions & 149 deletions xds/src/main/java/io/grpc/xds/XdsServerWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<ScheduledExecutorService> timeServicePool;
private final ScheduledExecutorService timeService;
private final FilterRegistry filterRegistry;
private final ThreadSafeRandom random = ThreadSafeRandomImpl.instance;
Expand Down Expand Up @@ -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<String, ?> bootstrapOverride,
@Nullable Function<String, String> 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<String, ?> 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<String, ?> 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<String, ?> bootstrapOverride,
@Nullable Function<String, String> ldsResourceNameResolver,
FilterRegistry filterRegistry) {
this(
listenerAddress,
delegateBuilder,
listener,
filterChainSelectorManager,
xdsClientPoolFactory,
bootstrapOverride,
ldsResourceNameResolver,
filterRegistry,
SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE),
builder -> { });
sharedTimeService = true;
}

@VisibleForTesting
XdsServerWrapper(
String listenerAddress,
Expand All @@ -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<String, ?> bootstrapOverride,
@Nullable Function<String, String> 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<String, ?> bootstrapOverride,
FilterRegistry filterRegistry,
ScheduledExecutorService timeService,
ChannelConfigurator channelConfigurator) {
this(
listenerAddress,
delegateBuilder,
listener,
filterChainSelectorManager,
xdsClientPoolFactory,
bootstrapOverride,
null,
filterRegistry,
timeService,
channelConfigurator);
}

@VisibleForTesting
XdsServerWrapper(
String listenerAddress,
Expand All @@ -311,7 +170,7 @@ public void uncaughtException(Thread t, Throwable e) {
@Nullable Map<String, ?> bootstrapOverride,
@Nullable Function<String, String> ldsResourceNameResolver,
FilterRegistry filterRegistry,
ScheduledExecutorService timeService,
ObjectPool<ScheduledExecutorService> timeServicePool,
ChannelConfigurator channelConfigurator) {
this.listenerAddress = checkNotNull(listenerAddress, "listenerAddress");
this.delegateBuilder = checkNotNull(delegateBuilder, "delegateBuilder");
Expand All @@ -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");
Expand Down Expand Up @@ -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();
}
Expand Down
4 changes: 3 additions & 1 deletion xds/src/test/java/io/grpc/xds/XdsClientWrapperForServerSdsTestMisc.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down
20 changes: 13 additions & 7 deletions xds/src/test/java/io/grpc/xds/XdsServerWrapperTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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<Server> start = SettableFuture.create();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -2001,7 +2006,8 @@ private FilterRegistry filterStateTestFilterRegistry(
private SettableFuture<Server> 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<Server> serverStart = SettableFuture.create();
scheduleServerStart(xdsServerWrapper, serverStart);
return serverStart;
Expand Down Expand Up @@ -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 {
Expand Down
Loading