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_all_gamestate(self):
142 """Send out game state data relevant to clients."""
144 def stringify_yx(tuple_):
145 """Transform tuple (y,x) into string 'Y:'+str(y)+',X:'+str(x)."""
146 return 'Y:' + str(tuple_[0]) + ',X:' + str(tuple_[1])
149 """Quote & escape string so client interprets it as single token."""
157 return ''.join(quoted)
159 self.send_all('NEW_TURN ' + str(self.world.turn))
160 self.send_all('MAP_SIZE ' + stringify_yx(self.world.map_size))
161 for y in range(self.world.map_size[0]):
162 width = self.world.map_size[1]
163 terrain_line = self.world.terrain_map[y * width:(y + 1) * width]
164 self.send_all('TERRAIN_LINE %5s %s' % (y, quoted(terrain_line)))
165 for thing in self.world.things:
166 self.send_all('THING_TYPE %s %s' % (thing.id_, thing.type_))
167 self.send_all('THING_POS %s %s' % (thing.id_,
168 stringify_yx(thing.position)))
171 """Send turn finish signal, run game world, send new world data.
173 First sends 'TURN_FINISHED' message, then runs game world
174 until new player input is needed, then sends game state.
176 self.send_all('TURN_FINISHED ' + str(self.world.turn))
177 self.world.proceed_to_next_player_turn()
178 msg = str(self.world.get_player().last_task_result)
179 self.send_all('LAST_PLAYER_TASK_RESULT ' + msg)
180 self.send_all_gamestate()
182 def cmd_FIB(self, numbers, connection_id):
183 """Reply with n-th Fibonacci numbers, n taken from tokens[1:].
185 Numbers are calculated in parallel as far as possible, using fib().
186 A 'CALCULATING …' message is sent to caller before the result.
188 self.send_to(connection_id, 'CALCULATING …')
189 results = self.pool.map(fib, numbers)
190 reply = ' '.join([str(r) for r in results])
191 self.send_to(connection_id, reply)
192 cmd_FIB.argtypes = 'seq:int:nonneg'
194 def cmd_INC_P(self, connection_id):
195 """Increment world.turn, send game turn data to everyone.
197 To simulate game processing waiting times, a one second delay between
198 TURN_FINISHED and NEW_TURN occurs; after NEW_TURN, some expensive
199 calculations are started as pool processes that need to be finished
200 until a further INC finishes the turn.
202 This is just a demo structure for how the game loop could work when
203 parallelized. One might imagine a two-step game turn, with a non-action
204 step determining actor tasks (the AI determinations would take the
205 place of the fib calculations here), and an action step wherein these
206 tasks are performed (where now sleep(1) is).
208 from time import sleep
209 if self.pool_result is not None:
210 self.pool_result.wait()
211 self.send_all('TURN_FINISHED ' + str(self.world.turn))
214 self.send_all_gamestate()
215 self.pool_result = self.pool.map_async(fib, (35, 35))
218 def io_loop(q, commander):
219 """Handle commands coming through queue q, send results back.
221 Commands from q are expected to be tuples, with the first element either
222 'ADD_QUEUE', 'COMMAND', or 'KILL_QUEUE', the second element a UUID, and
223 an optional third element of arbitrary type. The UUID identifies a
224 receiver for replies.
226 An 'ADD_QUEUE' command should contain as third element a queue through
227 which to send messages back to the sender of the command. A 'KILL_QUEUE'
228 command removes the queue for that receiver from the list of queues through
229 which to send replies.
231 A 'COMMAND' command is specified in greater detail by a string that is the
232 tuple's third element. The commander CommandHandler takes care of processing
233 this and sending out replies.
239 content = None if len(x) == 2 else x[2]
240 if command_type == 'ADD_QUEUE':
241 commander.queues_out[connection_id] = content
242 elif command_type == 'COMMAND':
243 commander.handle_input(content, connection_id)
244 elif command_type == 'KILL_QUEUE':
245 del commander.queues_out[connection_id]
248 if len(sys.argv) != 2:
249 print('wrong number of arguments, expected one (game file)')
251 game_file_name = sys.argv[1]
252 commander = CommandHandler(game_file_name)
253 if os.path.exists(game_file_name):
254 if not os.path.isfile(game_file_name):
255 print('game file name does not refer to a valid game file')
257 with open(game_file_name, 'r') as f:
258 lines = f.readlines()
259 for i in range(len(lines)):
261 print("FILE INPUT LINE %s: %s" % (i, line), end='')
262 commander.handle_input(line, store=False)
264 commander.handle_input('MAP_SIZE Y:5,X:5')
265 commander.handle_input('TERRAIN_LINE 0 "xxxxx"')
266 commander.handle_input('TERRAIN_LINE 1 "x...x"')
267 commander.handle_input('TERRAIN_LINE 2 "x.X.x"')
268 commander.handle_input('TERRAIN_LINE 3 "x...x"')
269 commander.handle_input('TERRAIN_LINE 4 "xxxxx"')
270 commander.handle_input('THING_TYPE 0 human')
271 commander.handle_input('THING_POS 0 Y:3,X:3')
272 commander.handle_input('THING_TYPE 1 monster')
273 commander.handle_input('THING_POS 1 Y:1,X:1')
275 c = threading.Thread(target=io_loop, daemon=True, args=(q, commander))
277 server = Server(q, ('localhost', 5000), IO_Handler)
279 server.serve_forever()
280 except KeyboardInterrupt:
283 print('Killing server')
284 server.server_close()