Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Router latency tuner
#!/usr/bin/env python3
import argparse
import time
import logging
import psutil
FILENAME_FORMAT = "/sys/devices/system/cpu/cpu{0:d}/power/pm_qos_resume_latency_us"
logger = logging.getLogger(__name__)
class IFaceStats:
__slots__ = ('name', 'r_packets', 't_packets')
def __init__(self, name):
self.name = name
self.r_packets = None
self.t_packets = None
def set_latency(val):
ok = True
for i in range(psutil.cpu_count()):
filename = FILENAME_FORMAT.format(i)
try:
with open(filename, "w") as f:
f.write(str(val))
except OSError:
logger.exception("Can't set latency for cpu: {0:d}".format(i))
ok = False
return ok
def start(ifaces, interval=10, alpha=0.3, rate=140):
ifaces = frozenset(ifaces)
state = {}
if alpha > 1 or alpha <= 0:
raise ValueError("Alpha can't be higher than 1")
set_latency(0) # validate that we can change this setting
rev_alpha = 1.0 - alpha
low_latency_threshold = rate * interval
powersave_threshold = low_latency_threshold * 0.7
for ifname, ifstats in psutil.net_io_counters(pernic=True).items():
if ifname not in ifaces:
continue
obj = IFaceStats(ifname)
obj.r_packets = ifstats.packets_recv
obj.t_packets = ifstats.packets_sent
state[ifname] = obj
if not state:
raise ValueError("No matched interfaces")
exp_counter = 0.0
low_latency = False
try:
while True:
total_diff = 0
time.sleep(interval)
for ifname, ifstats in psutil.net_io_counters(pernic=True).items():
if ifname not in ifaces:
continue
if_state = state[ifname]
r_diff = ifstats.packets_recv - if_state.r_packets
t_diff = ifstats.packets_sent - if_state.t_packets
if_state.r_packets = ifstats.packets_recv
if_state.t_packets = ifstats.packets_sent
total_diff += r_diff + t_diff
exp_counter = exp_counter * rev_alpha + total_diff * alpha
if exp_counter > low_latency_threshold and not low_latency:
set_latency(100)
low_latency = True
# print("set low latency")
elif exp_counter < powersave_threshold and low_latency:
set_latency(0)
low_latency = False
# print("set power saving")
except KeyboardInterrupt:
print("Quit")
finally:
set_latency(0)
def main(args=None):
parser = argparse.ArgumentParser(description="Set system C states (for latency)")
parser.add_argument('-i', '--interface', type=str, required=True,
action='append', help='Interface for stats')
parser.add_argument('-t', '--interval', type=float, default=10)
parser.add_argument('-r', '--rate', type=int, default=140)
parser.add_argument('--alpha', type=float, default=0.3)
params = parser.parse_args(args=args)
start(params.interface, interval=params.interval, rate=params.rate,
alpha=params.alpha)
if __name__ == '__main__':
main()
[Unit]
Description=Set Router latency to optimize power consumption
After=network.target
[Service]
Type=simple
User=root
Group=root
ProtectKernelModules=yes
ExecStart=/usr/local/bin/router_latency.py -i eth1 -i home -t 10 -r 132 --alpha 0.35
Restart=always
KillSignal=SIGINT
StandardInput=null
StandardOutput=syslog
StandardError=syslog
CapabilityBoundingSet=~CAP_SYS_PTRACE
SystemCallFilter=~@debug @keyring @module @mount @chown @cpu-emulation
PrivateTmp=yes
ProtectHome=yes
ProtectSystem=strict
PrivateDevices=yes
NoNewPrivileges=yes
LimitNPROC=45
TasksMax=10
Nice=19
ProtectKernelTunables=no
InaccessiblePaths=/boot /mnt /media /opt /var
ReadOnlyPaths=/proc
#SELinuxContext=system_u:system_r:router_latency_t
[Install]
WantedBy=multi-user.target
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.