Skip to content

Commit 842f64e

Browse files
nimfnimf
authored
fix(bigtable): recycle channel on consecutive new stream failures (googleapis#13245)
1 parent dcc2a68 commit 842f64e

2 files changed

Lines changed: 232 additions & 9 deletions

File tree

‎java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/ChannelPoolDpImpl.java‎

Lines changed: 64 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import com.google.bigtable.v2.TelemetryConfiguration;
2424
import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DebugTagTracer;
2525
import com.google.common.annotations.VisibleForTesting;
26+
import com.google.common.base.Ticker;
2627
import com.google.common.collect.HashMultiset;
2728
import com.google.common.collect.Multiset;
2829
import io.grpc.CallOptions;
@@ -65,6 +66,11 @@ public class ChannelPoolDpImpl implements ChannelPool {
6566
private static final String DEFAULT_LOG_NAME = "pool";
6667
private static final AtomicInteger INDEX = new AtomicInteger();
6768

69+
// TODO: Move to client configuration.
70+
private static final int CONSECUTIVE_OPEN_SESSION_FAILURE_THRESHOLD = 5;
71+
private static final Duration INITIAL_RECYCLE_BACKOFF = Duration.ofMillis(1);
72+
private static final Duration MAX_RECYCLE_BACKOFF = Duration.ofMinutes(1);
73+
6874
private final String poolLogId;
6975

7076
@VisibleForTesting volatile int minGroups;
@@ -95,12 +101,27 @@ public class ChannelPoolDpImpl implements ChannelPool {
95101
@GuardedBy("this")
96102
private boolean closed = false;
97103

104+
private final Ticker ticker;
105+
106+
@GuardedBy("this")
107+
private long lastRecycleNano = 0;
108+
109+
@GuardedBy("this")
110+
private Duration recycleBackoff = INITIAL_RECYCLE_BACKOFF;
111+
98112
public ChannelPoolDpImpl(
99113
Supplier<ManagedChannel> channelSupplier,
100114
ChannelPoolConfiguration config,
101115
DebugTagTracer debugTagTracer,
102116
ScheduledExecutorService executor) {
103-
this(channelSupplier, config, DEFAULT_LOG_NAME, debugTagTracer, executor, Clock.systemUTC());
117+
this(
118+
channelSupplier,
119+
config,
120+
DEFAULT_LOG_NAME,
121+
debugTagTracer,
122+
executor,
123+
Clock.systemUTC(),
124+
Ticker.systemTicker());
104125
}
105126

106127
public ChannelPoolDpImpl(
@@ -109,7 +130,14 @@ public ChannelPoolDpImpl(
109130
String logName,
110131
DebugTagTracer debugTagTracer,
111132
ScheduledExecutorService executor) {
112-
this(channelSupplier, config, logName, debugTagTracer, executor, Clock.systemUTC());
133+
this(
134+
channelSupplier,
135+
config,
136+
logName,
137+
debugTagTracer,
138+
executor,
139+
Clock.systemUTC(),
140+
Ticker.systemTicker());
113141
}
114142

115143
public ChannelPoolDpImpl(
@@ -119,8 +147,20 @@ public ChannelPoolDpImpl(
119147
DebugTagTracer debugTagTracer,
120148
ScheduledExecutorService executor,
121149
Clock clock) {
150+
this(channelSupplier, config, logName, debugTagTracer, executor, clock, Ticker.systemTicker());
151+
}
152+
153+
public ChannelPoolDpImpl(
154+
Supplier<ManagedChannel> channelSupplier,
155+
ChannelPoolConfiguration config,
156+
String logName,
157+
DebugTagTracer debugTagTracer,
158+
ScheduledExecutorService executor,
159+
Clock clock,
160+
Ticker ticker) {
122161
this.poolLogId = String.format("%d-%s", INDEX.getAndIncrement(), logName);
123162
this.clock = clock;
163+
this.ticker = ticker;
124164
this.channelSupplier = channelSupplier;
125165
this.executor = executor;
126166
updateConfig(config);
@@ -221,6 +261,8 @@ public void start(Listener responseListener, Metadata headers) {
221261
public void onBeforeSessionStart(PeerInfo peerInfo) {
222262
afeId = AfeId.extract(peerInfo);
223263
synchronized (ChannelPoolDpImpl.this) {
264+
channelWrapper.consecutiveFailures = 0;
265+
recycleBackoff = INITIAL_RECYCLE_BACKOFF;
224266
rehomeChannel(channelWrapper, afeId);
225267
sessionsPerAfeId.add(afeId);
226268
}
@@ -232,6 +274,8 @@ public void onClose(Status status, Metadata trailers) {
232274
synchronized (ChannelPoolDpImpl.this) {
233275
if (afeId != null) {
234276
sessionsPerAfeId.remove(afeId);
277+
} else if (!status.isOk() && status.getCode() != Code.CANCELLED) {
278+
channelWrapper.consecutiveFailures++;
235279
}
236280
releaseChannel(channelWrapper, status);
237281
}
@@ -306,12 +350,12 @@ private void releaseChannel(ChannelWrapper channelWrapper, Status status) {
306350
channelWrapper.group.numStreams--;
307351
channelWrapper.numOutstanding--;
308352

309-
if (shouldRecycleChannel(status)) {
353+
if (shouldRecycleChannel(channelWrapper, status)) {
310354
recycleChannel(channelWrapper);
311355
}
312356
}
313357

314-
private static boolean shouldRecycleChannel(Status status) {
358+
private static boolean shouldRecycleChannel(ChannelWrapper channelWrapper, Status status) {
315359
if (status.getCode() == Code.UNIMPLEMENTED) {
316360
return true;
317361
}
@@ -322,6 +366,10 @@ private static boolean shouldRecycleChannel(Status status) {
322366
return true;
323367
}
324368

369+
if (channelWrapper.consecutiveFailures >= CONSECUTIVE_OPEN_SESSION_FAILURE_THRESHOLD) {
370+
return true;
371+
}
372+
325373
return false;
326374
}
327375

@@ -332,6 +380,17 @@ private void recycleChannel(ChannelWrapper channelWrapper) {
332380
return;
333381
}
334382

383+
long nowNano = ticker.read();
384+
if (nowNano - lastRecycleNano < recycleBackoff.toNanos()) {
385+
return;
386+
}
387+
388+
lastRecycleNano = nowNano;
389+
recycleBackoff = recycleBackoff.multipliedBy(2);
390+
if (recycleBackoff.compareTo(MAX_RECYCLE_BACKOFF) > 0) {
391+
recycleBackoff = MAX_RECYCLE_BACKOFF;
392+
}
393+
335394
channelWrapper.group.channels.remove(channelWrapper);
336395
channelWrapper.channel.shutdown();
337396
// Checking for starting group because we don't want to delete the stating group.
@@ -359,9 +418,6 @@ private synchronized void serviceChannelsSafe() {
359418
log(Level.FINE, "Servicing channels");
360419
dumpState();
361420

362-
Instant now = Instant.now(clock);
363-
Instant createdAtThreshold = now.minus(Duration.ofMinutes(50));
364-
365421
// Thin out the channels in each group, so that each AFEGroup only has 1 channel
366422
for (AfeChannelGroup group : channelGroups) {
367423
if (LOGGER.isLoggable(Level.FINEST) && group.channels.size() > 1) {
@@ -480,6 +536,7 @@ static class ChannelWrapper {
480536
private final ManagedChannel channel;
481537
private final Instant createdAt;
482538
private int numOutstanding = 0;
539+
private int consecutiveFailures = 0;
483540

484541
public ChannelWrapper(AfeChannelGroup group, ManagedChannel channel, Clock clock) {
485542
this.group = group;

‎java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/channels/ChannelPoolDpImplTest.java‎

Lines changed: 168 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import com.google.cloud.bigtable.data.v2.internal.channels.SessionStream.Listener;
3131
import com.google.cloud.bigtable.data.v2.internal.csm.NoopMetrics;
3232
import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DebugTagTracer;
33+
import com.google.common.base.Ticker;
3334
import io.grpc.Attributes;
3435
import io.grpc.CallOptions;
3536
import io.grpc.ClientCall;
@@ -41,6 +42,8 @@
4142
import java.util.Base64;
4243
import java.util.List;
4344
import java.util.concurrent.ScheduledExecutorService;
45+
import java.util.concurrent.TimeUnit;
46+
import java.util.concurrent.atomic.AtomicLong;
4447
import java.util.function.Supplier;
4548
import org.junit.jupiter.api.Test;
4649
import org.junit.jupiter.api.extension.ExtendWith;
@@ -235,7 +238,6 @@ void testDownsize() {
235238
listener.onClose(Status.OK, new Metadata());
236239
}
237240

238-
when(clock.instant()).thenReturn(Instant.now());
239241
pool.serviceChannels();
240242
verify(channel, times(numChannels - pool.minGroups)).shutdown();
241243

@@ -295,7 +297,6 @@ void testDownsizeToOptimal() {
295297
// I.e. dumpState
296298
// FINE: ChannelPool channelGroups: 5, channels: 5, starting channels: 0, totalStreams: 19,
297299
// AFEs: 5, distribution: [4, 4, 4, 4, 3]
298-
when(clock.instant()).thenReturn(Instant.now());
299300
pool.serviceChannels();
300301

301302
// Should scale down to 4 channels. 19 / 5 round up = 4.
@@ -469,4 +470,169 @@ void testRecycledChannelDoesNotRejoinPool() throws InterruptedException {
469470

470471
pool.close();
471472
}
473+
474+
@Test
475+
void testRecycleChannelOnConsecutiveFailures() {
476+
when(channelSupplier.get()).thenReturn(channel);
477+
when(channel.newCall(any(), any())).thenReturn(clientCall);
478+
doNothing().when(clientCall).start(listener.capture(), any());
479+
480+
ChannelPoolDpImpl pool =
481+
new ChannelPoolDpImpl(channelSupplier, defaultConfig, debugTagTracer, bgExecutor);
482+
483+
for (int i = 0; i < 4; i++) {
484+
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
485+
.start(mock(Listener.class), new Metadata());
486+
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
487+
488+
// Should not be recycled yet
489+
verify(channel, times(0)).shutdown();
490+
verify(channelSupplier, times(1)).get();
491+
}
492+
493+
// 5th failure
494+
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
495+
.start(mock(Listener.class), new Metadata());
496+
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
497+
498+
// Now it should be recycled
499+
verify(channel, times(1)).shutdown();
500+
verify(channelSupplier, times(2)).get();
501+
502+
pool.close();
503+
}
504+
505+
@Test
506+
void testResetConsecutiveFailuresOnSuccess() {
507+
when(channelSupplier.get()).thenReturn(channel);
508+
when(channel.newCall(any(), any())).thenReturn(clientCall);
509+
doNothing().when(clientCall).start(listener.capture(), any());
510+
doReturn(Attributes.EMPTY).when(clientCall).getAttributes();
511+
512+
ChannelPoolDpImpl pool =
513+
new ChannelPoolDpImpl(channelSupplier, defaultConfig, debugTagTracer, bgExecutor);
514+
515+
// 4 failures
516+
for (int i = 0; i < 4; i++) {
517+
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
518+
.start(mock(Listener.class), new Metadata());
519+
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
520+
}
521+
verify(channel, times(0)).shutdown();
522+
523+
// A success: onHeaders (which calls onBeforeSessionStart)
524+
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
525+
.start(mock(Listener.class), new Metadata());
526+
527+
PeerInfo peerInfo = PeerInfo.newBuilder().setApplicationFrontendId(555).build();
528+
Metadata headers = new Metadata();
529+
headers.put(
530+
SessionStreamImpl.PEER_INFO_KEY,
531+
Base64.getEncoder().encodeToString(peerInfo.toByteArray()));
532+
listener.getValue().onHeaders(headers);
533+
listener.getValue().onClose(Status.OK, new Metadata());
534+
535+
// Another 4 failures - should still not recycle because counter was reset
536+
for (int i = 0; i < 4; i++) {
537+
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
538+
.start(mock(Listener.class), new Metadata());
539+
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
540+
}
541+
verify(channel, times(0)).shutdown();
542+
543+
pool.close();
544+
}
545+
546+
@Test
547+
void testCancelledDoesNotIncrementFailures() {
548+
when(channelSupplier.get()).thenReturn(channel);
549+
when(channel.newCall(any(), any())).thenReturn(clientCall);
550+
doNothing().when(clientCall).start(listener.capture(), any());
551+
552+
ChannelPoolDpImpl pool =
553+
new ChannelPoolDpImpl(channelSupplier, defaultConfig, debugTagTracer, bgExecutor);
554+
555+
for (int i = 0; i < 10; i++) {
556+
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
557+
.start(mock(Listener.class), new Metadata());
558+
listener.getValue().onClose(Status.CANCELLED, new Metadata());
559+
}
560+
561+
// Should never be recycled
562+
verify(channel, times(0)).shutdown();
563+
verify(channelSupplier, times(1)).get();
564+
565+
pool.close();
566+
}
567+
568+
@Test
569+
void testRecycleChannelBackoff() {
570+
when(channelSupplier.get()).thenReturn(channel);
571+
when(channel.newCall(any(), any())).thenReturn(clientCall);
572+
doNothing().when(clientCall).start(listener.capture(), any());
573+
574+
Ticker ticker = mock(Ticker.class);
575+
long startNanos = TimeUnit.SECONDS.toNanos(1);
576+
final AtomicLong time = new AtomicLong(startNanos);
577+
when(ticker.read()).thenAnswer(invocation -> time.get());
578+
579+
ChannelPoolDpImpl pool =
580+
new ChannelPoolDpImpl(
581+
channelSupplier,
582+
defaultConfig,
583+
"pool",
584+
debugTagTracer,
585+
bgExecutor,
586+
Clock.systemUTC(),
587+
ticker);
588+
589+
// --- First Recycle ---
590+
for (int i = 0; i < 5; i++) {
591+
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
592+
.start(mock(Listener.class), new Metadata());
593+
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
594+
}
595+
// Should be recycled once
596+
verify(channel, times(1)).shutdown();
597+
verify(channelSupplier, times(2)).get(); // 1 initial + 1 recycle
598+
599+
// --- Second Recycle (Immediate, same time) ---
600+
// Time has not advanced. Backoff is now 2ms.
601+
for (int i = 0; i < 5; i++) {
602+
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
603+
.start(mock(Listener.class), new Metadata());
604+
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
605+
}
606+
// Should NOT be recycled again because of backoff
607+
verify(channel, times(1)).shutdown();
608+
verify(channelSupplier, times(2)).get();
609+
610+
// --- Third Recycle (After partial backoff, still blocked) ---
611+
// Advance time by 1ms (backoff is 2ms, so still blocked)
612+
time.addAndGet(TimeUnit.MILLISECONDS.toNanos(1));
613+
614+
for (int i = 0; i < 5; i++) {
615+
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
616+
.start(mock(Listener.class), new Metadata());
617+
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
618+
}
619+
// Should still NOT be recycled
620+
verify(channel, times(1)).shutdown();
621+
verify(channelSupplier, times(2)).get();
622+
623+
// --- Fourth Recycle (After full backoff) ---
624+
// Advance time by another 2ms (total 3ms from last recycle, which is > 2ms backoff)
625+
time.addAndGet(TimeUnit.MILLISECONDS.toNanos(2));
626+
627+
for (int i = 0; i < 5; i++) {
628+
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
629+
.start(mock(Listener.class), new Metadata());
630+
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
631+
}
632+
// Now it should be recycled again
633+
verify(channel, times(2)).shutdown();
634+
verify(channelSupplier, times(3)).get();
635+
636+
pool.close();
637+
}
472638
}

0 commit comments

Comments
 (0)