Support for CFS bandwidth limiting.

Also trying to fix NOX cmdline opt, but broken at the moment.

bin/mn

@@ -20,11 +20,27 @@ from mininet.clean import cleanup
 from mininet.cli import CLI
 from mininet.log import lg, LEVELS, info
 from mininet.net import Mininet, init
-from mininet.node import Host, Controller, ControllerParams, NOX
+from mininet.node import Host, CPULimitedHost, Controller, NOX
 from mininet.node import RemoteController, UserSwitch, OVSKernelSwitch
+from mininet.link import Intf, TCIntf
 from mininet.topo import SingleSwitchTopo, LinearTopo, SingleSwitchReversedTopo
 from mininet.topolib import TreeTopo
-from mininet.util import makeNumeric
+from mininet.util import makeNumeric, custom
+
+def customNode( constructors, argStr ):
+    "Return custom Node constructor based on argStr"
+    cname, noargs, kwargs = splitArgs( argStr )
+    constructor = constructors.get( cname, None )
+    if noargs:
+        raise Exception( "please specify keyword  arguments for " + cname )
+    if not constructor:
+        raise Exception( "error: %s is unknown - please specify one of %s" %
+               ( cname, constructors.keys() ) )
+    def custom( *args, **params ):
+        params.update( kwargs )
+        print 'CONSTRUCTOR', constructor, 'ARGS', args, 'PARAMS', params
+        return constructor( *args, **params )
+    return custom
 
 # built in topologies, created only when run
 TOPODEF = 'minimal'
@@ -38,17 +54,22 @@ SWITCHDEF = 'ovsk'
 SWITCHES = { 'user': UserSwitch,
             'ovsk': OVSKernelSwitch }
 
-HOSTDEF = 'process'
+HOSTDEF = 'proc'
-HOSTS = { 'process': Host }
+HOSTS = { 'proc': Host,
+          'rt': custom( CPULimitedHost, sched='rt' ),
+          'cfs': custom( CPULimitedHost, sched='cfs' ) }
 
 CONTROLLERDEF = 'ref'
-# a and b are the name and inNamespace params.
 CONTROLLERS = { 'ref': Controller,
-               'nox_dump': lambda name: NOX( name, 'packetdump' ),
+               'nox': NOX, 
-               'nox_pysw': lambda name: NOX( name, 'pyswitch' ),
+               'remote': RemoteController,
-               'remote': lambda name: None,
                'none': lambda name: None }
 
+INTFDEF = 'default'
+INTFS = { 'default': Intf,
+          'tc': TCIntf }
+
+
 # optional tests to run
 TESTS = [ 'cli', 'build', 'pingall', 'pingpair', 'iperf', 'all', 'iperfudp',
          'none' ]
