13 # Avoid "Address already in use" errors.
14 socketserver.TCPServer.allow_reuse_address = True
17 class Server(socketserver.ThreadingTCPServer):
18 """Bind together threaded IO handling server and message queue."""
20 def __init__(self, queue, *args, **kwargs):
21 super().__init__(*args, **kwargs)
22 self.queue_out = queue
23 self.daemon_threads = True # Else, server's threads have daemon=False.
26 class IO_Handler(socketserver.BaseRequestHandler):
29 """Move messages between network socket and main thread via queues.
31 On start, sets up new queue, sends it via self.server.queue_out to
32 main thread, and from then on receives messages to send back from the
33 main thread via that new queue.
35 At the same time, loops over socket's recv to get messages from the
36 outside via self.server.queue_out into the main thread. Ends connection
37 once a 'QUIT' message is received from socket, and then also kills its
40 All messages to the main thread are tuples, with the first element a
41 meta command ('ADD_QUEUE' for queue creation, 'KILL_QUEUE' for queue
42 deletion, and 'COMMAND' for everything else), the second element a UUID
43 that uniquely identifies the thread (so that the main thread knows whom
44 to send replies back to), and optionally a third element for further
49 def caught_send(socket, message):
50 """Send message by socket, catch broken socket connection error."""
52 plom_socket_io.send(socket, message)
53 except plom_socket_io.BrokenSocketConnection:
56 def send_queue_messages(socket, queue_in, thread_alive):
57 """Send messages via socket from queue_in while thread_alive[0]."""
58 while thread_alive[0]:
60 msg = queue_in.get(timeout=1)
63 caught_send(socket, msg)
66 print('CONNECTION FROM:', str(self.client_address))
67 connection_id = uuid.uuid4()
68 queue_in = queue.Queue()
69 self.server.queue_out.put(('ADD_QUEUE', connection_id, queue_in))
71 t = threading.Thread(target=send_queue_messages,
72 args=(self.request, queue_in, thread_alive))
74 for message in plom_socket_io.recv(self.request):
76 caught_send(self.request, 'BAD MESSAGE')
77 elif 'QUIT' == message:
78 caught_send(self.request, 'BYE')
81 self.server.queue_out.put(('COMMAND', connection_id, message))
82 self.server.queue_out.put(('KILL_QUEUE', connection_id))
83 thread_alive[0] = False
84 print('CONNECTION CLOSED FROM:', str(self.client_address))
89 """Calculate n-th Fibonacci number. Very inefficiently."""
93 return fib(n-1) + fib(n-2)
96 class CommandHandler(game_common.Commander, server_.game.Commander):
98 def __init__(self, game_file_name):
100 self.world = server_.game.World()
101 self.parser = parser.Parser(self)
102 self.game_file_name = game_file_name
103 # self.pool and self.pool_result are currently only needed by the FIB
104 # command and the demo of a parallelized game loop in cmd_inc_p.
105 from multiprocessing import Pool
107 self.pool_result = None
109 def handle_input(self, input_, connection_id=None, store=True):
110 """Process input_ to command grammar, call command handler if found."""
111 from inspect import signature
113 command = self.parser.parse(input_)
115 self.send_to(connection_id, 'UNHANDLED INPUT')
117 if 'connection_id' in list(signature(command).parameters):
118 command(connection_id=connection_id)
122 with open(self.game_file_name, 'a') as f:
123 f.write(input_ + '\n')
124 except parser.ArgError as e:
125 self.send_to(connection_id, 'ARGUMENT ERROR: ' + str(e))
126 except server_.game.GameError as e:
127 self.send_to(connection_id, 'GAME ERROR: ' + str(e))
129 def send_to(self, connection_id, msg):
130 """Send msg to client of connection_id; if no later, print instead."""
132 self.queues_out[connection_id].put(msg)
136 def send_all(self, msg):
137 """Send msg to all clients."""
138 for connection_id in self.queues_out:
139 self.send_to(connection_id, msg)
141 def send_gamestate(self, connection_id=None):
142 """Send out game state data relevant to clients."""
143 from functools import partial
145 def stringify_yx(tuple_):
146 """Transform tuple (y,x) into string 'Y:'+str(y)+',X:'+str(x)."""
147 return 'Y:' + str(tuple_[0]) + ',X:' + str(tuple_[1])
150 """Quote & escape string so client interprets it as single token."""
158 return ''.join(quoted)
162 send=partial(self.send_to, connection_id)
163 send('NEW_TURN ' + str(self.world.turn))
164 send('MAP_SIZE ' + stringify_yx(self.world.map_size))
165 for y in range(self.world.map_size[0]):
166 width = self.world.map_size[1]
167 terrain_line = self.world.terrain_map[y * width:(y + 1) * width]
168 send('TERRAIN_LINE %5s %s' % (y, quoted(terrain_line)))
169 for thing in self.world.things:
170 send('THING_TYPE %s %s' % (thing.id_, thing.type_))
171 send('THING_POS %s %s' % (thing.id_,
172 stringify_yx(thing.position)))
175 """Send turn finish signal, run game world, send new world data.
177 First sends 'TURN_FINISHED' message, then runs game world
178 until new player input is needed, then sends game state.
180 self.send_all('TURN_FINISHED ' + str(self.world.turn))
181 self.world.proceed_to_next_player_turn()
182 msg = str(self.world.get_player().last_task_result)
183 self.send_all('LAST_PLAYER_TASK_RESULT ' + msg)
184 self.send_gamestate()
186 def cmd_FIB(self, numbers, connection_id):
187 """Reply with n-th Fibonacci numbers, n taken from tokens[1:].
189 Numbers are calculated in parallel as far as possible, using fib().
190 A 'CALCULATING …' message is sent to caller before the result.
192 self.send_to(connection_id, 'CALCULATING …')
193 results = self.pool.map(fib, numbers)
194 reply = ' '.join([str(r) for r in results])
195 self.send_to(connection_id, reply)
196 cmd_FIB.argtypes = 'seq:int:nonneg'
198 def cmd_INC_P(self, connection_id):
199 """Increment world.turn, send game turn data to everyone.
201 To simulate game processing waiting times, a one second delay between
202 TURN_FINISHED and NEW_TURN occurs; after NEW_TURN, some expensive
203 calculations are started as pool processes that need to be finished
204 until a further INC finishes the turn.
206 This is just a demo structure for how the game loop could work when
207 parallelized. One might imagine a two-step game turn, with a non-action
208 step determining actor tasks (the AI determinations would take the
209 place of the fib calculations here), and an action step wherein these
210 tasks are performed (where now sleep(1) is).
212 from time import sleep
213 if self.pool_result is not None:
214 self.pool_result.wait()
215 self.send_all('TURN_FINISHED ' + str(self.world.turn))
218 self.send_gamestate()
219 self.pool_result = self.pool.map_async(fib, (35, 35))
222 def io_loop(q, commander):
223 """Handle commands coming through queue q, send results back.
225 Commands from q are expected to be tuples, with the first element either
226 'ADD_QUEUE', 'COMMAND', or 'KILL_QUEUE', the second element a UUID, and
227 an optional third element of arbitrary type. The UUID identifies a
228 receiver for replies.
230 An 'ADD_QUEUE' command should contain as third element a queue through
231 which to send messages back to the sender of the command. A 'KILL_QUEUE'
232 command removes the queue for that receiver from the list of queues through
233 which to send replies.
235 A 'COMMAND' command is specified in greater detail by a string that is the
236 tuple's third element. The commander CommandHandler takes care of processing
237 this and sending out replies.
243 content = None if len(x) == 2 else x[2]
244 if command_type == 'ADD_QUEUE':
245 commander.queues_out[connection_id] = content
246 elif command_type == 'COMMAND':
247 commander.handle_input(content, connection_id)
248 elif command_type == 'KILL_QUEUE':
249 del commander.queues_out[connection_id]
252 if len(sys.argv) != 2:
253 print('wrong number of arguments, expected one (game file)')
255 game_file_name = sys.argv[1]
256 commander = CommandHandler(game_file_name)
257 if os.path.exists(game_file_name):
258 if not os.path.isfile(game_file_name):
259 print('game file name does not refer to a valid game file')
261 with open(game_file_name, 'r') as f:
262 lines = f.readlines()
263 for i in range(len(lines)):
265 print("FILE INPUT LINE %s: %s" % (i, line), end='')
266 commander.handle_input(line, store=False)
268 commander.handle_input('MAP_SIZE Y:5,X:5')
269 commander.handle_input('TERRAIN_LINE 0 "xxxxx"')
270 commander.handle_input('TERRAIN_LINE 1 "x...x"')
271 commander.handle_input('TERRAIN_LINE 2 "x.X.x"')
272 commander.handle_input('TERRAIN_LINE 3 "x...x"')
273 commander.handle_input('TERRAIN_LINE 4 "xxxxx"')
274 commander.handle_input('THING_TYPE 0 human')
275 commander.handle_input('THING_POS 0 Y:3,X:3')
276 commander.handle_input('THING_TYPE 1 monster')
277 commander.handle_input('THING_POS 1 Y:1,X:1')
279 c = threading.Thread(target=io_loop, daemon=True, args=(q, commander))
281 server = Server(q, ('localhost', 5000), IO_Handler)
283 server.serve_forever()
284 except KeyboardInterrupt:
287 print('Killing server')
288 server.server_close()