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

import sys, os, threading, time, logging, select, queue, collections
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):
        p = select.poll()
        p.register(req, select.POLLOUT)
        while len(req.buffer) > 0:
            p.poll()
            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):
    cname = "single"

    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 dbg(*a):
    f = True
    for o in a:
        if not f:
            sys.stderr.write(" ")
        sys.stderr.write(str(a))
        f = False
    sys.stderr.write("\n")
    sys.stderr.flush()

class freethread(handler):
    cname = "free"

    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.registered = False
            th.start()
            while not th.registered:
                self.tcond.wait()

    def run(self, req):
        try:
            th = threading.current_thread()
            with self.lk:
                self.current.add(th)
                th.registered = True
                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):
    cname = "pool"

    def __init__(self, *, max=25, qsz=100, timeout=None, **kw):
        super().__init__(**kw)
        self.current = set()
        self.clk = threading.Lock()
        self.ccond = threading.Condition(self.clk)
        self.queue = collections.deque()
        self.waiting = set()
        self.waitlimit = 5
        self.wlstart = 0.0
        self.qlk = threading.Lock()
        self.qfcond = threading.Condition(self.qlk)
        self.qecond = threading.Condition(self.qlk)
        self.max = max
        self.qsz = qsz
        self.timeout = timeout

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

    def handle(self, req):
        spawn = False
        with self.qlk:
            if self.timeout is not None:
                now = start = time.time()
                while len(self.queue) >= self.qsz:
                    self.qecond.wait(start + self.timeout - now)
                    now = time.time()
                    if now - start > self.timeout:
                        os.abort()
            else:
                while len(self.queue) >= self.qsz:
                    self.qecond.wait()
            self.queue.append(req)
            self.qfcond.notify()
            if len(self.waiting) < 1:
                spawn = True
        if spawn:
            with self.clk:
                if len(self.current) < self.max:
                    th = reqthread(target=self.run)
                    th.registered = False
                    th.start()
                    while not th.registered:
                        self.ccond.wait()

    def handle1(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)

    def run(self):
        timeout = 10.0
        th = threading.current_thread()
        with self.clk:
            self.current.add(th)
            th.registered = True
            self.ccond.notify_all()
        try:
            while True:
                start = now = time.time()
                with self.qlk:
                    while len(self.queue) < 1:
                        if len(self.waiting) >= self.waitlimit and now - self.wlstart >= timeout:
                            return
                        self.waiting.add(th)
                        try:
                            if len(self.waiting) == self.waitlimit:
                                self.wlstart = now
                            self.qfcond.wait(start + timeout - now)
                        finally:
                            self.waiting.remove(th)
                        now = time.time()
                        if now - start > timeout:
                            return
                    req = self.queue.popleft()
                    self.qecond.notify()
                try:
                    self.handle1(req)
                finally:
                    req.close()
        finally:
            with self.clk:
                self.current.remove(th)

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

class resplex(handler):
    cname = "rplex"

    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
        self.cqueue = queue.Queue(5)
        self.cnpipe = os.pipe()
        self.rthread = reqthread(name="Response thread", target=self.handle2)
        self.rthread.start()

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

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

    def handle(self, req):
        with self.lk:
            if self.max is not None:
                while len(self.current) >= self.max:
                    self.tcond.wait()
            th = reqthread(target=self.handle1, args=[req])
            th.registered = False
            th.start()
            while not th.registered:
                self.tcond.wait()

    def handle1(self, req):
        try:
            th = threading.current_thread()
            with self.lk:
                self.current.add(th)
                th.registered = True
                self.tcond.notify_all()
            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:
                with self.lk:
                    self.current.remove(th)
                    self.tcond.notify_all()
        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
                        try:
                            req.flushreq()
                        except:
                            log.error("exception occurred when handling response data", exc_info=True)
                    except:
                        rem = True
                        log.error("exception occurred when iterating response", exc_info=True)
                    if not rem:
                        if data:
                            try:
                                req.flushreq()
                                req.writedata(data)
                            except:
                                log.error("exception occurred when handling response data", exc_info=True)
                                rem = True
                    if rem:
                        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 = {cls.cname: cls for cls in globals().values() if
         isinstance(cls, type) and
         issubclass(cls, handler) and
         hasattr(cls, "cname")}

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
ab3275e9	1	import sys, os, threading, time, logging, select, queue, collections
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):
19ff507b	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):
6e8b9b9d FT	78	p = select.poll()
6e8b9b9d FT	79	p.register(req, select.POLLOUT)
dd1e6b98	80	while len(req.buffer) > 0:
6e8b9b9d	81	p.poll()
dd1e6b98	82	req.flush()
46adc298 FT	83	def close(self):
	84	pass
	85
