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 import parser
   9 import server_.game
  10 import game_common
  11
  12
  13 # Avoid "Address already in use" errors.
  14 socketserver.TCPServer.allow_reuse_address = True
  15
  16
  17 class Server(socketserver.ThreadingTCPServer):
  18     """Bind together threaded IO handling server and message queue."""
  19
  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.
  24
  25
  26 class IO_Handler(socketserver.BaseRequestHandler):
  27
  28     def handle(self):
  29         """Move messages between network socket and main thread via queues.
  30
  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.
  34
  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
  38         own queue.
  39
  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
  45         instructions.
  46         """
  47         import plom_socket_io
  48
  49         def caught_send(socket, message):
  50             """Send message by socket, catch broken socket connection error."""
  51             try:
  52                 plom_socket_io.send(socket, message)
  53             except plom_socket_io.BrokenSocketConnection:
  54                 pass
  55
  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]:
  59                 try:
  60                     msg = queue_in.get(timeout=1)
  61                 except queue.Empty:
  62                     continue
  63                 caught_send(socket, msg)
  64
  65         import uuid
  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))
  70         thread_alive = [True]
  71         t = threading.Thread(target=send_queue_messages,
  72                              args=(self.request, queue_in, thread_alive))
  73         t.start()
  74         for message in plom_socket_io.recv(self.request):
  75             if message is None:
  76                 caught_send(self.request, 'BAD MESSAGE')
  77             elif 'QUIT' == message:
  78                 caught_send(self.request, 'BYE')
  79                 break
  80             else:
  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))
  85         self.request.close()
  86
  87
  88 def fib(n):
  89     """Calculate n-th Fibonacci number. Very inefficiently."""
  90     if n in (1, 2):
  91         return 1
  92     else:
  93         return fib(n-1) + fib(n-2)
  94
  95
  96 class CommandHandler(game_common.Commander, server_.game.Commander):
  97
  98     def __init__(self, game_file_name):
  99         self.queues_out = {}
 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
 106         self.pool = Pool()
 107         self.pool_result = None
 108
 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
 112
 113         def answer(connection_id, msg):
 114             if connection_id:
 115                 self.send(msg, connection_id)
 116             else:
 117                 print(msg)
 118
 119         try:
 120             command = self.parser.parse(input_)
 121             if command is None:
 122                 answer(connection_id, 'UNHANDLED INPUT')
 123             else:
 124                 if 'connection_id' in list(signature(command).parameters):
 125                     command(connection_id=connection_id)
 126                 else:
 127                     command()
 128                     if store:
 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))
 135
 136     def quote(self, string):
 137         """Quote & escape string so client interprets it as single token."""
 138         quoted = []
 139         quoted += ['"']
 140         for c in string:
 141             if c in {'"', '\\'}:
 142                 quoted += ['\\']
 143             quoted += [c]
 144         quoted += ['"']
 145         return ''.join(quoted)
 146
 147     def send(self, msg, connection_id=None):
 148         if connection_id:
 149             self.queues_out[connection_id].put(msg)
 150         else:
 151             for connection_id in self.queues_out:
 152                 self.queues_out[connection_id].put(msg)
 153
 154     def send_gamestate(self, connection_id=None):
 155         """Send out game state data relevant to clients."""
 156
 157         def stringify_yx(tuple_):
 158             """Transform tuple (y,x) into string 'Y:'+str(y)+',X:'+str(x)."""
 159             return 'Y:' + str(tuple_[0]) + ',X:' + str(tuple_[1])
 160
 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' %
 166                       (y, self.quote(visible_map.get_line(y))))
 167         visible_things = self.world.get_player().get_visible_things()
 168         for thing in visible_things:
 169             self.send('THING_TYPE %s %s' % (thing.id_, thing.type_))
 170             self.send('THING_POS %s %s' % (thing.id_,
 171                                            stringify_yx(thing.position)))
 172
 173     def proceed(self):
 174         """Send turn finish signal, run game world, send new world data.
 175
 176         First sends 'TURN_FINISHED' message, then runs game world
 177         until new player input is needed, then sends game state.
 178         """
 179         self.send('TURN_FINISHED ' + str(self.world.turn))
 180         self.world.proceed_to_next_player_turn()
 181         msg = str(self.world.get_player().last_task_result)
 182         self.send('LAST_PLAYER_TASK_RESULT ' + self.quote(msg))
 183         self.send_gamestate()
 184
 185     def cmd_FIB(self, numbers, connection_id):
 186         """Reply with n-th Fibonacci numbers, n taken from tokens[1:].
 187
 188         Numbers are calculated in parallel as far as possible, using fib().
 189         A 'CALCULATING …' message is sent to caller before the result.
 190         """
 191         self.send('CALCULATING …', connection_id)
 192         results = self.pool.map(fib, numbers)
 193         reply = ' '.join([str(r) for r in results])
 194         self.send(reply, connection_id)
 195     cmd_FIB.argtypes = 'seq:int:nonneg'
 196
 197     def cmd_INC_P(self, connection_id):
 198         """Increment world.turn, send game turn data to everyone.
 199
 200         To simulate game processing waiting times, a one second delay between
 201         TURN_FINISHED and NEW_TURN occurs; after NEW_TURN, some expensive
 202         calculations are started as pool processes that need to be finished
 203         until a further INC finishes the turn.
 204
 205         This is just a demo structure for how the game loop could work when
 206         parallelized. One might imagine a two-step game turn, with a non-action
 207         step determining actor tasks (the AI determinations would take the
 208         place of the fib calculations here), and an action step wherein these
 209         tasks are performed (where now sleep(1) is).
 210         """
 211         from time import sleep
 212         if self.pool_result is not None:
 213             self.pool_result.wait()
 214         self.send('TURN_FINISHED ' + str(self.world.turn))
 215         sleep(1)
 216         self.world.turn += 1
 217         self.send_gamestate()
 218         self.pool_result = self.pool.map_async(fib, (35, 35))
 219
 220
 221 def io_loop(q, commander):
 222     """Handle commands coming through queue q, send results back.
 223
 224     Commands from q are expected to be tuples, with the first element either
 225     'ADD_QUEUE', 'COMMAND', or 'KILL_QUEUE', the second element a UUID, and
 226     an optional third element of arbitrary type. The UUID identifies a
 227     receiver for replies.
 228
 229     An 'ADD_QUEUE' command should contain as third element a queue through
 230     which to send messages back to the sender of the command. A 'KILL_QUEUE'
 231     command removes the queue for that receiver from the list of queues through
 232     which to send replies.
 233
 234     A 'COMMAND' command is specified in greater detail by a string that is the
 235     tuple's third element. The commander CommandHandler takes care of processing
 236     this and sending out replies.
 237     """
 238     while True:
 239         x = q.get()
 240         command_type = x[0]
 241         connection_id = x[1]
 242         content = None if len(x) == 2 else x[2]
 243         if command_type == 'ADD_QUEUE':
 244             commander.queues_out[connection_id] = content
 245         elif command_type == 'COMMAND':
 246             commander.handle_input(content, connection_id)
 247         elif command_type == 'KILL_QUEUE':
 248             del commander.queues_out[connection_id]
 249
 250
 251 if len(sys.argv) != 2:
 252     print('wrong number of arguments, expected one (game file)')
 253     exit(1)
 254 game_file_name = sys.argv[1]
 255 commander = CommandHandler(game_file_name)
 256 if os.path.exists(game_file_name):
 257     if not os.path.isfile(game_file_name):
 258         print('game file name does not refer to a valid game file')
 259     else:
 260         with open(game_file_name, 'r') as f:
 261             lines = f.readlines()
 262         for i in range(len(lines)):
 263             line = lines[i]
 264             print("FILE INPUT LINE %s: %s" % (i, line), end='')
 265             commander.handle_input(line, store=False)
 266 else:
 267     commander.handle_input('MAP_SIZE Y:5,X:5')
 268     commander.handle_input('TERRAIN_LINE 0 "xxxxx"')
 269     commander.handle_input('TERRAIN_LINE 1 "x...x"')
 270     commander.handle_input('TERRAIN_LINE 2 "x.X.x"')
 271     commander.handle_input('TERRAIN_LINE 3 "x...x"')
 272     commander.handle_input('TERRAIN_LINE 4 "xxxxx"')
 273     commander.handle_input('THING_TYPE 0 human')
 274     commander.handle_input('THING_POS 0 Y:3,X:3')
 275     commander.handle_input('THING_TYPE 1 monster')
 276     commander.handle_input('THING_POS 1 Y:1,X:1')
 277 q = queue.Queue()
 278 c = threading.Thread(target=io_loop, daemon=True, args=(q, commander))
 279 c.start()
 280 server = Server(q, ('localhost', 5000), IO_Handler)
 281 try:
 282     server.serve_forever()
 283 except KeyboardInterrupt:
 284     pass
 285 finally:
 286     print('Killing server')
 287     server.server_close()