基于协程的状态机

Question

我有一个棘手而有趣的问题要问你。

在处理I/O任务时，例如通过Twisted、Tornado等某些传输层实现协议，我发现了一个类似的场景或模式。这种模式比抽象更普遍。例如，当您使用类似MODEM的设备时，您发送命令并接收结果。

然而，有时您需要对调制解调器对最后一个命令的响应做出反应，并使用新的命令。举个例子，假设调制解调器是M，->是通信运营商，它需要一个参数-消息密钥，并且服务器是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
    ...

看起来像是有限状态机的行为。希望能在使用流对象的非阻塞I/O（通过流对象）时，在tornado中实现此场景。 只需将跟踪方案作为输入，并覆盖描述输入中状态（事件）的处理程序，即可达到漂亮的有限状态机行为。

输入可能具有以下表示法：

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

这里的所有字母数字符号都是州名，每个键值对包含一个州名和可能的状态转换集合。

在tornado协程和futures中介绍的情况下，有哪些可能的FSM实现方式？请分享你们的想法和代码。

- Rustem K

---

@Veedrac 现在好一些了吗？ - Rustem K

Answer 1

我认为Twisted更适合协议实现。无论如何，在Python中，函数和方法都是一流对象，这意味着您可以将它们存储在字典中。您还可以使用functools.partial将带有参数的函数绑定到字典键上。您可以使用它来实现转换。每个状态应该是一个包含字典的函数，其中键是可能的输入状态，值是输出状态。然后，您可以轻松地从一个状态跳到另一个状态。为了利用Tornado循环的下一个状态，而不是直接调用，应该使用ioloop.IOLoop.instance().add_callback注册回调。

接受语言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"

使用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

- Nykakin