home · contact · privacy
e688f2cf41f7d903207d5e2d4d226a4af35b78b3
[plomrogue2-experiments] / 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, queues_out={}):
98         from multiprocessing import Pool
99         self.queues_out = 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:' + thing.type_ + ' '
161                           + 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 cmd_MOVE(self, direction, connection_id):
174         """Set player task to 'move' with direction arg, finish player turn."""
175         if direction not in {'UP', 'DOWN', 'RIGHT', 'LEFT'}:
176             raise ArgError('Move argument must be one of: '
177                            'UP, DOWN, RIGHT, LEFT')
178         self.world.player.set_task('move', direction=direction)
179         self.proceed()
180     cmd_MOVE.argtypes = 'string'
181
182     def cmd_WAIT(self, connection_id):
183         """Set player task to 'wait', finish player turn."""
184         self.world.player.set_task('wait')
185         self.proceed()
186
187     def cmd_GET_TURN(self, connection_id):
188         """Send world.turn to caller."""
189         self.send_to(connection_id, str(self.world.turn))
190
191     def cmd_ECHO(self, msg, connection_id):
192         """Send msg to caller."""
193         self.send_to(connection_id, msg)
194     cmd_ECHO.argtypes = 'string'
195
196     def cmd_ALL(self, msg, connection_id):
197         """Send msg to all clients."""
198         self.send_all(msg)
199     cmd_ALL.argtypes = 'string'
200
201     def cmd_FIB(self, numbers, connection_id):
202         """Reply with n-th Fibonacci numbers, n taken from tokens[1:].
203
204         Numbers are calculated in parallel as far as possible, using fib().
205         A 'CALCULATING …' message is sent to caller before the result.
206         """
207         self.send_to(connection_id, 'CALCULATING …')
208         results = self.pool.map(fib, numbers)
209         reply = ' '.join([str(r) for r in results])
210         self.send_to(connection_id, reply)
211     cmd_FIB.argtypes = 'seq:int:nonneg'
212
213     def cmd_INC_P(self, connection_id):
214         """Increment world.turn, send game turn data to everyone.
215
216         To simulate game processing waiting times, a one second delay between
217         TURN_FINISHED and NEW_TURN occurs; after NEW_TURN, some expensive
218         calculations are started as pool processes that need to be finished
219         until a further INC finishes the turn.
220
221         This is just a demo structure for how the game loop could work when
222         parallelized. One might imagine a two-step game turn, with a non-action
223         step determining actor tasks (the AI determinations would take the
224         place of the fib calculations here), and an action step wherein these
225         tasks are performed (where now sleep(1) is).
226         """
227         from time import sleep
228         if self.pool_result is not None:
229             self.pool_result.wait()
230         self.send_all('TURN_FINISHED ' + str(self.world.turn))
231         sleep(1)
232         self.world.turn += 1
233         self.send_all_gamestate()
234         self.pool_result = self.pool.map_async(fib, (35, 35))
235
236
237 def io_loop(q, commander):
238     """Handle commands coming through queue q, send results back.
239
240     Commands from q are expected to be tuples, with the first element either
241     'ADD_QUEUE', 'COMMAND', or 'KILL_QUEUE', the second element a UUID, and
242     an optional third element of arbitrary type. The UUID identifies a
243     receiver for replies.
244
245     An 'ADD_QUEUE' command should contain as third element a queue through
246     which to send messages back to the sender of the command. A 'KILL_QUEUE'
247     command removes the queue for that receiver from the list of queues through
248     which to send replies.
249
250     A 'COMMAND' command is specified in greater detail by a string that is the
251     tuple's third element. The commander CommandHandler takes care of processing
252     this and sending out replies.
253     """
254     while True:
255         x = q.get()
256         command_type = x[0]
257         connection_id = x[1]
258         content = None if len(x) == 2 else x[2]
259         if command_type == 'ADD_QUEUE':
260             commander.queues_out[connection_id] = content
261         elif command_type == 'COMMAND':
262             commander.handle_input(content, connection_id)
263         elif command_type == 'KILL_QUEUE':
264             del commander.queues_out[connection_id]
265
266
267 if len(sys.argv) != 2:
268     print('wrong number of arguments, expected one (game file)')
269     exit(1)
270 game_file_name = sys.argv[1]
271 commander = CommandHandler()
272 if os.path.exists(game_file_name):
273     if not os.path.isfile(game_file_name):
274         print('game file name does not refer to a valid game file')
275     else:
276         with open(game_file_name, 'r') as f:
277             lines = f.readlines()
278         for i in range(len(lines)):
279             line = lines[i]
280             print("FILE INPUT LINE %s: %s" % (i, line), end='')
281             commander.handle_input(line, abort_on_error=True)
282 q = queue.Queue()
283 c = threading.Thread(target=io_loop, daemon=True, args=(q, commander))
284 c.start()
285 server = Server(q, ('localhost', 5000), IO_Handler)
286 try:
287     server.serve_forever()
288 except KeyboardInterrupt:
289     pass
290 finally:
291     print('Killing server')
292     server.server_close()