From 8d7fe80bdb2da953aa3b8028719bba7fb1a6e712 Mon Sep 17 00:00:00 2001 From: Janne Valkealahti Date: Fri, 19 May 2017 07:52:31 +0100 Subject: [PATCH] NPE on sendEvent - Adding null check for current state - Remove synchronization from scheduleEventQueueProcessing method in executor. Looks like this sync is not really needed and indeed may cause jvm level deadlocks if threads are used for execution. - Change how some internals in AbstractStateMachine are synchronized to limit changes for deadlock. - Relates to #307 - Backport of #359 #360 --- .../buildtests/TimerSmokeTests.java | 143 ++++++++++++++++++ .../support/AbstractStateMachine.java | 66 ++++---- .../support/DefaultStateMachineExecutor.java | 2 +- .../action/ActionAndTimerTests.java | 96 +++++++++++- 4 files changed, 274 insertions(+), 33 deletions(-) create mode 100644 spring-statemachine-build-tests/src/test/java/org/springframework/statemachine/buildtests/TimerSmokeTests.java diff --git a/spring-statemachine-build-tests/src/test/java/org/springframework/statemachine/buildtests/TimerSmokeTests.java b/spring-statemachine-build-tests/src/test/java/org/springframework/statemachine/buildtests/TimerSmokeTests.java new file mode 100644 index 00000000..3beeb7ce --- /dev/null +++ b/spring-statemachine-build-tests/src/test/java/org/springframework/statemachine/buildtests/TimerSmokeTests.java @@ -0,0 +1,143 @@ +/* + * Copyright 2017 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.statemachine.buildtests; + +import org.junit.Test; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.config.StateMachineBuilder; +import org.springframework.statemachine.test.StateMachineTestPlan; +import org.springframework.statemachine.test.StateMachineTestPlanBuilder; + +public class TimerSmokeTests { + + private static ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); + { + taskExecutor.initialize(); + } + + private StateMachine buildMachine() throws Exception { + + StateMachineBuilder.Builder builder = StateMachineBuilder.builder(); + + builder.configureConfiguration() + .withConfiguration() + .taskExecutor(taskExecutor); + + builder.configureStates() + .withStates() + .initial("initial") + .end("end"); + + builder.configureTransitions() + .withExternal() + .source("initial") + .target("end") + .timerOnce(30) + .and() + .withLocal() + .source("initial") + .event("repeate"); + + return builder.build(); + } + + private StateMachine buildMachine2() throws Exception { + + StateMachineBuilder.Builder builder = StateMachineBuilder.builder(); + + builder.configureConfiguration() + .withConfiguration() + .taskExecutor(taskExecutor); + + builder.configureStates() + .withStates() + .initial("initial").end("end").and() + .withStates().parent("initial").initial("inner"); + + builder.configureTransitions() + .withExternal() + .source("initial") + .target("end") + .timerOnce(30) + .and() + .withExternal() + .source("inner") + .target("end") + .timerOnce(15) + .and() + .withLocal() + .source("inner") + .event("repeate"); + + return builder.build(); + } + + @Test + public void testNPE() throws Exception { + StateMachine stateMachine; + for (int i = 0; i < 20; i++) { + stateMachine = buildMachine(); + stateMachine.start(); + while (!stateMachine.isComplete()) { + stateMachine.sendEvent("repeate"); + } + } + } + + @Test + public void testNPE2() throws Exception { + + StateMachine stateMachine; + + for (int i = 0; i < 20; i++) { + stateMachine = buildMachine2(); + stateMachine.start(); + while(!stateMachine.isComplete()) { + stateMachine.sendEvent("repeate"); + } + stateMachine.stop(); + } + } + + @Test + public void testDeadlock() throws Exception { + StateMachineTestPlan plan; + for (int i = 0; i < 20; i++) { + plan = StateMachineTestPlanBuilder. builder() + .defaultAwaitTime(1) + .stateMachine(buildMachine()) + .step() + .expectStateMachineStarted(1) + .expectStateEntered(1) + .expectStateEntered("initial") + .and() + .step() + .sendEvent("repeate") + .expectStates("initial") + .and() + .step() + .expectStateEntered(1) + .expectStateEntered("end") + .and() + .step() + .expectStateMachineStopped(1) + .and() + .build(); + plan.test(); + } + } +} diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java index aca39024..2a8aea9e 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java @@ -116,6 +116,7 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo private volatile Message forwardedInitialEvent; private final Object lock = new Object(); + private final Object lock2 = new Object(); private StateMachine parentMachine; @@ -211,30 +212,9 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo @Override public boolean sendEvent(Message event) { - if (hasStateMachineError()) { - // TODO: should we throw exception? - notifyEventNotAccepted(buildStateContext(Stage.EVENT_NOT_ACCEPTED, event, null, getRelayStateMachine(), getState(), null)); - return false; + synchronized (lock2) { + return sendEventInternal(event); } - - try { - event = getStateMachineInterceptors().preEvent(event, this); - } catch (Exception e) { - log.info("Event " + event + " threw exception in interceptors, not accepting event"); - notifyEventNotAccepted(buildStateContext(Stage.EVENT_NOT_ACCEPTED, event, null, getRelayStateMachine(), getState(), null)); - return false; - } - - if (isComplete() || !isRunning()) { - notifyEventNotAccepted(buildStateContext(Stage.EVENT_NOT_ACCEPTED, event, null, getRelayStateMachine(), getState(), null)); - return false; - } - boolean accepted = acceptEvent(event); - stateMachineExecutor.execute(); - if (!accepted) { - notifyEventNotAccepted(buildStateContext(Stage.EVENT_NOT_ACCEPTED, event, null, getRelayStateMachine(), getState(), null)); - } - return accepted; } @Override @@ -555,6 +535,33 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo forwardedInitialEvent = message; } + private boolean sendEventInternal(Message event) { + if (hasStateMachineError()) { + // TODO: should we throw exception? + notifyEventNotAccepted(buildStateContext(Stage.EVENT_NOT_ACCEPTED, event, null, getRelayStateMachine(), getState(), null)); + return false; + } + + try { + event = getStateMachineInterceptors().preEvent(event, this); + } catch (Exception e) { + log.info("Event " + event + " threw exception in interceptors, not accepting event"); + notifyEventNotAccepted(buildStateContext(Stage.EVENT_NOT_ACCEPTED, event, null, getRelayStateMachine(), getState(), null)); + return false; + } + + if (isComplete() || !isRunning()) { + notifyEventNotAccepted(buildStateContext(Stage.EVENT_NOT_ACCEPTED, event, null, getRelayStateMachine(), getState(), null)); + return false; + } + boolean accepted = acceptEvent(event); + stateMachineExecutor.execute(); + if (!accepted) { + notifyEventNotAccepted(buildStateContext(Stage.EVENT_NOT_ACCEPTED, event, null, getRelayStateMachine(), getState(), null)); + } + return accepted; + } + private StateMachine getRelayStateMachine() { return relay != null ? relay : this; } @@ -758,7 +765,7 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo State source = transition.getSource(); Trigger trigger = transition.getTrigger(); - if (StateMachineUtils.containsAtleastOne(source.getIds(), currentState.getIds())) { + if (currentState != null && StateMachineUtils.containsAtleastOne(source.getIds(), currentState.getIds())) { if (trigger != null && trigger.evaluate(new DefaultTriggerContext(message.getPayload()))) { stateMachineExecutor.queueEvent(message); return true; @@ -924,11 +931,18 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo return null; } - synchronized void setCurrentState(State state, Message message, Transition transition, boolean exit, StateMachine stateMachine) { + void setCurrentState(State state, Message message, Transition transition, boolean exit, StateMachine stateMachine) { setCurrentState(state, message, transition, exit, stateMachine, null, null); } - synchronized void setCurrentState(State state, Message message, Transition transition, boolean exit, + void setCurrentState(State state, Message message, Transition transition, boolean exit, + StateMachine stateMachine, Collection> sources, Collection> targets) { + synchronized (lock2) { + setCurrentStateInternal(state, message, transition, exit, stateMachine, sources, targets); + } + } + + private void setCurrentStateInternal(State state, Message message, Transition transition, boolean exit, StateMachine stateMachine, Collection> sources, Collection> targets) { State findDeep = findDeepParent(state); boolean isTargetSubOf = false; diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java index 2fd87fd8..acf465ec 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java @@ -263,7 +263,7 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im stateMachineExecutorTransit.transit(tran, stateContext, queuedMessage); } - private synchronized void scheduleEventQueueProcessing() { + private void scheduleEventQueueProcessing() { TaskExecutor executor = getTaskExecutor(); if (executor == null) { return; diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/action/ActionAndTimerTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/action/ActionAndTimerTests.java index 177fd753..b02c4c17 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/action/ActionAndTimerTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/action/ActionAndTimerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -27,6 +27,8 @@ import org.junit.Test; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.statemachine.AbstractStateMachineTests; import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; @@ -53,15 +55,48 @@ public class ActionAndTimerTests extends AbstractStateMachineTests { assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S1)); machine.sendEvent(TestEvents.E1); assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S2)); - // sleep so that action with timerOnce(1000) is fired before event is send - // event sending is happening on a main thread, but actions are executed on a same - // pool than the DefaultStateMachineExecutor is using. existing action execution is interrupted. - Thread.sleep(2000); + + assertThat(testTimerAction.latch.await(4, TimeUnit.SECONDS), is(true)); + assertThat(testTimerAction.e, nullValue()); + + // need to sleep for TimerTrigger not causing + // next event to get handled with threads, thus + // causing interrupt + Thread.sleep(1000); + machine.sendEvent(TestEvents.E2); assertThat(testListener.s3EnteredLatch.await(2, TimeUnit.SECONDS), is(true)); assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S3)); - assertThat(testTimerAction.latch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(testExitAction.latch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(testExitAction.e, nullValue()); + } + + @SuppressWarnings("unchecked") + @Test + public void testExitActionWithTimerOnceThreadPoolTaskScheduler() throws Exception { + context.register(Config2.class); + context.refresh(); + StateMachine machine = context.getBean(StateMachine.class); + TestTimerAction testTimerAction = context.getBean(TestTimerAction.class); + TestExitAction testExitAction = context.getBean(TestExitAction.class); + TestListener testListener = new TestListener(); + machine.addStateListener(testListener); + machine.start(); + assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S1)); + machine.sendEvent(TestEvents.E1); + assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S2)); + + assertThat(testTimerAction.latch.await(4, TimeUnit.SECONDS), is(true)); assertThat(testTimerAction.e, nullValue()); + + // need to sleep for TimerTrigger not causing + // next event to get handled with threads, thus + // causing interrupt + Thread.sleep(1000); + + machine.sendEvent(TestEvents.E2); + assertThat(testListener.s3EnteredLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S3)); assertThat(testExitAction.latch.await(2, TimeUnit.SECONDS), is(true)); assertThat(testExitAction.e, nullValue()); } @@ -109,6 +144,55 @@ public class ActionAndTimerTests extends AbstractStateMachineTests { } } + @Configuration + @EnableStateMachine + static class Config2 extends EnumStateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial(TestStates.S1) + .state(TestStates.S2, null, testExitAction()) + .state(TestStates.S3); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) throws Exception { + transitions + .withExternal() + .source(TestStates.S1) + .target(TestStates.S2) + .event(TestEvents.E1) + .and() + .withExternal() + .source(TestStates.S2) + .target(TestStates.S3) + .event(TestEvents.E2) + .and() + .withInternal() + .source(TestStates.S2) + .action(testTimerAction()) + .timerOnce(1000); + } + + @Bean + public TaskScheduler taskScheduler() { + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + return taskScheduler; + } + + @Bean + public TestExitAction testExitAction() { + return new TestExitAction(); + } + + @Bean + public TestTimerAction testTimerAction() { + return new TestTimerAction(); + } + } + private static class TestListener extends StateMachineListenerAdapter { volatile CountDownLatch s3EnteredLatch = new CountDownLatch(1);