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 def answer(connection_id, msg):
115 self.send(msg, connection_id)
120 command = self.parser.parse(input_)
122 answer(connection_id, 'UNHANDLED INPUT')
124 if 'connection_id' in list(signature(command).parameters):
125 command(connection_id=connection_id)
129 with open(self.game_file_name, 'a') as f:
130 f.write(input_ + '\n')
131 except parser.ArgError as e:
132 answer(connection_id, 'ARGUMENT ERROR: ' + str(e))
133 except server_.game.GameError as e:
134 answer(connection_id, 'GAME ERROR: ' + str(e))
136 def send(self, msg, connection_id=None):
138 self.queues_out[connection_id].put(msg)
140 for connection_id in self.queues_out:
141 self.queues_out[connection_id].put(msg)
143 def send_gamestate(self, connection_id=None):
144 """Send out game state data relevant to clients."""
146 def stringify_yx(tuple_):
147 """Transform tuple (y,x) into string 'Y:'+str(y)+',X:'+str(x)."""
148 return 'Y:' + str(tuple_[0]) + ',X:' + str(tuple_[1])
151 """Quote & escape string so client interprets it as single token."""
159 return ''.join(quoted)
161 self.send('NEW_TURN ' + str(self.world.turn))
162 self.send('MAP_SIZE ' + stringify_yx(self.world.map_.size))
163 visible_map = self.world.get_player().get_visible_map()
164 for y in range(self.world.map_.size[0]):
165 self.send('VISIBLE_MAP_LINE %5s "%s"' % (y, visible_map.get_line(y)))
166 visible_things = self.world.get_player().get_visible_things()
167 for thing in visible_things:
168 self.send('THING_TYPE %s %s' % (thing.id_, thing.type_))
169 self.send('THING_POS %s %s' % (thing.id_,
170 stringify_yx(thing.position)))
173 """Send turn finish signal, run game world, send new world data.
175 First sends 'TURN_FINISHED' message, then runs game world
176 until new player input is needed, then sends game state.
178 self.send('TURN_FINISHED ' + str(self.world.turn))
179 self.world.proceed_to_next_player_turn()
180 msg = str(self.world.get_player().last_task_result)
181 self.send('LAST_PLAYER_TASK_RESULT ' + msg)
182 self.send_gamestate()
184 def cmd_FIB(self, numbers, connection_id):
185 """Reply with n-th Fibonacci numbers, n taken from tokens[1:].
187 Numbers are calculated in parallel as far as possible, using fib().
188 A 'CALCULATING …' message is sent to caller before the result.
190 self.send('CALCULATING …', connection_id)
191 results = self.pool.map(fib, numbers)
192 reply = ' '.join([str(r) for r in results])
193 self.send(reply, connection_id)
194 cmd_FIB.argtypes = 'seq:int:nonneg'
196 def cmd_INC_P(self, connection_id):
197 """Increment world.turn, send game turn data to everyone.
199 To simulate game processing waiting times, a one second delay between
200 TURN_FINISHED and NEW_TURN occurs; after NEW_TURN, some expensive
201 calculations are started as pool processes that need to be finished
202 until a further INC finishes the turn.
204 This is just a demo structure for how the game loop could work when
205 parallelized. One might imagine a two-step game turn, with a non-action
206 step determining actor tasks (the AI determinations would take the
207 place of the fib calculations here), and an action step wherein these
208 tasks are performed (where now sleep(1) is).
210 from time import sleep
211 if self.pool_result is not None:
212 self.pool_result.wait()
213 self.send('TURN_FINISHED ' + str(self.world.turn))
216 self.send_gamestate()
217 self.pool_result = self.pool.map_async(fib, (35, 35))
220 def io_loop(q, commander):
221 """Handle commands coming through queue q, send results back.
223 Commands from q are expected to be tuples, with the first element either
224 'ADD_QUEUE', 'COMMAND', or 'KILL_QUEUE', the second element a UUID, and
225 an optional third element of arbitrary type. The UUID identifies a
226 receiver for replies.
228 An 'ADD_QUEUE' command should contain as third element a queue through
229 which to send messages back to the sender of the command. A 'KILL_QUEUE'
230 command removes the queue for that receiver from the list of queues through
231 which to send replies.
233 A 'COMMAND' command is specified in greater detail by a string that is the
234 tuple's third element. The commander CommandHandler takes care of processing
235 this and sending out replies.
241 content = None if len(x) == 2 else x[2]
242 if command_type == 'ADD_QUEUE':
243 commander.queues_out[connection_id] = content
244 elif command_type == 'COMMAND':
245 commander.handle_input(content, connection_id)
246 elif command_type == 'KILL_QUEUE':
247 del commander.queues_out[connection_id]
250 if len(sys.argv) != 2:
251 print('wrong number of arguments, expected one (game file)')
253 game_file_name = sys.argv[1]
254 commander = CommandHandler(game_file_name)
255 if os.path.exists(game_file_name):
256 if not os.path.isfile(game_file_name):
257 print('game file name does not refer to a valid game file')
259 with open(game_file_name, 'r') as f:
260 lines = f.readlines()
261 for i in range(len(lines)):
263 print("FILE INPUT LINE %s: %s" % (i, line), end='')
264 commander.handle_input(line, store=False)
266 commander.handle_input('MAP_SIZE Y:5,X:5')
267 commander.handle_input('TERRAIN_LINE 0 "xxxxx"')
268 commander.handle_input('TERRAIN_LINE 1 "x...x"')
269 commander.handle_input('TERRAIN_LINE 2 "x.X.x"')
270 commander.handle_input('TERRAIN_LINE 3 "x...x"')
271 commander.handle_input('TERRAIN_LINE 4 "xxxxx"')
272 commander.handle_input('THING_TYPE 0 human')
273 commander.handle_input('THING_POS 0 Y:3,X:3')
274 commander.handle_input('THING_TYPE 1 monster')
275 commander.handle_input('THING_POS 1 Y:1,X:1')
277 c = threading.Thread(target=io_loop, daemon=True, args=(q, commander))
279 server = Server(q, ('localhost', 5000), IO_Handler)
281 server.serve_forever()
282 except KeyboardInterrupt:
285 print('Killing server')
286 server.server_close()