[ashd.git] / python3 / ashd / serve.py

import sys, os, threading, time, logging, select, queue
from . import perf

log = logging.getLogger("ashd.serve")
seq = 1
seqlk = threading.Lock()

def reqseq():
    global seq
    with seqlk:
        s = seq
        seq += 1
        return s

class closed(IOError):
    def __init__(self):
        super().__init__("The client has closed the connection.")

class reqthread(threading.Thread):
    def __init__(self, *, name=None, **kw):
        if name is None:
            name = "Request handler %i" % reqseq()
        super().__init__(name=name, **kw)

class wsgirequest(object):
    def __init__(self, handler):
        self.status = None
        self.headers = []
        self.respsent = False
        self.handler = handler
        self.buffer = bytearray()

    def handlewsgi(self):
        raise Exception()
    def fileno(self):
        raise Exception()
    def writehead(self, status, headers):
        raise Exception()
    def flush(self):
        raise Exception()
    def close(self):
        pass
    def writedata(self, data):
        self.buffer.extend(data)

    def flushreq(self):
        if not self.respsent:
            if not self.status:
                raise Exception("Cannot send response body before starting response.")
            self.respsent = True
            self.writehead(self.status, self.headers)

    def write(self, data):
        if not data:
            return
        self.flushreq()
        self.writedata(data)
        self.handler.ckflush(self)

    def startreq(self, status, headers, exc_info=None):
        if self.status:
            if exc_info:
                try:
                    if self.respsent:
                        raise exc_info[1]
                finally:
                    exc_info = None
            else:
                raise Exception("Can only start responding once.")
        self.status = status
        self.headers = headers
        return self.write

class handler(object):
    def handle(self, request):
        raise Exception()
    def ckflush(self, req):
        while len(req.buffer) > 0:
            rls, wls, els = select.select([], [req], [req])
            req.flush()
    def close(self):
        pass

    @classmethod
    def parseargs(cls, **args):
        if len(args) > 0:
            raise ValueError("unknown handler argument: " + next(iter(args)))
        return {}

class single(handler):
    def handle(self, req):
        try:
            env = req.mkenv()
            with perf.request(env) as reqevent:
                respiter = req.handlewsgi(env, req.startreq)
                for data in respiter:
                    req.write(data)
                if req.status:
                    reqevent.response([req.status, req.headers])
                    req.flushreq()
                self.ckflush(req)
        except closed:
            pass
        except:
            log.error("exception occurred when handling request", exc_info=True)
        finally:
            req.close()

class freethread(handler):
    def __init__(self, *, max=None, timeout=None, **kw):
        super().__init__(**kw)
        self.current = set()
        self.lk = threading.Lock()
        self.tcond = threading.Condition(self.lk)
        self.max = max
        self.timeout = timeout

    @classmethod
    def parseargs(cls, *, max=None, abort=None, **args):
        ret = super().parseargs(**args)
        if max:
            ret["max"] = int(max)
        if abort:
            ret["timeout"] = int(abort)
        return ret

    def handle(self, req):
        with self.lk:
            if self.max is not None:
                if self.timeout is not None:
                    now = start = time.time()
                    while len(self.current) >= self.max:
                        self.tcond.wait(start + self.timeout - now)
                        now = time.time()
                        if now - start > self.timeout:
                            os.abort()
                else:
                    while len(self.current) >= self.max:
                        self.tcond.wait()
            th = reqthread(target=self.run, args=[req])
            th.start()
            while th.is_alive() and th not in self.current:
                self.tcond.wait()

    def run(self, req):
        try:
            th = threading.current_thread()
            with self.lk:
                self.current.add(th)
                self.tcond.notify_all()
            try:
                env = req.mkenv()
                with perf.request(env) as reqevent:
                    respiter = req.handlewsgi(env, req.startreq)
                    for data in respiter:
                        req.write(data)
                    if req.status:
                        reqevent.response([req.status, req.headers])
                        req.flushreq()
                    self.ckflush(req)
            except closed:
                pass
            except:
                log.error("exception occurred when handling request", exc_info=True)
            finally:
                with self.lk:
                    self.current.remove(th)
                    self.tcond.notify_all()
        finally:
            req.close()

    def close(self):
        while True:
            with self.lk:
                if len(self.current) > 0:
                    th = next(iter(self.current))
                else:
                    return
            th.join()

