Bugfix/task scheduler lifecycle

[RenanSFreitas]

ConcurrentTaskScheduler doesn't expose any way for it's inner executor to be shutdown, when it's instance is created with the default constructor. This can lead to a scenario where a library user ends up with many unused threads with no clear cues about where they came from. The unmanaged creation of those threads can result in an OutOfMemoryError for a client application.

This commit replaces the usage of ConcurrentTaskScheduler with ThreadPoolTaskScheduler, a type of similar semantics but better API, which includes methods for managing its internal resources lifecycle. Also it implements spring-beans interfaces which makes it more suitable to be managed by a Spring application context.

• Fixes https://github.com/spring-projects/spring-statemachine/issues/624

[RenanSFreitas]

This pull requests aims to fix the scenario exemplified in the code below, where usage of spring-statemachine 2.5.1 leads to a hanging JVM with many spawned threads. Increasing the number of iterations in the main loop results in more threads being created, which can result in an OOM scenario.

package com.example.springstatemachinethreadpoolbug;

import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.statemachine.StateMachine;
import org.springframework.statemachine.config.StateMachineBuilder;
import org.springframework.statemachine.config.configurers.StateConfigurer;
import org.springframework.statemachine.support.DefaultStateMachineContext;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;

@SpringBootApplication
public class Application implements CommandLineRunner {
    enum StateMachineEvent {
        E0, E1, E2
    }

    enum StateMachineState {
        S0, S1, S2
    }
    static final EnumSet<StateMachineState> ALL_STATES = EnumSet.allOf(StateMachineState.class);

    public static void main(String[] args) {
        final SpringApplication app = new SpringApplication(Application.class);
        app.run();
    }

    @Override
    public void run(final String... args) {
        runTestCase();
    }

    private class Iteration {
        StateMachineState iterationInitialState;
        StateMachineEvent iterationInitialEvent;
        StateMachineEvent iterationEventToPush;

        public Iteration(final StateMachineState iterationInitialState,
                         final StateMachineEvent iterationInitialEvent,
                         final StateMachineEvent iterationEventToPush) {
            this.iterationInitialState = iterationInitialState;
            this.iterationInitialEvent = iterationInitialEvent;
            this.iterationEventToPush = iterationEventToPush;
        }
    }
    private void runTestCase() {
        final var beforeThreadCount = Thread.activeCount();

        // Do the same thing a few times, to demonstrate the increase in the thread count
        for (int i = 0; i < 100; i++) {
            for (final var iteration : List.of(
                    new Iteration(StateMachineState.S0, StateMachineEvent.E0, StateMachineEvent.E0),
                    new Iteration(StateMachineState.S1, StateMachineEvent.E1, StateMachineEvent.E1)
            )) {
                final var iterationInitialState = iteration.iterationInitialState;
                final var iterationInitialEvent = iteration.iterationInitialEvent;
                final var stateMachine = buildStateMachine(iterationInitialState, iterationInitialEvent);
                stateMachine.start();
                stateMachine.sendEvent(MessageBuilder.withPayload(iteration.iterationEventToPush)
                        .setHeader("HEADER_A", "HEADER_A_VALUE")
                        .build());
                stateMachine.stop();
            }
        }

        System.out.println("Before thread count: " + beforeThreadCount);
        System.out.println("After thread count: " + Thread.activeCount());
    }

    public StateMachine<StateMachineState, StateMachineEvent> buildStateMachine(StateMachineState state, StateMachineEvent event) {
        try {
            final StateMachineBuilder.Builder<StateMachineState, StateMachineEvent> builder = buildStateMachine();

            final StateConfigurer<StateMachineState, StateMachineEvent> stateConfigurer = builder.configureStates()
                    .withStates()
                    .initial(StateMachineState.S0)
                    .states(ALL_STATES)
                    .end(StateMachineState.S2);

            builder.configureTransitions()
                    .withExternal()
                    .source(StateMachineState.S0)
                    .target(StateMachineState.S1)
                    .event(StateMachineEvent.E0)
                    .and()
                    .withExternal()
                    .source(StateMachineState.S1)
                    .target(StateMachineState.S2)
                    .event(StateMachineEvent.E1);

            stateConfigurer.state(StateMachineState.S2, context -> System.out.println("State machine action at " + StateMachineState.S2));
            // it happens with stateDo too
//          stateConfigurer.stateDo(StateMachineState.STATE_2, context -> System.out.println("State machine action at " + StateMachineState.S2));

            final var stateMachine = builder.build();

            stateMachine.getStateMachineAccessor()
                    .doWithAllRegions((function) -> function.resetStateMachine(new DefaultStateMachineContext<>(state,
                            event,
                            Map.of(),
                            stateMachine.getExtendedState(),
                            null,
                            stateMachine.getId()))
                    );

            return stateMachine;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private StateMachineBuilder.Builder<StateMachineState, StateMachineEvent> buildStateMachine()
            throws Exception {
        final StateMachineBuilder.Builder<StateMachineState, StateMachineEvent> result = StateMachineBuilder.builder();

        result.configureConfiguration()
                .withConfiguration()
                .autoStartup(false)
                .machineId("STATE_MACHINE_ID");

        return result;
    }
}

[pdalfarr]

@RenanSFreitas Did you receive any answer about this PR ?