6 from parser import ArgError, Parser
9 # Avoid "Address already in use" errors.
10 socketserver.TCPServer.allow_reuse_address = True
13 class Server(socketserver.ThreadingTCPServer):
14 """Bind together threaded IO handling server and message queue."""
16 def __init__(self, queue, *args, **kwargs):
17 super().__init__(*args, **kwargs)
18 self.queue_out = queue
19 self.daemon_threads = True # Else, server's threads have daemon=False.
22 class IO_Handler(socketserver.BaseRequestHandler):
25 """Move messages between network socket and main thread via queues.
27 On start, sets up new queue, sends it via self.server.queue_out to
28 main thread, and from then on receives messages to send back from the
29 main thread via that new queue.
31 At the same time, loops over socket's recv to get messages from the
32 outside via self.server.queue_out into the main thread. Ends connection
33 once a 'QUIT' message is received from socket, and then also kills its
36 All messages to the main thread are tuples, with the first element a
37 meta command ('ADD_QUEUE' for queue creation, 'KILL_QUEUE' for queue
38 deletion, and 'COMMAND' for everything else), the second element a UUID
39 that uniquely identifies the thread (so that the main thread knows whom
40 to send replies back to), and optionally a third element for further
45 def caught_send(socket, message):
46 """Send message by socket, catch broken socket connection error."""
48 plom_socket_io.send(socket, message)
49 except plom_socket_io.BrokenSocketConnection:
52 def send_queue_messages(socket, queue_in, thread_alive):
53 """Send messages via socket from queue_in while thread_alive[0]."""
54 while thread_alive[0]:
56 msg = queue_in.get(timeout=1)
59 caught_send(socket, msg)
62 print('CONNECTION FROM:', str(self.client_address))
63 connection_id = uuid.uuid4()
64 queue_in = queue.Queue()
65 self.server.queue_out.put(('ADD_QUEUE', connection_id, queue_in))
67 t = threading.Thread(target=send_queue_messages,
68 args=(self.request, queue_in, thread_alive))
70 for message in plom_socket_io.recv(self.request):
72 caught_send(self.request, 'BAD MESSAGE')
73 elif 'QUIT' == message:
74 caught_send(self.request, 'BYE')
77 self.server.queue_out.put(('COMMAND', connection_id, message))
78 self.server.queue_out.put(('KILL_QUEUE', connection_id))
79 thread_alive[0] = False
80 print('CONNECTION CLOSED FROM:', str(self.client_address))
86 def __init__(self, name, args=(), kwargs={}):
95 def __init__(self, type_, position):
97 self.position = position
98 self.task = Task('wait')
103 def task_move(self, direction):
104 if direction == 'UP':
105 self.position[0] -= 1
106 elif direction == 'DOWN':
107 self.position[0] += 1
108 elif direction == 'RIGHT':
109 self.position[1] += 1
110 elif direction == 'LEFT':
111 self.position[1] -= 1
113 def decide_task(self):
114 if self.position[1] > 1:
115 self.set_task('move', 'LEFT')
116 elif self.position[1] < 3:
117 self.set_task('move', 'RIGHT')
119 self.set_task('wait')
121 def set_task(self, task, *args, **kwargs):
122 self.task = Task(task, args, kwargs)
124 def proceed(self, is_AI=True):
125 """Further the thing in its tasks.
127 Decrements .task.todo; if it thus falls to <= 0, enacts method whose
128 name is 'task_' + self.task.name and sets .task = None. If is_AI, calls
129 .decide_task to decide a self.task.
132 if self.task.todo <= 0:
133 task = getattr(self, 'task_' + self.task.name)
134 task(*self.task.args, **self.task.kwargs)
136 if is_AI and self.task is None:
144 self.map_size = (5, 5)
145 self.map_ = 'xxxxx\n' +\
150 self.things = [Thing('human', [3, 3]), Thing('monster', [1, 1])]
152 self.player = self.things[self.player_i]
156 """Calculate n-th Fibonacci number. Very inefficiently."""
160 return fib(n-1) + fib(n-2)
163 class CommandHandler:
165 def __init__(self, queues_out):
166 from multiprocessing import Pool
167 self.queues_out = queues_out
169 self.parser = Parser(self)
170 # self.pool and self.pool_result are currently only needed by the FIB
171 # command and the demo of a parallelized game loop in cmd_inc_p.
173 self.pool_result = None
175 def send_to(self, connection_id, msg):
176 """Send msg to client of connection_id."""
177 self.queues_out[connection_id].put(msg)
179 def send_all(self, msg):
180 """Send msg to all clients."""
181 for connection_id in self.queues_out:
182 self.send_to(connection_id, msg)
184 def stringify_yx(self, tuple_):
185 """Transform tuple (y,x) into string 'Y:'+str(y)+',X:'+str(x)."""
186 return 'Y:' + str(tuple_[0]) + ',X:' + str(tuple_[1])
188 def quoted(self, string):
189 """Quote and escape string so client interprets it as single token."""
197 return ''.join(quoted)
199 def send_all_gamestate(self):
200 """Send out game state data relevant to clients."""
201 self.send_all('NEW_TURN ' + str(self.world.turn))
202 self.send_all('MAP_SIZE ' + self.stringify_yx(self.world.map_size))
203 self.send_all('TERRAIN\n' + self.quoted(self.world.map_))
204 for thing in self.world.things:
205 self.send_all('THING TYPE:' + thing.type + ' '
206 + self.stringify_yx(thing.position))
208 def proceed_to_next_player_turn(self, connection_id):
209 """Run game world turns until player can decide their next step.
211 Sends a 'TURN_FINISHED' message, then iterates through all non-player
212 things, on each step furthering them in their tasks (and letting them
213 decide new ones if they finish). The iteration order is: first all
214 things that come after the player in the world things list, then (after
215 incrementing the world turn) all that come before the player; then the
216 player's .proceed() is run, and if it does not finish his task, the
217 loop starts at the beginning. Once the player's task is finished, the
218 loop breaks, and client-relevant game data is sent.
220 self.send_all('TURN_FINISHED ' + str(self.world.turn))
222 for thing in self.world.things[self.world.player_i+1:]:
225 for thing in self.world.things[:self.world.player_i]:
227 self.world.player.proceed(is_AI=False)
228 if self.world.player.task is None:
230 self.send_all_gamestate()
232 def cmd_FIB(self, numbers, connection_id):
233 """Reply with n-th Fibonacci numbers, n taken from tokens[1:].
235 Numbers are calculated in parallel as far as possible, using fib().
236 A 'CALCULATING …' message is sent to caller before the result.
238 self.send_to(connection_id, 'CALCULATING …')
239 results = self.pool.map(fib, numbers)
240 reply = ' '.join([str(r) for r in results])
241 self.send_to(connection_id, reply)
242 cmd_FIB.argtypes = 'seq:int:nonneg'
244 def cmd_INC_P(self, connection_id):
245 """Increment world.turn, send game turn data to everyone.
247 To simulate game processing waiting times, a one second delay between
248 TURN_FINISHED and NEW_TURN occurs; after NEW_TURN, some expensive
249 calculations are started as pool processes that need to be finished
250 until a further INC finishes the turn.
252 This is just a demo structure for how the game loop could work when
253 parallelized. One might imagine a two-step game turn, with a non-action
254 step determining actor tasks (the AI determinations would take the
255 place of the fib calculations here), and an action step wherein these
256 tasks are performed (where now sleep(1) is).
258 from time import sleep
259 if self.pool_result is not None:
260 self.pool_result.wait()
261 self.send_all('TURN_FINISHED ' + str(self.world.turn))
264 self.send_all_gamestate()
265 self.pool_result = self.pool.map_async(fib, (35, 35))
267 def cmd_GET_TURN(self, connection_id):
268 """Send world.turn to caller."""
269 self.send_to(connection_id, str(self.world.turn))
271 def cmd_MOVE(self, direction, connection_id):
272 """Set player task to 'move' with direction arg, finish player turn."""
273 if direction not in {'UP', 'DOWN', 'RIGHT', 'LEFT'}:
274 raise ArgError('Move argument must be one of: '
275 'UP, DOWN, RIGHT, LEFT')
276 self.world.player.set_task('move', direction=direction)
277 self.proceed_to_next_player_turn(connection_id)
278 cmd_MOVE.argtypes = 'string'
280 def cmd_WAIT(self, connection_id):
281 """Set player task to 'wait', finish player turn."""
282 self.world.player.set_task('wait')
283 self.proceed_to_next_player_turn(connection_id)
285 def cmd_ECHO(self, msg, connection_id):
286 """Send msg to caller."""
287 self.send_to(connection_id, msg)
288 cmd_ECHO.argtypes = 'string'
290 def cmd_ALL(self, msg, connection_id):
291 """Send msg to all clients."""
293 cmd_ALL.argtypes = 'string'
295 def handle_input(self, input_, connection_id):
296 """Process input_ to command grammar, call command handler if found."""
298 command = self.parser.parse(input_)
300 self.send_to(connection_id, 'UNHANDLED INPUT')
302 command(connection_id=connection_id)
303 except ArgError as e:
304 self.send_to(connection_id, 'ARGUMENT ERROR: ' + str(e))
308 """Handle commands coming through queue q, send results back.
310 Commands from q are expected to be tuples, with the first element either
311 'ADD_QUEUE', 'COMMAND', or 'KILL_QUEUE', the second element a UUID, and
312 an optional third element of arbitrary type. The UUID identifies a
313 receiver for replies.
315 An 'ADD_QUEUE' command should contain as third element a queue through
316 which to send messages back to the sender of the command. A 'KILL_QUEUE'
317 command removes the queue for that receiver from the list of queues through
318 which to send replies.
320 A 'COMMAND' command is specified in greater detail by a string that is the
321 tuple's third element. CommandHandler takes care of processing this and
325 command_handler = CommandHandler(queues_out)
330 content = None if len(x) == 2 else x[2]
331 if command_type == 'ADD_QUEUE':
332 queues_out[connection_id] = content
333 elif command_type == 'COMMAND':
334 command_handler.handle_input(content, connection_id)
335 elif command_type == 'KILL_QUEUE':
336 del queues_out[connection_id]
340 c = threading.Thread(target=io_loop, daemon=True, args=(q,))
342 server = Server(q, ('localhost', 5000), IO_Handler)
344 server.serve_forever()
345 except KeyboardInterrupt:
348 print('Killing server')
349 server.server_close()