=xj<ddlZddlZGddeZGddZGddZGdd ZGd d ZGd d ZGddZ GddZ dS)NceZdZfdZxZS)RequestExceededExceptioncv||_||_d|d|}t|dS)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 zRequest amount z) exceeded the amount available. Retry in N) requested_amt retry_timesuper__init__)selfrrmsg __class__s EC:\Projects\Historical\venv\Lib\site-packages\s3transfer/bandwidth.pyr z!RequestExceededException.__init__sH+$d ddXbdd )__name__ __module__ __qualname__r __classcell__)r s@r rrs8rrceZdZdZdS) RequestTokenzDA token to pass as an identifier when consuming from the LeakyBucketN)rrr__doc__rr rr%sNNDrrceZdZdZdZdS) TimeUtilsc(tjS)zgGet the current time back :rtype: float :returns: The current time in seconds )timer s r rzTimeUtils.time,s y{{rc*tj|S)zwSleep for a designated time :type value: float :param value: The time to sleep for in seconds )rsleep)r values r rzTimeUtils.sleep4s z%   rN)rrrrrrrr rr+s2!!!!!rrc eZdZddZ ddZdS)BandwidthLimiterNcP||_||_|t|_dSdS)aLimits bandwidth for shared S3 transfers :type leaky_bucket: LeakyBucket :param leaky_bucket: The leaky bucket to use limit bandwidth :type time_utils: TimeUtils :param time_utils: Time utility to use for interacting with time. N) _leaky_bucket _time_utilsr)r leaky_bucket time_utilss r r zBandwidthLimiter.__init__>s4*%  ({{D     rTcjt||j||j}|s||S)aWraps a fileobj in a bandwidth limited stream wrapper :type fileobj: file-like obj :param fileobj: The file-like obj to wrap :type transfer_coordinator: s3transfer.futures.TransferCoordinator param transfer_coordinator: The coordinator for the general transfer that the wrapped stream is a part of :type enabled: boolean :param enabled: Whether bandwidth limiting should be enabled to start )BandwidthLimitedStreamr"r#disable_bandwidth_limiting)r fileobjtransfer_coordinatorenabledstreams r get_bandwith_limited_streamz,BandwidthLimiter.get_bandwith_limited_streamLsD( T')=t?O   0  - - / / / rN)T)rrrr r-rrr r r =sA + + + +6:rr c^eZdZ ddZdZdZdZdZdZd Z dd Z d Z d Z dZ dZdS)r'Nc||_||_||_||_|t |_d|_t |_d|_||_ dS)a[Limits bandwidth for reads on a wrapped stream :type fileobj: file-like object :param fileobj: The file like object to wrap :type leaky_bucket: LeakyBucket :param leaky_bucket: The leaky bucket to use to throttle reads on the stream :type transfer_coordinator: s3transfer.futures.TransferCoordinator param transfer_coordinator: The coordinator for the general transfer that the wrapped stream is a part of :type time_utils: TimeUtils :param time_utils: The time utility to use for interacting with time NTr) _fileobjr"_transfer_coordinatorr#r_bandwidth_limiting_enabledr_request_token _bytes_seen_bytes_threshold)r r)r$r*r%bytes_thresholds r r zBandwidthLimitedStream.__init__ds_0  )%9"%  ({{D +/(*nn /rcd|_dS)z0Enable bandwidth limiting on reads to the streamTNr4rs r enable_bandwidth_limitingz0BandwidthLimitedStream.enable_bandwidth_limitings+/(((rcd|_dS)z1Disable bandwidth limiting on reads to the streamFNr:rs r r(z1BandwidthLimitedStream.disable_bandwidth_limitings+0(((rc|js|j|S|xj|z c_|j|jkr|j|S||j|S)zhRead a specified amount Reads will only be throttled if bandwidth limiting is enabled. )r4r2readr6r7_consume_through_leaky_bucket)r amounts r r>zBandwidthLimitedStream.reads / .=%%f-- - F"  d3 3 3=%%f-- - **,,,}!!&)))rc|jjsq |j|j|jd|_dS#t $r)}|j|j Yd}~nd}~wwxYw|jjq|jj)Nr) r3 exceptionr"consumer6r5rr#rr)r es r r?z4BandwidthLimitedStream._consume_through_leaky_buckets ,6 7 5"**$d&9$% + 5 5 5 &&q|44444444 5,6 7,6 6s,< A/A**A/c.|dS)z6Signal that data being read is being transferred to S3N)r;rs r signal_transferringz*BandwidthLimitedStream.signal_transferrings &&(((((rc.|dS)z:Signal that data being read is not being transferred to S3N)r(rs r signal_not_transferringz.BandwidthLimitedStream.signal_not_transferrings '')))))rrc<|j||dSr.)r2seek)r wherewhences r rJzBandwidthLimitedStream.seeks  5&)))))rc4|jSr.)r2tellrs r rNzBandwidthLimitedStream.tells}!!###rc||jr|jr||jdSr.)r4r6r?r2closers r rPzBandwidthLimitedStream.closesE  + 10@ 1  . . 0 0 0 rc|Sr.rrs r __enter__z BandwidthLimitedStream.__enter__s rc.|dSr.)rP)r argskwargss r __exit__zBandwidthLimitedStream.__exit__s r)Nr0)r)rrrr r;r(r>r?rFrHrJrNrPrRrVrrr r'r'cs " !0!0!0!0F000111***(777$)))*******$$$rr'c:eZdZ ddZdZdZdZdZdZdS) LeakyBucketNc t||_||_|t|_t j|_||_|t|_||_ |t|_ dSdS)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 r r zLeakyBucket.__init__s.x%  ({{D ^%% )  !5!7!7D &;# (*>*@*@D ' ' ' ) (rc|j5|j}|j|r#||||cdddS|||r||||n"|||cdddS ddddS#1swxYwYdS)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 r rCzLeakyBucket.consumes$Z B B',,..H*77 FF BHH B B B B B B B B 33CBB B66223AA B B B B B B B B B B B B B B B B B B B B B B B B B BsA B=AB==CCcN|j||}||jkSr.)r_get_projected_rater[)r rmroprojected_rates r riz)LeakyBucket._projected_to_exceed_max_rates'+>>sHMM..rcb|j||||Sr.)raprocess_scheduled_consumptionrkrls r rhz8LeakyBucket._release_requested_amt_for_scheduled_requests: #AA    **3999rc|t|jz }|j|||}t ||)N)rr)rZr[raschedule_consumptionr)r rmrnroallocated_timers r rjz-LeakyBucket._raise_request_exceeded_exception#sSuT^4440EE   '*    rc<|j|||Sr.)r_record_consumption_rate)r rmros r rkz"LeakyBucket._release_requested_amt,s  223AAA r)NNN) rrrr rCrirhrjrkrrr rXrXs" !A!A!A!AFBBB>///:::   rrXc&eZdZdZdZdZdZdS)rbc"i|_d|_dS)z*Schedules when to consume a desired amountrN) _tokens_to_scheduled_consumption _total_waitrs r r zConsumptionScheduler.__init__2s02-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 r rgz!ConsumptionScheduler.is_scheduled7s===rcT|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 r rvz)ConsumptionScheduler.schedule_consumption@s@& O+!-.8 8 -e4rc|j|}t|j|dz d|_dS)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 r rtz2ConsumptionScheduler.process_scheduled_consumptionZsE?CCEJJ  /@A A1  rN)rrrr rgrvrtrrr rbrb1sP >>>   4      rrbcDeZdZd dZedZdZdZdZdZ dS) r`皙?c0||_d|_d|_dS)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 r r zBandwidthRateTracker.__init__hs  !rc"|jdS|jS)zmThe current transfer rate :rtype: float :returns: The current tracked transfer rate N)rrrs r current_ratez!BandwidthRateTracker.current_ratexs ? "3!!rc@|jdS|||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 Nr)r*_calculate_exponential_moving_average_rater rmtime_at_consumptions r rqz'BandwidthRateTracker.get_projected_rates/ ? "3>> $   rcx|j||_d|_dS||||_||_dS)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 r ryz,BandwidthRateTracker.record_consumption_ratesK ? "1DO!$D  F!LL $  .rcJ||jz }|dkrtdS||z S)Nrinf)rrZ)r rmr time_deltas r _calculate_ratez$BandwidthRateTracker._calculate_rates0(4?: ?? << j!!rch|||}|j|zd|jz |jzzS)N)rrr)r rmrnew_rates r rz?BandwidthRateTracker._calculate_exponential_moving_average_rates:''-@AA{X%T[Drs y(        !!!!!!!!$########LnnnnnnnnbZZZZZZZZz3 3 3 3 3 3 3 3 lNONONONONONONONONONOr