a mzfa<@sdZddlZddlZddlZddlZddlZddlZddlZddlm Z dZ GdddZ e Z e ZGdddejZGd d d ZdS) z8A thread-based worker pool. .. spelling:: joinable N) pass_none) WorkerThread ThreadPoolc@s eZdZdZddZddZdS) TrueyZerozDObject which equals and does math like the integer 0 but evals True.cCs|SNselfotherrrH/var/www/media/lib/python3.9/site-packages/cheroot/workers/threadpool.py__add__szTrueyZero.__add__cCs|Srrrrrr __radd__szTrueyZero.__radd__N)__name__ __module__ __qualname____doc__r r rrrr rsrc@s4eZdZdZdZdZdZddZddZdd Z dS) raFThread which continuously polls a Queue for Connection objects. Due to the timing issues of polling a Queue, a WorkerThread does not check its own 'ready' flag after it has started. To stop the thread, it is necessary to stick a _SHUTDOWNREQUEST object onto the Queue (one for each running WorkerThread). NFcsvd_|_d_d_d_d_d_fddfddfddfddd dd dd _tj dS) zInitialize WorkerThread instance. Args: server (cheroot.server.HTTPServer): web server object receiving this request FrNcsjjdurtpjjSr) requests_seen start_time trueyzeroconnsr rr Is  z'WorkerThread.__init__..csjjdurtpjjjSr) bytes_readrrrrfilerrrr rNs  csjjdurtpjjjSr) bytes_writtenrrrwfilerrrr rSs  cs"jjdurtptjSr) work_timerrtimerrrr rXs   cSs|d||d|pdS)N Bytes Read Work Timeư>rrrrr r]s cSs|d||d|pdS)N Bytes Writtenr!r"rrrrr r`s )ZRequestsr r#r!zRead ThroughputzWrite Throughput) readyserverrrrrrstats threadingThread__init__)r r%rrr r)9s    zWorkerThread.__init__c Cs|j|jjd|j<d|_zz |Wnttfyz}z6|jpD|}|jjd|t j d||j_ WYd}~nRd}~0t y}z2|jjd|dt j dd||j_ WYd}~n d}~00Wd |_nd |_0dS) a0Set up incoming HTTP connection processing loop. This is the thread's entry-point. It performs lop-layer exception handling and interrupt processing. :exc:`KeyboardInterrupt` and :exc:`SystemExit` bubbling up from the inner-layer code constitute a global server interrupt request. When they happen, the worker thread exits. :raises BaseException: when an unexpected non-interrupt exception leaks from the inner layers # noqa: DAR401 KeyboardInterrupt SystemExit zWorker ThreadsTz%Setting the server interrupt flag to levelNzAA fatal exception happened. Setting the server interrupt flag to z and giving up. Please, report this on the Cheroot tracker at , providing a full reproducer with as much context and details as possible.r+ tracebackF)r&r%namer$&_process_connections_until_interruptedKeyboardInterrupt SystemExit __cause__ error_logloggingDEBUGZ interrupt BaseExceptionCRITICAL)r Z interrupt_excZinterrupt_causeZunderlying_excrrr runfs,   zWorkerThread.runc CsV|jj}|turdS||_|jjd}|r8t|_d}zz|}Wnt y*}zd}|jj d|j d|j d|dt jdWYd}~W|r|j|n||r |j|j7_|j|jj7_|j|jj7_|jt|j7_d|_d|_qWYd}~n4d}~0ttfy}zHd}|jj d |j d|j d|dt jdtt||WYd}~nd}~0tyT}z|jj d |t jd d WYd}~W|r|j|n||r8|j|j7_|j|jj7_|j|jj7_|jt|j7_d|_d|_qWYd}~n d}~00W|rl|j|n||r|j|j7_|j|jj7_|j|jj7_|jt|j7_d|_d|_q|r|j|n||rH|j|j7_|j|jj7_|j|jj7_|jt|j7_d|_d|_0qdS) a Process incoming HTTP connections in an infinite loop. Retrieves incoming connections from thread pool, processing them one by one. :raises SystemExit: on the internal requests to stop the server instance NZEnabledFz8Got a connection error while handling a connection from :z ()r*z?Got a server shutdown request while handling a connection from z8Unhandled error while processing an incoming connection Tr,)r%requestsget_SHUTDOWNREQUESTrr&rr communicateConnectionErrorr3Z remote_addrZ remote_portr4INFOZput_conncloserrrrrrr0r1r5strr6ERROR)r rZis_stats_enabledZkeep_conn_openZconnection_errorshutdown_requestZunhandled_errorrrr r/s    &  z3WorkerThread._process_connections_until_interrupted) rrrrrr%r$r)r8r/rrrr r%s-,rc@seZdZdZdddZddZedd Zd d Zd d Z ddZ ddZ ddZ dddZ eeddZddZeddZdS) rzA Request Queue for an HTTPServer which pools threads. ThreadPool objects must provide min, get(), put(obj), start() and stop(timeout) attributes. cCs|dkrtd|d|tdkr&n4t|tr8|dkrJtd|dn|dkrZtd}||krxtd|d |d ||_||_||_g|_t j |d |_ ||_ |j j |_ t|_d S) anInitialize HTTP requests queue instance. Args: server (cheroot.server.HTTPServer): web server object receiving this request min (int): minimum number of worker threads max (int): maximum number of worker threads (-1/inf for no max) accepted_queue_size (int): maximum number of active requests in queue accepted_queue_timeout (int): timeout for putting request into queue :raises ValueError: if the min/max values are invalid :raises TypeError: if the max is not an integer or inf zmin=z must be > 0infrzIExpected an integer or the infinity value for the `max` argument but got .zmax=z must be > min=z (or infinity for no max))maxsizeN) ValueErrorfloat isinstanceint TypeErrorr%minmax_threadsqueueQueue_queue_queue_put_timeoutr< collectionsdeque_pending_shutdowns)r r%rPrQZaccepted_queue_sizeZaccepted_queue_timeoutrrr r)s0  zThreadPool.__init__cCs|jrtd||jdS)zaStart the pool of threads. :raises RuntimeError: if the pool is already started z%Threadpools can only be started once.N)rR RuntimeErrorgrowrPrrrr startszThreadPool.startcCs(tdd|jD}t|t|jdS)z3Number of worker threads which are idle. Read-only.cSsg|]}|jdur|qSrr.0trrr #z#ThreadPool.idle..r)lenrRrQrY)r Zidlesrrr idle szThreadPool.idlecCs|jj|d|jddS)zPut request into queue. Args: obj (:py:class:`~cheroot.server.HTTPConnection`): HTTP connection waiting to be processed T)blocktimeoutN)rUputrV)r objrrr rg&szThreadPool.putc CsHdd|jDD]2}|j|z|jWqty@Yq0qdS)NcSsg|]}|s|qSr)is_aliver^rrr ra1rbz2ThreadPool._clear_dead_threads..)rRremoverYpopleft IndexError)r r`rrr _clear_dead_threads/s   zThreadPool._clear_dead_threadscsbtjtjd}t||}fddt|D}|D]}|js:tdq>q:j|dS)z.Spawn new worker threads (not above self.max).rcsg|] }qSr) _spawn_worker)r_irrr ra=rbz#ThreadPool.grow..g?N) rQrcrRrPranger$rsleepextend)r amountZbudgetZn_newworkersworkerrrr r[8s zThreadPool.growcCs&t|j}dj|jd|_||S)NzCP Server {worker_name!s})Z worker_name)rr%formatr.r\)r rurrr rnCs zThreadPool._spawn_workercCsl|t|j8}||dkr"dStt|j|jd}t||}t|D]}|jd|j t qJdS)z-Kill off worker threads (not below self.min).rN) rcrYrmrQrRrPrpappendrUrgr=)r rsZn_extraZ n_to_remove_rrr shrinkLs   zThreadPool.shrinkc Cs|dur"|dkr"d}tjddd|dur6t|}|jD]}|jtqs z,ThreadPool._clear_threads..)rR)r threadsrrr rszThreadPool._clear_threadscCs |jS)zReturn the queue size.)rUqsizerrrr rszThreadPool.qsizeN)rErFrFrE)rz)rrrrr)r\propertyrdrgrmr[rnryr staticmethodrrrrrrrr rs& .       +  r)rrWr4r'rrr}rSZjaraco.functoolsr__all__rrobjectr=r(rrrrrr s  >