Coroutine based state machines

Question

I have a tricky and interesting question to You.

While working on I/O tasks such as protocol implementation via some transport layer in Twisted, Tornado, I found a similar scenario or pattern. The pattern is rather generic than abstract. For example, when you are working with MODEM-like device, you send him commands and receive the results.

However, sometimes you need to react on the response of a modem on last command with new command(s). For example, suppose that modem is M, -> is communication operator that takes one parameter, message key, and server is S.

    1. s ->(a) M
       1.1 M ->(b) S # modem reacts on `a` as `b`; so next we should send him command B
       1.2 M ->(c) S # modem responses on `a` as `c`; so next we should send him C
    2. s ->(b) M
       2.1 M ->(g) S
       2.2 M -> (f) S
       ...
       2.N M -> (x) S
    ...

So, it looks like FSM behaviour. It would be nice to implement this scenario in tornado, while working with non-blocking I/O (via streams objects). By simply providing tracking scenario as an input and overriding handlers to the states(events) described in input we can reach nice finite state machine behaviour.

Input may have the following notation:

{
  a: (b, c, d),
  b: (c, 'exit|silence'),
  c: (a, 'exit|silence'),
  d: (b)
}

where all these alphanumeric signs are state names. each key-value pair is state name and possible set of state transitions.

What is possible implementation of FSM using introduced in tornado coroutines and futures ? Please share your minds and code.

Answer 1

I think that Twisted is more suited for protocol implementations. Anyway, in Python functions and methods are objects of first class which means that you can store them inside of dictionaries. You can also use functools.partial to bind a function with arguments to a dictionary key. You can use it to implement transitions. Each state should be a function containing a dictionary in which keys are possible input states and values are output states. Then you can easily hoop from one state to another. To make use of Tornado loop next states, instead of be called directly, should be registered as a callback using ioloop.IOLoop.instance().add_callback.

An example implementation of automata accepting language a*b*c:

import errno
import functools
import socket
from tornado import ioloop, iostream

class Communicator(object):
    def connection_ready(self, sock, fd, events):
        while True:
            try:
                connection, address = sock.accept()
            except socket.error, e:
                if e[0] not in (errno.EWOULDBLOCK, errno.EAGAIN):
                    raise
                return
            connection.setblocking(0)
            self.stream = iostream.IOStream(connection)
            self.stream.read_until(delimiter='\n', callback=self.initial_state) 

    def initial_state(self, msg):
        msg = msg.rstrip()
        print "entering initial state with message: %s" % msg
        transitions = {
            'a' : functools.partial(ioloop.IOLoop.instance().add_callback, self.state_a, msg),
            'b' : functools.partial(ioloop.IOLoop.instance().add_callback, self.state_b, msg),
            'c' : functools.partial(ioloop.IOLoop.instance().add_callback, self.final_state, msg)
        }
        try:
            transitions[msg[0]]()
        except:
            self.stream.write("Aborted (wrong input)\n", self.stream.close)

    def state_a(self, msg):
        print "entering state a with message: %s" % msg
        transitions = {
            'a' : functools.partial(ioloop.IOLoop.instance().add_callback, self.stream.write, "got a\n", functools.partial(self.state_a, msg[1:])),
            'b' : functools.partial(ioloop.IOLoop.instance().add_callback, self.state_b, msg),
            'c' : functools.partial(ioloop.IOLoop.instance().add_callback, self.final_state, msg[1:])
        }
        try:
            transitions[msg[0]]()
        except:
            self.stream.write("Aborted (wrong input)\n", self.stream.close)

    def state_b(self, msg):
        print "entering state b with message: %s" % msg
        transitions = {
            'a' : functools.partial(ioloop.IOLoop.instance().add_callback, self.state_a, msg),
            'b' : functools.partial(ioloop.IOLoop.instance().add_callback, self.stream.write, "got b\n", functools.partial(self.state_a, msg[1:])),
            'c' : functools.partial(ioloop.IOLoop.instance().add_callback, self.final_state, msg[1:])}
        try:
            transitions[msg[0]]()
        except:
            self.stream.write("Aborted (wrong input)\n" , self.stream.close)

    def final_state(self, msg):
        print "entering final state with message: %s" % msg
        self.stream.write("Finished properly with message %s\n" % msg, self.stream.close)

if __name__ == '__main__':
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.setblocking(0)
    sock.bind(("", 8000))
    sock.listen(5000)

    communicator = Communicator()
    io_loop = ioloop.IOLoop.instance()
    callback = functools.partial(communicator.connection_ready, sock)
    io_loop.add_handler(sock.fileno(), callback, io_loop.READ)
    try:
        io_loop.start()
    except KeyboardInterrupt:
        io_loop.stop()
        print "exited cleanly"

Session using Netcat:

$ nc localhost 8000
aaaaa
got a
got a
got a
got a
got a
Aborted (wrong input)
$ nc localhost 8000
abababab
got a
got b
got a
got b
got a
got b
got a
got b
Aborted (wrong input)
$ nc localhost 8000
aaabbbc
got a
got a
got a
got b
got b
got b
Finished properly with message 
$ nc localhost 8000
abcabc
got a
got b
Finished properly with message abc