Skip to content

Commit bfce1d5

Browse files
kinsaurraldekinsaurralde
committed
refactor(grpc-gcp): defer per channel recovery
1 parent 5731104 commit bfce1d5

4 files changed

Lines changed: 79 additions & 318 deletions

File tree

‎grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannel.java‎

Lines changed: 38 additions & 91 deletions
Original file line numberDiff line numberDiff line change
@@ -33,9 +33,7 @@
3333
import java.util.concurrent.ScheduledExecutorService;
3434
import java.util.concurrent.ScheduledFuture;
3535
import java.util.concurrent.TimeUnit;
36-
import java.util.concurrent.atomic.AtomicBoolean;
3736
import java.util.concurrent.atomic.AtomicLong;
38-
import java.util.concurrent.locks.ReentrantLock;
3937
import java.util.logging.Logger;
4038
import javax.annotation.Nullable;
4139

@@ -55,11 +53,9 @@ public class GcpFallbackChannel extends ManagedChannel {
5553
private final boolean ownsFallbackState;
5654
private final GcpFallbackOpenTelemetry openTelemetry;
5755

58-
private final AtomicBoolean localInFallbackMode = new AtomicBoolean(false);
59-
private final AtomicLong localGeneration = new AtomicLong(0);
56+
private final AtomicLong localProbeGeneration = new AtomicLong(0);
6057
private final AtomicLong localProbeSuccesses = new AtomicLong(0);
6158
private final AtomicLong localFirstPrimaryProbeSuccessNanos = new AtomicLong(0);
62-
private final ReentrantLock stateLock = new ReentrantLock();
6359

6460
private final ScheduledExecutorService execService;
6561
private volatile ScheduledFuture<?> primaryProbeFuture = null;
@@ -94,9 +90,7 @@ public GcpFallbackChannel(
9490
this.ownsFallbackState = false;
9591
} else {
9692
this.fallbackState =
97-
execService != null
98-
? new GcpFallbackState(execService)
99-
: new GcpFallbackState();
93+
execService != null ? new GcpFallbackState(execService) : new GcpFallbackState();
10094
this.ownsFallbackState = true;
10195
}
10296
this.execService = fallbackState.getOrCreateExecutorService(options);
@@ -166,9 +160,7 @@ public GcpFallbackChannel(
166160
this.ownsFallbackState = false;
167161
} else {
168162
this.fallbackState =
169-
execService != null
170-
? new GcpFallbackState(execService)
171-
: new GcpFallbackState();
163+
execService != null ? new GcpFallbackState(execService) : new GcpFallbackState();
172164
this.ownsFallbackState = true;
173165
}
174166
this.execService = fallbackState.getOrCreateExecutorService(options);
@@ -190,55 +182,32 @@ public GcpFallbackChannel(
190182
init();
191183
}
192184

193-
private void syncFallbackModeState() {
194-
if (localGeneration.get() < fallbackState.getGeneration()) {
195-
stateLock.lock();
196-
try {
197-
synchronized (fallbackState) {
198-
long poolGen = fallbackState.getGeneration();
199-
if (localGeneration.get() < poolGen) {
200-
localInFallbackMode.set(fallbackState.getTargetFallbackMode());
201-
localProbeSuccesses.set(0);
202-
localFirstPrimaryProbeSuccessNanos.set(0);
203-
localGeneration.set(poolGen);
204-
}
205-
}
206-
} finally {
207-
stateLock.unlock();
208-
}
185+
private void resetProbeStatsIfGenerationChanged() {
186+
long currentGen = fallbackState.getGeneration();
187+
if (localProbeGeneration.get() != currentGen) {
188+
localProbeSuccesses.set(0);
189+
localFirstPrimaryProbeSuccessNanos.set(0);
190+
localProbeGeneration.set(currentGen);
209191
}
210192
}
211193

212194
public boolean isInFallbackMode() {
213-
syncFallbackModeState();
214-
return (localInFallbackMode.get() && fallbackChannel != null) || primaryChannel == null;
195+
return (fallbackState.isInFallbackMode() && fallbackChannel != null) || primaryChannel == null;
215196
}
216197

217198
@VisibleForTesting
218199
GcpFallbackState getFallbackState() {
219200
return fallbackState;
220201
}
221202

222-
@VisibleForTesting
223-
AtomicBoolean getLocalInFallbackMode() {
224-
return localInFallbackMode;
225-
}
226-
227-
@VisibleForTesting
228-
AtomicLong getLocalGeneration() {
229-
return localGeneration;
230-
}
231-
232203
@VisibleForTesting
233204
AtomicLong getLocalProbeSuccesses() {
205+
resetProbeStatsIfGenerationChanged();
234206
return localProbeSuccesses;
235207
}
236208

237209
private void init() {
238-
synchronized (fallbackState) {
239-
localGeneration.set(fallbackState.getGeneration());
240-
localInFallbackMode.set(fallbackState.getTargetFallbackMode());
241-
}
210+
localProbeGeneration.set(fallbackState.getGeneration());
242211
if (options.getPrimaryProbingFunction() != null) {
243212
this.primaryProbeFuture =
244213
fallbackState.scheduleTask(
@@ -261,7 +230,7 @@ private void init() {
261230
}
262231

263232
private void processPrimaryStatusCode(Status.Code statusCode) {
264-
if (fallbackChannel != null && !localInFallbackMode.get()) {
233+
if (fallbackChannel != null && !fallbackState.isInFallbackMode()) {
265234
if (options.getErroneousStates().contains(statusCode)) {
266235
fallbackState.getPrimaryFailures().incrementAndGet();
267236
} else {
@@ -281,63 +250,41 @@ private void processFallbackStatusCode(Status.Code statusCode) {
281250
}
282251

283252
private void probePrimary() {
284-
if (!localInFallbackMode.get() && primaryChannel != null) {
285-
return;
286-
}
287-
syncFallbackModeState();
288-
if (!localInFallbackMode.get() && primaryChannel != null) {
253+
if (!fallbackState.isInFallbackMode() && primaryChannel != null) {
289254
return;
290255
}
291-
long probeStartGen = localGeneration.get();
256+
resetProbeStatsIfGenerationChanged();
257+
long probeStartGen = localProbeGeneration.get();
292258
String result = "";
293259
if (primaryDelegateChannel == null) {
294260
result = INIT_FAILURE_REASON;
295261
} else {
296262
result = options.getPrimaryProbingFunction().apply(primaryDelegateChannel);
297263
}
298-
stateLock.lock();
299-
try {
300-
syncFallbackModeState();
301-
if (localGeneration.get() == probeStartGen
302-
&& (localInFallbackMode.get() || primaryChannel == null)) {
303-
if ("OK".equals(result)) {
304-
long nowNanos = System.nanoTime();
305-
long firstSuccessNanos =
306-
localFirstPrimaryProbeSuccessNanos.updateAndGet(prev -> prev == 0 ? nowNanos : prev);
307-
long primaryProbeSuccessCount = localProbeSuccesses.incrementAndGet();
308-
309-
boolean durationSatisfied = true;
310-
if (options.getMinPrimaryProbeSuccessDuration() != null
311-
&& !options.getMinPrimaryProbeSuccessDuration().isZero()
312-
&& !options.getMinPrimaryProbeSuccessDuration().isNegative()) {
313-
long elapsedNanos = nowNanos - firstSuccessNanos;
314-
durationSatisfied =
315-
elapsedNanos >= options.getMinPrimaryProbeSuccessDuration().toNanos();
316-
}
317-
318-
if (options.isEnableRecovery()
319-
&& fallbackChannel != null
320-
&& primaryProbeSuccessCount >= options.getMinPrimaryProbeSuccessCount()
321-
&& durationSatisfied) {
322-
long recoveredGen =
323-
fallbackState.recordRecovery(
324-
probeStartGen, options.isEnablePerChannelRecovery());
325-
if (recoveredGen != -1) {
326-
localInFallbackMode.set(false);
327-
localProbeSuccesses.set(0);
328-
localFirstPrimaryProbeSuccessNanos.set(0);
329-
localGeneration.set(recoveredGen);
330-
} else {
331-
syncFallbackModeState();
332-
}
333-
}
334-
} else {
335-
localProbeSuccesses.set(0);
336-
localFirstPrimaryProbeSuccessNanos.set(0);
264+
if ("".equals(result) && fallbackState.getGeneration() == probeStartGen) {
265+
if (options.isEnableRecovery() && fallbackChannel != null) {
266+
long nowNanos = System.nanoTime();
267+
long firstSuccessNanos =
268+
localFirstPrimaryProbeSuccessNanos.updateAndGet(prev -> prev == 0 ? nowNanos : prev);
269+
long primaryProbeSuccessCount = localProbeSuccesses.incrementAndGet();
270+
271+
boolean durationSatisfied = true;
272+
if (options.getMinPrimaryProbeSuccessDuration() != null
273+
&& !options.getMinPrimaryProbeSuccessDuration().isZero()
274+
&& !options.getMinPrimaryProbeSuccessDuration().isNegative()) {
275+
long elapsedNanos = nowNanos - firstSuccessNanos;
276+
durationSatisfied =
277+
elapsedNanos >= options.getMinPrimaryProbeSuccessDuration().toNanos();
278+
}
279+
280+
if (primaryProbeSuccessCount >= options.getMinPrimaryProbeSuccessCount()
281+
&& durationSatisfied) {
282+
fallbackState.recordRecovery(probeStartGen);
337283
}
338284
}
339-
} finally {
340-
stateLock.unlock();
285+
} else {
286+
localProbeSuccesses.set(0);
287+
localFirstPrimaryProbeSuccessNanos.set(0);
341288
}
342289
// Report metric based on result.
343290
openTelemetry.getModule().reportProbeResult(options.getPrimaryChannelName(), result);

‎grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannelOptions.java‎

Lines changed: 0 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,6 @@ public class GcpFallbackChannelOptions {
4141
private final int minPrimaryProbeSuccessCount;
4242
private final Duration minPrimaryProbeSuccessDuration;
4343
private final boolean enableRecovery;
44-
private final boolean enablePerChannelRecovery;
4544
private final String primaryChannelName;
4645
private final String fallbackChannelName;
4746
private final GcpFallbackOpenTelemetry openTelemetry;
@@ -61,7 +60,6 @@ public GcpFallbackChannelOptions(Builder builder) {
6160
this.minPrimaryProbeSuccessCount = builder.minPrimaryProbeSuccessCount;
6261
this.minPrimaryProbeSuccessDuration = builder.minPrimaryProbeSuccessDuration;
6362
this.enableRecovery = builder.enableRecovery;
64-
this.enablePerChannelRecovery = builder.enablePerChannelRecovery;
6563
this.primaryChannelName = builder.primaryChannelName;
6664
this.fallbackChannelName = builder.fallbackChannelName;
6765
this.openTelemetry = builder.openTelemetry;
@@ -121,10 +119,6 @@ public boolean isEnableRecovery() {
121119
return enableRecovery;
122120
}
123121

124-
public boolean isEnablePerChannelRecovery() {
125-
return enablePerChannelRecovery;
126-
}
127-
128122
public String getPrimaryChannelName() {
129123
return primaryChannelName;
130124
}
@@ -162,7 +156,6 @@ public static class Builder {
162156
private int minPrimaryProbeSuccessCount = 10;
163157
private Duration minPrimaryProbeSuccessDuration = Duration.ZERO;
164158
private boolean enableRecovery = false;
165-
private boolean enablePerChannelRecovery = false;
166159

167160
private String primaryChannelName = "primary";
168161
private String fallbackChannelName = "fallback";
@@ -243,11 +236,6 @@ public Builder setEnableRecovery(boolean enableRecovery) {
243236
return this;
244237
}
245238

246-
public Builder setEnablePerChannelRecovery(boolean enablePerChannelRecovery) {
247-
this.enablePerChannelRecovery = enablePerChannelRecovery;
248-
return this;
249-
}
250-
251239
public Builder setPrimaryChannelName(String primaryChannelName) {
252240
this.primaryChannelName = primaryChannelName;
253241
return this;

‎grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackState.java‎

Lines changed: 14 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -26,40 +26,21 @@
2626
import java.util.concurrent.atomic.AtomicLong;
2727

2828
/**
29-
* Shared state that coordinates failover across a pool of channels. All channels in a pool share
30-
* this instance and its background executor, so probing and error rate evaluation run once for the
31-
* whole pool.
32-
*
33-
* targetFallbackMode: whether channels should route to the fallback channel.
34-
* generation: incremented on every pool-wide transition. A channel re-reads targetFallbackMode when
35-
* generation advances past the value it last saw.
36-
* inFallbackMode: whether all traffic is on the fallback channel. Gates error rate evaluation.
37-
*
38-
* Failover: the primary error rate crossing the threshold sets targetFallbackMode and increments
39-
* generation. All channels switch to the fallback channel.
40-
*
41-
* Recovery, per-channel disabled: the first channel whose probes succeed clears targetFallbackMode
42-
* and increments generation. The whole pool returns to the primary channel.
43-
*
44-
* Recovery, per-channel enabled: a recovering channel updates only its own state, leaving
45-
* targetFallbackMode and generation unchanged. Other channels stay on the fallback channel until
46-
* their own probes succeed. inFallbackMode and targetFallbackMode diverge until then.
47-
*
48-
* A shared instance passed to setSharedState must be shut down by the caller. A state a channel
49-
* created for itself is shut down with that channel.
29+
* Shared thread-safe state that coordinates failover, recovery, and periodic error evaluation
30+
* across a pool of GcpFallbackChannel instances.
5031
*/
5132
public class GcpFallbackState {
5233
private final AtomicLong primarySuccesses = new AtomicLong(0);
5334
private final AtomicLong primaryFailures = new AtomicLong(0);
5435
private final AtomicLong fallbackSuccesses = new AtomicLong(0);
5536
private final AtomicLong fallbackFailures = new AtomicLong(0);
5637
private final AtomicLong generation = new AtomicLong(0);
57-
private final AtomicBoolean targetFallbackMode = new AtomicBoolean(false);
5838
private final AtomicBoolean inFallbackMode = new AtomicBoolean(false);
5939
private final AtomicBoolean evaluationStarted = new AtomicBoolean(false);
6040

6141
private ScheduledExecutorService execService = null;
6242
private boolean ownsExecutor = false;
43+
private boolean isShutdown = false;
6344
private volatile ScheduledFuture<?> scheduledEvaluationFuture = null;
6445

6546
public GcpFallbackState() {}
@@ -91,15 +72,7 @@ AtomicLong getFallbackFailures() {
9172
return fallbackFailures;
9273
}
9374

94-
/**
95-
* Returns whether periodic error-rate evaluation is armed, i.e. whether any traffic is currently
96-
* reaching the primary channel.
97-
*
98-
* <p>This is <em>not</em> "are the pool's channels routing to the fallback channel". Under
99-
* per-channel recovery this returns {@code false} as soon as the first channel recovers, while
100-
* other channels may still be in fallback. To ask whether a specific channel is in fallback, call
101-
* {@link GcpFallbackChannel#isInFallbackMode()} on that channel.
102-
*/
75+
/** Returns whether the pool is currently in fallback mode. */
10376
boolean isInFallbackMode() {
10477
return inFallbackMode.get();
10578
}
@@ -108,34 +81,25 @@ long getGeneration() {
10881
return generation.get();
10982
}
11083

111-
boolean getTargetFallbackMode() {
112-
return targetFallbackMode.get();
113-
}
114-
115-
/** Bumps the generation counter with a directive to target fallback mode. */
84+
/** Bumps the generation counter and transitions the pool to fallback mode. */
11685
synchronized void triggerFallback() {
11786
inFallbackMode.set(true);
118-
targetFallbackMode.set(true);
11987
generation.incrementAndGet();
12088
}
12189

12290
/**
123-
* Records channel recovery by clearing primary error counts and updating fallback mode.
91+
* Records pool recovery by clearing primary error counts and updating fallback mode.
12492
*
12593
* @param expectedGen the generation at which the recovery probe started.
126-
* @param perChannelRecovery whether recovery is scoped per channel rather than pool-wide.
12794
* @return the resulting pool generation, or -1 if the pool generation changed concurrently.
12895
*/
129-
synchronized long recordRecovery(long expectedGen, boolean perChannelRecovery) {
96+
synchronized long recordRecovery(long expectedGen) {
13097
if (generation.get() != expectedGen) {
13198
return -1;
13299
}
133-
if (inFallbackMode.get()) {
100+
if (inFallbackMode.compareAndSet(true, false)) {
134101
primaryFailures.set(0);
135102
primarySuccesses.set(0);
136-
inFallbackMode.set(false);
137-
}
138-
if (!perChannelRecovery && targetFallbackMode.compareAndSet(true, false)) {
139103
generation.incrementAndGet();
140104
}
141105
return generation.get();
@@ -167,7 +131,7 @@ synchronized ScheduledExecutorService getOrCreateExecutorService(
167131
/** Schedules a periodic task (e.g., probe) on the shared background executor service. */
168132
synchronized ScheduledFuture<?> scheduleTask(
169133
Runnable command, long initialDelay, long period, TimeUnit unit) {
170-
if (this.execService == null || this.execService.isShutdown()) {
134+
if (isShutdown || this.execService == null || this.execService.isShutdown()) {
171135
return null;
172136
}
173137
return this.execService.scheduleAtFixedRate(command, initialDelay, period, unit);
@@ -180,7 +144,8 @@ synchronized ScheduledFuture<?> scheduleTask(
180144
* @param options the fallback channel configuration options.
181145
*/
182146
synchronized void startPeriodicEvaluation(GcpFallbackChannelOptions options) {
183-
if (options == null
147+
if (isShutdown
148+
|| options == null
184149
|| !options.isEnableFallback()
185150
|| options.getPeriod() == null
186151
|| options.getPeriod().toMillis() <= 0) {
@@ -212,8 +177,7 @@ synchronized void startPeriodicEvaluation(GcpFallbackChannelOptions options) {
212177
* @param options the fallback channel configuration options.
213178
* @param openTelemetry telemetry module for recording error metrics.
214179
*/
215-
void checkErrorRates(
216-
GcpFallbackChannelOptions options, GcpFallbackOpenTelemetry openTelemetry) {
180+
void checkErrorRates(GcpFallbackChannelOptions options, GcpFallbackOpenTelemetry openTelemetry) {
217181
float primaryErrRate = 0f;
218182
boolean fallbackTriggered = false;
219183
boolean currentInFallback;
@@ -271,6 +235,7 @@ synchronized void stopPeriodicEvaluation() {
271235

272236
/** Shuts down the state, cancelling evaluation and shutting down internal executor if owned. */
273237
public synchronized void shutdown() {
238+
isShutdown = true;
274239
stopPeriodicEvaluation();
275240
if (ownsExecutor && execService != null && !execService.isShutdown()) {
276241
execService.shutdown();
@@ -282,6 +247,7 @@ public synchronized void shutdown() {
282247
* owned.
283248
*/
284249
public synchronized void shutdownNow() {
250+
isShutdown = true;
285251
stopPeriodicEvaluation();
286252
if (ownsExecutor && execService != null && !execService.isShutdown()) {
287253
execService.shutdownNow();

0 commit comments

Comments
 (0)