<s> """<STR_LIT>""" <EOL> import random <EOL> from . heap import HeapBalancerSink <EOL> from . . async import AsyncResult <EOL> from . . constants import ( ChannelState , SinkProperties , SinkRole ) <EOL> from . . sink import SinkProvider <EOL> from . . timer_queue import LOW_RESOLUTION_TIMER_QUEUE , LOW_RESOLUTION_TIME_SOURCE <EOL> from . . varz import ( <EOL> Ema , <EOL> Gauge , <EOL> MonoClock , <EOL> Source , <EOL> VarzBase <EOL> ) <EOL> class ApertureBalancerSink ( HeapBalancerSink ) : <EOL> """<STR_LIT>""" <EOL> class ApertureVarz ( VarzBase ) : <EOL> """<STR_LIT>""" <EOL> _VARZ_BASE_NAME = '<STR_LIT>' <EOL> _VARZ = { <EOL> '<STR_LIT>' : Gauge , <EOL> '<STR_LIT>' : Gauge , <EOL> '<STR_LIT>' : Gauge <EOL> } <EOL> def __init__ ( self , next_provider , sink_properties , global_properties ) : <EOL> self . _idle_endpoints = set ( ) <EOL> self . _total = <NUM_LIT:0> <EOL> self . _ema = Ema ( <NUM_LIT:5> ) <EOL> self . _time = MonoClock ( ) <EOL> self . _min_size = sink_properties . min_size <EOL> self . _max_size = sink_properties . max_size <EOL> self . _min_load = sink_properties . min_load <EOL> self . _max_load = sink_properties . max_load <EOL> self . _jitter_min = sink_properties . jitter_min_sec <EOL> self . _jitter_max = sink_properties . jitter_max_sec <EOL> service_name = global_properties [ SinkProperties . Label ] <EOL> self . __varz = self . ApertureVarz ( Source ( service = service_name ) ) <EOL> self . _pending_endpoints = set ( ) <EOL> super ( ApertureBalancerSink , self ) . __init__ ( next_provider , sink_properties , global_properties ) <EOL> if self . _jitter_min > <NUM_LIT:0> : <EOL> self . _ScheduleNextJitter ( ) <EOL> def _UpdateSizeVarz ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __varz . active ( self . _size ) <EOL> self . __varz . idle ( len ( self . _idle_endpoints ) ) <EOL> def _AddSink ( self , endpoint , sink_factory ) : <EOL> """<STR_LIT>""" <EOL> num_healthy = len ( [ c for c in self . _heap [ <NUM_LIT:1> : ] if c . channel . is_open ] ) <EOL> if num_healthy < self . _min_size : <EOL> super ( ApertureBalancerSink , self ) . _AddSink ( endpoint , sink_factory ) <EOL> else : <EOL> self . _idle_endpoints . add ( endpoint ) <EOL> self . _UpdateSizeVarz ( ) <EOL> def _RemoveSink ( self , endpoint ) : <EOL> """<STR_LIT>""" <EOL> removed = super ( ApertureBalancerSink , self ) . _RemoveSink ( endpoint ) <EOL> if removed : <EOL> self . _TryExpandAperture ( ) <EOL> if endpoint in self . _idle_endpoints : <EOL> self . _idle_endpoints . discard ( endpoint ) <EOL> self . _UpdateSizeVarz ( ) <EOL> def _TryExpandAperture ( self , leave_pending = False ) : <EOL> """<STR_LIT>""" <EOL> endpoints = list ( self . _idle_endpoints ) <EOL> added_node = None <EOL> new_endpoint = None <EOL> if endpoints : <EOL> new_endpoint = random . choice ( endpoints ) <EOL> self . _idle_endpoints . discard ( new_endpoint ) <EOL> self . _log . debug ( '<STR_LIT>' % str ( new_endpoint ) ) <EOL> new_sink = self . _servers [ new_endpoint ] <EOL> self . _pending_endpoints . add ( new_endpoint ) <EOL> added_node = super ( ApertureBalancerSink , self ) . _AddSink ( new_endpoint , new_sink ) <EOL> self . _UpdateSizeVarz ( ) <EOL> if added_node : <EOL> if not leave_pending : <EOL> added_node . ContinueWith ( <EOL> lambda ar : self . _pending_endpoints . discard ( new_endpoint ) ) <EOL> return added_node , new_endpoint <EOL> else : <EOL> return AsyncResult . Complete ( ) , None <EOL> def _ContractAperture ( self , force = False ) : <EOL> """<STR_LIT>""" <EOL> if self . _pending_endpoints and not force : <EOL> return <EOL> num_healthy = len ( [ c for c in self . _heap [ <NUM_LIT:1> : ] if c . channel . is_open ] ) <EOL> if num_healthy > self . _min_size : <EOL> least_loaded_endpoint = None <EOL> for n in self . _heap [ <NUM_LIT:1> : ] : <EOL> if n . channel . is_closed and n . endpoint not in self . _pending_endpoints : <EOL> least_loaded_endpoint = n . endpoint <EOL> break <EOL> if not least_loaded_endpoint : <EOL> for n in self . _heap [ <NUM_LIT:1> : ] : <EOL> if n . endpoint not in self . _pending_endpoints : <EOL> least_loaded_endpoint = n . endpoint <EOL> break <EOL> if least_loaded_endpoint : <EOL> self . _idle_endpoints . add ( least_loaded_endpoint ) <EOL> super ( ApertureBalancerSink , self ) . _RemoveSink ( least_loaded_endpoint ) <EOL> self . _log . debug ( '<STR_LIT>' % str ( least_loaded_endpoint ) ) <EOL> self . _UpdateSizeVarz ( ) <EOL> def _OnNodeDown ( self , node ) : <EOL> """<STR_LIT>""" <EOL> if node . channel . state != ChannelState . Idle : <EOL> ar , _ = self . _TryExpandAperture ( ) <EOL> return ar <EOL> else : <EOL> return AsyncResult . Complete ( ) <EOL> def _OnGet ( self , node ) : <EOL> """<STR_LIT>""" <EOL> self . _AdjustAperture ( <NUM_LIT:1> ) <EOL> def _OnPut ( self , node ) : <EOL> """<STR_LIT>""" <EOL> self . _AdjustAperture ( - <NUM_LIT:1> ) <EOL> def _ScheduleNextJitter ( self ) : <EOL> """<STR_LIT>""" <EOL> next_jitter = random . randint ( self . _jitter_min , self . _jitter_max ) <EOL> now = LOW_RESOLUTION_TIME_SOURCE . now <EOL> self . _next_jitter = LOW_RESOLUTION_TIMER_QUEUE . Schedule ( <EOL> now + next_jitter , self . _Jitter ) <EOL> def _Jitter ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> ar , endpoint = self . _TryExpandAperture ( True ) <EOL> if endpoint : <EOL> try : <EOL> ar . wait ( ) <EOL> if not ar . exception : <EOL> self . _ContractAperture ( True ) <EOL> finally : <EOL> self . _pending_endpoints . discard ( endpoint ) <EOL> finally : <EOL> self . _ScheduleNextJitter ( ) <EOL> def _AdjustAperture ( self , amount ) : <EOL> """<STR_LIT>""" <EOL> self . _total += amount <EOL> avg = self . _ema . Update ( self . _time . Sample ( ) , self . _total ) <EOL> aperture_size = self . _size <EOL> if aperture_size == <NUM_LIT:0> : <EOL> aperture_load = self . _max_load <EOL> else : <EOL> aperture_load = avg / aperture_size <EOL> self . __varz . load_average ( aperture_load ) <EOL> if ( aperture_load >= self . _max_load <EOL> and self . _idle_endpoints <EOL> and aperture_size < self . _max_size ) : <EOL> self . _TryExpandAperture ( ) <EOL> elif aperture_load <= self . _min_load and aperture_size > self . _min_size : <EOL> self . _ContractAperture ( ) <EOL> ApertureBalancerSink . Builder = SinkProvider ( <EOL> ApertureBalancerSink , <EOL> SinkRole . LoadBalancer , <EOL> smoothing_window = <NUM_LIT:5> , <EOL> min_size = <NUM_LIT:1> , <EOL> max_size = <NUM_LIT:2> ** <NUM_LIT> , <EOL> min_load = <NUM_LIT:0.5> , <EOL> max_load = <NUM_LIT> , <EOL> server_set_provider = None , <EOL> jitter_min_sec = <NUM_LIT> , <EOL> jitter_max_sec = <NUM_LIT> ) </s>
