7 # Avoid "Address already in use" errors.
8 socketserver.TCPServer.allow_reuse_address = True
11 class Server(socketserver.ThreadingTCPServer):
12 """Bind together threaded IO handling server and message queue."""
14 def __init__(self, queue, *args, **kwargs):
15 super().__init__(*args, **kwargs)
16 self.queue_out = queue
17 self.daemon_threads = True # Else, server's threads have daemon=False.
20 class IO_Handler(socketserver.BaseRequestHandler):
23 """Move messages between network socket and main thread via queues.
25 On start, sets up new queue, sends it via self.server.queue_out to
26 main thread, and from then on receives messages to send back from the
27 main thread via that new queue.
29 At the same time, loops over socket's recv to get messages from the
30 outside via self.server.queue_out into the main thread. Ends connection
31 once a 'QUIT' message is received from socket, and then also kills its
34 All messages to the main thread are tuples, with the first element a
35 meta command ('ADD_QUEUE' for queue creation, 'KILL_QUEUE' for queue
36 deletion, and 'COMMAND' for everything else), the second element a UUID
37 that uniquely identifies the thread (so that the main thread knows whom
38 to send replies back to), and optionally a third element for further
43 def caught_send(socket, message):
44 """Send message by socket, catch broken socket connection error."""
46 plom_socket_io.send(socket, message)
47 except plom_socket_io.BrokenSocketConnection:
50 def send_queue_messages(socket, queue_in, thread_alive):
51 """Send messages via socket from queue_in while thread_alive[0]."""
52 while thread_alive[0]:
54 msg = queue_in.get(timeout=1)
57 caught_send(socket, msg)
60 print('CONNECTION FROM:', str(self.client_address))
61 connection_id = uuid.uuid4()
62 queue_in = queue.Queue()
63 self.server.queue_out.put(('ADD_QUEUE', connection_id, queue_in))
65 t = threading.Thread(target=send_queue_messages,
66 args=(self.request, queue_in, thread_alive))
68 for message in plom_socket_io.recv(self.request):
70 caught_send(self.request, 'BAD MESSAGE')
71 elif 'QUIT' == message:
72 caught_send(self.request, 'BYE')
75 self.server.queue_out.put(('COMMAND', connection_id, message))
76 self.server.queue_out.put(('KILL_QUEUE', connection_id))
77 thread_alive[0] = False
78 print('CONNECTION CLOSED FROM:', str(self.client_address))
87 """Calculate n-th Fibonacci number. Very inefficiently."""
91 return fib(n-1) + fib(n-2)
96 def __init__(self, queues_out):
97 from multiprocessing import Pool
98 self.queues_out = queues_out
101 self.pool_result = None
103 def send_to(self, connection_id, msg):
104 """Send msg to client of connection_id."""
105 self.queues_out[connection_id].put(msg)
107 def send_all(self, msg):
108 """Send msg to all clients."""
109 for connection_id in self.queues_out:
110 self.send_to(connection_id, msg)
112 def cmd_fib(self, tokens, connection_id):
113 """Reply with n-th Fibonacci numbers, n taken from tokens[1:].
115 Numbers are calculated in parallel as far as possible, using fib().
116 A 'CALCULATING …' message is sent to caller before the result.
118 fib_fail = 'MALFORMED FIB REQUEST'
120 self.send_to(connection_id, fib_fail)
123 for token in tokens[1:]:
124 if token != '0' and token.isdigit():
125 numbers += [int(token)]
127 self.send_to(connection_id, fib_fail)
129 self.send_to(connection_id, 'CALCULATING …')
130 results = self.pool.map(fib, numbers)
131 reply = ' '.join([str(r) for r in results])
132 self.send_to(connection_id, reply)
134 def cmd_inc(self, connection_id):
135 """Increment world.turn, send TURN_FINISHED, NEW_TURN to everyone.
137 To simulate game processing waiting times, a one second delay between
138 TURN_FINISHED and NEW_TURN occurs; after NEW_TURN, some expensive
139 calculations are started as pool processes that need to be finished
140 until a further INC finishes the turn.
142 from time import sleep
143 if self.pool_result is not None:
144 self.pool_result.wait()
145 self.send_all('TURN_FINISHED ' + str(self.world.turn))
148 self.send_all('NEW_TURN ' + str(self.world.turn))
149 self.pool_result = self.pool.map_async(fib, (35,35))
151 def cmd_get_turn(self, connection_id):
152 """Send world.turn to caller."""
153 self.send_to(connection_id, str(self.world.turn))
155 def cmd_echo(self, tokens, input_, connection_id):
156 """Send message in input_ beyond tokens[0] to caller."""
157 msg = input_[len(tokens[0]) + 1:]
158 self.send_to(connection_id, msg)
160 def cmd_all(self, tokens, input_):
161 """Send message in input_ beyond tokens[0] to all clients."""
162 msg = input_[len(tokens[0]) + 1:]
165 def handle_input(self, input_, connection_id):
166 """Process input_ to command grammar, call command handler if found."""
167 tokens = [token for token in input_.split(' ') if len(token) > 0]
169 self.send_to(connection_id, 'EMPTY COMMAND')
170 elif len(tokens) == 1 and tokens[0] == 'INC':
171 self.cmd_inc(connection_id)
172 elif len(tokens) == 1 and tokens[0] == 'GET_TURN':
173 self.cmd_get_turn(connection_id)
174 elif len(tokens) >= 1 and tokens[0] == 'ECHO':
175 self.cmd_echo(tokens, input_, connection_id)
176 elif len(tokens) >= 1 and tokens[0] == 'ALL':
177 self.cmd_all(tokens, input_)
178 elif len(tokens) >= 1 and tokens[0] == 'FIB':
179 # TODO: Should this really block the whole loop?
180 self.cmd_fib(tokens, connection_id)
182 self.send_to(connection_id, 'UNKNOWN COMMAND')
186 """Handle commands coming through queue q, send results back.
188 Commands from q are expected to be tuples, with the first element either
189 'ADD_QUEUE', 'COMMAND', or 'KILL_QUEUE', the second element a UUID, and
190 an optional third element of arbitrary type. The UUID identifies a
191 receiver for replies.
193 An 'ADD_QUEUE' command should contain as third element a queue through
194 which to send messages back to the sender of the command. A 'KILL_QUEUE'
195 command removes the queue for that receiver from the list of queues through
196 which to send replies.
198 A 'COMMAND' command is specified in greater detail by a string that is the
199 tuple's third element. CommandHandler takes care of processing this and
203 command_handler = CommandHandler(queues_out)
208 content = None if len(x) == 2 else x[2]
209 if command_type == 'ADD_QUEUE':
210 queues_out[connection_id] = content
211 elif command_type == 'COMMAND':
212 command_handler.handle_input(content, connection_id)
213 elif command_type == 'KILL_QUEUE':
214 del queues_out[connection_id]
218 c = threading.Thread(target=io_loop, daemon=True, args=(q,))
220 server = Server(q, ('localhost', 5000), IO_Handler)
222 server.serve_forever()
223 except KeyboardInterrupt:
226 print('Killing server')
227 server.server_close()