Help with state de-duplication

[thedrow]

I have the following state machine:

class ServiceRestartState(Enum):
    starting = auto()
    stopping = auto()
    stopped = auto()

class ServiceState(Enum):
    initialized = auto()
    starting = auto()
    started = auto()
    restarting = ServiceRestartState
    stopping = auto()
    stopped = auto()

class Service(HierarchicalAsyncMachine):
    def __init__(self):
        transitions = [
            ['starting', [ServiceState.initialized, ServiceState.stopped], ServiceState.starting],
            ['started', [ServiceState.starting, ServiceRestartState.starting], ServiceState.started],
            ['restarting', ServiceState.started, 'restarting'],
            ['stopping', 'restarting', ServiceRestartState.stopping],
            ['stopped', ServiceRestartState.stopping, ServiceRestartState.stopped],
            ['starting', ServiceRestartState.stopped, ServiceRestartState.starting],
            ['stopping', ServiceState.started, ServiceState.stopping],
            ['stopped', ServiceState.stopping, ServiceState.stopped],
        ]
        super().__init__(states=ServiceState,
                         transitions=transitions,
                         initial=ServiceState.initialized,
                         auto_transitions=False)

The problem with it is that ServiceState and ServiceRestartState do not share the same callbacks when they are essentially in the same state. I looked into parallel states and they do not seem to do what I want.

What I want is to have a state restarting which may or may not be active in parallel to the starting, started, stopping, and stopped states.

Is there a way to currently define such a state machine? Should I just introduce a flag in the class instead?

[aleneum]

I am not sure I got this completely right. As far as I can tell you want to reference states instead of copying them. An instance of ServiceRestartState should trigger the same callbacks as a nested state in ServiceState. When you pass enums, they are currently treated just as strings in the sense that they act as blueprints for the state creation. You can either pass pre-constructed states to your Service or pass an instance of the initialized ServiceRestartState machine.

from enum import Enum, auto
from transitions.extensions import HierarchicalAsyncMachine
import asyncio

class ServiceRestartState(Enum):
    starting = auto()
    stopping = auto()
    stopped = auto()

class ServiceState(Enum):
    initialized = auto()
    starting = auto()
    started = auto()
    stopping = auto()
    stopped = auto()

def on_restarting():
    print("restarting")

class Service(HierarchicalAsyncMachine):
    def __init__(self):
        transitions = [
            ['starting', [ServiceState.initialized, ServiceState.stopped], ServiceState.starting],
            ['started', [ServiceState.starting, ServiceRestartState.starting], ServiceState.started],
        ]
        super().__init__(states=ServiceState,
                         transitions=transitions,
                         initial=ServiceState.initialized,
                         auto_transitions=False)
        self.init_restarting()

    def init_restarting(self):
        self.nested = HierarchicalAsyncMachine(states=ServiceRestartState, initial=ServiceRestartState.starting)
        self.nested.on_enter_starting(on_restarting)
        self.add_states(dict(name='restarting', children=self.nested))
        self.add_transition('restarting', ServiceState.started, 'restarting')
        self.add_transition('stopping', 'restarting', ServiceRestartState.stopping)

async def main():
    s = Service()
    print("Run Service ...")
    await s.starting()
    await s.started()
    await s.restarting()
    await s.stopping()
    assert s.state == ServiceRestartState.stopping
    print("Test nested ...")
    await s.nested.to_starting()

asyncio.run(main())

[thedrow]

Let me rephrase:

A service can transition to starting if it is either stopped or initialized. It can transition to restarting if it is already at the started state. At that stage, all callbacks that prepare the service for a restart will be called. Afterward, the state machine will automatically progress to the stopping state using self.add_transition(..., after=self.stopping). The same callbacks should be applied when stopping occurs but additional callbacks may also be applied since we're both in the restarting state and the stopping state.

What I think I'm asking for is a way to specify an optional parallel state that is triggered in each step of the restart process but isn't triggered during normal startup/shutdown.

The service should return True for both self.is_restarting() and self.is_stopping() if the service is currently restarting and False for self.is_restarting() but True for self.is_stopping() if the service is simply stopping.

I used nested states before since I couldn't find a way to express this correctly.

[aleneum]

Hmm.. you can skip initial states of nested states and override is_state to deal with parallel states.

from transitions.extensions import HierarchicalMachine
from transitions.core import listify

states = [
    'initialized',
    {'name': 'busy', 'parallel': [{
        'name': 'booting',
        'children': ['stopping', 'stopped', 'starting'],
        'initial': 'stopping',
        'transitions': [
            ['stopped', 'stopping', 'stopped'],
            ['start', 'stopped', 'starting']
        ]
    }, 'restarting']},
    'started'
]

transitions = [
    ['start', 'initialized', 'busy_booting_starting'],
    ['started', 'busy_booting_starting', 'started'],
    ['start', 'started', 'busy']
]

class HSM(HierarchicalMachine):

    def is_state(self, state_name, model):
        return any([current == state_name for current in listify(getattr(model, self.model_attribute))])

    def on_enter_busy_booting(self):
        print("-> booting")

    def on_enter_busy_restarting(self):
        print("-> restarting")

s = HSM(states=states, transitions=transitions, initial='initialized')
s.start()       # >> -> booting
print(s.state)  # >> busy_booting_starting
assert s.is_busy_booting_starting() and not s.is_busy_restarting()
s.started()
print(s.state)  # >> started
s.start()       # >> -> booting -> restarting
print(s.state)  # >> ('busy_booting_stopping', 'busy_restarting')
s.stopped()
print(s.state)  # >> ('busy_booting_stopped', 'busy_restarting')
s.start()
print(s.state)  # >> ('busy_booting_starting', 'busy_restarting')
assert s.is_busy_booting_starting() and s.is_busy_restarting()
s.started()
print(s.state)  # >>  started

[thedrow]

Thanks!