`2f=ddlZddlZGddeZGddZGddZGdd ZGd d ZGd d ZGddZ GddZ y)NceZdZfdZxZS)RequestExceededExceptioncd||_||_dj||}t||y)aError when requested amount exceeds what is allowed The request that raised this error should be retried after waiting the time specified by ``retry_time``. :type requested_amt: int :param requested_amt: The originally requested byte amount :type retry_time: float :param retry_time: The length in time to wait to retry for the requested amount zzBandwidthLimitedStream.reads //==%%f- - F"   d33 3==%%f- - **,}}!!&))rcr|jjs9 |jj|j|j d|_y|jj#t $r/}|jj|jYd}~nd}~wwxYw|jjsi)Nr) r3 exceptionr#consumer6r5rr$rr)r es rr?z4BandwidthLimitedStream._consume_through_leaky_buckets ,,66 5""**$$d&9&9$% ,,66 6, 5  &&q||44 5,,66s7A&& B/%BBc$|jy)z6Signal that data being read is being transferred to S3N)r;rs rsignal_transferringz*BandwidthLimitedStream.signal_transferrings &&(rc$|jy)z:Signal that data being read is not being transferred to S3N)r)rs rsignal_not_transferringz.BandwidthLimitedStream.signal_not_transferrings '')rc<|jj||yr/)r2seek)r wherewhences rrJzBandwidthLimitedStream.seeks 5&)rc6|jjSr/)r2tellrs rrNzBandwidthLimitedStream.tells}}!!##rc|jr|jr|j|jj yr/)r4r6r?r2closers rrPzBandwidthLimitedStream.closes1  + +0@0@  . . 0 rc|Sr/rrs r __enter__z BandwidthLimitedStream.__enter__s rc$|jyr/)rP)r argskwargss r__exit__zBandwidthLimitedStream.__exit__s  r)Ni)r)rrrr r;r)r>r?rFrHrJrNrPrRrVrrrr(r(esG " !0F01*(7$)**$rr(c8eZdZ ddZdZdZdZdZdZy) LeakyBucketNct||_||_|t|_t j |_||_|t|_||_ |t|_ yy)a9A leaky bucket abstraction to limit bandwidth consumption :type rate: int :type rate: The maximum rate to allow. This rate is in terms of bytes per second. :type time_utils: TimeUtils :param time_utils: The time utility to use for interacting with time :type rate_tracker: BandwidthRateTracker :param rate_tracker: Tracks bandwidth consumption :type consumption_scheduler: ConsumptionScheduler :param consumption_scheduler: Schedules consumption retries when necessary N) float _max_rater$r threadingLock_lock _rate_trackerBandwidthRateTracker_consumption_schedulerConsumptionScheduler)r max_rater& rate_trackerconsumption_schedulers rr zLeakyBucket.__init__sl.x%  ({D ^^% )  !5!7D &;# (*>*@D ' )rcl|j5|jj}|jj |r|j |||cdddS|j ||r|j|||n|j||cdddS dddy#1swYyxYw)acConsume an a requested amount :type amt: int :param amt: The amount of bytes to request to consume :type request_token: RequestToken :param request_token: The token associated to the consumption request that is used to identify the request. So if a RequestExceededException is raised the token should be used in subsequent retry consume() request. :raises RequestExceededException: If the consumption amount would exceed the maximum allocated bandwidth :rtype: int :returns: The amount consumed N) r^r$rra is_scheduled,_release_requested_amt_for_scheduled_request_projected_to_exceed_max_rate!_raise_request_exceeded_exception_release_requested_amtr amt request_tokentime_nows rrCzLeakyBucket.consumes$ZZ B'',,.H**77 FHH B B 33CB66223A B B B B BsAB*7B**B3cX|jj||}||jkDSr/)r_get_projected_rater[)r rmroprojected_rates rriz)LeakyBucket._projected_to_exceed_max_rates)++>>sHM..rc\|jj||j||Sr/)raprocess_scheduled_consumptionrkrls rrhz8LeakyBucket._release_requested_amt_for_scheduled_requests0 ##AA  **399rc|t|jz }|jj|||}t ||)N)rr)rZr[raschedule_consumptionr)r rmrnroallocated_timers rrjz-LeakyBucket._raise_request_exceeded_exception%sFuT^^4400EE  '*  rc>|jj|||Sr/)r_record_consumption_rate)r rmros rrkz"LeakyBucket._release_requested_amt.s 223A r)NNN) rrrr rCrirhrjrkrrrrXrXs." !AFB>/: rrXc$eZdZdZdZdZdZy)rbc i|_d|_y)z*Schedules when to consume a desired amountrN) _tokens_to_scheduled_consumption _total_waitrs rr zConsumptionScheduler.__init__4s02-rc||jvS)zIndicates if a consumption request has been scheduled :type token: RequestToken :param token: The token associated to the consumption request that is used to identify the request. )r|)r tokens rrgz!ConsumptionScheduler.is_scheduled9s====rc||xj|z c_|j|d|j|<|jS)aSchedules a wait time to be able to consume an amount :type amt: int :param amt: The amount of bytes scheduled to be consumed :type token: RequestToken :param token: The token associated to the consumption request that is used to identify the request. :type time_to_consume: float :param time_to_consume: The desired time it should take for that specific request amount to be consumed in regardless of previously scheduled consumption requests :rtype: float :returns: The amount of time to wait for the specific request before actually consuming the specified amount. ) wait_durationtime_to_consume)r}r|)r rmrrs rrvz)ConsumptionScheduler.schedule_consumptionBsA& O+!--.8 --e4rc||jj|}t|j|dz d|_y)zProcesses a scheduled consumption request that has completed :type token: RequestToken :param token: The token associated to the consumption request that is used to identify the request. rrN)r|popmaxr})r rscheduled_retrys rrtz2ConsumptionScheduler.process_scheduled_consumption\s=??CCEJ   /@A A1 rN)rrrr rgrvrtrrrrbrb3s > 4  rrbc<eZdZddZedZdZdZdZdZ y) r`c.||_d|_d|_y)aTracks the rate of bandwidth consumption :type a: float :param a: The constant to use in calculating the exponentional moving average of the bandwidth rate. Specifically it is used in the following calculation: current_rate = alpha * new_rate + (1 - alpha) * current_rate This value of this constant should be between 0 and 1. N)_alpha _last_time _current_rate)r alphas rr zBandwidthRateTracker.__init__js !rc4|jy|jS)zmThe current transfer rate :rtype: float :returns: The current tracked transfer rate )rrrs r current_ratez!BandwidthRateTracker.current_ratezs ?? "!!!rc@|jy|j||S)aZGet the projected rate using a provided amount and time :type amt: int :param amt: The proposed amount to consume :type time_at_consumption: float :param time_at_consumption: The proposed time to consume at :rtype: float :returns: The consumption rate if that amt and time were consumed r)r*_calculate_exponential_moving_average_rater rmtime_at_consumptions rrqz'BandwidthRateTracker.get_projected_rates* ?? ">> $  rcv|j||_d|_y|j|||_||_y)aRecord the consumption rate based off amount and time point :type amt: int :param amt: The amount that got consumed :type time_at_consumption: float :param time_at_consumption: The time at which the amount was consumed Nr)rrrrs rryz,BandwidthRateTracker.record_consumption_ratesB ?? "1DO!$D  !LL $ .rcJ||jz }|dkr tdS||z S)Nrinf)rrZ)r rmr time_deltas r_calculate_ratez$BandwidthRateTracker._calculate_rates.(4??: ? < j!!rc~|j||}|j|zd|jz |jzzS)N)rrr)r rmrnew_rates rrz?BandwidthRateTracker._calculate_exponential_moving_average_rates>''-@A{{X%T[[Drsk y,  !!$##LnnbZZz3 3 lNONOr