plomlompom.com Git - plomrogue2-experiments/blob - server.py

   1 #!/usr/bin/env python3
   2
   3 import socketserver
   4 import threading
   5 import queue
   6 import sys
   7 import os
   8 from parser import ArgError, Parser
   9 from server_.game import World, GameError
  10
  11
  12 # Avoid "Address already in use" errors.
  13 socketserver.TCPServer.allow_reuse_address = True
  14
  15
  16 class Server(socketserver.ThreadingTCPServer):
  17     """Bind together threaded IO handling server and message queue."""
  18
  19     def __init__(self, queue, *args, **kwargs):
  20         super().__init__(*args, **kwargs)
  21         self.queue_out = queue
  22         self.daemon_threads = True  # Else, server's threads have daemon=False.
  23
  24
  25 class IO_Handler(socketserver.BaseRequestHandler):
  26
  27     def handle(self):
  28         """Move messages between network socket and main thread via queues.
  29
  30         On start, sets up new queue, sends it via self.server.queue_out to
  31         main thread, and from then on receives messages to send back from the
  32         main thread via that new queue.
  33
  34         At the same time, loops over socket's recv to get messages from the
  35         outside via self.server.queue_out into the main thread. Ends connection
  36         once a 'QUIT' message is received from socket, and then also kills its
  37         own queue.
  38
  39         All messages to the main thread are tuples, with the first element a
  40         meta command ('ADD_QUEUE' for queue creation, 'KILL_QUEUE' for queue
  41         deletion, and 'COMMAND' for everything else), the second element a UUID
  42         that uniquely identifies the thread (so that the main thread knows whom
  43         to send replies back to), and optionally a third element for further
  44         instructions.
  45         """
  46         import plom_socket_io
  47
  48         def caught_send(socket, message):
  49             """Send message by socket, catch broken socket connection error."""
  50             try:
  51                 plom_socket_io.send(socket, message)
  52             except plom_socket_io.BrokenSocketConnection:
  53                 pass
  54
  55         def send_queue_messages(socket, queue_in, thread_alive):
  56             """Send messages via socket from queue_in while thread_alive[0]."""
  57             while thread_alive[0]:
  58                 try:
  59                     msg = queue_in.get(timeout=1)
  60                 except queue.Empty:
  61                     continue
  62                 caught_send(socket, msg)
  63
  64         import uuid
  65         print('CONNECTION FROM:', str(self.client_address))
  66         connection_id = uuid.uuid4()
  67         queue_in = queue.Queue()
  68         self.server.queue_out.put(('ADD_QUEUE', connection_id, queue_in))
  69         thread_alive = [True]
  70         t = threading.Thread(target=send_queue_messages,
  71                              args=(self.request, queue_in, thread_alive))
  72         t.start()
  73         for message in plom_socket_io.recv(self.request):
  74             if message is None:
  75                 caught_send(self.request, 'BAD MESSAGE')
  76             elif 'QUIT' == message:
  77                 caught_send(self.request, 'BYE')
  78                 break
  79             else:
  80                 self.server.queue_out.put(('COMMAND', connection_id, message))
  81         self.server.queue_out.put(('KILL_QUEUE', connection_id))
  82         thread_alive[0] = False
  83         print('CONNECTION CLOSED FROM:', str(self.client_address))
  84         self.request.close()
  85
  86
  87 def fib(n):
  88     """Calculate n-th Fibonacci number. Very inefficiently."""
  89     if n in (1, 2):
  90         return 1
  91     else:
  92         return fib(n-1) + fib(n-2)
  93
  94
  95 class CommandHandler:
  96
  97     def __init__(self):
  98         from multiprocessing import Pool
  99         self.queues_out = {}
 100         self.world = World()
 101         self.parser = Parser(self)
 102         # self.pool and self.pool_result are currently only needed by the FIB
 103         # command and the demo of a parallelized game loop in cmd_inc_p.
 104         self.pool = Pool()
 105         self.pool_result = None
 106
 107     def handle_input(self, input_, connection_id=None, abort_on_error=False):
 108         """Process input_ to command grammar, call command handler if found."""
 109         try:
 110             command = self.parser.parse(input_)
 111             if command is None:
 112                 self.send_to(connection_id, 'UNHANDLED INPUT')
 113             else:
 114                 command(connection_id=connection_id)
 115         except ArgError as e:
 116             self.send_to(connection_id, 'ARGUMENT ERROR: ' + str(e))
 117             if abort_on_error:
 118                 exit(1)
 119         except GameError as e:
 120             self.send_to(connection_id, 'GAME ERROR: ' + str(e))
 121             if abort_on_error:
 122                 exit(1)
 123
 124     def send_to(self, connection_id, msg):
 125         """Send msg to client of connection_id; if no later, print instead."""
 126         if connection_id:
 127             self.queues_out[connection_id].put(msg)
 128         else:
 129             print(msg)
 130
 131     def send_all(self, msg):
 132         """Send msg to all clients."""
 133         for connection_id in self.queues_out:
 134             self.send_to(connection_id, msg)
 135
 136     def stringify_yx(self, tuple_):
 137         """Transform tuple (y,x) into string 'Y:'+str(y)+',X:'+str(x)."""
 138         return 'Y:' + str(tuple_[0]) + ',X:' + str(tuple_[1])
 139
 140     def quoted(self, string):
 141         """Quote and escape string so client interprets it as single token."""
 142         quoted = []
 143         quoted += ['"']
 144         for c in string:
 145             if c in {'"', '\\'}:
 146                 quoted += ['\\']
 147             quoted += [c]
 148         quoted += ['"']
 149         return ''.join(quoted)
 150
 151     def send_all_gamestate(self):
 152         """Send out game state data relevant to clients."""
 153         self.send_all('NEW_TURN ' + str(self.world.turn))
 154         self.send_all('MAP_SIZE ' + self.stringify_yx(self.world.map_size))
 155         for y in range(self.world.map_size[0]):
 156             width = self.world.map_size[1]
 157             terrain_line = self.world.map_[y * width:(y + 1) * width]
 158             self.send_all('TERRAIN_LINE %5s %s' % (y, self.quoted(terrain_line)))
 159         for thing in self.world.things:
 160             self.send_all('THING_TYPE %s %s' % (thing.id_, thing.type_))
 161             self.send_all('THING_POS %s %s' % (thing.id_, self.stringify_yx(thing.position)))
 162
 163     def proceed(self):
 164         """Send turn finish signal, run game world, send new world data.
 165
 166         First sends 'TURN_FINISHED' message, then runs game world
 167         until new player input is needed, then sends game state.
 168         """
 169         self.send_all('TURN_FINISHED ' + str(self.world.turn))
 170         self.world.proceed_to_next_player_turn()
 171         self.send_all_gamestate()
 172
 173     def get_player(self):
 174         return self.world.get_thing(self.world.player_id)
 175
 176     def cmd_MOVE(self, direction, connection_id):
 177         """Set player task to 'move' with direction arg, finish player turn."""
 178         if direction not in {'UP', 'DOWN', 'RIGHT', 'LEFT'}:
 179             raise ArgError('Move argument must be one of: '
 180                            'UP, DOWN, RIGHT, LEFT')
 181         self.get_player().set_task('move', direction=direction)
 182         self.proceed()
 183     cmd_MOVE.argtypes = 'string'
 184
 185     def cmd_WAIT(self, connection_id):
 186         """Set player task to 'wait', finish player turn."""
 187         self.get_player().set_task('wait')
 188         self.proceed()
 189
 190     def cmd_MAP_SIZE(self, yx, connection_id):
 191         self.world.set_map_size(yx)
 192     cmd_MAP_SIZE.argtypes = 'yx_tuple:nonneg'
 193
 194     def cmd_TERRAIN_LINE(self, y, line, connection_id):
 195         self.world.set_map_line(y, line)
 196     cmd_TERRAIN_LINE.argtypes = 'int:nonneg string'
 197
 198     def cmd_THING_TYPE(self, i, type_, connection_id):
 199         t = self.world.get_thing(i)
 200         t.type_ = type_
 201     cmd_THING_TYPE.argtypes = 'int:nonneg string'
 202
 203     def cmd_THING_POS(self, i, yx, connection_id):
 204         t = self.world.get_thing(i)
 205         t.position = list(yx)
 206     cmd_THING_POS.argtypes = 'int:nonneg yx_tuple:nonneg'
 207
 208     def cmd_GET_TURN(self, connection_id):
 209         """Send world.turn to caller."""
 210         self.send_to(connection_id, str(self.world.turn))
 211
 212     def cmd_ECHO(self, msg, connection_id):
 213         """Send msg to caller."""
 214         self.send_to(connection_id, msg)
 215     cmd_ECHO.argtypes = 'string'
 216
 217     def cmd_ALL(self, msg, connection_id):
 218         """Send msg to all clients."""
 219         self.send_all(msg)
 220     cmd_ALL.argtypes = 'string'
 221
 222     def cmd_FIB(self, numbers, connection_id):
 223         """Reply with n-th Fibonacci numbers, n taken from tokens[1:].
 224
 225         Numbers are calculated in parallel as far as possible, using fib().
 226         A 'CALCULATING …' message is sent to caller before the result.
 227         """
 228         self.send_to(connection_id, 'CALCULATING …')
 229         results = self.pool.map(fib, numbers)
 230         reply = ' '.join([str(r) for r in results])
 231         self.send_to(connection_id, reply)
 232     cmd_FIB.argtypes = 'seq:int:nonneg'
 233
 234     def cmd_INC_P(self, connection_id):
 235         """Increment world.turn, send game turn data to everyone.
 236
 237         To simulate game processing waiting times, a one second delay between
 238         TURN_FINISHED and NEW_TURN occurs; after NEW_TURN, some expensive
 239         calculations are started as pool processes that need to be finished
 240         until a further INC finishes the turn.
 241
 242         This is just a demo structure for how the game loop could work when
 243         parallelized. One might imagine a two-step game turn, with a non-action
 244         step determining actor tasks (the AI determinations would take the
 245         place of the fib calculations here), and an action step wherein these
 246         tasks are performed (where now sleep(1) is).
 247         """
 248         from time import sleep
 249         if self.pool_result is not None:
 250             self.pool_result.wait()
 251         self.send_all('TURN_FINISHED ' + str(self.world.turn))
 252         sleep(1)
 253         self.world.turn += 1
 254         self.send_all_gamestate()
 255         self.pool_result = self.pool.map_async(fib, (35, 35))
 256
 257
 258 def io_loop(q, commander):
 259     """Handle commands coming through queue q, send results back.
 260
 261     Commands from q are expected to be tuples, with the first element either
 262     'ADD_QUEUE', 'COMMAND', or 'KILL_QUEUE', the second element a UUID, and
 263     an optional third element of arbitrary type. The UUID identifies a
 264     receiver for replies.
 265
 266     An 'ADD_QUEUE' command should contain as third element a queue through
 267     which to send messages back to the sender of the command. A 'KILL_QUEUE'
 268     command removes the queue for that receiver from the list of queues through
 269     which to send replies.
 270
 271     A 'COMMAND' command is specified in greater detail by a string that is the
 272     tuple's third element. The commander CommandHandler takes care of processing
 273     this and sending out replies.
 274     """
 275     while True:
 276         x = q.get()
 277         command_type = x[0]
 278         connection_id = x[1]
 279         content = None if len(x) == 2 else x[2]
 280         if command_type == 'ADD_QUEUE':
 281             commander.queues_out[connection_id] = content
 282         elif command_type == 'COMMAND':
 283             commander.handle_input(content, connection_id)
 284         elif command_type == 'KILL_QUEUE':
 285             del commander.queues_out[connection_id]
 286
 287
 288 if len(sys.argv) != 2:
 289     print('wrong number of arguments, expected one (game file)')
 290     exit(1)
 291 game_file_name = sys.argv[1]
 292 commander = CommandHandler()
 293 if os.path.exists(game_file_name):
 294     if not os.path.isfile(game_file_name):
 295         print('game file name does not refer to a valid game file')
 296     else:
 297         with open(game_file_name, 'r') as f:
 298             lines = f.readlines()
 299         for i in range(len(lines)):
 300             line = lines[i]
 301             print("FILE INPUT LINE %s: %s" % (i, line), end='')
 302             commander.handle_input(line, abort_on_error=True)
 303 else:
 304     commander.handle_input('MAP_SIZE Y:5,X:5')
 305     commander.handle_input('TERRAIN_LINE 0 "xxxxx"')
 306     commander.handle_input('TERRAIN_LINE 1 "x...x"')
 307     commander.handle_input('TERRAIN_LINE 2 "x.X.x"')
 308     commander.handle_input('TERRAIN_LINE 3 "x...x"')
 309     commander.handle_input('TERRAIN_LINE 4 "xxxxx"')
 310     commander.handle_input('THING_TYPE 0 human')
 311     commander.handle_input('THING_POS 0 Y:3,X:3')
 312     commander.handle_input('THING_TYPE 1 monster')
 313     commander.handle_input('THING_POS 1 Y:1,X:1')
 314 q = queue.Queue()
 315 c = threading.Thread(target=io_loop, daemon=True, args=(q, commander))
 316 c.start()
 317 server = Server(q, ('localhost', 5000), IO_Handler)
 318 try:
 319     server.serve_forever()
 320 except KeyboardInterrupt:
 321     pass
 322 finally:
 323     print('Killing server')
 324     server.server_close()