8db41888 FT	86	@classmethod
	87	def parseargs(cls, **args):
	88	if len(args) > 0:
	89	raise ValueError("unknown handler argument: " + next(iter(args)))
	90	return {}
	91
dd1e6b98	92	class single(handler):
5345dcaf FT	93	cname = "single"
5345dcaf FT	94
dd1e6b98 FT	95	def handle(self, req):
	96	try:
	97	env = req.mkenv()
	98	with perf.request(env) as reqevent:
	99	respiter = req.handlewsgi(env, req.startreq)
	100	for data in respiter:
	101	req.write(data)
	102	if req.status:
	103	reqevent.response([req.status, req.headers])
	104	req.flushreq()
	105	self.ckflush(req)
	106	except closed:
	107	pass
	108	except:
	109	log.error("exception occurred when handling request", exc_info=True)
	110	finally:
	111	req.close()
	112
76ff6c4d FT	113	def dbg(*a):
	114	f = True
	115	for o in a:
	116	if not f:
	117	sys.stderr.write(" ")
	118	sys.stderr.write(str(a))
	119	f = False
	120	sys.stderr.write("\n")
	121	sys.stderr.flush()
	122
46adc298	123	class freethread(handler):
5345dcaf FT	124	cname = "free"
5345dcaf FT	125
3a3b78e3	126	def __init__(self, , max=None, timeout=None, *kw):
46adc298 FT	127	super().__init__(**kw)
	128	self.current = set()
	129	self.lk = threading.Lock()
002ee932 FT	130	self.tcond = threading.Condition(self.lk)
002ee932 FT	131	self.max = max
3a3b78e3	132	self.timeout = timeout
002ee932 FT	133
002ee932 FT	134	@classmethod
3a3b78e3	135	def parseargs(cls, , max=None, abort=None, *args):
002ee932 FT	136	ret = super().parseargs(**args)
	137	if max:
	138	ret["max"] = int(max)
3a3b78e3 FT	139	if abort:
3a3b78e3 FT	140	ret["timeout"] = int(abort)
002ee932	141	return ret
46adc298 FT	142
46adc298 FT	143	def handle(self, req):
002ee932	144	with self.lk:
3a3b78e3 FT	145	if self.max is not None:
	146	if self.timeout is not None:
	147	now = start = time.time()
	148	while len(self.current) >= self.max:
	149	self.tcond.wait(start + self.timeout - now)
	150	now = time.time()
	151	if now - start > self.timeout:
	152	os.abort()
	153	else:
	154	while len(self.current) >= self.max:
	155	self.tcond.wait()
002ee932	156	th = reqthread(target=self.run, args=[req])
76ff6c4d	157	th.registered = False
002ee932	158	th.start()
76ff6c4d FT	159	while not th.registered:
76ff6c4d FT	160	self.tcond.wait()
46adc298	161
46adc298	162	def run(self, req):
552a70bf	163	try:
46adc298 FT	164	th = threading.current_thread()
	165	with self.lk:
	166	self.current.add(th)
76ff6c4d	167	th.registered = True
002ee932	168	self.tcond.notify_all()
552a70bf	169	try:
46adc298 FT	170	env = req.mkenv()
	171	with perf.request(env) as reqevent:
	172	respiter = req.handlewsgi(env, req.startreq)
	173	for data in respiter:
	174	req.write(data)
	175	if req.status:
	176	reqevent.response([req.status, req.headers])
	177	req.flushreq()
	178	self.ckflush(req)
	179	except closed:
	180	pass
	181	except:
	182	log.error("exception occurred when handling request", exc_info=True)
552a70bf	183	finally:
46adc298 FT	184	with self.lk:
46adc298 FT	185	self.current.remove(th)
002ee932	186	self.tcond.notify_all()
46adc298 FT	187	finally:
	188	req.close()
	189
	190	def close(self):
	191	while True:
	192	with self.lk:
	193	if len(self.current) > 0:
	194	th = next(iter(self.current))
	195	else:
8db41888	196	return
46adc298 FT	197	th.join()
46adc298 FT	198
ab3275e9 FT	199	class threadpool(handler):
	200	cname = "pool"
	201
	202	def __init__(self, , max=25, qsz=100, timeout=None, *kw):
	203	super().__init__(**kw)
	204	self.current = set()
	205	self.clk = threading.Lock()
	206	self.ccond = threading.Condition(self.clk)
	207	self.queue = collections.deque()
	208	self.waiting = set()
