python3/ashd/serve.py

   1 import sys, os, threading, time, logging, select
   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         raise Exception()
  79     def close(self):
  80         pass
  81
  82     @classmethod
  83     def parseargs(cls, **args):
  84         if len(args) > 0:
  85             raise ValueError("unknown handler argument: " + next(iter(args)))
  86         return {}
  87
  88 class freethread(handler):
  89     def __init__(self, *, max=None, **kw):
  90         super().__init__(**kw)
  91         self.current = set()
  92         self.lk = threading.Lock()
  93         self.tcond = threading.Condition(self.lk)
  94         self.max = max
  95
  96     @classmethod
  97     def parseargs(cls, *, max=None, **args):
  98         ret = super().parseargs(**args)
  99         if max:
 100             ret["max"] = int(max)
 101         return ret
 102
 103     def handle(self, req):
 104         with self.lk:
 105             while self.max is not None and len(self.current) >= self.max:
 106                 self.tcond.wait()
 107             th = reqthread(target=self.run, args=[req])
 108             th.start()
 109             while th.is_alive() and th not in self.current:
 110                 self.tcond.wait()
 111
 112     def ckflush(self, req):
 113         while len(req.buffer) > 0:
 114             rls, wls, els = select.select([], [req], [req])
 115             req.flush()
 116
 117     def run(self, req):
 118         try:
 119             th = threading.current_thread()
 120             with self.lk:
 121                 self.current.add(th)
 122                 self.tcond.notify_all()
 123             try:
 124                 env = req.mkenv()
 125                 with perf.request(env) as reqevent:
 126                     respiter = req.handlewsgi(env, req.startreq)
 127                     for data in respiter:
 128                         req.write(data)
 129                     if req.status:
 130                         reqevent.response([req.status, req.headers])
 131                         req.flushreq()
 132                     self.ckflush(req)
 133             except closed:
 134                 pass
 135             except:
 136                 log.error("exception occurred when handling request", exc_info=True)
 137             finally:
 138                 with self.lk:
 139                     self.current.remove(th)
 140                     self.tcond.notify_all()
 141         finally:
 142             req.close()
 143
 144     def close(self):
 145         while True:
 146             with self.lk:
 147                 if len(self.current) > 0:
 148                     th = next(iter(self.current))
 149                 else:
 150                     return
 151             th.join()
 152
 153 class threadpool(handler):
 154     def __init__(self, *, min=0, max=20, live=300, **kw):
 155         super().__init__(**kw)
 156         self.current = set()
 157         self.free = set()
 158         self.lk = threading.RLock()
 159         self.pcond = threading.Condition(self.lk)
 160         self.rcond = threading.Condition(self.lk)
 161         self.wreq = None
 162         self.min = min
 163         self.max = max
 164         self.live = live
 165         for i in range(self.min):
 166             self.newthread()
 167
 168     @classmethod
 169     def parseargs(cls, *, min=None, max=None, live=None, **args):
 170         ret = super().parseargs(**args)
 171         if min:
 172             ret["min"] = int(min)
 173         if max:
 174             ret["max"] = int(max)
 175         if live:
 176             ret["live"] = int(live)
 177         return ret
 178
 179     def newthread(self):
 180         with self.lk:
 181             th = reqthread(target=self.loop)
 182             th.start()
 183             while not th in self.current:
 184                 self.pcond.wait()
 185
 186     def ckflush(self, req):
 187         while len(req.buffer) > 0:
 188             rls, wls, els = select.select([], [req], [req])
 189             req.flush()
 190
 191     def _handle(self, req):
 192         try:
 193             env = req.mkenv()
 194             with perf.request(env) as reqevent:
 195                 respiter = req.handlewsgi(env, req.startreq)
 196                 for data in respiter:
 197                     req.write(data)
 198                 if req.status:
 199                     reqevent.response([req.status, req.headers])
 200                     req.flushreq()
 201                 self.ckflush(req)
 202         except closed:
 203             pass
 204         except:
 205             log.error("exception occurred when handling request", exc_info=True)
 206         finally:
 207             req.close()
 208
 209     def loop(self):
 210         th = threading.current_thread()
 211         with self.lk:
 212             self.current.add(th)
 213         try:
 214             while True:
 215                 with self.lk:
 216                     self.free.add(th)
 217                     try:
 218                         self.pcond.notify_all()
 219                         now = start = time.time()
 220                         while self.wreq is None:
 221                             self.rcond.wait(start + self.live - now)
 222                             now = time.time()
 223                             if now - start > self.live:
 224                                 if len(self.current) > self.min:
 225                                     self.current.remove(th)
 226                                     return
 227                                 else:
 228                                     start = now
 229                         req, self.wreq = self.wreq, None
 230                         self.pcond.notify_all()
 231                     finally:
 232                         self.free.remove(th)
 233                 self._handle(req)
 234                 req = None
 235         finally:
 236             with self.lk:
 237                 try:
 238                     self.current.remove(th)
 239                 except KeyError:
 240                     pass
 241                 self.pcond.notify_all()
 242
 243     def handle(self, req):
 244         while True:
 245             with self.lk:
 246                 if len(self.free) < 1 and len(self.current) < self.max:
 247                     self.newthread()
 248                 while self.wreq is not None:
 249                     self.pcond.wait()
 250                 if self.wreq is None:
 251                     self.wreq = req
 252                     self.rcond.notify(1)
 253                     return
 254
 255     def close(self):
 256         self.live = 0
 257         self.min = 0
 258         with self.lk:
 259             while len(self.current) > 0:
 260                 self.rcond.notify_all()
 261                 self.pcond.wait(1)
 262
 263 names = {"free": freethread,
 264          "pool": threadpool}
 265
 266 def parsehspec(spec):
 267     if ":" not in spec:
 268         return spec, {}
 269     nm, spec = spec.split(":", 1)
 270     args = {}
 271     while spec:
 272         if "," in spec:
 273             part, spec = spec.split(",", 1)
 274         else:
 275             part, spec = spec, None
 276         if "=" in part:
 277             key, val = part.split("=", 1)
 278         else:
 279             key, val = part, ""
 280         args[key] = val
 281     return nm, args