Skip to content

Commit c33ac2d

Browse files
sideshowcoderPhilipp Fehrebeeme1mrtoddbaert
authored
fix: possible event-related deadlocks with some providers (#1314)
* Move event emitting off the main thread to avoid deadlocks When stacking event emitting inside an EventProvider, when using sychronization the EventProvider can deadlock, to avoid this move the event emitting of the main thread. Signed-off-by: Philipp Fehre <[email protected]> * Test fixes Test provider should respect the init-delay during all test Signed-off-by: Philipp Fehre <[email protected]> * Add timeout to EventProviderTest With the events being executed on a different thread, we need to wait to make sure the thread is scheduled to have the events emitted. Signed-off-by: Philipp Fehre <[email protected]> * Don't reuse the JVM Process Signed-off-by: Philipp Fehre <[email protected]> --------- Signed-off-by: Philipp Fehre <[email protected]> Co-authored-by: Philipp Fehre <[email protected]> Co-authored-by: Michael Beemer <[email protected]> Co-authored-by: Todd Baert <[email protected]>
1 parent 08c38fb commit c33ac2d

File tree

6 files changed

+163
-13
lines changed

6 files changed

+163
-13
lines changed

‎pom.xml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -265,6 +265,8 @@
265265
<artifactId>maven-surefire-plugin</artifactId>
266266
<version>3.5.2</version>
267267
<configuration>
268+
<forkCount>1</forkCount>
269+
<reuseForks>false</reuseForks>
268270
<argLine>
269271
${surefireArgLine}
270272
</argLine>

‎src/main/java/dev/openfeature/sdk/EventProvider.java‎

Lines changed: 28 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
package dev.openfeature.sdk;
22

33
import dev.openfeature.sdk.internal.TriConsumer;
4+
import java.util.concurrent.ExecutorService;
5+
import java.util.concurrent.Executors;
6+
import java.util.concurrent.TimeUnit;
7+
import lombok.extern.slf4j.Slf4j;
48

59
/**
610
* Abstract EventProvider. Providers must extend this class to support events.
@@ -14,8 +18,10 @@
1418
*
1519
* @see FeatureProvider
1620
*/
21+
@Slf4j
1722
public abstract class EventProvider implements FeatureProvider {
1823
private EventProviderListener eventProviderListener;
24+
private final ExecutorService emitterExecutor = Executors.newCachedThreadPool();
1925

2026
void setEventProviderListener(EventProviderListener eventProviderListener) {
2127
this.eventProviderListener = eventProviderListener;
@@ -46,6 +52,24 @@ void detach() {
4652
this.onEmit = null;
4753
}
4854

55+
/**
56+
* Stop the event emitter executor and block until either termination has completed
57+
* or timeout period has elapsed.
58+
*/
59+
@Override
60+
public void shutdown() {
61+
emitterExecutor.shutdown();
62+
try {
63+
if (!emitterExecutor.awaitTermination(EventSupport.SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
64+
log.warn("Emitter executor did not terminate before the timeout period had elapsed");
65+
emitterExecutor.shutdownNow();
66+
}
67+
} catch (InterruptedException e) {
68+
emitterExecutor.shutdownNow();
69+
Thread.currentThread().interrupt();
70+
}
71+
}
72+
4973
/**
5074
* Emit the specified {@link ProviderEvent}.
5175
*
@@ -56,8 +80,10 @@ public void emit(ProviderEvent event, ProviderEventDetails details) {
5680
if (eventProviderListener != null) {
5781
eventProviderListener.onEmit(event, details);
5882
}
59-
if (this.onEmit != null) {
60-
this.onEmit.accept(this, event, details);
83+
84+
final TriConsumer<EventProvider, ProviderEvent, ProviderEventDetails> localOnEmit = this.onEmit;
85+
if (localOnEmit != null) {
86+
emitterExecutor.submit(() -> localOnEmit.accept(this, event, details));
6187
}
6288
}
6389

‎src/main/java/dev/openfeature/sdk/EventSupport.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,15 +19,15 @@
1919
@Slf4j
2020
class EventSupport {
2121

22+
public static final int SHUTDOWN_TIMEOUT_SECONDS = 3;
23+
2224
// we use a v4 uuid as a "placeholder" for anonymous clients, since
2325
// ConcurrentHashMap doesn't support nulls
2426
private static final String defaultClientUuid = UUID.randomUUID().toString();
25-
private static final int SHUTDOWN_TIMEOUT_SECONDS = 3;
2627
private final Map<String, HandlerStore> handlerStores = new ConcurrentHashMap<>();
2728
private final HandlerStore globalHandlerStore = new HandlerStore();
2829
private final ExecutorService taskExecutor = Executors.newCachedThreadPool(runnable -> {
2930
final Thread thread = new Thread(runnable);
30-
thread.setDaemon(true);
3131
return thread;
3232
});
3333

‎src/test/java/dev/openfeature/sdk/EventProviderTest.java‎

Lines changed: 23 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,18 @@
55
import static org.mockito.Mockito.*;
66

77
import dev.openfeature.sdk.internal.TriConsumer;
8+
import dev.openfeature.sdk.testutils.TestStackedEmitCallsProvider;
9+
import io.cucumber.java.AfterAll;
810
import lombok.SneakyThrows;
911
import org.junit.jupiter.api.BeforeEach;
1012
import org.junit.jupiter.api.DisplayName;
1113
import org.junit.jupiter.api.Test;
14+
import org.junit.jupiter.api.Timeout;
1215

1316
class EventProviderTest {
1417

18+
private static final int TIMEOUT = 300;
19+
1520
private TestEventProvider eventProvider;
1621

1722
@BeforeEach
@@ -21,6 +26,11 @@ void setup() {
2126
eventProvider.initialize(null);
2227
}
2328

29+
@AfterAll
30+
public static void resetDefaultProvider() {
31+
OpenFeatureAPI.getInstance().setProviderAndWait(new NoOpProvider());
32+
}
33+
2434
@Test
2535
@DisplayName("should run attached onEmit with emitters")
2636
void emitsEventsWhenAttached() {
@@ -34,10 +44,10 @@ void emitsEventsWhenAttached() {
3444
eventProvider.emitProviderStale(details);
3545
eventProvider.emitProviderError(details);
3646

37-
verify(onEmit, times(2)).accept(eventProvider, ProviderEvent.PROVIDER_READY, details);
38-
verify(onEmit, times(1)).accept(eventProvider, ProviderEvent.PROVIDER_CONFIGURATION_CHANGED, details);
39-
verify(onEmit, times(1)).accept(eventProvider, ProviderEvent.PROVIDER_STALE, details);
40-
verify(onEmit, times(1)).accept(eventProvider, ProviderEvent.PROVIDER_ERROR, details);
47+
verify(onEmit, timeout(TIMEOUT).times(2)).accept(eventProvider, ProviderEvent.PROVIDER_READY, details);
48+
verify(onEmit, timeout(TIMEOUT)).accept(eventProvider, ProviderEvent.PROVIDER_CONFIGURATION_CHANGED, details);
49+
verify(onEmit, timeout(TIMEOUT)).accept(eventProvider, ProviderEvent.PROVIDER_STALE, details);
50+
verify(onEmit, timeout(TIMEOUT)).accept(eventProvider, ProviderEvent.PROVIDER_ERROR, details);
4151
}
4252

4353
@Test
@@ -75,6 +85,15 @@ void doesNotThrowWhenOnEmitSame() {
7585
eventProvider.attach(onEmit2); // should not throw, same instance. noop
7686
}
7787

88+
@Test
89+
@SneakyThrows
90+
@Timeout(value = 2, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
91+
@DisplayName("should not deadlock on emit called during emit")
92+
void doesNotDeadlockOnEmitStackedCalls() {
93+
TestStackedEmitCallsProvider provider = new TestStackedEmitCallsProvider();
94+
OpenFeatureAPI.getInstance().setProviderAndWait(provider);
95+
}
96+
7897
static class TestEventProvider extends EventProvider {
7998

8099
private static final String NAME = "TestEventProvider";

‎src/test/java/dev/openfeature/sdk/EventsTest.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919

2020
class EventsTest {
2121

22-
private static final int TIMEOUT = 300;
22+
private static final int TIMEOUT = 500;
2323
private static final int INIT_DELAY = TIMEOUT / 2;
2424

2525
@AfterAll
@@ -601,13 +601,13 @@ void matchingStaleEventsMustRunImmediately() {
601601
OpenFeatureAPI api = OpenFeatureAPI.getInstance();
602602

603603
// provider which is already stale
604-
TestEventsProvider provider = TestEventsProvider.newInitializedTestEventsProvider();
604+
TestEventsProvider provider = new TestEventsProvider(INIT_DELAY);
605605
Client client = api.getClient(name);
606606
api.setProviderAndWait(name, provider);
607607
provider.emitProviderStale(ProviderEventDetails.builder().build());
608608
assertThat(client.getProviderState()).isEqualTo(ProviderState.STALE);
609609

610-
// should run even thought handler was added after stale
610+
// should run even though handler was added after stale
611611
client.onProviderStale(handler);
612612
verify(handler, timeout(TIMEOUT)).accept(any());
613613
}
@@ -623,13 +623,13 @@ void matchingErrorEventsMustRunImmediately() {
623623
OpenFeatureAPI api = OpenFeatureAPI.getInstance();
624624

625625
// provider which is already in error
626-
TestEventsProvider provider = new TestEventsProvider();
626+
TestEventsProvider provider = new TestEventsProvider(INIT_DELAY);
627627
Client client = api.getClient(name);
628628
api.setProviderAndWait(name, provider);
629629
provider.emitProviderError(ProviderEventDetails.builder().build());
630630
assertThat(client.getProviderState()).isEqualTo(ProviderState.ERROR);
631631

632-
// should run even thought handler was added after error
632+
// should run even though handler was added after error
633633
client.onProviderError(handler);
634634
verify(handler, timeout(TIMEOUT)).accept(any());
635635
}

‎src/test/java/dev/openfeature/sdk/testutils/TestStackedEmitCallsProvider.java‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
package dev.openfeature.sdk.testutils;
2+
3+
import dev.openfeature.sdk.EvaluationContext;
4+
import dev.openfeature.sdk.EventProvider;
5+
import dev.openfeature.sdk.Metadata;
6+
import dev.openfeature.sdk.ProviderEvaluation;
7+
import dev.openfeature.sdk.ProviderEvent;
8+
import dev.openfeature.sdk.ProviderEventDetails;
9+
import dev.openfeature.sdk.Value;
10+
import java.util.function.Consumer;
11+
12+
public class TestStackedEmitCallsProvider extends EventProvider {
13+
private final NestedBlockingEmitter nestedBlockingEmitter = new NestedBlockingEmitter(this::onProviderEvent);
14+
15+
@Override
16+
public Metadata getMetadata() {
17+
return () -> getClass().getSimpleName();
18+
}
19+
20+
@Override
21+
public void initialize(EvaluationContext evaluationContext) throws Exception {
22+
synchronized (nestedBlockingEmitter) {
23+
nestedBlockingEmitter.init();
24+
while (!nestedBlockingEmitter.isReady()) {
25+
try {
26+
nestedBlockingEmitter.wait();
27+
} catch (InterruptedException e) {
28+
}
29+
}
30+
}
31+
}
32+
33+
private void onProviderEvent(ProviderEvent providerEvent) {
34+
synchronized (nestedBlockingEmitter) {
35+
if (providerEvent == ProviderEvent.PROVIDER_READY) {
36+
nestedBlockingEmitter.setReady();
37+
/*
38+
* This line deadlocked in the original implementation without the emitterExecutor see
39+
* https://github.com/open-feature/java-sdk/issues/1299
40+
*/
41+
emitProviderReady(ProviderEventDetails.builder().build());
42+
}
43+
}
44+
}
45+
46+
@Override
47+
public ProviderEvaluation<Boolean> getBooleanEvaluation(String key, Boolean defaultValue, EvaluationContext ctx) {
48+
throw new UnsupportedOperationException("Unimplemented method 'getBooleanEvaluation'");
49+
}
50+
51+
@Override
52+
public ProviderEvaluation<String> getStringEvaluation(String key, String defaultValue, EvaluationContext ctx) {
53+
throw new UnsupportedOperationException("Unimplemented method 'getStringEvaluation'");
54+
}
55+
56+
@Override
57+
public ProviderEvaluation<Integer> getIntegerEvaluation(String key, Integer defaultValue, EvaluationContext ctx) {
58+
throw new UnsupportedOperationException("Unimplemented method 'getIntegerEvaluation'");
59+
}
60+
61+
@Override
62+
public ProviderEvaluation<Double> getDoubleEvaluation(String key, Double defaultValue, EvaluationContext ctx) {
63+
throw new UnsupportedOperationException("Unimplemented method 'getDoubleEvaluation'");
64+
}
65+
66+
@Override
67+
public ProviderEvaluation<Value> getObjectEvaluation(String key, Value defaultValue, EvaluationContext ctx) {
68+
throw new UnsupportedOperationException("Unimplemented method 'getObjectEvaluation'");
69+
}
70+
71+
static class NestedBlockingEmitter {
72+
73+
private final Consumer<ProviderEvent> emitProviderEvent;
74+
private volatile boolean isReady;
75+
76+
public NestedBlockingEmitter(Consumer<ProviderEvent> emitProviderEvent) {
77+
this.emitProviderEvent = emitProviderEvent;
78+
}
79+
80+
public void init() {
81+
// run init outside monitored thread
82+
new Thread(() -> {
83+
try {
84+
Thread.sleep(500);
85+
} catch (InterruptedException e) {
86+
throw new RuntimeException(e);
87+
}
88+
89+
emitProviderEvent.accept(ProviderEvent.PROVIDER_READY);
90+
})
91+
.start();
92+
}
93+
94+
public boolean isReady() {
95+
return isReady;
96+
}
97+
98+
public synchronized void setReady() {
99+
isReady = true;
100+
this.notifyAll();
101+
}
102+
}
103+
}

0 commit comments

Comments
 (0)