62220f39 FT	209	self.waitlimit = 5
62220f39 FT	210	self.wlstart = 0.0
ab3275e9	211	self.qlk = threading.Lock()
7aed82e3 FT	212	self.qfcond = threading.Condition(self.qlk)
7aed82e3 FT	213	self.qecond = threading.Condition(self.qlk)
ab3275e9 FT	214	self.max = max
	215	self.qsz = qsz
	216	self.timeout = timeout
	217
	218	@classmethod
	219	def parseargs(cls, , max=None, queue=None, abort=None, *args):
	220	ret = super().parseargs(**args)
	221	if max:
	222	ret["max"] = int(max)
	223	if queue:
	224	ret["qsz"] = int(queue)
	225	if abort:
	226	ret["timeout"] = int(abort)
	227	return ret
	228
	229	def handle(self, req):
bf8c09bf	230	spawn = False
ab3275e9 FT	231	with self.qlk:
	232	if self.timeout is not None:
	233	now = start = time.time()
	234	while len(self.queue) >= self.qsz:
7aed82e3	235	self.qecond.wait(start + self.timeout - now)
ab3275e9 FT	236	now = time.time()
	237	if now - start > self.timeout:
	238	os.abort()
	239	else:
7aed82e3 FT	240	while len(self.queue) >= self.qsz:
7aed82e3 FT	241	self.qecond.wait()
ab3275e9	242	self.queue.append(req)
7aed82e3	243	self.qfcond.notify()
ab3275e9	244	if len(self.waiting) < 1:
bf8c09bf FT	245	spawn = True
bf8c09bf FT	246	if spawn:
ab3275e9 FT	247	with self.clk:
	248	if len(self.current) < self.max:
	249	th = reqthread(target=self.run)
	250	th.registered = False
	251	th.start()
	252	while not th.registered:
	253	self.ccond.wait()
	254
	255	def handle1(self, req):
	256	try:
	257	env = req.mkenv()
	258	with perf.request(env) as reqevent:
	259	respiter = req.handlewsgi(env, req.startreq)
	260	for data in respiter:
	261	req.write(data)
	262	if req.status:
	263	reqevent.response([req.status, req.headers])
	264	req.flushreq()
	265	self.ckflush(req)
	266	except closed:
	267	pass
	268	except:
	269	log.error("exception occurred when handling request", exc_info=True)
	270
	271	def run(self):
	272	timeout = 10.0
	273	th = threading.current_thread()
	274	with self.clk:
	275	self.current.add(th)
	276	th.registered = True
	277	self.ccond.notify_all()
	278	try:
	279	while True:
	280	start = now = time.time()
	281	with self.qlk:
	282	while len(self.queue) < 1:
62220f39 FT	283	if len(self.waiting) >= self.waitlimit and now - self.wlstart >= timeout:
62220f39 FT	284	return
ab3275e9	285	self.waiting.add(th)
62220f39 FT	286	try:
	287	if len(self.waiting) == self.waitlimit:
	288	self.wlstart = now
7aed82e3	289	self.qfcond.wait(start + timeout - now)
62220f39 FT	290	finally:
62220f39 FT	291	self.waiting.remove(th)
ab3275e9 FT	292	now = time.time()
	293	if now - start > timeout:
	294	return
	295	req = self.queue.popleft()
7aed82e3	296	self.qecond.notify()
ab3275e9 FT	297	try:
	298	self.handle1(req)
	299	finally:
	300	req.close()
	301	finally:
	302	with self.clk:
	303	self.current.remove(th)
	304
	305	def close(self):
	306	while True:
62220f39	307	with self.clk:
ab3275e9 FT	308	if len(self.current) > 0:
	309	th = next(iter(self.current))
	310	else:
	311	return
	312	th.join()
	313
a962c944	314	class resplex(handler):
5345dcaf FT	315	cname = "rplex"
5345dcaf FT	316
48dc900a	317	def __init__(self, , max=None, *kw):
a962c944 FT	318	super().__init__(**kw)
	319	self.current = set()
	320	self.lk = threading.Lock()
48dc900a FT	321	self.tcond = threading.Condition(self.lk)
48dc900a FT	322	self.max = max
a962c944 FT	323	self.cqueue = queue.Queue(5)
	324	self.cnpipe = os.pipe()
	325	self.rthread = reqthread(name="Response thread", target=self.handle2)
	326	self.rthread.start()
	327
48dc900a FT	328	@classmethod
	329	def parseargs(cls, , max=None, *args):
	330	ret = super().parseargs(**args)
	331	if max:
	332	ret["max"] = int(max)
	333	return ret
	334
a962c944 FT	335	def ckflush(self, req):
	336	raise Exception("resplex handler does not support the write() function")
	337
	338	def handle(self, req):
48dc900a FT	339	with self.lk:
	340	if self.max is not None:
	341	while len(self.current) >= self.max:
	342	self.tcond.wait()
	343	th = reqthread(target=self.handle1, args=[req])
