added final bits for stateful sessions (drone keepalive)
[OpenSRF.git] / src / python / osrf / server.py
1 # -----------------------------------------------------------------------
2 # Copyright (C) 2008  Equinox Software, Inc.
3 # Bill Erickson <erickson@esilibrary.com>
4 #
5 # This program is free software; you can redistribute it and/or
6 # modify it under the terms of the GNU General Public License
7 # as published by the Free Software Foundation; either version 2
8 # of the License, or (at your option) any later version.
9 #
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13 # GNU General Public License for more details.
14 #
15 # You should have received a copy of the GNU General Public License
16 # along with this program; if not, write to the Free Software
17 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  
18 # 02110-1301, USA
19 # -----------------------------------------------------------------------
20
21 import os, sys, threading, logging, fcntl, socket, errno, signal, time
22 import osrf.log, osrf.conf, osrf.net, osrf.system, osrf.stack, osrf.app, osrf.const
23
24
25 # used to define the size of the PID/size leader in 
26 # status and data messages passed to and from children
27 SIZE_PAD = 12
28
29 class Controller(object):
30     ''' 
31         OpenSRF forking request server.  
32     '''
33
34     def __init__(self, service):
35         self.service = service # service name
36         self.application = None # the application we're serving
37         self.max_requests = 0 # max child requests
38         self.max_children = 0 # max num of child processes
39         self.min_childen = 0 # min num of child processes
40         self.num_children = 0 # current num children
41         self.child_idx = 0 # current index into the children array
42         self.children = [] # list of children
43         self.osrf_handle = None # xmpp handle
44         self.routers = [] # list of registered routers
45         self.keepalive = 0 # how long to wait for subsequent, stateful requests
46
47         # Global status socketpair.  All children relay their 
48         # availability info to the parent through this socketpair. 
49         self.read_status, self.write_status = socket.socketpair()
50
51     def load_app(self):
52         settings = osrf.set.get('activeapps.%s' % self.service)
53         
54
55     def cleanup(self):
56         ''' Closes management sockets, kills children, reaps children, exits '''
57
58         osrf.log.log_info("Shutting down...")
59         self.cleanup_routers()
60
61         self.read_status.shutdown(socket.SHUT_RDWR)
62         self.write_status.shutdown(socket.SHUT_RDWR)
63         self.read_status.close()
64         self.write_status.close()
65
66         for child in self.children:
67             child.read_data.shutdown(socket.SHUT_RDWR)
68             child.write_data.shutdown(socket.SHUT_RDWR)
69             child.read_data.close()
70             child.write_data.close()
71
72         os.kill(0, signal.SIGKILL)
73         self.reap_children(True)
74         os._exit(0)
75
76
77     def handle_signals(self):
78         ''' Installs SIGINT and SIGTERM handlers '''
79         def handler(signum, frame):
80             self.cleanup()
81         signal.signal(signal.SIGINT, handler)
82         signal.signal(signal.SIGTERM, handler)
83
84
85     def run(self):
86
87         osrf.net.get_network_handle().disconnect()
88         osrf.net.clear_network_handle()
89         self.spawn_children()
90         self.handle_signals()
91
92         time.sleep(.5) # give children a chance to connect before we start taking data
93         self.osrf_handle = osrf.system.System.net_connect(resource = '%s_listener' % self.service)
94
95         # clear the recv callback so inbound messages do not filter through the opensrf stack
96         self.osrf_handle.receive_callback = None
97
98         # connect to our listening routers
99         self.register_routers()
100
101         try:
102             osrf.log.log_debug("entering main server loop...")
103             while True: # main server loop
104
105                 self.reap_children()
106                 self.check_status()
107                 data = self.osrf_handle.recv(-1).to_xml()
108
109                 if self.try_avail_child(data):
110                     continue
111
112                 if self.try_new_child(data):
113                     continue
114
115                 self.wait_for_child()
116
117         except KeyboardInterrupt:
118             self.cleanup()
119         #except Exception, e: 
120             #osrf.log.log_error("server exiting with exception: %s" % e.message)
121             #self.cleanup()
122                 
123
124     def try_avail_child(self, data):
125         ''' Trys to send current request data to an available child process '''
126         ctr = 0
127         while ctr < self.num_children:
128
129             if self.child_idx >= self.num_children:
130                 self.child_idx = 0
131             child = self.children[self.child_idx]
132
133             if child.available:
134                 osrf.log.log_internal("sending data to available child")
135                 self.write_child(child, data)
136                 return True
137
138             ctr += 1
139             self.child_idx += 1
140         return False
141
142     def try_new_child(self, data):
143         ''' Tries to spawn a new child to send request data to '''
144         if self.num_children < self.max_children:
145             osrf.log.log_internal("spawning new child to handle data")
146             child = self.spawn_child()
147             self.write_child(child, data)
148             return True
149         return False
150
151     def try_wait_child(self, data):
152         ''' Waits for a child to become available '''
153         osrf.log.log_warn("No children available, waiting...")
154         child = self.check_status(True)
155         self.write_child(child, data)
156
157
158     def write_child(self, child, data):
159         ''' Sends data to the child process '''
160         child.available = False
161         child.write_data.sendall(str(len(data)).rjust(SIZE_PAD) + data)
162         self.child_idx += 1
163
164
165     def check_status(self, block=False):
166         ''' Checks to see if any children have indicated they are done with 
167             their current request.  If block is true, this will wait 
168             indefinitely for a child to be free. '''
169
170         pid = None
171         child = None
172         if block:
173             pid = self.read_status.recv(SIZE_PAD)
174         else:
175             try:
176                 self.read_status.setblocking(0)
177                 pid = self.read_status.recv(SIZE_PAD)
178             except socket.error, e:
179                 if e.args[0] != errno.EAGAIN:
180                     raise e
181             self.read_status.setblocking(1)
182                 
183         if pid:
184             pid = int(pid)
185             child = [c for c in self.children if c.pid == pid][0]
186             child.available = True
187
188         return child
189         
190
191     def reap_children(self, done=False):
192         ''' Uses waitpid() to reap the children.  If necessary, new children are spawned '''
193         options = 0
194         if not done: 
195             options = os.WNOHANG 
196
197         while True:
198             try:
199                 (pid, status) = os.waitpid(0, options)
200                 if pid == 0:
201                     if not done:
202                         self.spawn_children()
203                     return
204                 osrf.log.log_debug("reaping child %d" % pid)
205                 self.num_children -= 1
206                 self.children = [c for c in self.children if c.pid != pid]
207             except OSError:
208                 return
209         
210     def spawn_children(self):
211         ''' Launches up to min_children child processes '''
212         while self.num_children < self.min_children:
213             self.spawn_child()
214
215     def spawn_child(self):
216         ''' Spawns a new child process '''
217
218         child = Child(self)
219         child.read_data, child.write_data = socket.socketpair()
220         child.pid = os.fork()
221
222         if child.pid:
223             self.num_children += 1
224             self.children.append(child)
225             osrf.log.log_debug("spawned child %d : %d total" % (child.pid, self.num_children))
226             return child
227         else:
228             child.pid = os.getpid()
229             child.init()
230             child.run()
231             osrf.net.get_network_handle().disconnect()
232             osrf.log.log_internal("child exiting...")
233             os._exit(0)
234
235     def register_routers(self):
236         ''' Registers this application instance with all configured routers '''
237         routers = osrf.conf.get('routers.router')
238
239         if not isinstance(routers, list):
240             routers = [routers]
241
242         for router in routers:
243             if isinstance(router, dict):
244                 if not 'services' in router or \
245                         self.service in router['services']['service']:
246                     target = "%s@%s/router" % (router['name'], router['domain'])
247                     self.register_router(target)
248             else:
249                 router_name = osrf.conf.get('router_name')
250                 target = "%s@%s/router" % (router_name, router)
251                 self.register_router(target)
252
253
254     def register_router(self, target):
255         ''' Registers with a single router '''
256         osrf.log.log_info("registering with router %s" % target)
257         self.routers.append(target)
258
259         reg_msg = osrf.net.NetworkMessage(
260             recipient = target,
261             body = 'registering...',
262             router_command = 'register',
263             router_class = self.service
264         )
265
266         self.osrf_handle.send(reg_msg)
267
268     def cleanup_routers(self):
269         ''' Un-registers with all connected routers '''
270         for target in self.routers:
271             osrf.log.log_info("un-registering with router %s" % target)
272             unreg_msg = osrf.net.NetworkMessage(
273                 recipient = target,
274                 body = 'un-registering...',
275                 router_command = 'unregister',
276                 router_class = self.service
277             )
278             self.osrf_handle.send(unreg_msg)
279         
280
281 class Child(object):
282     ''' Models a single child process '''
283
284     def __init__(self, controller):
285         self.controller = controller # our Controller object
286         self.num_requests = 0 # how many requests we've served so far
287         self.read_data = None # the child reads data from the controller on this socket
288         self.write_data = None # the controller sends data to the child on this socket 
289         self.available = True # true if this child is not currently serving a request
290         self.pid = 0 # my process id
291
292
293     def run(self):
294         ''' Loops, processing data, until max_requests is reached '''
295         while True:
296             try:
297                 size = int(self.read_data.recv(SIZE_PAD) or 0)
298                 data = self.read_data.recv(size)
299                 osrf.log.log_internal("recv'd data " + data)
300                 session = osrf.stack.push(osrf.net.NetworkMessage.from_xml(data))
301                 self.keepalive_loop(session)
302                 self.num_requests += 1
303                 if self.num_requests == self.controller.max_requests:
304                     break
305                 self.send_status()
306             except KeyboardInterrupt:
307                 pass
308         # run the exit handler
309         osrf.app.Application.application.child_exit()
310
311     def keepalive_loop(self, session):
312         keepalive = self.controller.keepalive
313
314         while session.state == osrf.const.OSRF_APP_SESSION_CONNECTED:
315
316             start = time.time()
317             session.wait(keepalive)
318             end = time.time()
319
320             if session.state == osrf.const.OSRF_APP_SESSION_DISCONNECTED:
321                 osrf.log.log_internal("client sent disconnect, exiting keepalive")
322                 break
323
324             if (end - start) >= keepalive: # exceeded keepalive timeout
325
326                 osrf.log.log_info("No request was received in %d seconds, exiting stateful session" % int(keepalive));
327
328                 session.send_status(
329                     session.thread, 
330                     osrf.net_obj.NetworkObject.osrfConnectStatus({   
331                         'status' : 'Disconnected on timeout',
332                         'statusCode': osrf.const.OSRF_STATUS_TIMEOUT
333                     })
334                 )
335
336                 break
337
338         return
339
340     def send_status(self):
341         ''' Informs the controller that we are done processing this request '''
342         fcntl.lockf(self.controller.write_status.fileno(), fcntl.LOCK_EX)
343         try:
344             self.controller.write_status.sendall(str(self.pid).rjust(SIZE_PAD))
345         finally:
346             fcntl.lockf(self.controller.write_status.fileno(), fcntl.LOCK_UN)
347
348     def init(self):
349         ''' Connects the opensrf xmpp handle '''
350         osrf.net.clear_network_handle()
351         osrf.system.System.net_connect(resource = '%s_drone' % self.controller.service)
352         osrf.app.Application.application.child_init()
353