@@ -56,24 +77,31 @@ TESTS = [ 'cli', 'build', 'pingall', 'pingpair', 'iperf', 'all', 'iperfudp',
 ALTSPELLING = { 'pingall': 'pingAll', 'pingpair': 'pingPair',
     'iperfudp': 'iperfUdp', 'iperfUDP': 'iperfUdp', 'prefixlen': 'prefixLen' }
 
-def buildTopo( topo ):
-    "Create topology from string with format (object, arg1, arg2,...)."
-    topo_split = topo.split( ',' )
-    topo_name = topo_split[ 0 ]
-    topo_params = topo_split[ 1: ]
 
-    # Convert int and float args; removes the need for every topology to
+def splitArgs( argstr ):
-    # be flexible with input arg formats.
+    """Split argument string into usable python arguments
-    topo_seq_params = [ s for s in topo_params if '=' not in s ]
+       argstr: argument string with format fn,arg2,kw1=arg3...
-    topo_seq_params = [ makeNumeric( s ) for s in topo_seq_params ]
+       returns: fn, args, kwargs"""
-    topo_kw_params = {}
+    split = argstr.split( ',' )
-    for s in [ p for p in topo_params if '=' in p ]:
+    fn = split[ 0 ]
+    params = split[ 1: ]
+    # Convert int and float args; removes the need for function
+    # to be flexible with input arg formats.
+    args = [ s for s in params if '=' not in s ]
+    args = map( makeNumeric, args )
+    kwargs = {}
+    for s in [ p for p in params if '=' in p ]:
         key, val = s.split( '=' )
-        topo_kw_params[ key ] = makeNumeric( val )
+        kwargs[ key ] = makeNumeric( val )
+    return fn, args, kwargs
 
-    if topo_name not in TOPOS.keys():
+
-        raise Exception( 'Invalid topo_name %s' % topo_name )
+def buildTopo( topoStr ):
-    return TOPOS[ topo_name ]( *topo_seq_params, **topo_kw_params )
+    "Create topology from string with format (object, arg1, arg2,...)."
+    topo, args, kwargs = splitArgs( topoStr )
+    if topo not in TOPOS:
+        raise Exception( 'Invalid topo name %s' % topo )
+    return TOPOS[ topo ]( *args, **kwargs )
 
 
 def addDictOption( opts, choicesDict, default, name, helpStr=None ):
@@ -87,10 +115,9 @@ def addDictOption( opts, choicesDict, default, name, helpStr=None ):
         raise Exception( 'Invalid  default %s for choices dict: %s' %
                         ( default, name ) )
     if not helpStr:
-        helpStr = '[' + ' '.join( choicesDict.keys() ) + ']'
+        helpStr = '|'.join( sorted( choicesDict.keys() ) ) + '[,param=value...]'
     opts.add_option( '--' + name,
-                    type='choice',
+                    type='string',
-                    choices=choicesDict.keys(),
                     default = default,
                     help = helpStr )
 
@@ -135,7 +162,6 @@ class MininetRunner( object ):
         """Parse command-line args and return options object.
            returns: opts parse options dict"""
         if '--custom' in sys.argv:
-            print "custom in sys.argv"
             index = sys.argv.index( '--custom' )
             if len( sys.argv ) > index + 1:
                 custom = sys.argv[ index + 1 ]
@@ -147,46 +173,41 @@ class MininetRunner( object ):
         addDictOption( opts, SWITCHES, SWITCHDEF, 'switch' )
         addDictOption( opts, HOSTS, HOSTDEF, 'host' )
         addDictOption( opts, CONTROLLERS, CONTROLLERDEF, 'controller' )
+        addDictOption( opts, INTFS, INTFDEF, 'intf' )
+        addDictOption( opts, TOPOS, TOPODEF, 'topo' )
 
-        opts.add_option( '--topo', type='string', default=TOPODEF,
-                        help='[' + ' '.join( TOPOS.keys() ) + '],arg1,arg2,'
-                        '...argN')
         opts.add_option( '--clean', '-c', action='store_true',
                         default=False, help='clean and exit' )
         opts.add_option( '--custom', type='string', default=None,
                         help='read custom topo and node params from .py file' )
         opts.add_option( '--test', type='choice', choices=TESTS,
                         default=TESTS[ 0 ],
-                        help='[' + ' '.join( TESTS ) + ']' )
+                        help='|'.join( TESTS ) )
         opts.add_option( '--xterms', '-x', action='store_true',
                         default=False, help='spawn xterms for each node' )
         opts.add_option( '--mac', action='store_true',
-                        default=False, help='set MACs equal to DPIDs' )
+                        default=False, help='automatically set host MACs' )
         opts.add_option( '--arp', action='store_true',
                         default=False, help='set all-pairs ARP entries' )
         opts.add_option( '--verbosity', '-v', type='choice',
                         choices=LEVELS.keys(), default = 'info',
-                        help = '[' + ' '.join( LEVELS.keys() ) + ']' )
+                        help = '|'.join( LEVELS.keys() )  )
         opts.add_option( '--ip', type='string', default='127.0.0.1',
-                        help='[ip address as a dotted decimal string for a'
+                        help='ip address as a dotted decimal string for a'
-                        'remote controller]' )
+                        'remote controller' )
-        opts.add_option( '--port', type='int', default=6633,
-                        help='[port integer for a listening remote'
-                        ' controller]' )
         opts.add_option( '--innamespace', action='store_true',
                         default=False, help='sw and ctrl in namespace?' )
         opts.add_option( '--listenport', type='int', default=6634,
-                        help='[base port for passive switch listening'
+                         help='base port for passive switch listening' )
-                        ' controller]' )
         opts.add_option( '--nolistenport', action='store_true',
                         default=False, help="don't use passive listening port")
         opts.add_option( '--pre', type='string', default=None,
-                        help='[CLI script to run before tests]' )
+                        help='CLI script to run before tests' )
         opts.add_option( '--post', type='string', default=None,
-                        help='[CLI script to run after tests]' )
+                        help='CLI script to run after tests' )
         opts.add_option( '--prefixlen', type='int', default=8,
-                        help='[prefix length (e.g. /8) for automatic '
+                        help='prefix length (e.g. /8) for automatic '
-                        'network configuration]' )
+                        'network configuration' )
 
         self.options, self.args = opts.parse_args()
 
