Throttling or slowing down network interfaces on Ubuntu

Michael Mahemoff, on Google Plus, points out an idea from the Rails people to slow down your network connection to your local machine in order to simulate the experience of using the web, where slow connections are common (especially on mobile).

The commands they mention are for OS X, though. To do this in Ubuntu you want the following:

Slow down your network connection to localhost by adding a 500ms delay:

sudo tc qdisc add dev lo root handle 1:0 netem delay 500msec

If you do this, then ping localhost, you’ll see that packets now take a second to return (because the 500ms delay applies on the way out and the way back).

To remove this delay:

sudo tc qdisc del dev lo root

Since those commands are pretty alarmingly impenetrable, I put together a tiny little app to do it for you instead. Grab the Python code below and run it, and then you can enable throttling by just ticking a box, and drag the sliders to set the amount you want to slow down your connection. Try it next time you’re building an app which talks to the network — you may find it enlightening (although depressing) how badly your app (or the framework you’re using) deals with slow connections… and half your users will have those slow connections. So we need to get better at dealing with them.

from gi.repository import Gtk, GLib
import socket, fcntl, struct, array, sys, os

def which(program):
    import os
    def is_exe(fpath):
        return os.path.isfile(fpath) and os.access(fpath, os.X_OK)

    fpath, fname = os.path.split(program)
    if fpath:
        if is_exe(program):
            return program
        for path in os.environ["PATH"].split(os.pathsep):
            path = path.strip('"')
            exe_file = os.path.join(path, program)
            if is_exe(exe_file):
                return exe_file
    return None

def all_interfaces():
    max_possible = 128 # arbitrary. raise if needed.
    bytes = max_possible * 32
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    names = array.array('B', '\0' * bytes)
    outbytes = struct.unpack('iL', fcntl.ioctl(
    0x8912, # SIOCGIFCONF
    struct.pack('iL', bytes, names.buffer_info()[0])
    namestr = names.tostring()
    lst = {}
    for i in range(0, outbytes, 40):
        name = namestr[i:i+16].split('\0', 1)[0]
        ip = namestr[i+20:i+24]
        friendly = ""
        if name == "lo": friendly = "localhost"
        if name.startswith("eth"): friendly = "Wired connection %s" % (name.replace("eth", ""))
        if name.startswith("wlan"): friendly = "Wifi connection %s" % (name.replace("eth", ""))
        lst[name] =(
            {"friendly": friendly, "action_timer": None, "current_real_value": 0,
             "toggled_by_code": False}
    return lst

class App(object):
    def __init__(self):
        win = Gtk.Window()
        win.set_size_request(300, 200)
        win.connect("destroy", Gtk.main_quit)
        win.set_title("Network Throttle")
        self.ifs = all_interfaces()
        tbl = Gtk.Table(rows=len(self.ifs.keys())+1, columns=4)
        tbl.attach(Gtk.Label("Throttled"), 2, 3, 0, 1)
        delay_label = Gtk.Label("Delay")
        delay_label.set_size_request(150, 40)
        tbl.attach(delay_label, 3, 4, 0, 1)
        row = 1
        for k, v in self.ifs.items():
            tbl.attach(Gtk.Label(k), 0, 1, row, row+1)
            tbl.attach(Gtk.Label(v["friendly"]), 1, 2, row, row+1)
            tb = Gtk.CheckButton()
            tbl.attach(tb, 2, 3, row, row+1)
            tb.connect("toggled", self.toggle_button, k)
            self.ifs[k]["checkbox"] = tb
            sl = Gtk.HScale()
            sl.set_increments(20, 100)
            sl.set_range(20, 980)
            sl.connect("value_changed", self.value_changed, k)
            tbl.attach(sl, 3, 4, row, row+1)
            self.ifs[k]["slider"] = sl
            row += 1
        box = Gtk.Box(spacing=6)
        box.pack_start(tbl, True, True, 6)

    def toggle_button(self, button, interface):
        if self.ifs[interface]["toggled_by_code"]:
            print "ignoring toggle button because it was toggled by code, not user"
            self.ifs[interface]["toggled_by_code"] = False
        print "toggled to", button.get_active()
        if button.get_active():

    def value_changed(self, slider, interface):
        print "value_changed", slider.get_value()
        if slider.get_value() == self.ifs[interface]["current_real_value"]:
            print "Not setting if because it already is that value"

    def get_tc(self):
        print "getting tc"
        self.throttled_ifs = {}
        def get_tc_output(io, condition):
            print "got tc output", condition
            line = io.readline()
            print "got tc line", line
            parts = line.split()
            if len(parts) > 2 and parts[0] == "qdisc" and parts[1] == "netem":
                if len(parts) == 12:
                    self.throttled_ifs[parts[4]] = {"delay": parts[11].replace("ms", "")}
            if condition == GLib.IO_IN:
                return True
            elif condition == GLib.IO_HUP|GLib.IO_IN:
                print "throttled IFs are", self.throttled_ifs
                return False

        pid, stdin, stdout, stderr = GLib.spawn_async(
            ["tc", "qdisc"],
        io = GLib.IOChannel(stdout)
        self.source_id = io.add_watch(GLib.IO_IN|GLib.IO_HUP, get_tc_output, 

    def actually_turn_on_throttling(self, interface, value):
        print "actually throttling", interface, "to", value
        self.ifs[interface]["action_timer"] = None
        cmd = "pkexec tc qdisc replace dev %s root handle 1:0 netem delay %smsec" % (interface, int(value),)
        print cmd

    def turn_on_throttling(self, interface):
        val = self.ifs[interface]["slider"].get_value()
        if self.ifs[interface]["action_timer"] is not None:
            print "aborting previous throttle request for", interface
        print "throttling", interface, "to", val
        source_id = GLib.timeout_add_seconds(1, self.actually_turn_on_throttling, interface, val)
        self.ifs[interface]["action_timer"] = source_id

    def actually_turn_off_throttling(self, interface):
        print "actually unthrottling", interface
        self.ifs[interface]["action_timer"] = None
        cmd = "pkexec tc qdisc del dev %s root" % (interface,)
        print cmd

    def turn_off_throttling(self, interface):
        if self.ifs[interface]["action_timer"] is not None:
            print "aborting previous throttle request for", interface
        print "unthrottling", interface
        source_id = GLib.timeout_add_seconds(1, self.actually_turn_off_throttling, interface)
        self.ifs[interface]["action_timer"] = source_id

    def update_throttled_list(self, throttled_ifs):
        for k, v in self.ifs.items():
            if k in throttled_ifs:
                current = v["checkbox"].get_active()
                if not current:
                    self.ifs[k]["toggled_by_code"] = True
                    delay = float(throttled_ifs[k]["delay"])
                    self.ifs[k]["current_real_value"] = delay
                current = v["checkbox"].get_active()
                if current:

if __name__ == "__main__":
    if not which("pkexec"): 
        print "You need pkexec installed."
    app = App()
I'm currently available for hire, to help you plan, architect, and build new systems, and for technical writing and articles. You can take a look at some projects I've worked on and some of my writing. If you'd like to talk about your upcoming project, do get in touch.

More in the discussion (powered by webmentions)

  • (no mentions, yet.)