Quick Start

The following section will help you get started.

The fastest way to get started is to pip install the package:

python3 -m pip install dummynet

After this you are ready to go:

  1import dummynet
  2import logging
  3import sys
  4import argparse
  5
  6
  7def run():
  8
  9    parser = argparse.ArgumentParser(description="Program with debugger option")
 10    parser.add_argument("--debug", action="store_true", help="Enable log debugger")
 11    args = parser.parse_args()
 12
 13    log = logging.getLogger("dummynet")
 14    log.setLevel(logging.DEBUG)
 15
 16    if args.debug:
 17        console_handler = logging.StreamHandler(sys.stdout)
 18        console_handler.setLevel(logging.DEBUG)
 19        log.addHandler(console_handler)
 20
 21    process_monitor = dummynet.ProcessMonitor(log=log)
 22    shell = dummynet.HostShell(log=log, sudo=True, process_monitor=process_monitor)
 23    net = dummynet.DummyNet(shell=shell)
 24
 25    cgroup0 = net.add_cgroup(
 26        name="test_cgroup0",
 27        shell=shell,
 28        log=log,
 29        controllers={"cpu.max": 0.5, "memory.high": 200000000},
 30        pid=None,
 31    )
 32    cgroup0 = dummynet.CGroup.build_cgroup(cgroup0, force=True)
 33
 34    cgroup1 = net.add_cgroup(
 35        name="test_cgroup1",
 36        shell=shell,
 37        log=log,
 38        controllers={"cpu.max": 0.2, "memory.high": 100000000},
 39    )
 40    cgroup1.delete_cgroup(force=True)
 41    cgroup1.make_cgroup()
 42    cgroup1.input_validation()
 43    cgroup1.set_limit(cgroup1.controllers)
 44
 45    try:
 46
 47        # Get a list of the current namespaces
 48        namespaces = net.netns_list()
 49        assert namespaces == []
 50
 51        # create two namespaces
 52        demo0 = net.netns_add(name="demo0")
 53        demo1 = net.netns_add(name="demo1")
 54
 55        net.link_veth_add(p1_name="demo0-eth0", p2_name="demo1-eth0")
 56
 57        # Move the interfaces to the namespaces
 58        net.link_set(namespace="demo0", interface="demo0-eth0")
 59        net.link_set(namespace="demo1", interface="demo1-eth0")
 60
 61        # Bind an IP-address to the two peers in the link.
 62        demo0.addr_add(ip="10.0.0.1/24", interface="demo0-eth0")
 63        demo1.addr_add(ip="10.0.0.2/24", interface="demo1-eth0")
 64
 65        # Activate the interfaces.
 66        demo0.up(interface="demo0-eth0")
 67        demo1.up(interface="demo1-eth0")
 68        demo0.up(interface="lo")
 69        demo1.up(interface="lo")
 70
 71        # Test will run until last non-daemon process is done.
 72        proc0 = demo0.run_async(cmd="ping -c 20 10.0.0.2", daemon=True)
 73        proc1 = demo1.run_async(cmd="ping -c 10 10.0.0.1")
 74
 75        # # Add the processes to the cgroup.
 76        cgroup0.add_pid(proc0.pid)
 77        cgroup1.add_pid(proc1.pid)
 78
 79        # Print output as we go (optional)
 80        def _proc0_stdout(data):
 81            print("proc0: {}".format(data))
 82
 83        def _proc1_stdout(data):
 84            print("proc1: {}".format(data))
 85
 86        proc0.stdout_callback = _proc0_stdout
 87        proc1.stdout_callback = _proc1_stdout
 88
 89        while process_monitor.run():
 90            pass
 91
 92        # Check that the ping succeeded.
 93        proc1.match(stdout="10 packets transmitted*", stderr=None)
 94
 95        # Since proc0 is a daemon we automatically kill it when the last
 96        # non-daemon process is done. However we can still see the output
 97        # it generated.
 98        print(f"proc0: {proc0.stdout}")
 99
100    finally:
101
102        # Clean up.
103        net.cleanup()
104        net.cgroup_cleanup()
105
106
107if __name__ == "__main__":
108    run()