similar to recent opensrf-c changes, keep active and idle child procs in separate...
[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, 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.max_requests = 0 # max child requests
37         self.max_children = 0 # max num of child processes
38         self.min_childen = 0 # min num of child processes
39         self.num_children = 0 # current num children
40         self.osrf_handle = None # xmpp handle
41         self.routers = [] # list of registered routers
42         self.keepalive = 0 # how long to wait for subsequent, stateful requests
43         self.active_list = [] # list of active children
44         self.idle_list = [] # list of idle children
45
46         # Global status socketpair.  All children relay their 
47         # availability info to the parent through this socketpair. 
48         self.read_status, self.write_status = socket.socketpair()
49
50     def load_app(self):
51         settings = osrf.set.get('activeapps.%s' % self.service)
52         
53
54     def cleanup(self):
55         ''' Closes management sockets, kills children, reaps children, exits '''
56
57         osrf.log.log_info("Shutting down...")
58         self.cleanup_routers()
59
60         self.read_status.shutdown(socket.SHUT_RDWR)
61         self.write_status.shutdown(socket.SHUT_RDWR)
62         self.read_status.close()
63         self.write_status.close()
64
65         for child in self.idle_list + self.active_list:
66             child.read_data.shutdown(socket.SHUT_RDWR)
67             child.write_data.shutdown(socket.SHUT_RDWR)
68             child.read_data.close()
69             child.write_data.close()
70
71         os.kill(0, signal.SIGKILL)
72         self.reap_children(True)
73         os._exit(0)
74
75
76     def handle_signals(self):
77         ''' Installs SIGINT and SIGTERM handlers '''
78         def handler(signum, frame):
79             self.cleanup()
80         signal.signal(signal.SIGINT, handler)
81         signal.signal(signal.SIGTERM, handler)
82
83
84     def run(self):
85
86         osrf.net.get_network_handle().disconnect()
87         osrf.net.clear_network_handle()
88         self.spawn_children()
89         self.handle_signals()
90
91         time.sleep(.5) # give children a chance to connect before we start taking data
92         self.osrf_handle = osrf.system.System.net_connect(
93             resource = '%s_listener' % self.service,
94             service = self.service
95         )
96
97         # clear the recv callback so inbound messages do not filter through the opensrf stack
98         self.osrf_handle.receive_callback = None
99
100         # connect to our listening routers
101         self.register_routers()
102
103         try:
104             osrf.log.log_debug("entering main server loop...")
105             while True: # main server loop
106
107                 self.reap_children()
108                 self.check_status()
109                 data = self.osrf_handle.recv(-1).to_xml()
110
111                 if self.try_avail_child(data):
112                     continue
113
114                 if self.try_new_child(data):
115                     continue
116
117                 self.wait_for_child()
118
119         except KeyboardInterrupt:
120             self.cleanup()
121         #except Exception, e: 
122             #osrf.log.log_error("server exiting with exception: %s" % e.message)
123             #self.cleanup()
124                 
125
126     def try_avail_child(self, data):
127         ''' Trys to send current request data to an available child process '''
128
129         if len(self.idle_list) == 0:
130             return False
131
132         child = self.idle_list.pop(0) # remove from idle list
133         osrf.log.log_internal("sending data to available child %d" % child.pid)
134         self.write_child(child, data)
135         self.active_list.insert(0, child) # add to active list
136         return True
137
138     def try_new_child(self, data):
139         ''' Tries to spawn a new child to send request data to '''
140
141         if self.num_children < self.max_children:
142             osrf.log.log_internal("spawning new child to handle data")
143             child = self.spawn_child(True)
144             self.write_child(child, data)
145             return True
146         return False
147
148     def try_wait_child(self, data):
149         ''' Waits for a child to become available '''
150
151         osrf.log.log_warn("No children available, waiting...")
152         child = self.check_status(True)
153         self.write_child(child, data)
154
155
156     def write_child(self, child, data):
157         ''' Sends data to the child process '''
158         # Do we need to watch for sigpipe, etc?
159         child.write_data.sendall(str(len(data)).rjust(SIZE_PAD) + data)
160
161
162     def check_status(self, block=False):
163         ''' Checks to see if any children have indicated they are done with 
164             their current request.  If block is true, this will wait 
165             indefinitely for a child to be free. '''
166
167         pid = None
168         if block:
169             pid = self.read_status.recv(SIZE_PAD)
170         else:
171             try:
172                 self.read_status.setblocking(0)
173                 pid = self.read_status.recv(SIZE_PAD)
174             except socket.error, e:
175                 if e.args[0] != errno.EAGAIN:
176                     raise e
177             self.read_status.setblocking(1)
178                 
179         if pid:
180             pid = int(pid)
181             child = [c for c in self.active_list if c.pid == pid][0]
182             self.active_list.remove(child)
183             self.idle_list.insert(0, child)
184             return child
185
186         return None
187         
188
189     def reap_children(self, done=False):
190         ''' Uses waitpid() to reap the children.  If necessary, new children are spawned '''
191
192         options = 0
193         if not done: 
194             options = os.WNOHANG 
195
196         while True:
197             try:
198                 (pid, status) = os.waitpid(0, options)
199                 if pid == 0:
200                     if not done:
201                         self.spawn_children()
202                     return
203
204                 osrf.log.log_debug("reaping child %d" % pid)
205                 self.num_children -= 1
206
207                 # locate the child in the active or idle list and remove it
208                 # Note: typically, a dead child will be in the active list, since 
209                 # exiting children do not send a cleanup status to the controller
210
211                 child = [c for c in self.active_list if c.pid == pid]
212                 if len(child) > 0:
213                     self.active_list.remove(child[0])
214                 else:
215                     child = [c for c in self.idle_list if c.pid == pid]
216                     self.idle_list.remove(child[0])
217
218             except OSError:
219                 return
220         
221     def spawn_children(self):
222         ''' Launches up to min_children child processes '''
223         while self.num_children < self.min_children:
224             self.spawn_child()
225
226     def spawn_child(self, active=False):
227         ''' Spawns a new child process '''
228
229         child = Child(self)
230         child.read_data, child.write_data = socket.socketpair()
231         child.pid = os.fork()
232
233         if child.pid:
234             self.num_children += 1
235             if active:
236                 self.active_list.insert(0, child)
237             else:
238                 self.idle_list.insert(0, child)
239             osrf.log.log_debug("spawned child %d : %d total" % (child.pid, self.num_children))
240             return child
241         else:
242             child.pid = os.getpid()
243             child.init()
244             child.run()
245             osrf.net.get_network_handle().disconnect()
246             osrf.log.log_internal("child exiting...")
247             os._exit(0)
248
249     def register_routers(self):
250         ''' Registers this application instance with all configured routers '''
251         routers = osrf.conf.get('routers.router')
252
253         if not isinstance(routers, list):
254             routers = [routers]
255
256         for router in routers:
257             if isinstance(router, dict):
258                 if not 'services' in router or \
259                         self.service in router['services']['service']:
260                     target = "%s@%s/router" % (router['name'], router['domain'])
261                     self.register_router(target)
262             else:
263                 router_name = osrf.conf.get('router_name')
264                 target = "%s@%s/router" % (router_name, router)
265                 self.register_router(target)
266
267
268     def register_router(self, target):
269         ''' Registers with a single router '''
270
271         osrf.log.log_info("registering with router %s" % target)
272         self.routers.append(target)
273
274         reg_msg = osrf.net.NetworkMessage(
275             recipient = target,
276             body = 'registering...',
277             router_command = 'register',
278             router_class = self.service
279         )
280
281         self.osrf_handle.send(reg_msg)
282
283     def cleanup_routers(self):
284         ''' Un-registers with all connected routers '''
285
286         for target in self.routers:
287             osrf.log.log_info("un-registering with router %s" % target)
288             unreg_msg = osrf.net.NetworkMessage(
289                 recipient = target,
290                 body = 'un-registering...',
291                 router_command = 'unregister',
292                 router_class = self.service
293             )
294             self.osrf_handle.send(unreg_msg)
295         
296
297 class Child(object):
298     ''' Models a single child process '''
299
300     def __init__(self, controller):
301         self.controller = controller # our Controller object
302         self.num_requests = 0 # how many requests we've served so far
303         self.read_data = None # the child reads data from the controller on this socket
304         self.write_data = None # the controller sends data to the child on this socket 
305         self.pid = 0 # my process id
306
307     def run(self):
308         ''' Loops, processing data, until max_requests is reached '''
309
310         while True:
311             try:
312                 size = int(self.read_data.recv(SIZE_PAD) or 0)
313                 data = self.read_data.recv(size)
314                 osrf.log.log_internal("recv'd data " + data)
315                 osrf.net.get_network_handle().flush_inbound_data()
316                 session = osrf.stack.push(osrf.net.NetworkMessage.from_xml(data))
317                 self.keepalive_loop(session)
318                 self.num_requests += 1
319                 if self.num_requests == self.controller.max_requests:
320                     break
321                 self.send_status()
322             except KeyboardInterrupt:
323                 pass
324
325         # run the exit handler
326         osrf.app.Application.application.child_exit()
327
328     def keepalive_loop(self, session):
329         keepalive = self.controller.keepalive
330
331         while session.state == osrf.const.OSRF_APP_SESSION_CONNECTED:
332
333             status = session.wait(keepalive)
334
335             if session.state == osrf.const.OSRF_APP_SESSION_DISCONNECTED:
336                 osrf.log.log_internal("client sent disconnect, exiting keepalive")
337                 break
338
339             if status is None: # no msg received before keepalive timeout expired
340
341                 osrf.log.log_info("No request was received in %d seconds, exiting stateful session" % int(keepalive));
342
343                 session.send_status(
344                     session.thread, 
345                     osrf.net_obj.NetworkObject.osrfConnectStatus({   
346                         'status' : 'Disconnected on timeout',
347                         'statusCode': osrf.const.OSRF_STATUS_TIMEOUT
348                     })
349                 )
350
351                 break
352
353         session.cleanup()
354         return
355
356     def send_status(self):
357         ''' Informs the controller that we are done processing this request '''
358         fcntl.lockf(self.controller.write_status.fileno(), fcntl.LOCK_EX)
359         try:
360             self.controller.write_status.sendall(str(self.pid).rjust(SIZE_PAD))
361         finally:
362             fcntl.lockf(self.controller.write_status.fileno(), fcntl.LOCK_UN)
363
364     def init(self):
365         ''' Connects the opensrf xmpp handle '''
366         osrf.net.clear_network_handle()
367         osrf.system.System.net_connect(
368             resource = '%s_drone' % self.controller.service, 
369             service = self.controller.service
370         )
371         osrf.app.Application.application.child_init()
372