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

import sys, os, threading, time, logging, select
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):
        raise Exception()
    def close(self):
        pass

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

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

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

    def handle(self, req):
        with self.lk:
            while self.max is not None and 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 ckflush(self, req):
        while len(req.buffer) > 0:
            rls, wls, els = select.select([], [req], [req])
            req.flush()

    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 ckflush(self, req):
        while len(req.buffer) > 0:
            rls, wls, els = select.select([], [req], [req])
            req.flush()

    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)

names = {"free": freethread,
         "pool": threadpool}

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
46adc298 FT	1	import sys, os, threading, time, logging, select
46adc298 FT	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):
	78	raise Exception()
	79	def close(self):
	80	pass
	81
8db41888 FT	82	@classmethod
	83	def parseargs(cls, **args):
	84	if len(args) > 0:
	85	raise ValueError("unknown handler argument: " + next(iter(args)))
	86	return {}
	87
46adc298	88	class freethread(handler):
002ee932	89	def __init__(self, , max=None, *kw):
46adc298 FT	90	super().__init__(**kw)
	91	self.current = set()
	92	self.lk = threading.Lock()
002ee932 FT	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
46adc298 FT	102
46adc298 FT	103	def handle(self, req):
002ee932 FT	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()
46adc298 FT	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):
552a70bf	118	try:
46adc298 FT	119	th = threading.current_thread()
	120	with self.lk:
	121	self.current.add(th)
002ee932	122	self.tcond.notify_all()
552a70bf	123	try:
46adc298 FT	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)
552a70bf	137	finally:
46adc298 FT	138	with self.lk:
46adc298 FT	139	self.current.remove(th)
002ee932	140	self.tcond.notify_all()
46adc298 FT	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:
8db41888	150	return
46adc298 FT	151	th.join()
46adc298 FT	152
d570c3a5	153	class threadpool(handler):
8db41888	154	def __init__(self, , min=0, max=20, live=300, *kw):
d570c3a5 FT	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
8db41888 FT	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
d570c3a5 FT	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}
8db41888 FT	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