<s> from collections import deque <EOL> import heapq <EOL> import operator <EOL> import signal <EOL> from threading import RLock <EOL> import time <EOL> import pyuv <EOL> from . sync import add , sub , increment , atomic_read <EOL> from . util import nanotime <EOL> class ProcessTracker ( object ) : <EOL> def __init__ ( self , loop ) : <EOL> self . processes = [ ] <EOL> self . _done_cb = None <EOL> self . _check_timer = pyuv . Timer ( loop ) <EOL> self . _lock = RLock ( ) <EOL> def start ( self , interval = <NUM_LIT:0.1> ) : <EOL> self . _check_timer . start ( self . _on_check , interval , interval ) <EOL> self . _check_timer . unref ( ) <EOL> def on_done ( self , callback ) : <EOL> self . _done_cb = callback <EOL> def stop ( self ) : <EOL> self . _check_timer . stop ( ) <EOL> self . processes = [ ] <EOL> def close ( self ) : <EOL> self . processes = [ ] <EOL> self . _done_cb = None <EOL> if not self . _check_timer . closed : <EOL> self . _check_timer . close ( ) <EOL> def check ( self , process , graceful_timeout = <NUM_LIT> ) : <EOL> process . graceful_time = graceful_timeout + nanotime ( ) <EOL> heapq . heappush ( self . processes , process ) <EOL> def uncheck ( self , process ) : <EOL> if process in self . processes : <EOL> del self . processes [ operator . indexOf ( self . processes , process ) ] <EOL> def _on_check ( self , handle ) : <EOL> with self . _lock : <EOL> while True : <EOL> if not len ( self . processes ) : <EOL> if self . _done_cb is not None : <EOL> self . _done_cb ( ) <EOL> self . _donc_cb = None <EOL> break <EOL> p = heapq . heappop ( self . processes ) <EOL> now = nanotime ( ) <EOL> delta = p . graceful_time - now <EOL> if delta > <NUM_LIT:0> : <EOL> if p . active : <EOL> heapq . heappush ( self . processes , p ) <EOL> break <EOL> else : <EOL> try : <EOL> p . kill ( signal . SIGKILL ) <EOL> except : <EOL> pass <EOL> p . close ( ) <EOL> class FlappingInfo ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , attempts = <NUM_LIT:2> , window = <NUM_LIT:1.> , retry_in = <NUM_LIT> , max_retry = <NUM_LIT:5> ) : <EOL> self . attempts = attempts <EOL> self . window = window <EOL> self . retry_in = retry_in <EOL> self . max_retry = max_retry <EOL> self . history = deque ( maxlen = max_retry ) <EOL> self . retries = <NUM_LIT:0> <EOL> def reset ( self ) : <EOL> self . history . clear ( ) <EOL> self . retries = <NUM_LIT:0> <EOL> class ProcessState ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , config , sessionid , env = None ) : <EOL> self . config = config <EOL> self . sessionid = sessionid <EOL> self . env = env <EOL> self . running = deque ( ) <EOL> self . running_out = deque ( ) <EOL> self . stopped = False <EOL> self . setup ( ) <EOL> def setup ( self ) : <EOL> self . name = "<STR_LIT>" % ( self . sessionid , self . config . name ) <EOL> self . cmd = self . config . cmd <EOL> self . _numprocesses = self . config . get ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> self . flapping = self . config . get ( '<STR_LIT>' ) <EOL> if isinstance ( self . flapping , dict ) : <EOL> try : <EOL> self . flapping = FlappingInfo ( ** self . flapping ) <EOL> except TypeError : <EOL> self . flapping = None <EOL> self . flapping_timer = None <EOL> self . stopped = False <EOL> @ property <EOL> def active ( self ) : <EOL> return ( len ( self . running ) + len ( self . running_out ) ) > <NUM_LIT:0> <EOL> @ property <EOL> def graceful_timeout ( self ) : <EOL> return nanotime ( self . config . get ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % self . name <EOL> def make_process ( self , loop , id , on_exit ) : <EOL> """<STR_LIT>""" <EOL> return self . config . make_process ( loop , id , self . name , env = self . env , <EOL> on_exit = on_exit ) <EOL> def __get_numprocesses ( self ) : <EOL> return atomic_read ( self . _numprocesses ) <EOL> def __set_numprocesses ( self , n ) : <EOL> self . _numprocesses = n <EOL> numprocesses = property ( __get_numprocesses , __set_numprocesses , <EOL> doc = """<STR_LIT>""" ) <EOL> @ property <EOL> def pids ( self ) : <EOL> pids = [ p . pid for p in self . running ] <EOL> pids . extend ( [ p . pid for p in self . running_out ] ) <EOL> return pids <EOL> def reset ( self ) : <EOL> """<STR_LIT>""" <EOL> self . numprocesses = self . config . get ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> if self . flapping and self . flapping is not None : <EOL> self . flapping . reset ( ) <EOL> def update ( self , config , env = None ) : <EOL> """<STR_LIT>""" <EOL> self . config = config <EOL> self . env = env <EOL> self . numprocesses = max ( self . config . get ( '<STR_LIT>' , <NUM_LIT:1> ) , <EOL> self . numprocesses ) <EOL> def incr ( self , i = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> self . _numprocesses = add ( self . _numprocesses , i ) <EOL> return self . _numprocesses <EOL> def decr ( self , i = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> self . _numprocesses = sub ( self . _numprocesses , i ) <EOL> return self . _numprocesses <EOL> def queue ( self , process ) : <EOL> """<STR_LIT>""" <EOL> self . running . append ( process ) <EOL> def dequeue ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . running . popleft ( ) <EOL> def remove ( self , process ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . running . remove ( process ) <EOL> except ValueError : <EOL> pass <EOL> def list_processes ( self ) : <EOL> return list ( self . running ) <EOL> def check_flapping ( self ) : <EOL> """<STR_LIT>""" <EOL> f = self . flapping <EOL> f . history . append ( time . time ( ) ) <EOL> if len ( f . history ) >= f . attempts : <EOL> diff = f . history [ - <NUM_LIT:1> ] - f . history [ <NUM_LIT:0> ] <EOL> if diff > f . window : <EOL> f . reset ( ) <EOL> self . flapping = f <EOL> elif f . retries < f . max_retry : <EOL> f . retries = increment ( f . retries ) <EOL> self . flapping = f <EOL> return False , True <EOL> else : <EOL> f . reset ( ) <EOL> self . flapping = f <EOL> return False , False <EOL> self . flapping = f <EOL> return True , None </s>
