Merge branch 'py-reserve'
[ashd.git] / python / ashd / serve.py
CommitLineData
8f410ce7
FT
1import sys, os, threading, time, logging, select, Queue
2import perf
63090e5a
FT
3
4log = logging.getLogger("ashd.serve")
5seq = 1
6seqlk = threading.Lock()
7
8def reqseq():
9 global seq
8f410ce7 10 with seqlk:
63090e5a
FT
11 s = seq
12 seq += 1
13 return s
63090e5a
FT
14
15class closed(IOError):
16 def __init__(self):
17 super(closed, self).__init__("The client has closed the connection.")
18
8f410ce7
FT
19class reqthread(threading.Thread):
20 def __init__(self, name=None, **kw):
21 if name is None:
22 name = "Request handler %i" % reqseq()
23 super(reqthread, self).__init__(name=name, **kw)
24
25class wsgirequest(object):
26 def __init__(self, handler):
63090e5a
FT
27 self.status = None
28 self.headers = []
29 self.respsent = False
8f410ce7
FT
30 self.handler = handler
31 self.buffer = bytearray()
63090e5a
FT
32
33 def handlewsgi(self):
34 raise Exception()
8f410ce7
FT
35 def fileno(self):
36 raise Exception()
63090e5a
FT
37 def writehead(self, status, headers):
38 raise Exception()
8f410ce7 39 def flush(self):
63090e5a 40 raise Exception()
8f410ce7
FT
41 def close(self):
42 pass
43 def writedata(self, data):
44 self.buffer.extend(data)
63090e5a
FT
45
46 def flushreq(self):
47 if not self.respsent:
48 if not self.status:
49 raise Exception("Cannot send response body before starting response.")
50 self.respsent = True
51 self.writehead(self.status, self.headers)
52
8f410ce7
FT
53 def write(self, data):
54 if not data:
55 return
56 self.flushreq()
57 self.writedata(data)
58 self.handler.ckflush(self)
59
63090e5a
FT
60 def startreq(self, status, headers, exc_info=None):
61 if self.status:
8f410ce7 62 if exc_info:
63090e5a
FT
63 try:
64 if self.respsent:
8f410ce7 65 raise exc_info[1]
63090e5a 66 finally:
8f410ce7 67 exc_info = None
63090e5a
FT
68 else:
69 raise Exception("Can only start responding once.")
70 self.status = status
71 self.headers = headers
72 return self.write
8f410ce7
FT
73
74class handler(object):
75 def handle(self, request):
76 raise Exception()
77 def ckflush(self, req):
78 while len(req.buffer) > 0:
79 rls, wls, els = select.select([], [req], [req])
80 req.flush()
81 def close(self):
82 pass
83
84 @classmethod
85 def parseargs(cls, **args):
86 if len(args) > 0:
87 raise ValueError("unknown handler argument: " + iter(args).next())
88 return {}
89
90class single(handler):
91 cname = "single"
92
93 def handle(self, req):
63090e5a 94 try:
8f410ce7
FT
95 env = req.mkenv()
96 with perf.request(env) as reqevent:
97 respiter = req.handlewsgi(env, req.startreq)
63090e5a 98 for data in respiter:
8f410ce7
FT
99 req.write(data)
100 if req.status:
101 reqevent.response([req.status, req.headers])
102 req.flushreq()
103 self.ckflush(req)
63090e5a
FT
104 except closed:
105 pass
8f410ce7
FT
106 except:
107 log.error("exception occurred when handling request", exc_info=True)
108 finally:
109 req.close()
110
111class freethread(handler):
112 cname = "free"
63090e5a 113
8f410ce7
FT
114 def __init__(self, max=None, timeout=None, **kw):
115 super(freethread, self).__init__(**kw)
116 self.current = set()
117 self.lk = threading.Lock()
118 self.tcond = threading.Condition(self.lk)
119 self.max = max
120 self.timeout = timeout
63090e5a 121
8f410ce7
FT
122 @classmethod
123 def parseargs(cls, max=None, abort=None, **args):
124 ret = super(freethread, cls).parseargs(**args)
125 if max:
126 ret["max"] = int(max)
127 if abort:
128 ret["timeout"] = int(abort)
129 return ret
63090e5a 130
8f410ce7
FT
131 def handle(self, req):
132 with self.lk:
133 if self.max is not None:
134 if self.timeout is not None:
135 now = start = time.time()
136 while len(self.current) >= self.max:
137 self.tcond.wait(start + self.timeout - now)
138 now = time.time()
139 if now - start > self.timeout:
140 os.abort()
141 else:
142 while len(self.current) >= self.max:
143 self.tcond.wait()
144 th = reqthread(target=self.run, args=[req])
145 th.start()
146 while th.is_alive() and th not in self.current:
147 self.tcond.wait()
148
149 def run(self, req):
63090e5a 150 try:
8f410ce7
FT
151 th = threading.current_thread()
152 with self.lk:
153 self.current.add(th)
154 self.tcond.notify_all()
155 try:
156 env = req.mkenv()
157 with perf.request(env) as reqevent:
158 respiter = req.handlewsgi(env, req.startreq)
159 for data in respiter:
160 req.write(data)
161 if req.status:
162 reqevent.response([req.status, req.headers])
163 req.flushreq()
164 self.ckflush(req)
165 except closed:
166 pass
167 except:
168 log.error("exception occurred when handling request", exc_info=True)
169 finally:
170 with self.lk:
171 self.current.remove(th)
172 self.tcond.notify_all()
63090e5a 173 finally:
8f410ce7
FT
174 req.close()
175
176 def close(self):
177 while True:
178 with self.lk:
179 if len(self.current) > 0:
180 th = iter(self.current).next()
181 else:
182 return
183 th.join()
184
185class resplex(handler):
186 cname = "rplex"
187
188 def __init__(self, max=None, **kw):
189 super(resplex, self).__init__(**kw)
190 self.current = set()
191 self.lk = threading.Lock()
192 self.tcond = threading.Condition(self.lk)
193 self.max = max
194 self.cqueue = Queue.Queue(5)
195 self.cnpipe = os.pipe()
196 self.rthread = reqthread(name="Response thread", target=self.handle2)
197 self.rthread.start()
198
199 @classmethod
200 def parseargs(cls, max=None, **args):
201 ret = super(resplex, cls).parseargs(**args)
202 if max:
203 ret["max"] = int(max)
204 return ret
205
206 def ckflush(self, req):
207 raise Exception("resplex handler does not support the write() function")
63090e5a 208
8f410ce7
FT
209 def handle(self, req):
210 with self.lk:
211 if self.max is not None:
212 while len(self.current) >= self.max:
213 self.tcond.wait()
214 th = reqthread(target=self.handle1, args=[req])
215 th.start()
216 while th.is_alive() and th not in self.current:
217 self.tcond.wait()
218
219 def handle1(self, req):
63090e5a 220 try:
8f410ce7
FT
221 th = threading.current_thread()
222 with self.lk:
223 self.current.add(th)
224 self.tcond.notify_all()
225 try:
226 env = req.mkenv()
227 respobj = req.handlewsgi(env, req.startreq)
228 respiter = iter(respobj)
229 if not req.status:
230 log.error("request handler returned without calling start_request")
231 if hasattr(respiter, "close"):
232 respiter.close()
233 return
234 else:
235 self.cqueue.put((req, respiter))
236 os.write(self.cnpipe[1], " ")
237 req = None
238 finally:
239 with self.lk:
240 self.current.remove(th)
241 self.tcond.notify_all()
242 except closed:
243 pass
244 except:
245 log.error("exception occurred when handling request", exc_info=True)
63090e5a 246 finally:
8f410ce7
FT
247 if req is not None:
248 req.close()
63090e5a 249
8f410ce7 250 def handle2(self):
63090e5a 251 try:
8f410ce7
FT
252 rp = self.cnpipe[0]
253 current = {}
e0c88e73 254
8f410ce7
FT
255 def closereq(req):
256 respiter = current[req]
257 try:
258 if respiter is not None and hasattr(respiter, "close"):
259 respiter.close()
260 except:
261 log.error("exception occurred when closing iterator", exc_info=True)
262 try:
263 req.close()
264 except:
265 log.error("exception occurred when closing request", exc_info=True)
266 del current[req]
267 def ckiter(req):
268 respiter = current[req]
269 if respiter is not None:
270 rem = False
271 try:
272 data = respiter.next()
273 except StopIteration:
274 rem = True
275 try:
276 req.flushreq()
277 except:
278 log.error("exception occurred when handling response data", exc_info=True)
279 except:
280 rem = True
281 log.error("exception occurred when iterating response", exc_info=True)
282 if not rem:
283 if data:
284 try:
285 req.flushreq()
286 req.writedata(data)
287 except:
288 log.error("exception occurred when handling response data", exc_info=True)
289 rem = True
290 if rem:
291 current[req] = None
292 try:
293 if hasattr(respiter, "close"):
294 respiter.close()
295 except:
296 log.error("exception occurred when closing iterator", exc_info=True)
297 respiter = None
298 if respiter is None and not req.buffer:
299 closereq(req)
300
301 while True:
302 bufl = list(req for req in current.iterkeys() if req.buffer)
303 rls, wls, els = select.select([rp], bufl, [rp] + bufl)
304 if rp in rls:
305 ret = os.read(rp, 1024)
306 if not ret:
307 os.close(rp)
308 return
309 try:
310 while True:
311 req, respiter = self.cqueue.get(False)
312 current[req] = respiter
313 ckiter(req)
314 except Queue.Empty:
315 pass
316 for req in wls:
317 try:
318 req.flush()
319 except closed:
320 closereq(req)
321 except:
322 log.error("exception occurred when writing response", exc_info=True)
323 closereq(req)
324 else:
325 if len(req.buffer) < 65536:
326 ckiter(req)
327 except:
328 log.critical("unexpected exception occurred in response handler thread", exc_info=True)
e0c88e73 329 os.abort()
8f410ce7
FT
330
331 def close(self):
332 while True:
333 with self.lk:
334 if len(self.current) > 0:
335 th = iter(self.current).next()
336 else:
337 break
338 th.join()
339 os.close(self.cnpipe[1])
340 self.rthread.join()
341
342names = dict((cls.cname, cls) for cls in globals().itervalues() if
343 isinstance(cls, type) and
344 issubclass(cls, handler) and
345 hasattr(cls, "cname"))
346
347def parsehspec(spec):
348 if ":" not in spec:
349 return spec, {}
350 nm, spec = spec.split(":", 1)
351 args = {}
352 while spec:
353 if "," in spec:
354 part, spec = spec.split(",", 1)
355 else:
356 part, spec = spec, None
357 if "=" in part:
358 key, val = part.split("=", 1)
359 else:
360 key, val = part, ""
361 args[key] = val
362 return nm, args