@@ -214,23 +235,14 @@ class MininetRunner( object ):
         start = time.time()
 
         topo = buildTopo( self.options.topo )
-        switch = SWITCHES[ self.options.switch ]
+        switch = customNode( SWITCHES,  self.options.switch )
-        host = HOSTS[ self.options.host ]
+        host = customNode( HOSTS, self.options.host )
-        controller = CONTROLLERS[ self.options.controller ]
+        controller = customNode( CONTROLLERS, self.options.controller )
-        if self.options.controller == 'remote':
+        intf = customNode( INTFS, self.options.intf )
-            controller = lambda a: RemoteController( a,
-                             defaultIP=self.options.ip,
-                             port=self.options.port )
 
         if self.validate:
             self.validate( self.options )
 
-        # We should clarify what this is actually for...
-        # It seems like it should be default values for the
-        # *data* network, so it may be misnamed.
-        controllerParams = ControllerParams( '10.0.0.0',
-            self.options.prefixlen)
-
         inNamespace = self.options.innamespace
         xterms = self.options.xterms
         mac = self.options.mac
@@ -238,7 +250,9 @@ class MininetRunner( object ):
         listenPort = None
         if not self.options.nolistenport:
             listenPort = self.options.listenport
-        mn = Mininet( topo, switch, host, controller, controllerParams,
+        mn = Mininet( topo=topo, 
+                      switch=switch, host=host, controller=controller, 
+                      intf=intf,
                       inNamespace=inNamespace,
                       xterms=xterms, autoSetMacs=mac,
                       autoStaticArp=arp, listenPort=listenPort )

examples/limit.py

@@ -5,19 +5,20 @@ limit.py: example of using link and CPU limits
 """
 
 from mininet.net import Mininet
-from mininet.link import TCIntf, Link
+from mininet.link import TCIntf
 from mininet.node import CPULimitedHost
 from mininet.topolib import TreeTopo
 from mininet.util import custom, quietRun
 from mininet.log import setLogLevel
 from time import sleep
 
-def testLinkLimit( net ):
+def testLinkLimit( net, bw ):
-    print '*** Testing network bandwidth limit'
+    print '*** Testing network %.2f Mbps bandwidth limit' % bw
-    net.iperf()
+    net.iperf( )
 
-def testCpuLimit( net ):
+def testCpuLimit( net, cpu ):
-    print '*** Testing CPU bandwidth limit'
+    pct = cpu * 100
+    print '*** Testing CPU %.0f%% bandwidth limit' % pct
     h1, h2 = net.hosts
     h1.cmd( 'while true; do a=1; done &' )
     h2.cmd( 'while true; do a=1; done &' )
@@ -26,25 +27,23 @@ def testCpuLimit( net ):
     cmd = 'ps -p %s,%s -o pid,%%cpu,args' % ( pid1, pid2 )
     for i in range( 0, 5):
         sleep( 1 ) 
-        print quietRun( cmd )
+        print quietRun( cmd ).strip()
     h1.cmd( 'kill %1')
     h2.cmd( 'kill %1')
 
-def limit():
+def limit( bw=1, cpu=.3 ):
-    "Example/test of link and CPU bandwidth limits"
+    """Example/test of link and CPU bandwidth limits
-    # 1 Mbps interfaces limited using tc
+       bw: interface bandwidth limit in Mbps
-    intf1Mbps = custom( TCIntf, bw=1 )
+       cpu: cpu limit as fraction of overall CPU time"""
-    # Links consisting of two 10 Mbps interfaces
+    intf = custom( TCIntf, bw=1 )
-    link1Mbps = custom( Link, intf=intf1Mbps, cls2=TCIntf )
-    # Hosts with 30% of system bandwidth
-    host30pct = custom( CPULimitedHost, cpu=.3 )
     myTopo = TreeTopo( depth=1, fanout=2 )
-    net = Mininet( topo=myTopo,
+    for sched in 'rt', 'cfs':
-                   link=link1Mbps, 
+        print '*** Testing with', sched, 'bandwidth limiting'
-                   host=host30pct )
+        host = custom( CPULimitedHost, sched=sched, cpu=cpu )
+        net = Mininet( topo=myTopo, intf=intf, host=host )
         net.start()
-    testLinkLimit( net )
+        testLinkLimit( net, bw=bw )
-    testCpuLimit( net )
+        testCpuLimit( net, cpu=cpu )
         net.stop()
 
 if __name__ == '__main__':

mininet/net.py

@@ -104,7 +104,7 @@ class Mininet( object ):
     "Network emulation with hosts spawned in network namespaces."
 
     def __init__( self, topo=None, switch=OVSKernelSwitch, host=Host,
-                 controller=Controller, link=Link,
+                 controller=Controller, link=Link, intf=None, 
                  build=True, xterms=False, cleanup=False,
                  inNamespace=False,
                  autoSetMacs=False, autoStaticArp=False, listenPort=None ):
@@ -114,6 +114,7 @@ class Mininet( object ):
            host: default Host class/constructor
            controller: default Controller class/constructor
            link: default Link class/constructor
+           intf: default Intf class/constructor
            ipBase: base IP address for hosts,
            build: build now from topo?
            xterms: if build now, spawn xterms?
@@ -123,11 +124,12 @@ class Mininet( object ):
            autoStaticArp: set all-pairs static MAC addrs?
            listenPort: base listening port to open; will be incremented for
                each additional switch in the net if inNamespace=False"""
+        self.topo = topo
         self.switch = switch
         self.host = host
         self.controller = controller
         self.link = link
-        self.topo = topo
+        self.intf = intf
         self.inNamespace = inNamespace
         self.xterms = xterms
         self.cleanup = cleanup
@@ -199,12 +201,12 @@ class Mininet( object ):
     def configHosts( self ):
         "Configure a set of hosts."
         for host in self.hosts:
+            info( host.name + ' ' )
             host.configDefault( defaultRoute=host.defaultIntf )
             # You're low priority, dude!
             # BL: do we want to do this here or not?
             # May not make sense if we have CPU lmiting...
             # quietRun( 'renice +18 -p ' + repr( host.pid ) )
-            info( host.name + ' ' )
         info( '\n' )
 
     def buildFromTopo( self, topo=None ):
@@ -235,6 +237,8 @@ class Mininet( object ):
             ei = topo.edgeInfo( srcId, dstId )
             link = getattr( ei, 'cls', link )
             params = ei.params
+            if self.intf and not 'intf' in params:
+                params[ 'intf' ] = self.intf
             if not link:
                 link = self.link
             info( '(%s, %s) ' % ( src.name, dst.name ) )
@@ -447,6 +451,8 @@ class Mininet( object ):
             error( 'could not parse iperf output: ' + iperfOutput )
             return ''
 
+    # XXX This should be cleaned up
+
     def iperf( self, hosts=None, l4Type='TCP', udpBw='10M' ):
         """Run iperf between two hosts.
            hosts: list of hosts; if None, uses opposite hosts

mininet/node.py

@@ -127,9 +127,12 @@ class Node( object ):
         if self.shell:
             error( "%s: shell is already running" )
             return
+        # mnexec: (c)lose descriptors, (d)etach from tty,
+        # (p)rint pid, and run in (n)amespace 
         opts = '-cdp'
         if self.inNamespace:
             opts += 'n'
+        # bash -m: enable job control
         cmd = [ 'mnexec', opts, 'bash', '-m' ]
         self.shell = Popen( cmd, stdin=PIPE, stdout=PIPE, stderr=STDOUT,
             close_fds=True )
@@ -405,10 +408,14 @@ class Node( object ):
     # annoying, but at least the information is there!
 
     def setParam( self, results, method, **param ):
-        """Internal method: configure single parameter"""
+        """Internal method: configure a *single* parameter
+           results: dict of results to update
+           method: config method name
+           param: arg=value (ignore if value=None)
+           value may also be list or dict"""
         name, value = param.items()[ 0 ]
         f = getattr( self, method, None )
-        if not value or not f:
+        if not f or value is None:
             return
         if type( value ) is list:
             result = f( *value )
@@ -417,6 +424,7 @@ class Node( object ):
         else:
             result = f( value )
         results[ name ] = result
+        return result
 
     def config( self, mac=None, ip=None, ifconfig=None, 
                 defaultRoute=None, **params):
@@ -462,7 +470,12 @@ class Node( object ):
             self.name, self.IP(), ','.join( self.intfNames() ), self.pid )
 
 
-class CPULimitedHost( Node ):
+class Host( Node ):
+    "A host is simply a Node"
+    pass
+
+
+class CPULimitedHost( Host ):
 
     "CPU limited host"
 
@@ -472,9 +485,8 @@ class CPULimitedHost( Node ):
         cgroup = 'cpu,cpuacct:/' + self.name
         errFail( 'cgcreate -g ' + cgroup )
         errFail( 'cgclassify -g %s %s' % ( cgroup, self.pid ) )
-        self.sched = 'rt'
+        self.period_us = kwargs.get( 'period_us', 10000 )
-        self.period_us = 10000
+        self.sched = kwargs.get( 'sched', 'rt' )
-        self.rtset = False
 
     def cgroupSet( self, param, value, resource='cpu' ):
         "Set a cgroup parameter and return its value"
@@ -495,40 +507,73 @@ class CPULimitedHost( Node ):
         lastword = firstline.split( ' ' )[ -1 ]
         return lastword
 
-    def setCPUFrac( self, f=-1 ):
+    # BL comment:
-        "Set overall CPU fraction for this host"
+    # This may not be the right API, 
-        if ( f < 0 or f is None):
+    # since it doesn't specify CPU bandwidth in "absolute"
-            # Reset to unlimited
+    # units the way link bandwidth is specified.
-            f = -1
+    # We should use MIPS or SPECINT or something instead.
-        # Set new period and quota
+    # Alternatively, we should change from system fraction
+    # to CPU seconds per second, essentially assuming that
+    # all CPUs are the same.
+    
+    def setCPUFrac( self, f=-1, sched=None):
+        """Set overall CPU fraction for this host
+           f: CPU bandwidth limit (fraction)
+           sched: 'rt' or 'cfs'
+           Note 'cfs' requires CONFIG_CFS_BANDWIDTH"""
+        if not f:
+            return
+        if not sched:
+            sched = self.sched
+        period = self.period_us
+        if sched == 'rt':
             pstr, qstr = 'rt_period_us', 'rt_runtime_us'
+            # RT uses system time for period and quota
+            quota = int( period * f * numCores() )
+        elif sched == 'cfs':
+            pstr, qstr = 'cfs_period_us', 'cfs_quota_us'
+            # CFS uses wall clock time for period and CPU time for quota.
             quota = int( self.period_us * f * numCores() )
-        self.cgroupSet( pstr, self.period_us )
+            if f > 0 and quota < 1000:
+                info( '*** setCPUFrac: quota too small - adjusting period\n' )
+                quota = 1000
+                period = int( quota / f / numCores() )
+        else:
+            return
+        if quota < 0:
+            # Reset to unlimited
+            quota = -1
+        # Set cgroup's period and quota
+        self.cgroupSet( pstr, period )
         nquota = int ( self.cgroupGet( qstr ) )
         self.cgroupSet( qstr, quota )
         nperiod = int( self.cgroupGet( pstr ) )
-        # Set RT priority
+        # Make sure it worked
-        nchrt = self.chrt( prio=20 )
-        # Check to make sure it worked
-        if 'SCHED_RR' not in nchrt:
-            error( '*** error: could not assign SCHED_RR to %s\n' % self.name )
         if nperiod != self.period_us:
             error( '*** error: period is %s rather than %s\n' % (
                     nperiod, self.period_us ) )
         if nquota != quota:
             error( '*** error: quota is %s rather than %s\n' % (
                     nquota, quota ) )
+        if sched == 'rt':
+            # Set RT priority if necessary
+            nchrt = self.chrt( prio=20 )
+            # Nake sure it worked
+            if sched == 'SCHED_RR' not in nchrt:
+                error( '*** error: could not assign SCHED_RR to %s\n' % self.name )
+            info( '( period', nperiod, 'quota', nquota, nchrt, ') ' )
+        else:
+            info( '( period', nperiod, 'quota', nquota, ') ' )
 
-    def config( self, cpu=None, **params ):
+    def config( self, cpu=None, sched=None, **params ):
         """cpu: desired overall system CPU fraction
            params: parameters for Node.config()"""
         r = Node.config( self, **params )
+        # Was considering cpu={'cpu': cpu , 'sched': sched}, but
+        # that seems redundant
         self.setParam( r, 'setCPUFrac', cpu=cpu )
         return r
 
-Host = CPULimitedHost
-
-
 # Some important things to note:
 #
 # The "IP" address which we assign to the switch is not
@@ -805,7 +850,7 @@ class ControllerParams( object ):
 class NOX( Controller ):
     "Controller to run a NOX application."
 
-    def __init__( self, name, noxArgs=None, **kwargs ):
+    def __init__( self, name, noxArgs=[], **kwargs ):
         """Init.
            name: name to give controller
            noxArgs: list of args, or single arg, to pass to NOX"""