class threadpool(handler):
    def __init__(self, *, min=0, max=20, live=300, **kw):
        super().__init__(**kw)
        self.current = set()
        self.free = set()
        self.lk = threading.RLock()
        self.pcond = threading.Condition(self.lk)
        self.rcond = threading.Condition(self.lk)
        self.wreq = None
        self.min = min
        self.max = max
        self.live = live
        for i in range(self.min):
            self.newthread()

    @classmethod
    def parseargs(cls, *, min=None, max=None, live=None, **args):
        ret = super().parseargs(**args)
        if min:
            ret["min"] = int(min)
        if max:
            ret["max"] = int(max)
        if live:
            ret["live"] = int(live)
        return ret

    def newthread(self):
        with self.lk:
            th = reqthread(target=self.loop)
            th.start()
            while not th in self.current:
                self.pcond.wait()

    def _handle(self, req):
        try:
            env = req.mkenv()
            with perf.request(env) as reqevent:
                respiter = req.handlewsgi(env, req.startreq)
                for data in respiter:
                    req.write(data)
                if req.status:
                    reqevent.response([req.status, req.headers])
                    req.flushreq()
                self.ckflush(req)
        except closed:
            pass
        except:
            log.error("exception occurred when handling request", exc_info=True)
        finally:
            req.close()

    def loop(self):
        th = threading.current_thread()
        with self.lk:
            self.current.add(th)
        try:
            while True:
                with self.lk:
                    self.free.add(th)
                    try:
                        self.pcond.notify_all()
                        now = start = time.time()
                        while self.wreq is None:
                            self.rcond.wait(start + self.live - now)
                            now = time.time()
                            if now - start > self.live:
                                if len(self.current) > self.min:
                                    self.current.remove(th)
                                    return
                                else:
                                    start = now
                        req, self.wreq = self.wreq, None
                        self.pcond.notify_all()
                    finally:
                        self.free.remove(th)
                self._handle(req)
                req = None
        finally:
            with self.lk:
                try:
                    self.current.remove(th)
                except KeyError:
                    pass
                self.pcond.notify_all()

    def handle(self, req):
        while True:
            with self.lk:
                if len(self.free) < 1 and len(self.current) < self.max:
                    self.newthread()
                while self.wreq is not None:
                    self.pcond.wait()
                if self.wreq is None:
                    self.wreq = req
                    self.rcond.notify(1)
                    return

    def close(self):
        self.live = 0
        self.min = 0
        with self.lk:
            while len(self.current) > 0:
                self.rcond.notify_all()
                self.pcond.wait(1)

class resplex(handler):
    def __init__(self, **kw):
        super().__init__(**kw)
        self.current = set()
        self.lk = threading.Lock()
        self.cqueue = queue.Queue(5)
        self.cnpipe = os.pipe()
        self.rthread = reqthread(name="Response thread", target=self.handle2)
        self.rthread.start()

    def ckflush(self, req):
        raise Exception("resplex handler does not support the write() function")

    def handle(self, req):
        reqthread(target=self.handle1, args=[req]).start()

    def handle1(self, req):
        try:
            th = threading.current_thread()
            with self.lk:
                self.current.add(th)
            try:
                env = req.mkenv()
                respobj = req.handlewsgi(env, req.startreq)
                respiter = iter(respobj)
                if not req.status:
                    log.error("request handler returned without calling start_request")
                    if hasattr(respiter, "close"):
                        respiter.close()
                    return
                else:
                    self.cqueue.put((req, respiter))
                    os.write(self.cnpipe[1], b" ")
                    req = None
            finally:
                self.current.remove(th)
        except closed:
            pass
        except:
            log.error("exception occurred when handling request", exc_info=True)
        finally:
            if req is not None:
                req.close()

    def handle2(self):
        try:
            rp = self.cnpipe[0]
            current = {}

            def closereq(req):
                respiter = current[req]
                try:
                    if respiter is not None and hasattr(respiter, "close"):
                        respiter.close()
                except:
                    log.error("exception occurred when closing iterator", exc_info=True)
                try:
                    req.close()
                except:
                    log.error("exception occurred when closing request", exc_info=True)
                del current[req]
            def ckiter(req):
                respiter = current[req]
                if respiter is not None:
                    rem = False
                    try:
                        data = next(respiter)
                    except StopIteration:
                        rem = True
                        req.flushreq()
                    except:
                        rem = True
                        log.error("exception occurred when iterating response", exc_info=True)
                    if not rem:
                        if data:
                            req.flushreq()
                            req.writedata(data)
                    else:
                        current[req] = None
                        try:
                            if hasattr(respiter, "close"):
                                respiter.close()
                        except:
                            log.error("exception occurred when closing iterator", exc_info=True)
                        respiter = None
                if respiter is None and not req.buffer:
                    closereq(req)

            while True:
                bufl = list(req for req in current.keys() if req.buffer)
                rls, wls, els = select.select([rp], bufl, [rp] + bufl)
                if rp in rls:
                    ret = os.read(rp, 1024)
                    if not ret:
                        os.close(rp)
                        return
                    try:
                        while True:
                            req, respiter = self.cqueue.get(False)
                            current[req] = respiter
                            ckiter(req)
                    except queue.Empty:
                        pass
                for req in wls:
                    try:
                        req.flush()
                    except closed:
                        closereq(req)
                    except:
                        log.error("exception occurred when writing response", exc_info=True)
                        closereq(req)
                    else:
                        if len(req.buffer) < 65536:
                            ckiter(req)
        except:
            log.critical("unexpected exception occurred in response handler thread", exc_info=True)
            os.abort()

    def close(self):
        while True:
            with self.lk:
                if len(self.current) > 0:
                    th = next(iter(self.current))
                else:
                    break
            th.join()
        os.close(self.cnpipe[1])
        self.rthread.join()

