<s> from collections import deque <EOL> from datetime import datetime <EOL> import errno <EOL> from functools import partial <EOL> import os <EOL> import socket <EOL> import ssl <EOL> import sys <EOL> from threading import RLock <EOL> import time <EOL> from . . import http <EOL> from . . http import wsgi <EOL> from . . import util <EOL> from . import base <EOL> from . . import six <EOL> try : <EOL> import concurrent . futures as futures <EOL> except ImportError : <EOL> raise RuntimeError ( """<STR_LIT>""" ) <EOL> try : <EOL> from asyncio import selectors <EOL> except ImportError : <EOL> from gunicorn import selectors <EOL> class TConn ( object ) : <EOL> def __init__ ( self , cfg , listener , sock , addr ) : <EOL> self . cfg = cfg <EOL> self . listener = listener <EOL> self . sock = sock <EOL> self . addr = addr <EOL> self . timeout = None <EOL> self . parser = None <EOL> self . sock . setblocking ( False ) <EOL> def init ( self ) : <EOL> self . sock . setblocking ( True ) <EOL> if self . parser is None : <EOL> if self . cfg . is_ssl : <EOL> self . sock = ssl . wrap_socket ( self . sock , server_side = True , <EOL> ** self . cfg . ssl_options ) <EOL> self . parser = http . RequestParser ( self . cfg , self . sock ) <EOL> def set_timeout ( self ) : <EOL> self . timeout = time . time ( ) + self . cfg . keepalive <EOL> def close ( self ) : <EOL> util . close ( self . sock ) <EOL> def __lt__ ( self , other ) : <EOL> return self . timeout < other . timeout <EOL> __cmp__ = __lt__ <EOL> class ThreadWorker ( base . Worker ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( ThreadWorker , self ) . __init__ ( * args , ** kwargs ) <EOL> self . worker_connections = self . cfg . worker_connections <EOL> self . max_keepalived = self . cfg . worker_connections - self . cfg . threads <EOL> self . tpool = None <EOL> self . poller = None <EOL> self . _lock = None <EOL> self . futures = deque ( ) <EOL> self . _keep = deque ( ) <EOL> self . nr_conns = <NUM_LIT:0> <EOL> @ classmethod <EOL> def check_config ( cls , cfg , log ) : <EOL> max_keepalived = cfg . worker_connections - cfg . threads <EOL> if max_keepalived <= <NUM_LIT:0> and cfg . keepalive : <EOL> log . warning ( "<STR_LIT>" + <EOL> "<STR_LIT>" ) <EOL> def init_process ( self ) : <EOL> self . tpool = futures . ThreadPoolExecutor ( max_workers = self . cfg . threads ) <EOL> self . poller = selectors . DefaultSelector ( ) <EOL> self . _lock = RLock ( ) <EOL> super ( ThreadWorker , self ) . init_process ( ) <EOL> def handle_quit ( self , sig , frame ) : <EOL> self . alive = False <EOL> self . cfg . worker_int ( self ) <EOL> self . tpool . shutdown ( False ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> sys . exit ( <NUM_LIT:0> ) <EOL> def _wrap_future ( self , fs , conn ) : <EOL> fs . conn = conn <EOL> self . futures . append ( fs ) <EOL> fs . add_done_callback ( self . finish_request ) <EOL> def enqueue_req ( self , conn ) : <EOL> conn . init ( ) <EOL> fs = self . tpool . submit ( self . handle , conn ) <EOL> self . _wrap_future ( fs , conn ) <EOL> def accept ( self , listener ) : <EOL> try : <EOL> client , addr = listener . accept ( ) <EOL> conn = TConn ( self . cfg , listener , client , addr ) <EOL> self . nr_conns += <NUM_LIT:1> <EOL> self . enqueue_req ( conn ) <EOL> except socket . error as e : <EOL> if e . args [ <NUM_LIT:0> ] not in ( errno . EAGAIN , <EOL> errno . ECONNABORTED , errno . EWOULDBLOCK ) : <EOL> raise <EOL> def reuse_connection ( self , conn , client ) : <EOL> with self . _lock : <EOL> self . poller . unregister ( client ) <EOL> try : <EOL> self . _keep . remove ( conn ) <EOL> except ValueError : <EOL> return <EOL> self . enqueue_req ( conn ) <EOL> def murder_keepalived ( self ) : <EOL> now = time . time ( ) <EOL> while True : <EOL> with self . _lock : <EOL> try : <EOL> conn = self . _keep . popleft ( ) <EOL> except IndexError : <EOL> break <EOL> delta = conn . timeout - now <EOL> if delta > <NUM_LIT:0> : <EOL> with self . _lock : <EOL> self . _keep . appendleft ( conn ) <EOL> break <EOL> else : <EOL> self . nr_conns -= <NUM_LIT:1> <EOL> with self . _lock : <EOL> try : <EOL> self . poller . unregister ( conn . sock ) <EOL> except socket . error as e : <EOL> if e . args [ <NUM_LIT:0> ] != errno . EBADF : <EOL> raise <EOL> conn . close ( ) <EOL> def is_parent_alive ( self ) : <EOL> if self . ppid != os . getppid ( ) : <EOL> self . log . info ( "<STR_LIT>" , self ) <EOL> return False <EOL> return True <EOL> def run ( self ) : <EOL> for s in self . sockets : <EOL> s . setblocking ( False ) <EOL> self . poller . register ( s , selectors . EVENT_READ , self . accept ) <EOL> timeout = self . cfg . timeout or <NUM_LIT:0.5> <EOL> while self . alive : <EOL> self . notify ( ) <EOL> if self . nr_conns < self . worker_connections : <EOL> events = self . poller . select ( <NUM_LIT> ) <EOL> for key , mask in events : <EOL> callback = key . data <EOL> callback ( key . fileobj ) <EOL> if not self . is_parent_alive ( ) : <EOL> break <EOL> self . murder_keepalived ( ) <EOL> if len ( self . futures ) < self . cfg . threads : <EOL> continue <EOL> result = futures . wait ( self . futures , timeout = timeout , <EOL> return_when = futures . FIRST_COMPLETED ) <EOL> if not result . done : <EOL> break <EOL> else : <EOL> [ self . futures . remove ( f ) for f in result . done ] <EOL> self . tpool . shutdown ( False ) <EOL> self . poller . close ( ) <EOL> def finish_request ( self , fs ) : <EOL> if fs . cancelled ( ) : <EOL> fs . conn . close ( ) <EOL> return <EOL> try : <EOL> ( keepalive , conn ) = fs . result ( ) <EOL> if keepalive : <EOL> conn . sock . setblocking ( False ) <EOL> conn . set_timeout ( ) <EOL> with self . _lock : <EOL> self . _keep . append ( conn ) <EOL> self . poller . register ( conn . sock , selectors . EVENT_READ , <EOL> partial ( self . reuse_connection , conn ) ) <EOL> else : <EOL> self . nr_conns -= <NUM_LIT:1> <EOL> conn . close ( ) <EOL> except : <EOL> self . nr_conns -= <NUM_LIT:1> <EOL> fs . conn . close ( ) <EOL> def handle ( self , conn ) : <EOL> keepalive = False <EOL> req = None <EOL> try : <EOL> req = six . next ( conn . parser ) <EOL> if not req : <EOL> return ( False , conn ) <EOL> keepalive = self . handle_request ( req , conn ) <EOL> if keepalive : <EOL> return ( keepalive , conn ) <EOL> except http . errors . NoMoreData as e : <EOL> self . log . debug ( "<STR_LIT>" , e ) <EOL> except StopIteration as e : <EOL> self . log . debug ( "<STR_LIT>" , e ) <EOL> except ssl . SSLError as e : <EOL> if e . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_EOF : <EOL> self . log . debug ( "<STR_LIT>" ) <EOL> conn . sock . close ( ) <EOL> else : <EOL> self . log . debug ( "<STR_LIT>" ) <EOL> self . handle_error ( req , conn . sock , conn . addr , e ) <EOL> except socket . error as e : <EOL> if e . args [ <NUM_LIT:0> ] not in ( errno . EPIPE , errno . ECONNRESET ) : <EOL> self . log . exception ( "<STR_LIT>" ) <EOL> else : <EOL> if e . args [ <NUM_LIT:0> ] == errno . ECONNRESET : <EOL> self . log . debug ( "<STR_LIT>" ) <EOL> else : <EOL> self . log . debug ( "<STR_LIT>" ) <EOL> except Exception as e : <EOL> self . handle_error ( req , conn . sock , conn . addr , e ) <EOL> return ( False , conn ) <EOL> def handle_request ( self , req , conn ) : <EOL> environ = { } <EOL> resp = None <EOL> try : <EOL> self . cfg . pre_request ( self , req ) <EOL> request_start = datetime . now ( ) <EOL> resp , environ = wsgi . create ( req , conn . sock , conn . addr , <EOL> conn . listener . getsockname ( ) , self . cfg ) <EOL> environ [ "<STR_LIT>" ] = True <EOL> self . nr += <NUM_LIT:1> <EOL> if self . alive and self . nr >= self . max_requests : <EOL> self . log . info ( "<STR_LIT>" ) <EOL> resp . force_close ( ) <EOL> self . alive = False <EOL> if not self . cfg . keepalive : <EOL> resp . force_close ( ) <EOL> elif len ( self . _keep ) >= self . max_keepalived : <EOL> resp . force_close ( ) <EOL> respiter = self . wsgi ( environ , resp . start_response ) <EOL> try : <EOL> if isinstance ( respiter , environ [ '<STR_LIT>' ] ) : <EOL> resp . write_file ( respiter ) <EOL> else : <EOL> for item in respiter : <EOL> resp . write ( item ) <EOL> resp . close ( ) <EOL> request_time = datetime . now ( ) - request_start <EOL> self . log . access ( resp , req , environ , request_time ) <EOL> finally : <EOL> if hasattr ( respiter , "<STR_LIT>" ) : <EOL> respiter . close ( ) <EOL> if resp . should_close ( ) : <EOL> self . log . debug ( "<STR_LIT>" ) <EOL> return False <EOL> except socket . error : <EOL> exc_info = sys . exc_info ( ) <EOL> six . reraise ( exc_info [ <NUM_LIT:0> ] , exc_info [ <NUM_LIT:1> ] , exc_info [ <NUM_LIT:2> ] ) <EOL> except Exception : <EOL> if resp and resp . headers_sent : <EOL> self . log . exception ( "<STR_LIT>" ) <EOL> try : <EOL> conn . sock . shutdown ( socket . SHUT_RDWR ) <EOL> conn . sock . close ( ) <EOL> except socket . error : <EOL> pass <EOL> raise StopIteration ( ) <EOL> raise <EOL> finally : <EOL> try : <EOL> self . cfg . post_request ( self , req , environ , resp ) <EOL> except Exception : <EOL> self . log . exception ( "<STR_LIT>" ) <EOL> return True </s>
