Skip to content

Commit 4d26645

Browse files
rahul2393rahul2393
authored
feat(spanner): prime scaled-up dynamic channel pool channels with SELECT 1 (#14254)
The grpc-gcp dynamic channel pool (DCP) now supports a GcpChannelPrimer that runs before a scaled-up channel is published to the pool, but java-spanner never registered one, so channels added under load were published cold: the first requests on them paid for the TCP handshake, TLS session, and server-side connection setup at the moment the pool was already saturated. The Go Spanner client primes scaled-up channels by executing SELECT 1 with the multiplexed session through the new channel. This change brings java-spanner to parity.
1 parent 6383f81 commit 4d26645

14 files changed

Lines changed: 2486 additions & 48 deletions

‎java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/MultiplexedSessionDatabaseClient.java‎

Lines changed: 81 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,10 @@
2828
import com.google.cloud.spanner.Options.UpdateOption;
2929
import com.google.cloud.spanner.SessionClient.SessionConsumer;
3030
import com.google.cloud.spanner.SpannerException.ResourceNotFoundException;
31+
import com.google.cloud.spanner.spi.v1.SpannerRpc.ChannelPrimeSessionSource;
3132
import com.google.common.annotations.VisibleForTesting;
3233
import com.google.common.base.Preconditions;
34+
import com.google.common.util.concurrent.Futures;
3335
import com.google.spanner.v1.BatchWriteResponse;
3436
import java.time.Clock;
3537
import java.time.Duration;
@@ -38,6 +40,7 @@
3840
import java.util.EnumSet;
3941
import java.util.HashMap;
4042
import java.util.Map;
43+
import java.util.concurrent.CancellationException;
4144
import java.util.concurrent.ExecutionException;
4245
import java.util.concurrent.Executors;
4346
import java.util.concurrent.Future;
@@ -48,12 +51,14 @@
4851
import java.util.concurrent.atomic.AtomicInteger;
4952
import java.util.concurrent.atomic.AtomicLong;
5053
import java.util.concurrent.atomic.AtomicReference;
54+
import javax.annotation.Nullable;
5155

5256
/**
5357
* {@link DatabaseClient} implementation that uses a single multiplexed session to execute
5458
* transactions.
5559
*/
56-
final class MultiplexedSessionDatabaseClient extends AbstractMultiplexedSessionDatabaseClient {
60+
final class MultiplexedSessionDatabaseClient extends AbstractMultiplexedSessionDatabaseClient
61+
implements ChannelPrimeSessionSource {
5762
/**
5863
* The maximum number of attempts that the client will try to execute CreateSession for the
5964
* initial multiplexed session. This value is only used for the very first multiplexed session
@@ -186,7 +191,7 @@ private SharedChannelUsage(int numChannels) {
186191
*/
187192
private final AtomicInteger numCurrentSingleUseTransactions = new AtomicInteger();
188193

189-
private boolean isClosed;
194+
private volatile boolean isClosed;
190195

191196
/** The duration before we try to replace the multiplexed session. The default is 7 days. */
192197
private final Duration sessionExpirationDuration;
@@ -253,15 +258,23 @@ private SharedChannelUsage(int numChannels) {
253258
final SettableApiFuture<SessionReference> initialSessionReferenceFuture =
254259
SettableApiFuture.create();
255260
this.multiplexedSessionReference = new AtomicReference<>(initialSessionReferenceFuture);
261+
spanner.getRpc().registerChannelPrimeSessionSource(this);
256262

257-
Duration waitDuration =
258-
sessionClient.getSpanner().getOptions().getSessionPoolOptions().getWaitForMinSessions();
259-
int initialAttempts =
260-
waitDuration == null || waitDuration.isZero() ? MAX_INITIAL_CREATE_SESSION_ATTEMPTS : 1;
261-
asyncCreateMultiplexedSession(initialSessionReferenceFuture, initialAttempts);
262-
maybeWaitForSessionCreation(
263-
sessionClient.getSpanner().getOptions().getSessionPoolOptions(),
264-
initialSessionReferenceFuture);
263+
try {
264+
Duration waitDuration =
265+
sessionClient.getSpanner().getOptions().getSessionPoolOptions().getWaitForMinSessions();
266+
int initialAttempts =
267+
waitDuration == null || waitDuration.isZero() ? MAX_INITIAL_CREATE_SESSION_ATTEMPTS : 1;
268+
asyncCreateMultiplexedSession(initialSessionReferenceFuture, initialAttempts);
269+
maybeWaitForSessionCreation(
270+
sessionClient.getSpanner().getOptions().getSessionPoolOptions(),
271+
initialSessionReferenceFuture);
272+
} catch (Throwable t) {
273+
// The caller never receives this client, so it will never be closed; close() therefore undoes
274+
// the registrations.
275+
close();
276+
throw t;
277+
}
265278
}
266279

267280
private void asyncCreateMultiplexedSession(
@@ -270,10 +283,19 @@ private void asyncCreateMultiplexedSession(
270283
new SessionConsumer() {
271284
@Override
272285
public void onSessionReady(SessionImpl session) {
286+
synchronized (MultiplexedSessionDatabaseClient.this) {
287+
if (isClosed) {
288+
// The client was closed while the session was being created. Ignore the session:
289+
// it must neither be handed to waiters of a closed client nor keep a maintainer
290+
// running for a client that no longer exists.
291+
sessionReferenceFuture.setException(newClosedException());
292+
return;
293+
}
294+
// only start the maintainer if we actually managed to create a session in the first
295+
// place. Starting it under the lock guarantees that a concurrent close() stops it.
296+
maintainer.start();
297+
}
273298
sessionReferenceFuture.set(session.getSessionReference());
274-
// only start the maintainer if we actually managed to create a session in the first
275-
// place.
276-
maintainer.start();
277299
if (sessionClient
278300
.getSpanner()
279301
.getOptions()
@@ -296,9 +318,11 @@ public void onSessionCreateFailure(Throwable t, int createFailureForSessionCount
296318
(ResourceNotFoundException) spannerException);
297319
}
298320
// Set the exception to trigger an error for all waiters.
299-
// Then retry the session creation if the error is (potentially) transient.
321+
// Then retry the session creation if the error is (potentially) transient and the
322+
// client has not been closed in the meantime.
300323
sessionReferenceFuture.setException(t);
301324
if (remainingAttempts > 1
325+
&& !isClientClosed()
302326
&& RETRYABLE_ERROR_CODES.contains(spannerException.getErrorCode())) {
303327
final SettableApiFuture<SessionReference> future = SettableApiFuture.create();
304328
MultiplexedSessionDatabaseClient.this.multiplexedSessionReference.set(future);
@@ -363,6 +387,15 @@ boolean isValid() {
363387
return resourceNotFoundException.get() == null;
364388
}
365389

390+
private boolean isClientClosed() {
391+
return isClosed;
392+
}
393+
394+
private static SpannerException newClosedException() {
395+
return SpannerExceptionFactory.newSpannerException(
396+
ErrorCode.FAILED_PRECONDITION, "This client has been closed");
397+
}
398+
366399
AtomicLong getNumSessionsAcquired() {
367400
return this.numSessionsAcquired;
368401
}
@@ -371,6 +404,24 @@ AtomicLong getNumSessionsReleased() {
371404
return this.numSessionsReleased;
372405
}
373406

407+
@Override
408+
@Nullable
409+
public String getChannelPrimeSessionName() {
410+
if (isClosed || !isValid()) {
411+
return null;
412+
}
413+
ApiFuture<SessionReference> future = multiplexedSessionReference.get();
414+
if (!future.isDone() || future.isCancelled()) {
415+
return null;
416+
}
417+
try {
418+
SessionReference session = Futures.getDone(future);
419+
return isClosed || !isValid() ? null : session.getName();
420+
} catch (CancellationException | ExecutionException e) {
421+
return null;
422+
}
423+
}
424+
374425
void close() {
375426
boolean releaseChannelUsage = false;
376427
synchronized (this) {
@@ -381,6 +432,9 @@ void close() {
381432
}
382433
}
383434
if (releaseChannelUsage) {
435+
// The multiplexed session is no longer maintained, so this client must no longer be asked
436+
// for a session when priming dynamic channel pool channels.
437+
spanner.getRpc().unregisterChannelPrimeSessionSource(this);
384438
synchronized (CHANNEL_USAGE) {
385439
SharedChannelUsage sharedChannelUsage = CHANNEL_USAGE.get(this.spanner);
386440
if (sharedChannelUsage != null) {
@@ -659,12 +713,19 @@ void maintain() {
659713
new SessionConsumer() {
660714
@Override
661715
public void onSessionReady(SessionImpl session) {
662-
multiplexedSessionReference.set(
663-
ApiFutures.immediateFuture(session.getSessionReference()));
664-
expirationDate.set(
665-
clock
666-
.instant()
667-
.plus(MultiplexedSessionDatabaseClient.this.sessionExpirationDuration));
716+
synchronized (MultiplexedSessionDatabaseClient.this) {
717+
if (isClosed) {
718+
// The client was closed while the session was being refreshed. The refreshed
719+
// session belongs to a client that no longer exists and is ignored.
720+
return;
721+
}
722+
multiplexedSessionReference.set(
723+
ApiFutures.immediateFuture(session.getSessionReference()));
724+
expirationDate.set(
725+
clock
726+
.instant()
727+
.plus(MultiplexedSessionDatabaseClient.this.sessionExpirationDuration));
728+
}
668729
}
669730

670731
@Override

‎java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerImpl.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -298,7 +298,8 @@ public DatabaseClient getDatabaseClient(DatabaseId db) {
298298
checkClosed();
299299
String clientId = null;
300300
if (dbClients.containsKey(db) && !dbClients.get(db).isValid()) {
301-
// Close the invalidated client and remove it.
301+
// Close the invalidated client and remove it. Closing it unregisters its multiplexed
302+
// session source from the dynamic channel pool primer.
302303
dbClients.get(db).closeAsync(new ClosedException());
303304
clientId = dbClients.get(db).clientId;
304305
dbClients.remove(db);
@@ -360,8 +361,8 @@ void close(long timeout, TimeUnit unit) {
360361
}
361362
try {
362363
closureFutures = new ArrayList<>();
363-
for (DatabaseClientImpl dbClient : dbClients.values()) {
364-
closureFutures.add(dbClient.closeAsync(closedException));
364+
for (Map.Entry<DatabaseId, DatabaseClientImpl> dbClient : dbClients.entrySet()) {
365+
closureFutures.add(dbClient.getValue().closeAsync(closedException));
365366
}
366367
dbClients.clear();
367368
Futures.successfulAsList(closureFutures).get(timeout, unit);

‎java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerOptions.java‎

Lines changed: 45 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -178,6 +178,19 @@ public class SpannerOptions extends ServiceOptions<Spanner, SpannerOptions> {
178178
*/
179179
public static final Duration DEFAULT_DYNAMIC_POOL_CLEANUP_INTERVAL = Duration.ofMinutes(1);
180180

181+
/**
182+
* Default maximum time for one attempt to prime a channel that the dynamic channel pool adds
183+
* during scale-up. Scaled-up channels are primed by executing {@code SELECT 1} with a multiplexed
184+
* session before they are published to the pool.
185+
*/
186+
public static final Duration DEFAULT_DYNAMIC_POOL_CHANNEL_PRIME_TIMEOUT = Duration.ofSeconds(10);
187+
188+
/**
189+
* Default maximum number of attempts to prime a channel that the dynamic channel pool adds during
190+
* scale-up before the channel is discarded.
191+
*/
192+
public static final int DEFAULT_DYNAMIC_POOL_CHANNEL_PRIME_MAX_ATTEMPTS = 3;
193+
181194
/**
182195
* Creates a {@link GcpChannelPoolOptions} instance with Spanner-specific defaults for dynamic
183196
* channel pooling. These defaults are optimized for typical Spanner workloads.
@@ -193,8 +206,16 @@ public class SpannerOptions extends ServiceOptions<Spanner, SpannerOptions> {
193206
* <li>Scale down interval: 3 minutes
194207
* <li>Affinity key lifetime: 10 minutes
195208
* <li>Cleanup interval: 1 minute
209+
* <li>Channel prime timeout: 10 seconds
210+
* <li>Channel prime max attempts: {@value #DEFAULT_DYNAMIC_POOL_CHANNEL_PRIME_MAX_ATTEMPTS}
196211
* </ul>
197212
*
213+
* <p>Channels that the pool adds during scale-up are primed with {@code SELECT 1} on a
214+
* multiplexed session before they are published. The primer is registered by the Spanner client
215+
* when dynamic channel pooling is enabled, unless these options already contain a primer. Priming
216+
* rotates across available multiplexed sessions owned by live database clients of the {@link
217+
* Spanner} instance. Closed or invalid database clients do not supply sessions for priming.
218+
*
198219
* @return a new {@link GcpChannelPoolOptions} instance with Spanner defaults
199220
*/
200221
public static GcpChannelPoolOptions createDefaultDynamicChannelPoolOptions() {
@@ -208,13 +229,16 @@ public static GcpChannelPoolOptions createDefaultDynamicChannelPoolOptions() {
208229
DEFAULT_DYNAMIC_POOL_SCALE_DOWN_INTERVAL)
209230
.setAffinityKeyLifetime(DEFAULT_DYNAMIC_POOL_AFFINITY_KEY_LIFETIME)
210231
.setCleanupInterval(DEFAULT_DYNAMIC_POOL_CLEANUP_INTERVAL)
232+
.setChannelPrimeTimeout(DEFAULT_DYNAMIC_POOL_CHANNEL_PRIME_TIMEOUT)
233+
.setChannelPrimeMaxAttempts(DEFAULT_DYNAMIC_POOL_CHANNEL_PRIME_MAX_ATTEMPTS)
211234
.build();
212235
}
213236

214237
/**
215238
* Merges user-provided {@link GcpChannelPoolOptions} with Spanner-specific defaults. Any value
216239
* that the user has not explicitly set (i.e. left at the builder's default of 0 or null) will be
217-
* filled in from {@link #createDefaultDynamicChannelPoolOptions()}.
240+
* filled in from {@link #createDefaultDynamicChannelPoolOptions()}. A user-provided channel
241+
* primer, prime timeout, and prime attempt count are always preserved.
218242
*/
219243
static GcpChannelPoolOptions mergeWithDefaultChannelPoolOptions(
220244
GcpChannelPoolOptions userOptions) {
@@ -251,6 +275,13 @@ static GcpChannelPoolOptions mergeWithDefaultChannelPoolOptions(
251275
if (userOptions.getCleanupInterval() == null || userOptions.getCleanupInterval().isZero()) {
252276
merged.setCleanupInterval(defaults.getCleanupInterval());
253277
}
278+
if (userOptions.getChannelPrimeTimeout() == null
279+
|| userOptions.getChannelPrimeTimeout().isZero()) {
280+
merged.setChannelPrimeTimeout(defaults.getChannelPrimeTimeout());
281+
}
282+
if (userOptions.getChannelPrimeMaxAttempts() <= 0) {
283+
merged.setChannelPrimeMaxAttempts(defaults.getChannelPrimeMaxAttempts());
284+
}
254285
return merged.build();
255286
}
256287

@@ -2031,6 +2062,12 @@ public Builder disableGrpcGcpExtension() {
20312062
* Enables dynamic channel pooling. When enabled, the client will automatically scale the number
20322063
* of channels based on load. This requires the gRPC-GCP extension to be enabled.
20332064
*
2065+
* <p>Channels that the pool adds during scale-up are primed before they serve traffic: the
2066+
* client executes {@code SELECT 1} with a multiplexed session on the new channel, and the pool
2067+
* only publishes the channel once that succeeds. See {@link
2068+
* #createDefaultDynamicChannelPoolOptions()} for the prime timeout and attempt defaults, and
2069+
* {@link #setGcpChannelPoolOptions(GcpChannelPoolOptions)} to customize them.
2070+
*
20342071
* <p>Dynamic channel pooling is disabled by default. Use this method to explicitly enable it.
20352072
* Note that calling {@link #setNumChannels(int)} will disable dynamic channel pooling even if
20362073
* this method was called.
@@ -2068,7 +2105,13 @@ public Builder setGrpcGcpOtelMetricsEnabled(boolean enableGrpcGcpOtelMetrics) {
20682105
* channel pool behavior when {@link #enableDynamicChannelPool()} is enabled.
20692106
*
20702107
* <p>If not set, Spanner-specific defaults will be used (see {@link
2071-
* #createDefaultDynamicChannelPoolOptions()}).
2108+
* #createDefaultDynamicChannelPoolOptions()}). Values that are left unset in the given options
2109+
* are filled in from those defaults.
2110+
*
2111+
* <p>Channels that the pool adds during scale-up are primed with {@code SELECT 1} on a
2112+
* multiplexed session before they are published. A channel primer, prime timeout, or prime
2113+
* attempt count that is set in the given options takes precedence over the Spanner primer and
2114+
* its defaults.
20722115
*
20732116
* <p>Example usage:
20742117
*

0 commit comments

Comments
 (0)