names = {"single": single,
         "free": freethread,
         "pool": threadpool,
         "rplex": resplex}

def parsehspec(spec):
    if ":" not in spec:
        return spec, {}
    nm, spec = spec.split(":", 1)
    args = {}
    while spec:
        if "," in spec:
            part, spec = spec.split(",", 1)
        else:
            part, spec = spec, None
        if "=" in part:
            key, val = part.split("=", 1)
        else:
            key, val = part, ""
        args[key] = val
    return nm, args
Commit	Line	Data
a962c944	1	import sys, os, threading, time, logging, select, queue
46adc298	2	from . import perf
552a70bf FT	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
552a70bf FT	15	class closed(IOError):
	16	def __init__(self):
	17	super().__init__("The client has closed the connection.")
	18
46adc298 FT	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):
552a70bf FT	27	self.status = None
	28	self.headers = []
	29	self.respsent = False
46adc298 FT	30	self.handler = handler
46adc298 FT	31	self.buffer = bytearray()
552a70bf FT	32
	33	def handlewsgi(self):
	34	raise Exception()
46adc298 FT	35	def fileno(self):
46adc298 FT	36	raise Exception()
552a70bf FT	37	def writehead(self, status, headers):
552a70bf FT	38	raise Exception()
46adc298	39	def flush(self):
552a70bf	40	raise Exception()
46adc298 FT	41	def close(self):
	42	pass
	43	def writedata(self, data):
	44	self.buffer.extend(data)
552a70bf 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
46adc298 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
552a70bf FT	60	def startreq(self, status, headers, exc_info=None):
552a70bf FT	61	if self.status:
46adc298	62	if exc_info:
552a70bf FT	63	try:
	64	if self.respsent:
	65	raise exc_info[1]
	66	finally:
46adc298	67	exc_info = None
552a70bf FT	68	else:
	69	raise Exception("Can only start responding once.")
	70	self.status = status
	71	self.headers = headers
	72	return self.write
46adc298 FT	73
	74	class handler(object):
	75	def handle(self, request):
	76	raise Exception()
	77	def ckflush(self, req):
dd1e6b98 FT	78	while len(req.buffer) > 0:
	79	rls, wls, els = select.select([], [req], [req])
	80	req.flush()
46adc298 FT	81	def close(self):
	82	pass
	83
8db41888 FT	84	@classmethod
	85	def parseargs(cls, **args):
	86	if len(args) > 0:
	87	raise ValueError("unknown handler argument: " + next(iter(args)))
	88	return {}
	89
dd1e6b98 FT	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
46adc298	109	class freethread(handler):
3a3b78e3	110	def __init__(self, , max=None, timeout=None, *kw):
46adc298 FT	111	super().__init__(**kw)
	112	self.current = set()
	113	self.lk = threading.Lock()
002ee932 FT	114	self.tcond = threading.Condition(self.lk)
002ee932 FT	115	self.max = max
3a3b78e3	116	self.timeout = timeout
002ee932 FT	117
002ee932 FT	118	@classmethod
3a3b78e3	119	def parseargs(cls, , max=None, abort=None, *args):
002ee932 FT	120	ret = super().parseargs(**args)
	121	if max:
	122	ret["max"] = int(max)
3a3b78e3 FT	123	if abort:
3a3b78e3 FT	124	ret["timeout"] = int(abort)
002ee932	125	return ret
46adc298 FT	126
46adc298 FT	127	def handle(self, req):
002ee932	128	with self.lk:
3a3b78e3 FT	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()
002ee932 FT	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()
46adc298	144
46adc298	145	def run(self, req):
552a70bf	146	try:
46adc298 FT	147	th = threading.current_thread()
	148	with self.lk:
	149	self.current.add(th)
002ee932	150	self.tcond.notify_all()
552a70bf	151	try:
46adc298 FT	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)
552a70bf	165	finally:
46adc298 FT	166	with self.lk:
46adc298 FT	167	self.current.remove(th)
002ee932	168	self.tcond.notify_all()
46adc298 FT	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:
8db41888	178	return
46adc298 FT	179	th.join()
46adc298 FT	180
d570c3a5	181	class threadpool(handler):
8db41888	182	def __init__(self, , min=0, max=20, live=300, *kw):
d570c3a5 FT	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
8db41888 FT	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
d570c3a5 FT	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
d570c3a5 FT	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
a962c944 FT	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)
1b0caaa0	402	os.abort()
a962c944 FT	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
dd1e6b98 FT	415	names = {"single": single,
dd1e6b98 FT	416	"free": freethread,
a962c944 FT	417	"pool": threadpool,
a962c944 FT	418	"rplex": resplex}
8db41888 FT	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