From 133bb2002ad9e92197dbe735d91b4300d55fc0a9 Mon Sep 17 00:00:00 2001 From: Janne Valkealahti Date: Sat, 21 Nov 2020 09:33:56 +0000 Subject: [PATCH] Make emit non-blocking - Now doing triggerSink.emitNext with retry instead of parking which caused blocking errors from blockhound. - Retry is now hardcoded and settings for it could be exposed later. --- .../support/ReactiveStateMachineExecutor.java | 37 ++++++++++--------- .../support/StateMachineExecutor.java | 11 +----- .../statemachine/EventDeferTests.java | 22 ++++++----- .../statemachine/ReactiveTests.java | 5 ++- .../statemachine/TestUtils.java | 4 +- 5 files changed, 39 insertions(+), 40 deletions(-) diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java index f2f1ec35..4bb44094 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java @@ -15,6 +15,7 @@ */ package org.springframework.statemachine.support; +import java.time.Duration; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -28,7 +29,6 @@ import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.locks.LockSupport; import java.util.function.Function; import java.util.stream.Collectors; @@ -57,9 +57,11 @@ import reactor.core.Disposable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.Sinks; +import reactor.core.publisher.Sinks.EmitFailureHandler; import reactor.core.publisher.Sinks.Many; import reactor.util.concurrent.Queues; import reactor.util.context.Context; +import reactor.util.retry.Retry; /** * Default reactive implementation of a {@link StateMachineExecutor}. @@ -152,17 +154,6 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i return stopTriggers().and(mono); } - @Override - public void queueTrigger(Trigger trigger, Message message) { - if (log.isDebugEnabled()) { - log.debug("Queue trigger " + trigger); - } - TriggerQueueItem tqi = new TriggerQueueItem(trigger, message, null, null); - while (triggerSink.tryEmitNext(tqi).isFailure()) { - LockSupport.parkNanos(10); - } - } - @Override public void queueDeferredEvent(Message message) { if (log.isDebugEnabled()) { @@ -211,11 +202,10 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i return messages .flatMap(m -> handleEvent(m, callback, triggerCallback)) - .doOnNext(i -> { - while (triggerSink.tryEmitNext(i).isFailure()) { - LockSupport.parkNanos(10); - } - }) + .flatMap(tqi -> Mono.fromRunnable(() -> { + triggerSink.emitNext(tqi, EmitFailureHandler.FAIL_FAST); + }) + .retryWhen(Retry.fixedDelay(10, Duration.ofMillis(10)))) .then() .and(triggerCallbackSink); } @@ -464,7 +454,13 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i if (log.isDebugEnabled()) { log.debug("TimedTrigger triggered " + trigger); } - queueTrigger(trigger, null); + Mono.just(new TriggerQueueItem(trigger, null, null, null)) + .flatMap(tqi -> Mono.fromCallable(() -> { + triggerSink.emitNext(tqi, EmitFailureHandler.FAIL_FAST); + return null; + }) + .retryWhen(Retry.fixedDelay(10, Duration.ofNanos(10)))) + .subscribe(); } }); } @@ -514,5 +510,10 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i this.callback = callback; this.triggerCallback = triggerCallback; } + + @Override + public String toString() { + return "TriggerQueueItem [message=" + message + ", trigger=" + trigger + "]"; + } } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java index 91b6a8f0..28d68c2c 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,7 +23,6 @@ import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.access.StateMachineAccess; import org.springframework.statemachine.state.State; import org.springframework.statemachine.transition.Transition; -import org.springframework.statemachine.trigger.Trigger; import reactor.core.publisher.Mono; import reactor.core.publisher.MonoSink; @@ -47,14 +46,6 @@ public interface StateMachineExecutor extends StateMachineReactiveLifecycl */ Mono queueEvent(Mono> message, StateMachineExecutorCallback callback); - /** - * Queue trigger. - * - * @param trigger the trigger - * @param message the message - */ - void queueTrigger(Trigger trigger, Message message); - /** * Queue deferred event. * diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/EventDeferTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/EventDeferTests.java index de21a772..19c772ba 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/EventDeferTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/EventDeferTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -30,7 +30,10 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Configuration; import org.springframework.statemachine.config.EnableStateMachine; @@ -42,6 +45,8 @@ import org.springframework.statemachine.state.State; public class EventDeferTests extends AbstractStateMachineTests { + private static final Log log = LogFactory.getLog(EventDeferTests.class); + @Override protected AnnotationConfigApplicationContext buildContext() { return new AnnotationConfigApplicationContext(); @@ -65,6 +70,7 @@ public class EventDeferTests extends AbstractStateMachineTests { } @Test + @Timeout(value = 20, unit = TimeUnit.SECONDS) public void testDeferSmokeExecutorConcurrentModification() throws Exception { context.register(Config5.class); context.refresh(); @@ -73,20 +79,15 @@ public class EventDeferTests extends AbstractStateMachineTests { doStartAndAssert(machine); assertThat(machine.getState().getIds(), contains("S1")); - doSendEventAndConsumeAll(machine, "E2"); - - Object executor = TestUtils.readField("stateMachineExecutor", machine); - Collection readField = TestUtils.readField("deferList", executor); - assertThat(readField.size(), is(1)); - - AtomicReference error = new AtomicReference<>(); + AtomicReference error = new AtomicReference<>(); AtomicInteger i1 = new AtomicInteger(); Thread t1 = new Thread(() -> { while(i1.incrementAndGet() < 200) { try { doSendEventAndConsumeAll(machine, "E1"); doSendEventAndConsumeAll(machine, "E2"); - } catch (Exception e) { + } catch (Throwable e) { + log.error("T1: error " + e); error.set(e); break; } @@ -98,7 +99,8 @@ public class EventDeferTests extends AbstractStateMachineTests { try { doSendEventAndConsumeAll(machine, "E1"); doSendEventAndConsumeAll(machine, "E2"); - } catch (Exception e) { + } catch (Throwable e) { + log.error("T2: error " + e); error.set(e); break; } diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/ReactiveTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/ReactiveTests.java index 1726b22e..259b9adf 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/ReactiveTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/ReactiveTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 2019-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,8 +21,10 @@ import static org.hamcrest.Matchers.containsInAnyOrder; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Configuration; import org.springframework.messaging.Message; @@ -60,6 +62,7 @@ public class ReactiveTests extends AbstractStateMachineTests { @SuppressWarnings("unchecked") @Test + @Timeout(value = 10, unit = TimeUnit.SECONDS) public void testMonosAllAccepted() { context.register(Config1.class); context.refresh(); diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/TestUtils.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/TestUtils.java index db46bfb1..d7d2ea36 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/TestUtils.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/TestUtils.java @@ -21,6 +21,7 @@ import static org.springframework.statemachine.StateMachineSystemConstants.DEFAU import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.time.Duration; import org.springframework.beans.factory.BeanFactory; import org.springframework.messaging.Message; @@ -99,7 +100,8 @@ public class TestUtils { public static void doSendEventAndConsumeAll(StateMachine stateMachine, E event) { StepVerifier.create(stateMachine.sendEvent(eventAsMono(event))) .thenConsumeWhile(eventResult -> true) - .verifyComplete(); + .expectComplete() + .verify(Duration.ofSeconds(5)); } public static void doSendEventAndConsumeAll(StateMachine stateMachine, Message event) {