76ff6c4d	344	th.registered = False
48dc900a	345	th.start()
76ff6c4d FT	346	while not th.registered:
76ff6c4d FT	347	self.tcond.wait()
a962c944 FT	348
	349	def handle1(self, req):
	350	try:
	351	th = threading.current_thread()
	352	with self.lk:
	353	self.current.add(th)
76ff6c4d	354	th.registered = True
48dc900a	355	self.tcond.notify_all()
a962c944 FT	356	try:
	357	env = req.mkenv()
	358	respobj = req.handlewsgi(env, req.startreq)
	359	respiter = iter(respobj)
	360	if not req.status:
	361	log.error("request handler returned without calling start_request")
	362	if hasattr(respiter, "close"):
	363	respiter.close()
	364	return
	365	else:
	366	self.cqueue.put((req, respiter))
	367	os.write(self.cnpipe[1], b" ")
	368	req = None
	369	finally:
48dc900a FT	370	with self.lk:
	371	self.current.remove(th)
	372	self.tcond.notify_all()
a962c944 FT	373	except closed:
	374	pass
	375	except:
	376	log.error("exception occurred when handling request", exc_info=True)
	377	finally:
	378	if req is not None:
	379	req.close()
	380
	381	def handle2(self):
	382	try:
	383	rp = self.cnpipe[0]
	384	current = {}
	385
	386	def closereq(req):
	387	respiter = current[req]
	388	try:
	389	if respiter is not None and hasattr(respiter, "close"):
	390	respiter.close()
	391	except:
	392	log.error("exception occurred when closing iterator", exc_info=True)
	393	try:
	394	req.close()
	395	except:
	396	log.error("exception occurred when closing request", exc_info=True)
	397	del current[req]
	398	def ckiter(req):
	399	respiter = current[req]
	400	if respiter is not None:
	401	rem = False
	402	try:
	403	data = next(respiter)
	404	except StopIteration:
	405	rem = True
b0a7be65 FT	406	try:
	407	req.flushreq()
	408	except:
	409	log.error("exception occurred when handling response data", exc_info=True)
a962c944 FT	410	except:
	411	rem = True
	412	log.error("exception occurred when iterating response", exc_info=True)
	413	if not rem:
	414	if data:
b0a7be65 FT	415	try:
	416	req.flushreq()
	417	req.writedata(data)
	418	except:
	419	log.error("exception occurred when handling response data", exc_info=True)
	420	rem = True
	421	if rem:
a962c944 FT	422	current[req] = None
	423	try:
	424	if hasattr(respiter, "close"):
	425	respiter.close()
	426	except:
	427	log.error("exception occurred when closing iterator", exc_info=True)
	428	respiter = None
	429	if respiter is None and not req.buffer:
	430	closereq(req)
	431
	432	while True:
	433	bufl = list(req for req in current.keys() if req.buffer)
	434	rls, wls, els = select.select([rp], bufl, [rp] + bufl)
	435	if rp in rls:
	436	ret = os.read(rp, 1024)
	437	if not ret:
	438	os.close(rp)
	439	return
	440	try:
	441	while True:
	442	req, respiter = self.cqueue.get(False)
	443	current[req] = respiter
	444	ckiter(req)
	445	except queue.Empty:
	446	pass
	447	for req in wls:
	448	try:
	449	req.flush()
	450	except closed:
	451	closereq(req)
	452	except:
	453	log.error("exception occurred when writing response", exc_info=True)
	454	closereq(req)
	455	else:
	456	if len(req.buffer) < 65536:
	457	ckiter(req)
	458	except:
	459	log.critical("unexpected exception occurred in response handler thread", exc_info=True)
1b0caaa0	460	os.abort()
a962c944 FT	461
	462	def close(self):
	463	while True:
	464	with self.lk:
	465	if len(self.current) > 0:
	466	th = next(iter(self.current))
	467	else:
	468	break
	469	th.join()
	470	os.close(self.cnpipe[1])
	471	self.rthread.join()
	472
5345dcaf FT	473	names = {cls.cname: cls for cls in globals().values() if
	474	isinstance(cls, type) and
	475	issubclass(cls, handler) and
	476	hasattr(cls, "cname")}
8db41888 FT	477
	478	def parsehspec(spec):
	479	if ":" not in spec:
	480	return spec, {}
	481	nm, spec = spec.split(":", 1)
	482	args = {}
	483	while spec:
	484	if "," in spec:
	485	part, spec = spec.split(",", 1)
	486	else:
	487	part, spec = spec, None
	488	if "=" in part:
	489	key, val = part.split("=", 1)
	490	else:
	491	key, val = part, ""
	492	args[key] = val
	493	return nm, args