Spring/spring-statemachine
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix concurrency in DefaultStateMachineExecutor

Browse Source 
- Change deferList from LinkedList to ConcurrentLinkedQueue
  to fix ConcurrentModificationException with parallel events.
- Backport #736
- Relates #578
This commit is contained in:
Janne Valkealahti
2019-05-10 07:01:11 +01:00
parent 8021a82239
commit 246e415563
2 changed files with 92 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java
View File

@@ -20,9 +20,8 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
@@ -72,7 +71,7 @@ public class DefaultStateMachineExecutor<S, E> extends LifecycleObjectSupport im
private final Queue<Message<E>> eventQueue = new ConcurrentLinkedQueue<Message<E>>();
private final LinkedList<Message<E>> deferList = new LinkedList<Message<E>>();
private final Queue<Message<E>> deferList = new ConcurrentLinkedQueue<Message<E>>();
private final Queue<TriggerQueueItem> triggerQueue = new ConcurrentLinkedQueue<TriggerQueueItem>();
@@ -155,7 +154,7 @@ public class DefaultStateMachineExecutor<S, E> extends LifecycleObjectSupport im
if (log.isDebugEnabled()) {
log.debug("Deferring message " + message);
}
deferList.addLast(message);
deferList.add(message);
}
@Override
@@ -486,7 +485,7 @@ public class DefaultStateMachineExecutor<S, E> extends LifecycleObjectSupport im
if (log.isDebugEnabled()) {
log.debug("Process defer list, size=" + deferList.size());
}
ListIterator<Message<E>> iterator = deferList.listIterator();
Iterator<Message<E>> iterator = deferList.iterator();
State<S,E> currentState = stateMachine.getState();
while (iterator.hasNext()) {
Message<E> event = iterator.next();

99
spring-statemachine-core/src/test/java/org/springframework/statemachine/EventDeferTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015 the original author or authors.
* Copyright 2015-2019 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.
@@ -15,14 +15,17 @@
*/
package org.springframework.statemachine;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
import java.util.List;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -56,7 +59,7 @@ public class EventDeferTests extends AbstractStateMachineTests {
machine.sendEvent("E3");
machine.sendEvent("E1");
Object executor = TestUtils.readField("stateMachineExecutor", machine);
List<?> readField = TestUtils.readField("deferList", executor);
Collection<?> readField = TestUtils.readField("deferList", executor);
assertThat(readField.size(), is(1));
machine.sendEvent("E2");
assertThat(readField.size(), is(2));
@@ -81,12 +84,60 @@ public class EventDeferTests extends AbstractStateMachineTests {
machine.sendEvent("E1");
machine.sendEvent("E1");
Object executor = TestUtils.readField("stateMachineExecutor", machine);
List<?> readField = TestUtils.readField("deferList", executor);
Collection<?> readField = TestUtils.readField("deferList", executor);
assertThat(readField.size(), is(2));
machine.sendEvent("E2");
assertThat(readField.size(), is(3));
}
@Test
public void testDeferSmokeExecutorConcurrentModification() throws Exception {
context.register(Config5.class);
context.refresh();
@SuppressWarnings("unchecked")
StateMachine<String, String> machine = context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class);
machine.start();
assertThat(machine.getState().getIds(), contains("S1"));
machine.sendEvent("E2");
Object executor = TestUtils.readField("stateMachineExecutor", machine);
Collection<?> readField = TestUtils.readField("deferList", executor);
assertThat(readField.size(), is(1));
AtomicReference<Exception> error = new AtomicReference<>();
AtomicInteger i1 = new AtomicInteger();
Thread t1 = new Thread(() -> {
while(i1.incrementAndGet() < 1000) {
try {
machine.sendEvent("E1");
machine.sendEvent("E2");
} catch (Exception e) {
error.set(e);
break;
}
}
});
AtomicInteger i2 = new AtomicInteger();
Thread t2 = new Thread(() -> {
while(i2.incrementAndGet() < 1000) {
try {
machine.sendEvent("E1");
machine.sendEvent("E2");
} catch (Exception e) {
error.set(e);
break;
}
}
});
t1.start();
t2.start();
t1.join();
t2.join();
assertThat(error.get(), nullValue());
}
@Test
public void testDeferWithSubsSyncExecutor() throws Exception {
context.register(Config1.class);
@@ -108,7 +159,7 @@ public class EventDeferTests extends AbstractStateMachineTests {
machine.sendEvent("E1");
Object executor = TestUtils.readField("stateMachineExecutor", machine);
List<?> readField = TestUtils.readField("deferList", executor);
Collection<?> readField = TestUtils.readField("deferList", executor);
assertThat(readField.size(), is(1));
listener.reset(0, 0, 2, 0);
@@ -142,7 +193,7 @@ public class EventDeferTests extends AbstractStateMachineTests {
machine.sendEvent("E1");
Object executor = TestUtils.readField("stateMachineExecutor", machine);
List<?> readField = TestUtils.readField("deferList", executor);
Collection<?> readField = TestUtils.readField("deferList", executor);
assertThat(readField.size(), is(2));
listener.reset(0, 0, 3, 0);
@@ -203,7 +254,7 @@ public class EventDeferTests extends AbstractStateMachineTests {
// sub doesn't defer
machine.sendEvent("E15");
Object executor = TestUtils.readField("stateMachineExecutor", machine);
List<?> readField = TestUtils.readField("deferList", executor);
Collection<?> readField = TestUtils.readField("deferList", executor);
assertThat(readField.size(), is(0));
assertThat(machine.getState().getIds(), contains("SUB5"));
@@ -229,7 +280,7 @@ public class EventDeferTests extends AbstractStateMachineTests {
// sub defers
machine.sendEvent("E15");
Object executor = TestUtils.readField("stateMachineExecutor", machine);
List<?> readField = TestUtils.readField("deferList", executor);
Collection<?> readField = TestUtils.readField("deferList", executor);
assertThat(readField.size(), is(1));
assertThat(machine.getState().getIds(), contains("SUB1", "SUB12"));
@@ -260,7 +311,7 @@ public class EventDeferTests extends AbstractStateMachineTests {
// regions defers
machine.sendEvent("E3");
Object executor = TestUtils.readField("stateMachineExecutor", machine);
List<?> readField = TestUtils.readField("deferList", executor);
Collection<?> readField = TestUtils.readField("deferList", executor);
assertThat(readField.size(), is(0));
}
@@ -287,7 +338,7 @@ public class EventDeferTests extends AbstractStateMachineTests {
// regions defers
machine.sendEvent("E3");
Object executor = TestUtils.readField("stateMachineExecutor", machine);
List<?> readField = TestUtils.readField("deferList", executor);
Collection<?> readField = TestUtils.readField("deferList", executor);
assertThat(readField.size(), is(1));
}
@@ -308,7 +359,7 @@ public class EventDeferTests extends AbstractStateMachineTests {
// regions doesn't defer
machine.sendEvent("E3");
Object executor = TestUtils.readField("stateMachineExecutor", machine);
List<?> readField = TestUtils.readField("deferList", executor);
Collection<?> readField = TestUtils.readField("deferList", executor);
assertThat(readField.size(), is(0));
assertThat(machine.getState().getIds(), contains("SUB2"));
@@ -542,6 +593,32 @@ public class EventDeferTests extends AbstractStateMachineTests {
}
@Configuration
@EnableStateMachine
static class Config5 extends StateMachineConfigurerAdapter<String, String> {
@Override
public void configure(StateMachineStateConfigurer<String, String> states) throws Exception {
states
.withStates()
.initial("S1")
.state("S1", "E2")
.state("S2");
}
@Override
public void configure(StateMachineTransitionConfigurer<String, String> transitions) throws Exception {
transitions
.withExternal()
.source("S1").target("S2")
.event("E1")
.and()
.withExternal()
.source("S2").target("S1")
.event("E2");
}
}
@Configuration
static class ExecutorConfig {