0a4525915e9865a9ee31886933dae40192b3308e
[ashd.git] / python3 / ashd / serve.py
1 import sys, os, threading, time, logging, select, queue
2 from . import perf
3
4 log = logging.getLogger("ashd.serve")
5 seq = 1
6 seqlk = threading.Lock()
7
8 def reqseq():
9     global seq
10     with seqlk:
11         s = seq
12         seq += 1
13         return s
14
15 class closed(IOError):
16     def __init__(self):
17         super().__init__("The client has closed the connection.")
18
19 class reqthread(threading.Thread):
20     def __init__(self, *, name=None, **kw):
21         if name is None:
22             name = "Request handler %i" % reqseq()
23         super().__init__(name=name, **kw)
24
25 class wsgirequest(object):
26     def __init__(self, handler):
27         self.status = None
28         self.headers = []
29         self.respsent = False
30         self.handler = handler
31         self.buffer = bytearray()
32
33     def handlewsgi(self):
34         raise Exception()
35     def fileno(self):
36         raise Exception()
37     def writehead(self, status, headers):
38         raise Exception()
39     def flush(self):
40         raise Exception()
41     def close(self):
42         pass
43     def writedata(self, data):
44         self.buffer.extend(data)
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
53     def write(self, data):
54         if not data:
55             return
56         self.flushreq()
57         self.writedata(data)
58         self.handler.ckflush(self)
59
60     def startreq(self, status, headers, exc_info=None):
61         if self.status:
62             if exc_info:
63                 try:
64                     if self.respsent:
65                         raise exc_info[1]
66                 finally:
67                     exc_info = None
68             else:
69                 raise Exception("Can only start responding once.")
70         self.status = status
71         self.headers = headers
72         return self.write
73
74 class 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: " + next(iter(args)))
88         return {}
89
90 class single(handler):
91     def handle(self, req):
92         try:
93             env = req.mkenv()
94             with perf.request(env) as reqevent:
95                 respiter = req.handlewsgi(env, req.startreq)
96                 for data in respiter:
97                     req.write(data)
98                 if req.status:
99                     reqevent.response([req.status, req.headers])
100                     req.flushreq()
101                 self.ckflush(req)
102         except closed:
103             pass
104         except:
105             log.error("exception occurred when handling request", exc_info=True)
106         finally:
107             req.close()
108
109 class freethread(handler):
110     def __init__(self, *, max=None, timeout=None, **kw):
111         super().__init__(**kw)
112         self.current = set()
113         self.lk = threading.Lock()
114         self.tcond = threading.Condition(self.lk)
115         self.max = max
116         self.timeout = timeout
117
118     @classmethod
119     def parseargs(cls, *, max=None, abort=None, **args):
120         ret = super().parseargs(**args)
121         if max:
122             ret["max"] = int(max)
123         if abort:
124             ret["timeout"] = int(abort)
125         return ret
126
127     def handle(self, req):
128         with self.lk:
129             if self.max is not None:
130                 if self.timeout is not None:
131                     now = start = time.time()
132                     while len(self.current) >= self.max:
133                         self.tcond.wait(start + self.timeout - now)
134                         now = time.time()
135                         if now - start > self.timeout:
136                             os.abort()
137                 else:
138                     while len(self.current) >= self.max:
139                         self.tcond.wait()
140             th = reqthread(target=self.run, args=[req])
141             th.start()
142             while th.is_alive() and th not in self.current:
143                 self.tcond.wait()
144
145     def run(self, req):
146         try:
147             th = threading.current_thread()
148             with self.lk:
149                 self.current.add(th)
150                 self.tcond.notify_all()
151             try:
152                 env = req.mkenv()
153                 with perf.request(env) as reqevent:
154                     respiter = req.handlewsgi(env, req.startreq)
155                     for data in respiter:
156                         req.write(data)
157                     if req.status:
158                         reqevent.response([req.status, req.headers])
159                         req.flushreq()
160                     self.ckflush(req)
161             except closed:
162                 pass
163             except:
164                 log.error("exception occurred when handling request", exc_info=True)
165             finally:
166                 with self.lk:
167                     self.current.remove(th)
168                     self.tcond.notify_all()
169         finally:
170             req.close()
171
172     def close(self):
173         while True:
174             with self.lk:
175                 if len(self.current) > 0:
176                     th = next(iter(self.current))
177                 else:
178                     return
179             th.join()
180
181 class threadpool(handler):
182     def __init__(self, *, min=0, max=20, live=300, **kw):
183         super().__init__(**kw)
184         self.current = set()
185         self.free = set()
186         self.lk = threading.RLock()
187         self.pcond = threading.Condition(self.lk)
188         self.rcond = threading.Condition(self.lk)
189         self.wreq = None
190         self.min = min
191         self.max = max
192         self.live = live
193         for i in range(self.min):
194             self.newthread()
195
196     @classmethod
197     def parseargs(cls, *, min=None, max=None, live=None, **args):
198         ret = super().parseargs(**args)
199         if min:
200             ret["min"] = int(min)
201         if max:
202             ret["max"] = int(max)
203         if live:
204             ret["live"] = int(live)
205         return ret
206
207     def newthread(self):
208         with self.lk:
209             th = reqthread(target=self.loop)
210             th.start()
211             while not th in self.current:
212                 self.pcond.wait()
213
214     def _handle(self, req):
215         try:
216             env = req.mkenv()
217             with perf.request(env) as reqevent:
218                 respiter = req.handlewsgi(env, req.startreq)
219                 for data in respiter:
220                     req.write(data)
221                 if req.status:
222                     reqevent.response([req.status, req.headers])
223                     req.flushreq()
224                 self.ckflush(req)
225         except closed:
226             pass
227         except:
228             log.error("exception occurred when handling request", exc_info=True)
229         finally:
230             req.close()
231
232     def loop(self):
233         th = threading.current_thread()
234         with self.lk:
235             self.current.add(th)
236         try:
237             while True:
238                 with self.lk:
239                     self.free.add(th)
240                     try:
241                         self.pcond.notify_all()
242                         now = start = time.time()
243                         while self.wreq is None:
244                             self.rcond.wait(start + self.live - now)
245                             now = time.time()
246                             if now - start > self.live:
247                                 if len(self.current) > self.min:
248                                     self.current.remove(th)
249                                     return
250                                 else:
251                                     start = now
252                         req, self.wreq = self.wreq, None
253                         self.pcond.notify_all()
254                     finally:
255                         self.free.remove(th)
256                 self._handle(req)
257                 req = None
258         finally:
259             with self.lk:
260                 try:
261                     self.current.remove(th)
262                 except KeyError:
263                     pass
264                 self.pcond.notify_all()
265
266     def handle(self, req):
267         while True:
268             with self.lk:
269                 if len(self.free) < 1 and len(self.current) < self.max:
270                     self.newthread()
271                 while self.wreq is not None:
272                     self.pcond.wait()
273                 if self.wreq is None:
274                     self.wreq = req
275                     self.rcond.notify(1)
276                     return
277
278     def close(self):
279         self.live = 0
280         self.min = 0
281         with self.lk:
282             while len(self.current) > 0:
283                 self.rcond.notify_all()
284                 self.pcond.wait(1)
285
286 class resplex(handler):
287     def __init__(self, **kw):
288         super().__init__(**kw)
289         self.current = set()
290         self.lk = threading.Lock()
291         self.cqueue = queue.Queue(5)
292         self.cnpipe = os.pipe()
293         self.rthread = reqthread(name="Response thread", target=self.handle2)
294         self.rthread.start()
295
296     def ckflush(self, req):
297         raise Exception("resplex handler does not support the write() function")
298
299     def handle(self, req):
300         reqthread(target=self.handle1, args=[req]).start()
301
302     def handle1(self, req):
303         try:
304             th = threading.current_thread()
305             with self.lk:
306                 self.current.add(th)
307             try:
308                 env = req.mkenv()
309                 respobj = req.handlewsgi(env, req.startreq)
310                 respiter = iter(respobj)
311                 if not req.status:
312                     log.error("request handler returned without calling start_request")
313                     if hasattr(respiter, "close"):
314                         respiter.close()
315                     return
316                 else:
317                     self.cqueue.put((req, respiter))
318                     os.write(self.cnpipe[1], b" ")
319                     req = None
320             finally:
321                 self.current.remove(th)
322         except closed:
323             pass
324         except:
325             log.error("exception occurred when handling request", exc_info=True)
326         finally:
327             if req is not None:
328                 req.close()
329
330     def handle2(self):
331         try:
332             rp = self.cnpipe[0]
333             current = {}
334
335             def closereq(req):
336                 respiter = current[req]
337                 try:
338                     if respiter is not None and hasattr(respiter, "close"):
339                         respiter.close()
340                 except:
341                     log.error("exception occurred when closing iterator", exc_info=True)
342                 try:
343                     req.close()
344                 except:
345                     log.error("exception occurred when closing request", exc_info=True)
346                 del current[req]
347             def ckiter(req):
348                 respiter = current[req]
349                 if respiter is not None:
350                     rem = False
351                     try:
352                         data = next(respiter)
353                     except StopIteration:
354                         rem = True
355                         req.flushreq()
356                     except:
357                         rem = True
358                         log.error("exception occurred when iterating response", exc_info=True)
359                     if not rem:
360                         if data:
361                             req.flushreq()
362                             req.writedata(data)
363                     else:
364                         current[req] = None
365                         try:
366                             if hasattr(respiter, "close"):
367                                 respiter.close()
368                         except:
369                             log.error("exception occurred when closing iterator", exc_info=True)
370                         respiter = None
371                 if respiter is None and not req.buffer:
372                     closereq(req)
373
374             while True:
375                 bufl = list(req for req in current.keys() if req.buffer)
376                 rls, wls, els = select.select([rp], bufl, [rp] + bufl)
377                 if rp in rls:
378                     ret = os.read(rp, 1024)
379                     if not ret:
380                         os.close(rp)
381                         return
382                     try:
383                         while True:
384                             req, respiter = self.cqueue.get(False)
385                             current[req] = respiter
386                             ckiter(req)
387                     except queue.Empty:
388                         pass
389                 for req in wls:
390                     try:
391                         req.flush()
392                     except closed:
393                         closereq(req)
394                     except:
395                         log.error("exception occurred when writing response", exc_info=True)
396                         closereq(req)
397                     else:
398                         if len(req.buffer) < 65536:
399                             ckiter(req)
400         except:
401             log.critical("unexpected exception occurred in response handler thread", exc_info=True)
402             os.abort()
403
404     def close(self):
405         while True:
406             with self.lk:
407                 if len(self.current) > 0:
408                     th = next(iter(self.current))
409                 else:
410                     break
411             th.join()
412         os.close(self.cnpipe[1])
413         self.rthread.join()
414
415 names = {"single": single,
416          "free": freethread,
417          "pool": threadpool,
418          "rplex": resplex}
419
420 def parsehspec(spec):
421     if ":" not in spec:
422         return spec, {}
423     nm, spec = spec.split(":", 1)
424     args = {}
425     while spec:
426         if "," in spec:
427             part, spec = spec.split(",", 1)
428         else:
429             part, spec = spec, None
430         if "=" in part:
431             key, val = part.split("=", 1)
432         else:
433             key, val = part, ""
434         args[key] = val
435     return nm, args