Coroutine-based horizontal machines

Question

I have a difficult and interesting question for you.

While working on I / O tasks, such as implementing a protocol through some transport layer in Twisted, Tornado, I found a similar script or template. The pattern is rather general than abstract. For example, when you work with a MODEM device , you send commands to it and get the results.

However, sometimes you need to respond to the modem responding to the last command with a new command. For example, suppose a modem is M, → a communication operator that takes one parameter, a message key, and server 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.NM -> (x) S ... 

So, this is similar to the behavior of FSM. It would be nice to implement this scenario in a tornado, working with non-blocking I / O (through stream objects). By simply providing the tracking script as input and overriding handlers to the states (events) described in the input, we can achieve pleasant behavior on the final machine.

Input can have the following notation:

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

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

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

Answer 1

I think Twisted more suitable for implementing the protocol. Anyway, in Python, functions and methods are first class objects, which means you can store them inside 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 get engaged from one state to another. To use the Tornado loop, the following states, instead of a direct call, must be registered as a callback using ioloop.IOLoop.instance().add_callback .

An example implementation of automata that accept the 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