diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/FDv2DataSource.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/FDv2DataSource.java index 9b8d4056..87eae313 100644 --- a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/FDv2DataSource.java +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/FDv2DataSource.java @@ -28,6 +28,8 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; /** @@ -54,6 +56,15 @@ public interface DataSourceFactory { private static final String INITIALIZER_CANCELLED = "Initializer cancelled: {}"; private static final String INITIALIZER_INTERRUPTED = "Initializer interrupted: {}"; + /** + * A synchronizer session shorter than this never really connected to anything, so the rotation + * to the next synchronizer is paused rather than attempted immediately. + */ + private static final long MIN_SYNCHRONIZER_SESSION_MILLIS = 500; + + /** Upper bound for the growing pause between synchronizer sessions that keep ending at once. */ + private static final long MAX_ROTATION_PAUSE_MILLIS = 30_000; + private final List> cacheInitializers; private final SourceManager sourceManager; private final long fallbackTimeoutSeconds; @@ -510,13 +521,36 @@ private void maybeLogSynchronizerStatusChange(@Nullable String sourceName, @NonN logger.info("Synchronizer '{}' reported status: {}.", sourceName, state.name()); } + /** + * Waits before building the next synchronizer. Waiting on {@link #shutdownCause} rather than + * sleeping means a {@link #stop(Callback)} during the pause is acted on right away. + * + * @return false if the data source shut down while waiting, in which case the caller must stop + */ + private boolean pauseBeforeRotation(long pauseMillis) { + logger.debug("Waiting {}ms before trying the next synchronizer.", pauseMillis); + try { + shutdownCause.get(pauseMillis, TimeUnit.MILLISECONDS); + return false; + } catch (TimeoutException e) { + return true; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } catch (ExecutionException e) { + return false; + } + } + private void runSynchronizers( @NonNull LDContext context, @NonNull DataSourceUpdateSinkV2 sink ) { + long rotationPauseMillis = 0; try { Synchronizer synchronizer = sourceManager.getNextAvailableSynchronizerAndSetActive(); while (synchronizer != null) { + long sessionStartNanos = System.nanoTime(); String synchronizerName = synchronizer.name(); logger.info("Synchronizer '{}' is starting.", synchronizerName); resetSynchronizerStatusDedupe(); @@ -651,6 +685,22 @@ private void runSynchronizers( Thread.currentThread().interrupt(); return; } + + // A source that ends its session at once, such as one reporting SHUTDOWN as soon as + // it is built, must not be allowed to drive this rotation at CPU speed. The pause + // grows while sessions keep ending immediately and resets once one of them lasts. + if (TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - sessionStartNanos) + < MIN_SYNCHRONIZER_SESSION_MILLIS) { + rotationPauseMillis = rotationPauseMillis == 0 + ? MIN_SYNCHRONIZER_SESSION_MILLIS + : Math.min(rotationPauseMillis * 2, MAX_ROTATION_PAUSE_MILLIS); + if (!pauseBeforeRotation(rotationPauseMillis)) { + return; + } + } else { + rotationPauseMillis = 0; + } + synchronizer = sourceManager.getNextAvailableSynchronizerAndSetActive(); } if (!stopCalled.get()) { diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/FDv2DataSourceTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/FDv2DataSourceTest.java index 5b2fde21..a79df9d9 100644 --- a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/FDv2DataSourceTest.java +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/FDv2DataSourceTest.java @@ -517,22 +517,27 @@ public void emptyInitializerListSkipsToSynchronizers() throws Exception { @Test public void fallbackAndRecoveryTasksWellBehaved() throws Exception { - // First sync: changeset then INTERRUPTED; second sync: changeset; recovery brings back first - MockQueuedSynchronizer firstSync = new MockQueuedSynchronizer( - FDv2SourceResult.changeSet(makeChangeSet(false), false), - interrupted()); - MockQueuedSynchronizer secondSync = new MockQueuedSynchronizer( - FDv2SourceResult.changeSet(makeChangeSet(false), false)); - AtomicInteger firstCallCount = new AtomicInteger(0); AtomicInteger secondCallCount = new AtomicInteger(0); MockComponents.MockDataSourceUpdateSink sink = new MockComponents.MockDataSourceUpdateSink(); + // First sync: changeset then INTERRUPTED; second sync: changeset; recovery brings back first. + // Each factory must build a fresh synchronizer, because the data source closes the previous + // one when it switches, and a closed synchronizer only ever reports SHUTDOWN. FDv2DataSource dataSource = buildDataSource(sink, Collections.emptyList(), Arrays.asList( - () -> { firstCallCount.incrementAndGet(); return firstSync; }, - () -> { secondCallCount.incrementAndGet(); return secondSync; }), + () -> { + firstCallCount.incrementAndGet(); + return new MockQueuedSynchronizer( + FDv2SourceResult.changeSet(makeChangeSet(false), false), + interrupted()); + }, + () -> { + secondCallCount.incrementAndGet(); + return new MockQueuedSynchronizer( + FDv2SourceResult.changeSet(makeChangeSet(false), false)); + }), 1, 2); AwaitableCallback startCallback = startDataSource(dataSource); @@ -663,17 +668,22 @@ public void recoveryResetsToFirstAvailableSynchronizer() throws Exception { AtomicInteger firstCallCount = new AtomicInteger(0); AtomicInteger secondCallCount = new AtomicInteger(0); - MockQueuedSynchronizer firstSync = new MockQueuedSynchronizer( - FDv2SourceResult.changeSet(makeChangeSet(false), false), - interrupted()); - MockQueuedSynchronizer secondSync = new MockQueuedSynchronizer( - FDv2SourceResult.changeSet(makeChangeSet(false), false)); - + // Each factory must build a fresh synchronizer, because the data source closes the previous + // one when it switches, and a closed synchronizer only ever reports SHUTDOWN. FDv2DataSource dataSource = buildDataSource(sink, Collections.emptyList(), Arrays.asList( - () -> { firstCallCount.incrementAndGet(); return firstSync; }, - () -> { secondCallCount.incrementAndGet(); return secondSync; }), + () -> { + firstCallCount.incrementAndGet(); + return new MockQueuedSynchronizer( + FDv2SourceResult.changeSet(makeChangeSet(false), false), + interrupted()); + }, + () -> { + secondCallCount.incrementAndGet(); + return new MockQueuedSynchronizer( + FDv2SourceResult.changeSet(makeChangeSet(false), false)); + }), 1, 2); AwaitableCallback startCallback = startDataSource(dataSource); @@ -687,6 +697,37 @@ public void recoveryResetsToFirstAvailableSynchronizer() throws Exception { stopDataSource(dataSource); } + @Test + public void synchronizersThatShutDownImmediatelyDoNotSpin() throws Exception { + MockComponents.MockDataSourceUpdateSink sink = new MockComponents.MockDataSourceUpdateSink(); + AtomicInteger buildCount = new AtomicInteger(0); + + // Every session ends as soon as it starts, so the rotation has nothing to wait on and would + // run at CPU speed if it were not rate limited. + FDv2DataSource dataSource = buildDataSource(sink, + Collections.emptyList(), + Arrays.asList( + () -> { + buildCount.incrementAndGet(); + return new MockQueuedSynchronizer( + FDv2SourceResult.status(FDv2SourceResult.Status.shutdown(), false)); + }, + () -> { + buildCount.incrementAndGet(); + return new MockQueuedSynchronizer( + FDv2SourceResult.status(FDv2SourceResult.Status.shutdown(), false)); + })); + + startDataSource(dataSource); + Thread.sleep(1500); + + // Pauses of 500ms, 1s, 2s and so on allow only a handful of attempts in this window. + int builds = buildCount.get(); + assertTrue("expected rate limited rotation, but saw " + builds + " synchronizers built", + builds <= 6); + stopDataSource(dataSource); + } + @Test public void fallbackMovesToNextSynchronizer() throws Exception { MockComponents.MockDataSourceUpdateSink sink = new MockComponents.MockDataSourceUpdateSink(); @@ -2024,16 +2065,16 @@ public void orchestrationLogging_fallback_logsInfo() throws Exception { @Test public void orchestrationLogging_recovery_logsInfo() throws Exception { MockComponents.MockDataSourceUpdateSink sink = new MockComponents.MockDataSourceUpdateSink(); - MockQueuedSynchronizer firstSync = new MockQueuedSynchronizer( - FDv2SourceResult.changeSet(makeChangeSet(false), false), - interrupted()); - MockQueuedSynchronizer secondSync = new MockQueuedSynchronizer( - FDv2SourceResult.changeSet(makeChangeSet(false), false)); + // Fresh instances per build: the data source closes the previous synchronizer when it + // switches, and a closed synchronizer only ever reports SHUTDOWN. FDv2DataSource dataSource = buildDataSource(sink, Collections.emptyList(), Arrays.asList( - () -> firstSync, - () -> secondSync), + () -> new MockQueuedSynchronizer( + FDv2SourceResult.changeSet(makeChangeSet(false), false), + interrupted()), + () -> new MockQueuedSynchronizer( + FDv2SourceResult.changeSet(makeChangeSet(false), false))), 1, 2); AwaitableCallback startCallback = startDataSource(dataSource); try {