135 lines
5.6 KiB
Python
135 lines
5.6 KiB
Python
import asyncio
|
|
import time
|
|
|
|
import psutil
|
|
|
|
import util
|
|
|
|
|
|
class DynamicThrottle(object):
|
|
def __init__(self, **kwargs):
|
|
self.target_cpu_usage = kwargs.get("target_cpu_usage", 50)
|
|
self.sleep_interval = 0.0
|
|
|
|
self.sleep_increment = kwargs.get("sleep_increment", 0.01)
|
|
self.sleep_decrement = kwargs.get("sleep_decrement", 0.01)
|
|
|
|
self.sleep_max = kwargs.get("sleep_max", 0.1)
|
|
self.sleep_min = kwargs.get("sleep_min", 0.01)
|
|
|
|
self.psutil_interval = kwargs.get("psutil_interval", 0.1)
|
|
|
|
self.log = kwargs.get("log", util.get_logger(self.__class__.__name__))
|
|
|
|
self.consecutive_increments = 0
|
|
self.consecutive_decrements = 0
|
|
|
|
self.consecutive_divisor = kwargs.get("consecutive_divisor", 1)
|
|
|
|
self.last_was_increment = kwargs.get("start_increment", True)
|
|
|
|
if kwargs.get("use_async"):
|
|
self.wait = self.dynamic_throttle_async
|
|
else:
|
|
self.wait = self.dynamic_throttle
|
|
|
|
async def dynamic_throttle_async(self):
|
|
"""
|
|
Dynamically sleeps before a request if CPU usage is above our target.
|
|
"""
|
|
current_cpu_usage = psutil.cpu_percent(interval=self.psutil_interval)
|
|
|
|
if current_cpu_usage > self.target_cpu_usage:
|
|
if self.last_was_increment:
|
|
self.consecutive_increments += 1
|
|
# self.log.debug(f"High CPU consecutive increments: {self.consecutive_increments}")
|
|
else:
|
|
self.consecutive_increments = 0 # ?
|
|
self.consecutive_decrements = 0 # ?
|
|
# self.log.debug(f"High CPU alert reset.")
|
|
self.sleep_interval += self.sleep_increment * (
|
|
max(1, self.consecutive_increments) / self.consecutive_divisor
|
|
)
|
|
self.last_was_increment = True
|
|
if self.sleep_interval > self.sleep_max:
|
|
self.sleep_interval = self.sleep_max
|
|
# self.log.debug(f"High CPU, but not increasing above {self.sleep_max:.3f}s")
|
|
# self.log.debug(
|
|
# f"High CPU: {current_cpu_usage}% > {self.target_cpu_usage}%, "
|
|
# f"=> sleep {self.sleep_interval:.3f}s"
|
|
# )
|
|
elif current_cpu_usage < self.target_cpu_usage:
|
|
if not self.last_was_increment:
|
|
self.consecutive_decrements += 1
|
|
# self.log.debug(f"Low CPU consecutive decrements: {self.consecutive_decrements}")
|
|
else:
|
|
self.consecutive_decrements = 0 # ?
|
|
self.consecutive_increments = 0 # ?
|
|
# self.log.debug(f"Low CPU alert reset.")
|
|
self.sleep_interval -= self.sleep_decrement * (
|
|
max(1, self.consecutive_decrements) / self.consecutive_divisor
|
|
)
|
|
self.last_was_increment = False
|
|
if self.sleep_interval < self.sleep_min:
|
|
self.sleep_interval = self.sleep_min
|
|
# self.log.debug(f"Low CPU, but not decreasing below {self.sleep_min:.3f}s")
|
|
# self.log.debug(
|
|
# f"Low CPU: {current_cpu_usage}% < {self.target_cpu_usage}%, "
|
|
# f"=> sleep {self.sleep_interval:.3f}s"
|
|
# )
|
|
|
|
if self.sleep_interval > 0:
|
|
await asyncio.sleep(self.sleep_interval)
|
|
return self.sleep_interval
|
|
return 0.0
|
|
|
|
def dynamic_throttle(self):
|
|
"""
|
|
Dynamically sleeps before a request if CPU usage is above our target.
|
|
"""
|
|
current_cpu_usage = psutil.cpu_percent(interval=self.psutil_interval)
|
|
|
|
if current_cpu_usage > self.target_cpu_usage:
|
|
if self.last_was_increment:
|
|
self.consecutive_increments += 1
|
|
# self.log.debug(f"High CPU consecutive increments: {self.consecutive_increments}")
|
|
else:
|
|
self.consecutive_increments = 0 # ?
|
|
self.consecutive_decrements = 0 # ?
|
|
# self.log.debug(f"High CPU alert reset.")
|
|
self.sleep_interval += self.sleep_increment * (
|
|
max(1, self.consecutive_increments) / self.consecutive_divisor
|
|
)
|
|
self.last_was_increment = True
|
|
if self.sleep_interval > self.sleep_max:
|
|
self.sleep_interval = self.sleep_max
|
|
# self.log.debug(f"High CPU, but not increasing above {self.sleep_max:.3f}s")
|
|
# self.log.debug(
|
|
# f"High CPU: {current_cpu_usage}% > {self.target_cpu_usage}%, "
|
|
# f"=> sleep {self.sleep_interval:.3f}s"
|
|
# )
|
|
elif current_cpu_usage < self.target_cpu_usage:
|
|
if not self.last_was_increment:
|
|
self.consecutive_decrements += 1
|
|
# self.log.debug(f"Low CPU consecutive decrements: {self.consecutive_decrements}")
|
|
else:
|
|
self.consecutive_decrements = 0 # ?
|
|
self.consecutive_increments = 0 # ?
|
|
# self.log.debug(f"Low CPU alert reset.")
|
|
self.sleep_interval -= self.sleep_decrement * (
|
|
max(1, self.consecutive_decrements) / self.consecutive_divisor
|
|
)
|
|
self.last_was_increment = False
|
|
if self.sleep_interval < self.sleep_min:
|
|
self.sleep_interval = self.sleep_min
|
|
# self.log.debug(f"Low CPU, but not decreasing below {self.sleep_min:.3f}s")
|
|
# self.log.debug(
|
|
# f"Low CPU: {current_cpu_usage}% < {self.target_cpu_usage}%, "
|
|
# f"=> sleep {self.sleep_interval:.3f}s"
|
|
# )
|
|
|
|
if self.sleep_interval > 0:
|
|
time.sleep(self.sleep_interval)
|
|
return self.sleep_interval
|
|
return 0.0
|