Merge pull request #359 from mininet/devel/cluster

Cluster Support Prototype

bin/mn

@@ -22,7 +22,7 @@ if 'PYTHONPATH' in os.environ:
 
 from mininet.clean import cleanup
 from mininet.cli import CLI
-from mininet.log import lg, LEVELS, info, debug, error
+from mininet.log import lg, LEVELS, info, debug, warn, error
 from mininet.net import Mininet, MininetWithControlNet, VERSION
 from mininet.node import ( Host, CPULimitedHost, Controller, OVSController,
                            NOX, RemoteController, DefaultController,
@@ -35,6 +35,15 @@ from mininet.topolib import TreeTopo, TorusTopo
 from mininet.util import custom, customConstructor
 from mininet.util import buildTopo
 
+from functools import partial
+
+# Experimental! cluster edition prototype
+from mininet.examples.cluster import ( MininetCluster, RemoteHost,
+                                       RemoteOVSSwitch, RemoteLink,
+                                       SwitchBinPlacer, RandomPlacer )
+from mininet.examples.clustercli import DemoCLI as ClusterCLI
+
+PLACEMENT = { 'block': SwitchBinPlacer, 'random': RandomPlacer }
 
 # built in topologies, created only when run
 TOPODEF = 'minimal'
@@ -204,6 +213,14 @@ class MininetRunner( object ):
                          default=False, help="adds a NAT to the topology "
                          "that connects Mininet to the physical network" )
         opts.add_option( '--version', action='callback', callback=version )
+        opts.add_option( '--cluster', type='string', default=None,
+                         metavar='server1,server2...',
+                         help=( 'run on multiple servers (experimental!)' ) )
+        opts.add_option( '--placement', type='choice',
+                         choices=PLACEMENT.keys(), default='block',
+                         metavar='block|random',
+                         help=( 'node placement for --cluster '
+                                '(experimental!) ' ) )
 
         self.options, self.args = opts.parse_args()
 
@@ -226,6 +243,8 @@ class MininetRunner( object ):
     def begin( self ):
         "Create and run mininet."
 
+        global CLI
+        
         if self.options.clean:
             cleanup()
             exit()
@@ -241,8 +260,6 @@ class MininetRunner( object ):
         if self.validate:
             self.validate( self.options )
 
-        inNamespace = self.options.innamespace
-        Net = MininetWithControlNet if inNamespace else Mininet
         ipBase = self.options.ipbase
         xterms = self.options.xterms
         mac = self.options.mac
@@ -251,6 +268,22 @@ class MininetRunner( object ):
         listenPort = None
         if not self.options.nolistenport:
             listenPort = self.options.listenport
+
+        # Handle inNamespace, cluster options
+        inNamespace = self.options.innamespace
+        cluster = self.options.cluster
+        if inNamespace and cluster:
+            print "Please specify --innamespace OR --cluster"
+            exit()
+        Net = MininetWithControlNet if inNamespace else Mininet
+        if cluster:
+            warn( '*** WARNING: Experimental cluster mode!\n'
+              '*** Using RemoteHost, RemoteOVSSwitch, RemoteLink\n' )
+            host, switch, link = RemoteHost, RemoteOVSSwitch, RemoteLink
+            CLI = ClusterCLI
+            Net = partial( MininetCluster, servers=cluster.split( ',' ),
+                           placement=PLACEMENT[ self.options.placement ] )
+
         mn = Net( topo=topo,
                   switch=switch, host=host, controller=controller,
                   link=link,

examples/cluster.py

@@ -0,0 +1,839 @@
+#!/usr/bin/python
+
+"""
+cluster.py: prototyping/experimentation for distributed Mininet,
+            aka Mininet: Cluster Edition
+
+Author: Bob Lantz
+
+Core classes:
+
+    RemoteNode: a Node() running on a remote server
+    RemoteOVSSwitch(): an OVSSwitch() running on a remote server
+    RemoteLink: a Link() on a remote server
+    Tunnel: a Link() between a local Node() and a RemoteNode()
+
+These are largely interoperable with local objects.
+
+- One Mininet to rule them all
+
+It is important that the same topologies, APIs, and CLI can be used
+with minimal or no modification in both local and distributed environments.
+
+- Multiple placement models
+
+Placement should be as easy as possible. We should provide basic placement
+support and also allow for explicit placement.
+
+Questions:
+
+What is the basic communication mechanism?
+
+To start with? Probably a single multiplexed ssh connection between each
+pair of mininet servers that needs to communicate.
+
+How are tunnels created?
+
+We have several options including ssh, GRE, OF capsulator, socat, VDE, l2tp,
+etc..  It's not clear what the best one is.  For now, we use ssh tunnels since
+they are encrypted and semi-automatically shared.  We will probably want to
+support GRE as well because it's very easy to set up with OVS.
+
+How are tunnels destroyed?
+
+They are destroyed when the links are deleted in Mininet.stop()
+
+How does RemoteNode.popen() work?
+
+It opens a shared ssh connection to the remote server and attaches to
+the namespace using mnexec -a -g.
+
+Is there any value to using Paramiko vs. raw ssh?
+
+Maybe, but it doesn't seem to support L2 tunneling.
+
+Should we preflight the entire network, including all server-to-server
+connections?
+
+Yes! We don't yet do this with remote server-to-server connections yet.
+
+Should we multiplex the link ssh connections?
+
+Yes, this is done automatically with ControlMaster=auto.
+
+Note on ssh and DNS:
+Please add UseDNS: no to your /etc/ssh/sshd_config!!!
+
+Things to do:
+
+- asynchronous/pipelined/parallel startup
+- ssh debugging/profiling
+- make connections into real objects
+- support for other tunneling schemes
+- tests and benchmarks
+- hifi support (e.g. delay compensation)
+"""
+
+from mininet.node import Node, Host, OVSSwitch, Controller
+from mininet.link import Link, Intf
+from mininet.net import Mininet
+from mininet.topo import LinearTopo
+from mininet.topolib import TreeTopo
+from mininet.util import quietRun, makeIntfPair, errRun, retry
+from mininet.examples.clustercli import CLI
+from mininet.log import setLogLevel, debug, info, error
+
+from signal import signal, SIGINT, SIGHUP, SIG_IGN
+from subprocess import Popen, PIPE, STDOUT
+import os
+from random import randrange
+from sys import exit
+import re
+
+from distutils.version import StrictVersion
+
+# BL note: so little code is required for remote nodes,
+# we will probably just want to update the main Node()
+# class to enable it for remote access! However, there
+# are a large number of potential failure conditions with
+# remote nodes which we may want to detect and handle.
+# Another interesting point is that we could put everything
+# in a mix-in class and easily add cluster mode to 2.0.
+
+class RemoteMixin( object ):
+
+    "A mix-in class to turn local nodes into remote nodes"
+
+    # ssh base command
+    # -q: don't print stupid diagnostic messages
+    # BatchMode yes: don't ask for password
+    # ForwardAgent yes: forward authentication credentials
+    sshbase = [ 'ssh', '-q',
+                '-o', 'BatchMode=yes',
+                '-o', 'ForwardAgent=yes', '-tt' ]
+
+    def __init__( self, name, server=None, user=None, serverIP=None,
+                  controlPath='/tmp/mn-%r@%h:%p', splitInit=False, **kwargs):
+        """Instantiate a remote node
+           name: name of remote node
+           server: remote server (optional)
+           user: user on remote server (optional)
+           controlPath: ssh control path template (optional)
+           splitInit: split initialization?
+           **kwargs: see Node()"""
+        # We connect to servers by IP address
+        if server == 'localhost':
+            server = None
+        self.server = server
+        self.serverIP = serverIP if serverIP else self.findServerIP( server )
+        self.user = user if user else self.findUser()
+        if self.user and self.server:
+            self.dest = '%s@%s' % ( self.user, self.serverIP )
+        else:
+            self.dest = None
+        self.controlPath = controlPath
+        self.sshcmd = []
+        if self.dest:
+            self.sshcmd = [ 'sudo', '-E', '-u', self.user ] + self.sshbase
+            if self.controlPath:
+                self.sshcmd += [ '-o', 'ControlPath=' + self.controlPath,
+                                               '-o', 'ControlMaster=auto' ]
+            self.sshcmd = self.sshcmd + [ self.dest ]
+        self.splitInit = splitInit
+        super( RemoteMixin, self ).__init__( name, **kwargs )
+
+    @staticmethod
+    def findUser():
+        "Try to return logged-in (usually non-root) user"
+        try:
+            # If we're running sudo
+            return os.environ[ 'SUDO_USER' ]
+        except:
+            try:
+                # Logged-in user (if we have a tty)
+                return quietRun( 'who am i' ).split()[ 0 ]
+            except:
+                # Give up and return effective user
+                return quietRun( 'whoami' )
+
+    # Determine IP address of local host
+    _ipMatchRegex = re.compile( r'\d+\.\d+\.\d+\.\d+' )
+
+    @classmethod
+    def findServerIP( cls, server, intf='eth0' ):
+        "Return our server's IP address"
+        # Check for this server
+        if not server:
+            output = quietRun( 'ifconfig %s' % intf  )
+        # Otherwise, handle remote server
+        else:
+            # First, check for an IP address
+            if server:
+                ipmatch = cls._ipMatchRegex.findall( server )
+                if ipmatch:
+                    return ipmatch[ 0 ]
+            # Otherwise, look up remote server
+            output = quietRun( 'getent ahostsv4 %s' % server )
+        ips = cls._ipMatchRegex.findall( output )
+        ip = ips[ 0 ] if ips else None
+        return ip
+
+    # Command support via shell process in namespace
+    def startShell( self, *args, **kwargs ):
+        "Start a shell process for running commands"
+        if self.dest:
+            kwargs.update( mnopts='-c' )
+        super( RemoteMixin, self ).startShell( *args, **kwargs )
+        if self.splitInit:
+            self.sendCmd( 'echo $$' )
+        else:
+            self.pid = int( self.cmd( 'echo $$' ) )
+
+    def finishInit( self ):
+        self.pid = int( self.waitOutput() )
+
+    def rpopen( self, *cmd, **opts ):
+        "Return a Popen object on underlying server in root namespace"
+        params = { 'stdin': PIPE,
+                   'stdout': PIPE,
+                   'stderr': STDOUT,
+                   'sudo': True }
+        params.update( opts )
+        return self._popen( *cmd, **params )
+
+    def rcmd( self, *cmd, **opts):
+        """rcmd: run a command on underlying server
+           in root namespace
+           args: string or list of strings
+           returns: stdout and stderr"""
+        popen = self.rpopen( *cmd, **opts )
+        # print 'RCMD: POPEN:', popen
+        # These loops are tricky to get right.
+        # Once the process exits, we can read
+        # EOF twice if necessary.
+        result = ''
+        while True:
+            poll = popen.poll()
+            result += popen.stdout.read()
+            if poll is not None:
+                break
+        return result
+
+    @staticmethod
+    def _ignoreSignal():
+        "Detach from process group to ignore all signals"
+        os.setpgrp()
+
+    def _popen( self, cmd, sudo=True, tt=True, **params):
+        """Spawn a process on a remote node
+            cmd: remote command to run (list)
+            **params: parameters to Popen()
+            returns: Popen() object"""
+        if type( cmd ) is str:
+            cmd = cmd.split()
+        if self.dest:
+            if sudo:
+                cmd = [ 'sudo', '-E' ] + cmd
+            if tt:
+                cmd = self.sshcmd + cmd
+            else:
+                # Hack: remove -tt
+                sshcmd = list( self.sshcmd )
+                sshcmd.remove( '-tt' )
+                cmd = sshcmd + cmd
+        else:
+            if self.user and not sudo:
+                # Drop privileges
+                cmd = [ 'sudo', '-E', '-u', self.user ] + cmd
+        params.update( preexec_fn=self._ignoreSignal )
+        debug( '_popen', ' '.join(cmd), params )
+        popen = super( RemoteMixin, self )._popen( cmd, **params )
+        return popen
+
+    def popen( self, *args, **kwargs ):
+        "Override: disable -tt"
+        return super( RemoteMixin, self).popen( *args, tt=False, **kwargs )
+
+    def addIntf( self, *args, **kwargs ):
+        "Override: use RemoteLink.moveIntf"
+        return super( RemoteMixin, self).addIntf( *args,
+                        moveIntfFn=RemoteLink.moveIntf, **kwargs )
+
+
+class RemoteNode( RemoteMixin, Node ):
+    "A node on a remote server"
+    pass
+
+
+class RemoteHost( RemoteNode ):
+    "A RemoteHost is simply a RemoteNode"
+    pass
+
+
+class RemoteOVSSwitch( RemoteMixin, OVSSwitch ):
+    "Remote instance of Open vSwitch"
+    OVSVersions = {}
+    def isOldOVS( self ):
+        "Is remote switch using an old OVS version?"
+        cls = type( self )
+        if self.server not in cls.OVSVersions:
+            vers = self.cmd( 'ovs-vsctl --version' )
+            cls.OVSVersions[ self.server ] = re.findall( '\d+\.\d+', vers )[ 0 ]
+        return ( StrictVersion( cls.OVSVersions[ self.server ] ) <
+                StrictVersion( '1.10' ) )
+
+
+
+class RemoteLink( Link ):
+
+    "A RemoteLink is a link between nodes which may be on different servers"
+
+    def __init__( self, node1, node2, **kwargs ):
+        """Initialize a RemoteLink
+           see Link() for parameters"""
+        # Create links on remote node
+        self.node1 = node1
+        self.node2 = node2
+        self.tunnel = None
+        kwargs.setdefault( 'params1', {} )
+        kwargs.setdefault( 'params2', {} )
+        Link.__init__( self, node1, node2, **kwargs )
+
+    def stop( self ):
+        "Stop this link"
+        if self.tunnel:
+            self.tunnel.terminate()
+        self.tunnel = None
+
+    def makeIntfPair( self, intfname1, intfname2, addr1=None, addr2=None ):
+        """Create pair of interfaces
+            intfname1: name of interface 1
+            intfname2: name of interface 2
+            (override this method [and possibly delete()]
+            to change link type)"""
+        node1, node2 = self.node1, self.node2
+        server1 = getattr( node1, 'server', None )
+        server2 = getattr( node2, 'server', None )
+        if not server1 and not server2:
+            # Local link
+            return makeIntfPair( intfname1, intfname2, addr1, addr2 )
+        elif server1 == server2:
+            # Remote link on same remote server
+            return makeIntfPair( intfname1, intfname2, addr1, addr2,
+                                 run=node1.rcmd )
+        # Otherwise, make a tunnel
+        self.tunnel = self.makeTunnel( node1, node2, intfname1, intfname2, addr1, addr2 )
+        return self.tunnel
+
+    @staticmethod
+    def moveIntf( intf, node, printError=True ):
+        """Move remote interface from root ns to node
+            intf: string, interface
+            dstNode: destination Node
+            srcNode: source Node or None (default) for root ns
+            printError: if true, print error"""
+        intf = str( intf )
+        cmd = 'ip link set %s netns %s' % ( intf, node.pid )
+        node.rcmd( cmd )
+        links = node.cmd( 'ip link show' )
+        if not ( ' %s:' % intf ) in links:
+            if printError:
+                error( '*** Error: RemoteLink.moveIntf: ' + intf +
+                      ' not successfully moved to ' + node.name + '\n' )
+            return False
+        return True
+    
+    def makeTunnel( self, node1, node2, intfname1, intfname2,
+                    addr1=None, addr2=None ):
+        "Make a tunnel across switches on different servers"
+        # 1. Create tap interfaces
+        for node in node1, node2:
+            # For now we are hard-wiring tap9, which we will rename
+            node.rcmd( 'ip link delete tap9', stderr=PIPE )
+            cmd = 'ip tuntap add dev tap9 mode tap user ' + node.user
+            node.rcmd( cmd )
+            links = node.rcmd( 'ip link show' )
+            # print 'after add, links =', links
+            assert 'tap9' in links
+        # 2. Create ssh tunnel between tap interfaces
+        # -n: close stdin
+        dest = '%s@%s' % ( node2.user, node2.serverIP )
+        cmd = [ 'ssh', '-n', '-o', 'Tunnel=Ethernet', '-w', '9:9',
+                dest, 'echo @' ]
+        self.cmd = cmd
+        tunnel = node1.rpopen( cmd, sudo=False )
+        # When we receive the character '@', it means that our
+        # tunnel should be set up
+        debug( 'Waiting for tunnel to come up...\n' )
+        ch = tunnel.stdout.read( 1 )
+        if ch != '@':
+            error( 'makeTunnel:\n',
+                   'Tunnel setup failed for',
+                   '%s:%s' % ( node1, node1.dest ), 'to', 
+                   '%s:%s\n' % ( node2, node2.dest ),
+                  'command was:', cmd, '\n' )
+            tunnel.terminate()
+            tunnel.wait()
+            error( ch + tunnel.stdout.read() )
+            error( tunnel.stderr.read() )
+            exit( 1 )
+        # 3. Move interfaces if necessary
+        for node in node1, node2:
+            if node.inNamespace:
+                retry( 3, .01, RemoteLink.moveIntf, 'tap9', node )
+        # 4. Rename tap interfaces to desired names
+        for node, intf, addr in ( ( node1, intfname1, addr1 ),
+                            ( node2, intfname2, addr2 ) ):
+            if not addr:
+                node.cmd( 'ip link set tap9 name', intf )
+            else:
+                node.cmd( 'ip link set tap9 name', intf, 'address', addr )
+        for node, intf in ( ( node1, intfname1 ), ( node2, intfname2 ) ):
+            assert intf in node.cmd( 'ip link show' )
+        return tunnel
+
+    def status( self ):
+        "Detailed representation of link"
+        if self.tunnel:
+            if self.tunnel.poll() is not None:
+                status = "Tunnel EXITED %s" % self.tunnel.returncode
+            else:
+                status = "Tunnel Running (%s: %s)" % (
+                    self.tunnel.pid, self.cmd )
+        else:
+            status = "OK"
+        result = "%s %s" % ( Link.status( self ), status )
+        return result
+
+
+# Some simple placement algorithms for MininetCluster
+
+class Placer( object ):
+    "Node placement algorithm for MininetCluster"
+    
+    def __init__( self, servers=None, nodes=None, hosts=None,
+                 switches=None, controllers=None, links=None ):
+        """Initialize placement object
+           servers: list of servers
+           nodes: list of all nodes
+           hosts: list of hosts
+           switches: list of switches
+           controllers: list of controllers
+           links: list of links
+           (all arguments are optional)
+           returns: server"""
+        self.servers = servers or []
+        self.nodes = nodes or []
+        self.hosts = hosts or []
+        self.switches = switches or []
+        self.controllers = controllers or []
+        self.links = links or []
+
+    def place( self, node ):
+        "Return server for a given node"
+        # Default placement: run locally
+        return None
+
+
+class RandomPlacer( Placer ):
+    "Random placement"
+    def place( self, nodename ):
+        """Random placement function
+            nodename: node name"""
+        # This may be slow with lots of servers
+        return self.servers[ randrange( 0, len( self.servers ) ) ]
+
+
+class RoundRobinPlacer( Placer ):
+    """Round-robin placement
+       Note this will usually result in cross-server links between
+       hosts and switches"""
+    
+    def __init__( self, *args, **kwargs ):
+        Placer.__init__( self, *args, **kwargs )
+        self.next = 0
+
+    def place( self, nodename ):
+        """Round-robin placement function
+            nodename: node name"""
+        # This may be slow with lots of servers
+        server = self.servers[ self.next ]
+        self.next = ( self.next + 1 ) % len( self.servers )
+        return server
+
+
+class SwitchBinPlacer( Placer ):
+    """Place switches (and controllers) into evenly-sized bins,
+       and attempt to co-locate hosts and switches"""
+
+    def __init__( self, *args, **kwargs ):
+        Placer.__init__( self, *args, **kwargs )
+        # Easy lookup for servers and node sets
+        self.servdict = dict( enumerate( self.servers ) )
+        self.hset = frozenset( self.hosts )
+        self.sset = frozenset( self.switches )
+        self.cset = frozenset( self.controllers )
+        # Server and switch placement indices
+        self.placement =  self.calculatePlacement()
+
+    @staticmethod
+    def bin( nodes, servers ):
+        "Distribute nodes evenly over servers"
+        # Calculate base bin size
+        nlen = len( nodes )
+        slen = len( servers )
+        # Basic bin size
+        quotient = int( nlen / slen )
+        binsizes = { server: quotient for server in servers }
+        # Distribute remainder
+        remainder = nlen % slen
+        for server in servers[ 0 : remainder ]:
+            binsizes[ server ] += 1
+        # Create binsize[ server ] tickets for each server
+        tickets = sum( [ binsizes[ server ] * [ server ]
+                         for server in servers ], [] )
+        # And assign one ticket to each node
+        return { node: ticket for node, ticket in zip( nodes, tickets ) }
+
+    def calculatePlacement( self ):
+        "Pre-calculate node placement"
+        placement = {}
+        # Create host-switch connectivity map,
+        # associating host with last switch that it's
+        # connected to
+        switchFor = {}
+        for src, dst in self.links:
+            if src in self.hset and dst in self.sset:
+                switchFor[ src ] = dst
+            if dst in self.hset and src in self.sset:
+                switchFor[ dst ] = src
+        # Place switches
+        placement = self.bin( self.switches, self.servers )
+        # Place controllers and merge into placement dict
+        placement.update( self.bin( self.controllers, self.servers ) )
+        # Co-locate hosts with their switches
+        for h in self.hosts:
+            if h in placement:
+                # Host is already placed - leave it there
+                continue
+            if h in switchFor:
+                placement[ h ] = placement[ switchFor[ h ] ]
+            else:
+                raise Exception(
+                        "SwitchBinPlacer: cannot place isolated host " + h )
+        return placement
+
+    def place( self, node ):
+        """Simple placement algorithm:
+           place switches into evenly sized bins,
+           and place hosts near their switches"""
+        return self.placement[ node ]
+
+
+class HostSwitchBinPlacer( Placer ):
+    """Place switches *and hosts* into evenly-sized bins
+       Note that this will usually result in cross-server
+       links between hosts and switches"""
+
+    def __init__( self, *args, **kwargs ):
+        Placer.__init__( self, *args, **kwargs )
+        # Calculate bin sizes
+        scount = len( self.servers )
+        self.hbin = max( int( len( self.hosts ) / scount ), 1 )
+        self.sbin = max( int( len( self.switches ) / scount ), 1 )
+        self.cbin = max( int( len( self.controllers ) / scount ) , 1 )
+        info( 'scount:', scount )
+        info( 'bins:', self.hbin, self.sbin, self.cbin, '\n' )
+        self.servdict = dict( enumerate( self.servers ) )
+        self.hset = frozenset( self.hosts )
+        self.sset = frozenset( self.switches )
+        self.cset = frozenset( self.controllers )
+        self.hind, self.sind, self.cind = 0, 0, 0
+    
+    def place( self, nodename ):
+        """Simple placement algorithm:
+            place nodes into evenly sized bins"""
+        # Place nodes into bins
+        if nodename in self.hset:
+            server = self.servdict[ self.hind / self.hbin ]
+            self.hind += 1
+        elif nodename in self.sset:
+            server = self.servdict[ self.sind / self.sbin ]
+            self.sind += 1
+        elif nodename in self.cset:
+            server = self.servdict[ self.cind / self.cbin ]
+            self.cind += 1
+        else:
+            info( 'warning: unknown node', nodename )
+            server = self.servdict[ 0 ]
+        return server
+
+
+
+# The MininetCluster class is not strictly necessary.
+# However, it has several purposes:
+# 1. To set up ssh connection sharing/multiplexing
+# 2. To pre-flight the system so that everything is more likely to work
+# 3. To allow connection/connectivity monitoring
+# 4. To support pluggable placement algorithms
+
+class MininetCluster( Mininet ):
+
+    "Cluster-enhanced version of Mininet class"
+
+    # Default ssh command
+    # BatchMode yes: don't ask for password
+    # ForwardAgent yes: forward authentication credentials
+    sshcmd = [ 'ssh', '-o', 'BatchMode=yes', '-o', 'ForwardAgent=yes' ]
+
+    def __init__( self, *args, **kwargs ):
+        """servers: a list of servers to use (note: include
+           localhost or None to use local system as well)
+           user: user name for server ssh
+           placement: Placer() subclass"""
+        params = { 'host': RemoteHost,
+                   'switch': RemoteOVSSwitch,
+                   'link': RemoteLink,
+                   'precheck': True }
+        params.update( kwargs )
+        servers = params.pop( 'servers', [ None ] )
+        servers = [ s if s != 'localhost' else None for s in servers ]
+        self.servers = servers
+        self.serverIP = params.pop( 'serverIP', {} )
+        if not self.serverIP:
+            self.serverIP = { server: RemoteMixin.findServerIP( server )
+                              for server in self.servers }
+        self.user = params.pop( 'user', RemoteMixin.findUser() )
+        if params.pop( 'precheck' ):
+            self.precheck()
+        self.connections = {}
+        self.placement = params.pop( 'placement', SwitchBinPlacer )
+        # Make sure control directory exists
+        self.cdir = os.environ[ 'HOME' ] + '/.ssh/mn'
+        errRun( [ 'mkdir', '-p', self.cdir ] )
+        Mininet.__init__( self, *args, **params )
+
+    def popen( self, cmd ):
+        "Popen() for server connections"
+        old = signal( SIGINT, SIG_IGN )
+        conn = Popen( cmd, stdin=PIPE, stdout=PIPE, close_fds=True )
+        signal( SIGINT, old )
+        return conn
+
+    def baddLink( self, *args, **kwargs ):
+        "break addlink for testing"
+        pass
+
+    def precheck( self ):
+        """Pre-check to make sure connection works and that
+           we can call sudo without a password"""
+        result = 0
+        info( '*** Checking servers\n' )
+        for server in self.servers:
+            ip = self.serverIP[ server ]
+            if not server or server == 'localhost':
+                 continue
+            info( server, '' )
+            dest = '%s@%s' % ( self.user, ip )
+            cmd = [ 'sudo', '-E', '-u', self.user ]
+            cmd += self.sshcmd + [ '-n', dest, 'sudo true' ]
+            debug( ' '.join( cmd ), '\n' )
+            out, err, code = errRun( cmd )
+            if code != 0:
+                error( '\nstartConnection: server connection check failed '
+                       'to %s using command:\n%s\n'
+                        % ( server, ' '.join( cmd ) ) )
+            result |= code
+        if result:
+            error( '*** Server precheck failed.\n'
+                   '*** Make sure that the above ssh command works correctly.\n'
+                   '*** You may also need to run mn -c on all nodes, and/or\n'
+                   '*** use sudo -E.\n' )
+            exit( 1 )
+        info( '\n' )
+
+    def modifiedaddHost( self, *args, **kwargs ):
+        "Slightly modify addHost"
+        kwargs[ 'splitInit' ] = True
+        return Mininet.addHost( *args, **kwargs )
+
+
+    def placeNodes( self ):
+        """Place nodes on servers (if they don't have a server), and
+           start shell processes"""
+        if not self.servers or not self.topo:
+            # No shirt, no shoes, no service
+            return
+        nodes = self.topo.nodes()
+        placer = self.placement( servers=self.servers,
+                                 nodes=self.topo.nodes(),
+                                 hosts=self.topo.hosts(),
+                                 switches=self.topo.switches(),
+                                 links=self.topo.links() )
+        for node in nodes:
+            config = self.topo.node_info[ node ]
+            server = config.setdefault( 'server', placer.place( node ) )
+            if server:
+                config.setdefault( 'serverIP', self.serverIP[ server ] )
+            info( '%s:%s ' % ( node, server ) )
+            key = ( None, server )
+            _dest, cfile, _conn = self.connections.get(
+                        key, ( None, None, None ) )
+            if cfile:
+                config.setdefault( 'controlPath', cfile )
+
+    def addController( self, *args, **kwargs ):
+        "Patch to update IP address to global IP address"
+        controller = Mininet.addController( self, *args, **kwargs )
+        # Update IP address for controller that may not be local
+        if ( isinstance( controller, Controller)
+             and controller.IP() == '127.0.0.1'
+             and ' eth0:' in controller.cmd( 'ip link show' ) ):
+             Intf( 'eth0', node=controller ).updateIP()
+        return controller
+
+    def buildFromTopo( self, *args, **kwargs ):
+        "Start network"
+        info( '*** Placing nodes\n' )
+        self.placeNodes()
+        info( '\n' )
+        Mininet.buildFromTopo( self, *args, **kwargs )
+
+
+def testNsTunnels():
+    "Test tunnels between nodes in namespaces"
+    net = Mininet( host=RemoteHost, link=RemoteLink )
+    h1 = net.addHost( 'h1' )
+    h2 = net.addHost( 'h2', server='ubuntu2' )
+    net.addLink( h1, h2 )
+    net.start()
+    net.pingAll()
+    net.stop()
+
+# Manual topology creation with net.add*()
+#
+# This shows how node options may be used to manage
+# cluster placement using the net.add*() API
+
+def testRemoteNet( remote='ubuntu2' ):
+    "Test remote Node classes"
+    print '*** Remote Node Test'
+    net = Mininet( host=RemoteHost, switch=RemoteOVSSwitch,
+                   link=RemoteLink )
+    c0 = net.addController( 'c0' )
+    # Make sure controller knows its non-loopback address
+    Intf( 'eth0', node=c0 ).updateIP()
+    print "*** Creating local h1"
+    h1 = net.addHost( 'h1' )
+    print "*** Creating remote h2"
+    h2 = net.addHost( 'h2', server=remote )
+    print "*** Creating local s1"
+    s1 = net.addSwitch( 's1' )
+    print "*** Creating remote s2"
+    s2 = net.addSwitch( 's2', server=remote )
+    print "*** Adding links"
+    net.addLink( h1, s1 )
+    net.addLink( s1, s2 )
+    net.addLink( h2, s2 )
+    net.start()
+    print 'Mininet is running on', quietRun( 'hostname' ).strip()
+    for node in c0, h1, h2, s1, s2:
+        print 'Node', node, 'is running on', node.cmd( 'hostname' ).strip()
+    net.pingAll()
+    CLI( net )
+    net.stop()
+
+
+# High-level/Topo API example
+#
+# This shows how existing Mininet topologies may be used in cluster
+# mode by creating node placement functions and a controller which
+# can be accessed remotely. This implements a very compatible version
+# of cluster edition with a minimum of code!
+
+remoteHosts = [ 'h2' ]
+remoteSwitches = [ 's2' ]
+remoteServer = 'ubuntu2'
+
+def HostPlacer( name, *args, **params ):
+    "Custom Host() constructor which places hosts on servers"
+    if name in remoteHosts:
+        return RemoteHost( name, *args, server=remoteServer, **params )
+    else:
+        return Host( name, *args, **params )
+
+def SwitchPlacer( name, *args, **params ):
+    "Custom Switch() constructor which places switches on servers"
+    if name in remoteSwitches:
+        return RemoteOVSSwitch( name, *args, server=remoteServer, **params )
+    else:
+        return RemoteOVSSwitch( name, *args, **params )
+
+def ClusterController( *args, **kwargs):
+    "Custom Controller() constructor which updates its eth0 IP address"
+    controller = Controller( *args, **kwargs )
+    # Find out its IP address so that cluster switches can connect
+    Intf( 'eth0', node=controller ).updateIP()
+    return controller
+
+def testRemoteTopo():
+    "Test remote Node classes using Mininet()/Topo() API"
+    topo = LinearTopo( 2 )
+    net = Mininet( topo=topo, host=HostPlacer, switch=SwitchPlacer,
+                  link=RemoteLink, controller=ClusterController )
+    net.start()
+    net.pingAll()
+    net.stop()
+
+# Need to test backwards placement, where each host is on
+# a server other than its switch!! But seriously we could just
+# do random switch placement rather than completely random
+# host placement.
+
+def testRemoteSwitches():
+    "Test with local hosts and remote switches"
+    servers = [ 'localhost', 'ubuntu2']
+    topo = TreeTopo( depth=4, fanout=2 )
+    net = MininetCluster( topo=topo, servers=servers,
+                          placement=RoundRobinPlacer )
+    net.start()
+    net.pingAll()
+    net.stop()
+
+
+#
+# For testing and demo purposes it would be nice to draw the
+# network graph and color it based on server.
+
+# The MininetCluster() class integrates pluggable placement
+# functions, for maximum ease of use. MininetCluster() also
+# pre-flights and multiplexes server connections.
+
+def testMininetCluster():
+    "Test MininetCluster()"
+    servers = [ 'localhost', 'ubuntu2' ]
+    topo = TreeTopo( depth=3, fanout=3 )
+    net = MininetCluster( topo=topo, servers=servers,
+                          placement=SwitchBinPlacer )
+    net.start()
+    net.pingAll()
+    net.stop()
+
+def signalTest():
+    "Make sure hosts are robust to signals"
+    h = RemoteHost( 'h0', server='ubuntu1' )
+    h.shell.send_signal( SIGINT )
+    h.shell.poll()
+    if h.shell.returncode is None:
+        print 'OK: ', h, 'has not exited'
+    else:
+        print 'FAILURE:', h, 'exited with code', h.shell.returncode
+    h.stop()
+
+if __name__ == '__main__':
+    setLogLevel( 'info' )
+    # testRemoteTopo()
+    # testRemoteNet()
+    # testMininetCluster()
+    # testRemoteSwitches()
+    signalTest()

examples/clusterSanity.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python
+
+'''
+A sanity check for cluster edition
+'''
+
+from mininet.examples.cluster import MininetCluster
+from mininet.log import info, setLogLevel
+from mininet.examples.clustercli import DemoCLI as CLI
+from mininet.topo import SingleSwitchTopo
+
+def clusterSanity():
+    "Sanity check for cluster mode"
+    topo = SingleSwitchTopo()
+    net = MininetCluster( topo=topo ) 
+    net.start()
+    CLI( net )
+    net.stop()
+
+if __name__  == '__main__':
+    setLogLevel( 'info' )
+    clusterSanity()

examples/clustercli.py

@@ -0,0 +1,93 @@
+#!/usr/bin/python
+
+"CLI for Mininet Cluster Edition prototype demo"
+
+from mininet.cli import CLI
+from mininet.log import output, error
+
+nx, graphviz_layout, plt = None, None, None  # Will be imported on demand
+
+
+class DemoCLI( CLI ):
+    "CLI with additional commands for Cluster Edition demo"
+
+    @staticmethod
+    def colorsFor( seq ):
+        "Return a list of background colors for a sequence"
+        colors = [ 'red', 'lightgreen', 'cyan', 'yellow', 'orange',
+                  'magenta', 'pink', 'grey', 'brown',
+                  'white' ]
+        slen, clen = len( seq ), len( colors )
+        reps = max( 1, slen / clen )
+        colors = colors * reps
+        colors = colors[ 0 : slen ]
+        return colors
+    
+    def do_plot( self, line ):
+        "Plot topology colored by node placement"
+        # Import networkx if needed
+        global nx, plt
+        if not nx:
+            try:
+                import networkx as nx
+                import matplotlib.pyplot as plt
+                import pygraphviz
+            except:
+                error( 'plot requires networkx, matplotlib and pygraphviz - '
+                       'please install them and try again\n' )
+                return
+        # Make a networkx Graph
+        g = nx.Graph()
+        mn = self.mn
+        servers, hosts, switches = mn.servers, mn.hosts, mn.switches
+        hlen, slen = len( hosts ), len( switches )
+        nodes = hosts + switches
+        g.add_nodes_from( nodes )
+        links = [ ( link.intf1.node, link.intf2.node )
+                  for link in self.mn.links ]
+        g.add_edges_from( links )
+        # Pick some shapes and colors
+        # shapes = hlen * [ 's' ] + slen * [ 'o' ]
+        color = dict( zip( servers, self.colorsFor( servers ) ) )
+        # Plot it!
+        pos = nx.graphviz_layout( g )
+        opts = { 'ax': None, 'font_weight': 'bold',
+		 'width': 2, 'edge_color': 'darkblue' }
+        hcolors = [ color[ h.server ] for h in hosts ]
+        scolors = [ color[ s.server ] for s in switches ]
+        nx.draw_networkx( g, pos=pos, nodelist=hosts, node_size=800, label='host',
+                          node_color=hcolors, node_shape='s', **opts )
+        nx.draw_networkx( g, pos=pos, nodelist=switches, node_size=1000,
+                          node_color=scolors, node_shape='o', **opts )
+        # Get rid of axes, add title, and show
+        fig = plt.gcf()
+        ax = plt.gca()
+        ax.get_xaxis().set_visible( False )
+        ax.get_yaxis().set_visible( False )
+        fig.canvas.set_window_title( 'Mininet')
+        plt.title( 'Node Placement', fontweight='bold' )
+        plt.show()
+
+    def do_status( self, line ):
+        "Report on node shell status"
+        nodes = self.mn.hosts + self.mn.switches
+        for node in nodes:
+            node.shell.poll()
+        exited = [ node for node in nodes
+                   if node.shell.returncode is not None ]
+        if exited:
+            for node in exited:
+                output( '%s has exited with code %d\n'
+                        % ( node, node.shell.returncode ) )
+        else:
+            output( 'All nodes are still running.\n' )
+
+
+    def do_placement( self, line ):
+        "Describe node placement"
+        mn = self.mn
+        nodes = mn.hosts + mn.switches + mn.controllers
+        for server in mn.servers:
+            names = [ n.name for n in nodes if hasattr( n, 'server' )
+                      and n.server == server ]
+            output( '%s: %s\n' % ( server, ' '.join( names ) ) )

examples/clusterdemo.py

@@ -0,0 +1,23 @@
+#!/usr/bin/python
+
+"clusterdemo.py: demo of Mininet Cluster Edition prototype"
+
+from mininet.examples.cluster import MininetCluster, SwitchBinPlacer
+from mininet.topolib import TreeTopo
+from mininet.log import setLogLevel
+from mininet.examples.clustercli import DemoCLI as CLI
+
+def demo():
+    "Simple Demo of Cluster Mode"
+    servers = [ 'localhost', 'ubuntu2', 'ubuntu3' ]
+    topo = TreeTopo( depth=3, fanout=3 )
+    net = MininetCluster( topo=topo, servers=servers,
+                          placement=SwitchBinPlacer )
+    net.start()
+    CLI( net )
+    net.stop()
+
+if __name__ == '__main__':
+    setLogLevel( 'info' )
+    demo()
+

examples/test/test_clusterSanity.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python
+
+'''
+A simple sanity check test for cluster edition
+'''
+
+import unittest
+import pexpect
+
+class clusterSanityCheck( unittest.TestCase ):
+
+    prompt = 'mininet>'
+
+    def testClusterPingAll( self ):
+        p = pexpect.spawn( 'python -m mininet.examples.clusterSanity' )
+        p.expect( self.prompt )
+        p.sendline( 'pingall' )
+        p.expect ( '(\d+)% dropped' )
+        percent = int( p.match.group( 1 ) ) if p.match else -1
+        self.assertEqual( percent, 0 )
+        p.expect( self.prompt )
+        p.sendline( 'exit' )
+        p.wait()
+
+
+if __name__  == '__main__':
+    unittest.main()

mininet/clean.py

@@ -21,6 +21,21 @@ def sh( cmd ):
     info( cmd + '\n' )
     return Popen( [ '/bin/sh', '-c', cmd ], stdout=PIPE ).communicate()[ 0 ]
 
+def killprocs( pattern ):
+    "Reliably terminate processes matching a pattern (including args)"
+    sh( 'pkill -9 -f %s' % pattern )
+    # Make sure they are gone
+    while True:
+        try:
+            pids = co( 'pgrep -f %s' % pattern )
+        except:
+            pids = ''
+        if pids:
+            sh( 'pkill -f 9 mininet:' )
+            sleep( .5 )
+        else:
+            break
+
 def cleanup():
     """Clean up junk which might be left over from old runs;
        do fast stuff before slow dp and link removal!"""
@@ -70,17 +85,11 @@ def cleanup():
             sh( "ip link del " + link )
 
     info( "*** Killing stale mininet node processes\n" )
-    sh( 'pkill -9 -f mininet:' )
-    # Make sure they are gone
-    while True:
-        try:
-            pids = co( 'pgrep -f mininet:'.split() )
-        except:
-            pids = ''
-        if pids:
-            sh( 'pkill -f 9 mininet:' )
-            sleep( .5 )
-        else:
-            break
+    killprocs( 'mininet:' )
 
+    info ( "*** Shutting down stale tunnels\n" )
+    killprocs( 'Tunnel=Ethernet' )
+    killprocs( '.ssh/mn')
+    sh( 'rm -f ~/.ssh/mn/*' )
+    
     info( "*** Cleanup complete.\n" )

mininet/cli.py

@@ -75,7 +75,7 @@ class CLI( Cmd ):
                 for node in self.mn.values():
                     while node.waiting:
                         node.sendInt()
-                        node.monitor()
+                        node.waitOutput()
                 if self.isatty():
                     quietRun( 'stty echo sane intr "^C"' )
                 self.cmdloop()
@@ -331,6 +331,11 @@ class CLI( Cmd ):
         elapsed = time.time() - start
         self.stdout.write("*** Elapsed time: %0.6f secs\n" % elapsed)
 
+    def do_links( self, line ):
+        "Report on links"
+        for link in self.mn.links:
+            print link, link.status()
+
     def default( self, line ):
         """Called on an input line when the command prefix is not recognized.
         Overridden to run shell commands when a node is the first CLI argument.

mininet/link.py

@@ -32,7 +32,8 @@ class Intf( object ):
 
     "Basic interface object that can configure itself."
 
-    def __init__( self, name, node=None, port=None, link=None, mac=None, **params ):
+    def __init__( self, name, node=None, port=None, link=None,
+                  mac=None, srcNode=None, **params ):
         """name: interface name (e.g. h1-eth0)
            node: owning node (where this intf most likely lives)
            link: parent link if we're part of a link
@@ -189,6 +190,14 @@ class Intf( object ):
             # Link may have been dumped into root NS
             quietRun( 'ip link del ' + self.name )
 
+    def status( self ):
+        "Return intf status as a string"
+        links, err_, result_ = self.node.pexec( 'ip link show' )
+        if self.name in links:
+            return "OK"
+        else:
+            return "MISSING"
+
     def __repr__( self ):
         return '<%s %s>' % ( self.__class__.__name__, self.name )
 
@@ -367,16 +376,27 @@ class Link( object ):
            params1: parameters for interface 1
            params2: parameters for interface 2"""
         # This is a bit awkward; it seems that having everything in
-        # params would be more orthogonal, but being able to specify
-        # in-line arguments is more convenient!
-        if port1 is None:
-            port1 = node1.newPort()
-        if port2 is None:
-            port2 = node2.newPort()
+        # params is more orthogonal, but being able to specify
+        # in-line arguments is more convenient! So we support both.
+        if params1 is None:
+            params1 = {}
+        if params2 is None:
+            params2 = {}
+        # Allow passing in params1=params2
+        if params2 is params1:
+            params2 = dict( params1 )
+        if port1 is not None:
+            params1[ 'port' ] = port1
+        if port2 is not None:
+            params2[ 'port' ] = port2
+        if 'port' not in params1:
+            params1[ 'port' ] = node1.newPort()
+        if 'port' not in params2:
+            params2[ 'port' ] = node2.newPort()
         if not intfName1:
-            intfName1 = self.intfName( node1, port1 )
+            intfName1 = self.intfName( node1, params1[ 'port' ] )
         if not intfName2:
-            intfName2 = self.intfName( node2, port2 )
+            intfName2 = self.intfName( node2, params2[ 'port' ] )
 
         self.makeIntfPair( intfName1, intfName2, addr1, addr2 )
 
@@ -384,38 +404,41 @@ class Link( object ):
             cls1 = intf
         if not cls2:
             cls2 = intf
-        if not params1:
-            params1 = {}
-        if not params2:
-            params2 = {}
 
-        intf1 = cls1( name=intfName1, node=node1, port=port1,
+        intf1 = cls1( name=intfName1, node=node1,
                       link=self, mac=addr1, **params1  )
-        intf2 = cls2( name=intfName2, node=node2, port=port2,
+        intf2 = cls2( name=intfName2, node=node2,
                       link=self, mac=addr2, **params2 )
 
         # All we are is dust in the wind, and our two interfaces
         self.intf1, self.intf2 = intf1, intf2
 
-    @classmethod
-    def intfName( cls, node, n ):
+    def intfName( _self, node, n ):
         "Construct a canonical interface name node-ethN for interface n."
         return node.name + '-eth' + repr( n )
 
     @classmethod
-    def makeIntfPair( cls, intf1, intf2, addr1=None, addr2=None ):
+    def makeIntfPair( _cls, intfname1, intfname2, addr1=None, addr2=None ):
         """Create pair of interfaces
-           intf1: name of interface 1
-           intf2: name of interface 2
-           (override this class method [and possibly delete()]
+           intfname1: name of interface 1
+           intfname2: name of interface 2
+           (override this method [and possibly delete()]
            to change link type)"""
-        makeIntfPair( intf1, intf2, addr1, addr2 )
+        return makeIntfPair( intfname1, intfname2, addr1, addr2 )
 
     def delete( self ):
         "Delete this link"
         self.intf1.delete()
         self.intf2.delete()
 
+    def stop( self ):
+        "Override to stop and clean up link as needed"
+        pass
+
+    def status( self ):
+        "Return link status as a string"
+        return "(%s %s)" % ( self.intf1.status(), self.intf2.status() )
+
     def __str__( self ):
         return '%s<->%s' % ( self.intf1, self.intf2 )
 
@@ -430,4 +453,4 @@ class TCLink( Link ):
                        cls2=TCIntf,
                        addr1=addr1, addr2=addr2,
                        params1=params,
-                       params2=params)
+                       params2=params )

mininet/net.py

@@ -157,6 +157,7 @@ class Mininet( object ):
         self.hosts = []
         self.switches = []
         self.controllers = []
+        self.links = []
 
         self.nameToNode = {}  # name to Node (Host/Switch) objects
 
@@ -338,7 +339,9 @@ class Mininet( object ):
         defaults.update( params )
         if not cls:
             cls = self.link
-        return cls( node1, node2, **defaults )
+        link = cls( node1, node2, **defaults )
+        self.links.append( link )
+        return link
 
     def configHosts( self ):
         "Configure a set of hosts."
@@ -478,6 +481,11 @@ class Mininet( object ):
         for switch in self.switches:
             info( switch.name + ' ' )
             switch.stop()
+            switch.terminate()
+        info( '\n' )
+        info( '*** Stopping %i links\n' % len( self.links ) )
+        for link in self.links:
+            link.stop()
         info( '\n' )
         info( '*** Stopping %i hosts\n' % len( self.hosts ) )
         for host in self.hosts:

mininet/node.py

@@ -80,8 +80,8 @@ class Node( object ):
         # Make sure class actually works
         self.checkSetup()
 
-        self.name = name
-        self.inNamespace = inNamespace
+        self.name = params.get( 'name', name )
+        self.inNamespace = params.get( 'inNamespace', inNamespace )
 
         # Stash configuration parameters for future reference
         self.params = params
@@ -116,29 +116,28 @@ class Node( object ):
         return node or cls.inToNode.get( fd )
 
     # Command support via shell process in namespace
-
-    def startShell( self ):
+    def startShell( self, mnopts=None ):
         "Start a shell process for running commands"
         if self.shell:
-            error( "%s: shell is already running" )
+            error( "%s: shell is already running\n" % self.name )
             return
         # mnexec: (c)lose descriptors, (d)etach from tty,
         # (p)rint pid, and run in (n)amespace
-        opts = '-cd'
+        opts = '-cd' if mnopts is None else mnopts
         if self.inNamespace:
             opts += 'n'
         # bash -m: enable job control, i: force interactive
         # -s: pass $* to shell, and make process easy to find in ps
         # prompt is set to sentinel chr( 127 )
-        os.environ[ 'PS1' ] = chr( 127 )
-        cmd = [ 'mnexec', opts, 'bash', '--norc', '-mis', 'mininet:' + self.name ]
+        cmd = [ 'mnexec', opts, 'env',  'PS1=' + chr( 127 ),
+                'bash', '--norc', '-mis', 'mininet:' + self.name ]
         # Spawn a shell subprocess in a pseudo-tty, to disable buffering
         # in the subprocess and insulate it from signals (e.g. SIGINT)
         # received by the parent
         master, slave = pty.openpty()
-        self.shell = Popen( cmd, stdin=slave, stdout=slave, stderr=slave,
+        self.shell = self._popen( cmd, stdin=slave, stdout=slave, stderr=slave,
                                   close_fds=False )
-        self.stdin = os.fdopen( master )
+        self.stdin = os.fdopen( master, 'rw' )
         self.stdout = self.stdin
         self.pid = self.shell.pid
         self.pollOut = select.poll()
@@ -162,6 +161,12 @@ class Node( object ):
         self.cmd( 'stty -echo' )
         self.cmd( 'set +m' )
 
+    def _popen( self, cmd, **params ):
+        """Internal method: spawn and return a process
+            cmd: command to run (list)
+            params: parameters to Popen()"""
+        return Popen( cmd, **params )
+
     def cleanup( self ):
         "Help python collect its garbage."
         # Intfs may end up in root NS
@@ -206,7 +211,8 @@ class Node( object ):
     def terminate( self ):
         "Send kill signal to Node and clean up after it."
         if self.shell:
-            os.killpg( self.pid, signal.SIGHUP )
+            if self.shell.poll() is None:
+                os.killpg( self.shell.pid, signal.SIGHUP )
         self.cleanup()
 
     def stop( self ):
@@ -251,12 +257,14 @@ class Node( object ):
 
     def sendInt( self, intr=chr( 3 ) ):
         "Interrupt running command."
+        debug( 'sendInt: writing chr(%d)\n' % ord( intr ) )
         self.write( intr )
 
     def monitor( self, timeoutms=None, findPid=True ):
         """Monitor and return the output of a command.
            Set self.waiting to False if command has completed.
-           timeoutms: timeout in ms or None to wait indefinitely."""
+           timeoutms: timeout in ms or None to wait indefinitely
+           findPid: look for PID from mnexec -p"""
         self.waitReadable( timeoutms )
         data = self.read( 1024 )
         pidre = r'\[\d+\] \d+\r\n'
@@ -282,7 +290,7 @@ class Node( object ):
             data = data.replace( chr( 127 ), '' )
         return data
 
-    def waitOutput( self, verbose=False ):
+    def waitOutput( self, verbose=False, findPid=True ):
         """Wait for a command to complete.
            Completion is signaled by a sentinel character, ASCII(127)
            appearing in the output stream.  Wait for the sentinel and return
@@ -331,18 +339,19 @@ class Node( object ):
             # popen( cmd, arg1, arg2... )
             cmd = list( args )
         # Attach to our namespace  using mnexec -a
-        mncmd = defaults[ 'mncmd' ]
-        del defaults[ 'mncmd' ]
-        cmd = mncmd + cmd
+        cmd = defaults.pop( 'mncmd' ) + cmd
         # Shell requires a string, not a list!
         if defaults.get( 'shell', False ):
             cmd = ' '.join( cmd )
-        return Popen( cmd, **defaults )
+        popen = self._popen( cmd, **defaults )
+        return popen
 
     def pexec( self, *args, **kwargs ):
         """Execute a command using popen
            returns: out, err, exitcode"""
-        popen = self.popen( *args, **kwargs)
+        popen = self.popen( *args, stdin=PIPE, stdout=PIPE, stderr=PIPE,
+                           **kwargs )
+        # Warning: this can fail with large numbers of fds!
         out, err = popen.communicate()
         exitcode = popen.wait()
         return out, err, exitcode
@@ -361,20 +370,22 @@ class Node( object ):
             return max( self.ports.values() ) + 1
         return self.portBase
 
-    def addIntf( self, intf, port=None ):
+    def addIntf( self, intf, port=None, moveIntfFn=moveIntf ):
         """Add an interface.
            intf: interface
-           port: port number (optional, typically OpenFlow port number)"""
+           port: port number (optional, typically OpenFlow port number)
+           moveIntfFn: function to move interface (optional)"""
         if port is None:
             port = self.newPort()
         self.intfs[ port ] = intf
         self.ports[ intf ] = port
         self.nameToIntf[ intf.name ] = intf
         debug( '\n' )
-        debug( 'added intf %s:%d to node %s\n' % ( intf, port, self.name ) )
+        debug( 'added intf %s (%d) to node %s\n' % (
+                intf, port, self.name ) )
         if self.inNamespace:
             debug( 'moving', intf, 'into namespace for', self.name, '\n' )
-            moveIntf( intf.name, self )
+            moveIntfFn( intf.name, self  )
 
     def defaultIntf( self ):
         "Return interface for lowest port"
@@ -1105,7 +1116,8 @@ class OVSSwitch( Switch ):
         intfs = ' '.join( '-- add-port %s %s ' % ( self, intf ) +
                           '-- set Interface %s ' % intf +
                           'ofport_request=%s ' % self.ports[ intf ]
-                         for intf in self.intfList() if not intf.IP() )
+                         for intf in self.intfList()
+                         if self.ports[ intf ] and not intf.IP() )
         clist = ' '.join( '%s:%s:%d' % ( c.protocol, c.IP(), c.port )
                          for c in controllers )
         if self.listenPort:

mininet/util.py

@@ -145,21 +145,22 @@ isShellBuiltin.builtIns = None
 # live in the root namespace and thus do not have to be
 # explicitly moved.
 
-def makeIntfPair( intf1, intf2, addr1=None, addr2=None ):
+def makeIntfPair( intf1, intf2, addr1=None, addr2=None, run=quietRun ):
     """Make a veth pair connecting intf1 and intf2.
        intf1: string, interface
        intf2: string, interface
-       returns: success boolean"""
+       node: node to run on or None (default)
+       returns: ip link add result"""
     # Delete any old interfaces with the same names
-    quietRun( 'ip link del ' + intf1 )
-    quietRun( 'ip link del ' + intf2 )
+    run( 'ip link del ' + intf1 )
+    run( 'ip link del ' + intf2 )
     # Create new pair
     if addr1 is None and addr2 is None:
         cmd = 'ip link add name ' + intf1 + ' type veth peer name ' + intf2
     else:
         cmd = ( 'ip link add name ' + intf1 + ' address ' + addr1 +
                 ' type veth peer name ' + intf2 + ' address ' + addr2 )
-    cmdOutput = quietRun( cmd )
+    cmdOutput = run( cmd )
     if cmdOutput == '':
         return True
     else:
@@ -180,36 +181,32 @@ def retry( retries, delaySecs, fn, *args, **keywords ):
         error( "*** gave up after %i retries\n" % tries )
         exit( 1 )
 
-def moveIntfNoRetry( intf, dstNode, srcNode=None, printError=False ):
+def moveIntfNoRetry( intf, dstNode, printError=False ):
     """Move interface to node, without retrying.
        intf: string, interface
         dstNode: destination Node
-        srcNode: source Node or None (default) for root ns
         printError: if true, print error"""
     intf = str( intf )
     cmd = 'ip link set %s netns %s' % ( intf, dstNode.pid )
-    if srcNode:
-        cmdOutput = srcNode.cmd( cmd )
-    else:
-        cmdOutput = quietRun( cmd )
+    cmdOutput = quietRun( cmd )
     # If ip link set does not produce any output, then we can assume
     # that the link has been moved successfully.
     if cmdOutput:
         if printError:
             error( '*** Error: moveIntf: ' + intf +
-                   ' not successfully moved to ' + dstNode.name + '\n' )
+                   ' not successfully moved to ' + dstNode.name + ':\n',
+                   cmdOutput )
         return False
     return True
 
-def moveIntf( intf, dstNode, srcNode=None, printError=False,
+def moveIntf( intf, dstNode, srcNode=None, printError=True,
              retries=3, delaySecs=0.001 ):
     """Move interface to node, retrying on failure.
        intf: string, interface
        dstNode: destination Node
-       srcNode: source Node or None (default) for root ns
        printError: if true, print error"""
     retry( retries, delaySecs, moveIntfNoRetry, intf, dstNode,
-          srcNode=srcNode, printError=printError )
+           printError=printError )
 
 # Support for dumping network
 

util/clustersetup.sh

@@ -0,0 +1,182 @@
+#!/usr/bin/env bash
+
+# Mininet ssh authentication script for cluster edition
+# This script will create a single key pair, which is then
+# propagated throughout the entire cluster.
+# There are two options for setup; temporary setup
+# persistent setup. If no options are specified, and the script
+# is only given ip addresses or host names, it will default to
+# the temporary setup. An ssh directory is then created in
+# /tmp/mn/ssh on each node, and mounted with the keys over the
+# user's ssh directory. This setup can easily be torn down by running
+# clustersetup with the -c option.
+# If the -p option is used, the setup will be persistent. In this
+# case, the key pair will be be distributed directly to each node's
+# ssh directory, but will be called cluster_key. An option to
+# specify this key for use will be added to the config file in each
+# user's ssh directory.
+
+
+set -e
+num_options=0
+persistent=false
+showHelp=false
+clean=false
+declare -a hosts=()
+user=$(whoami)
+SSHDIR=/tmp/mn/ssh
+USERDIR=/home/$user/.ssh
+usage=$'./clustersetup.sh [ -p|h|c ] [ host1 ] [ host2 ] ...\n
+        Authenticate yourself and other cluster nodes to each other
+        via ssh for mininet cluster edition. By default, we use a
+        temporary ssh setup. An ssh directory is mounted over
+        /home/user/.ssh on each machine in the cluster.
+        
+                -h: display this help
+                -p: create a persistent ssh setup. This will add
+                    new ssh keys and known_hosts to each nodes
+                    /home/user/.ssh directory
+                -c: method to clean up a temporary ssh setup.
+                    Any hosts taken as arguments will be cleaned
+        '
+
+persistentSetup() {
+    echo "***creating key pair"
+    ssh-keygen -t rsa -C "Cluster_Edition_Key" -f $USERDIR/cluster_key -N '' &> /dev/null
+    cat $USERDIR/cluster_key.pub >> $USERDIR/authorized_keys
+    echo "***configuring ssh"
+    echo "IdentityFile $USERDIR/cluster_key" >> $USERDIR/config
+    echo "IdentityFile $USERDIR/id_rsa" >> $USERDIR/config
+
+    for host in $hosts; do
+        echo "***copying public key to $host"
+        ssh-copy-id  -i $USERDIR/cluster_key.pub $user@$host &> /dev/null
+        echo "***copying key pair to remote host"
+        scp $USERDIR/cluster_key $user@$host:$USERDIR
+        scp $USERDIR/cluster_key.pub $user@$host:$USERDIR
+        echo "***configuring remote host"
+        ssh -o ForwardAgent=yes  $user@$host "
+        echo 'IdentityFile $USERDIR/cluster_key' >> $USERDIR/config
+        echo 'IdentityFile $USERDIR/id_rsa' >> $USERDIR/config"
+    done
+
+    for host in $hosts; do
+        echo "***copying known_hosts to $host"
+        scp $USERDIR/known_hosts $user@$host:$USERDIR/cluster_known_hosts
+        ssh $user@$host "
+        cat $USERDIR/cluster_known_hosts >> $USERDIR/known_hosts
+        rm $USERDIR/cluster_known_hosts"
+    done
+}
+
+tempSetup() {
+    
+    echo "***creating temporary ssh directory"
+    mkdir -p $SSHDIR 
+    echo "***creating key pair"
+    ssh-keygen -t rsa -C "Cluster_Edition_Key" -f /tmp/mn/ssh/id_rsa -N '' &> /dev/null
+
+    echo "***mounting temporary ssh directory"
+    sudo mount --bind $SSHDIR /home/$user/.ssh
+    cp $SSHDIR/id_rsa.pub $SSHDIR/authorized_keys
+
+for host in $hosts; do
+    echo "***copying public key to $host"
+    ssh-copy-id $user@$host &> /dev/null
+    echo "***mounting remote temporary ssh directory for $host"
+    ssh -o ForwardAgent=yes  $user@$host "
+    mkdir -p /tmp/mn/ssh
+    cp /home/$user/.ssh/authorized_keys $SSHDIR/authorized_keys
+    sudo mount --bind $SSHDIR /home/$user/.ssh"
+    echo "***copying key pair to $host"
+    scp $SSHDIR/{id_rsa,id_rsa.pub} $user@$host:$SSHDIR
+done
+
+for host in $hosts; do
+    echo "***copying known_hosts to $host"
+    scp $SSHDIR/known_hosts $user@$host:$SSHDIR
+done
+}
+
+cleanup() {
+    
+    for host in $hosts; do
+    echo "***cleaning up $host"
+    ssh $user@$host "sudo umount /home/$user/.ssh
+                          sudo rm -rf $SSHDIR"
+    done
+
+    echo "**unmounting local directories"
+    sudo umount /home/$user/.ssh
+    echo "***removing temporary ssh directory"
+    sudo rm -rf $SSHDIR
+    echo "done!"
+
+}
+
+
+if [ $# -eq 0 ]; then
+    echo "ERROR: No Arguments"
+    echo "$usage"
+    exit
+else
+    while getopts 'hpc' OPTION
+    do
+        ((num_options+=1))
+        case $OPTION in
+        h)  showHelp=true;;
+        p)  persistent=true;;
+        c)  clean=true;;
+        ?)  showHelp=true;;
+        esac
+    done
+    shift $(($OPTIND - 1))
+fi
+
+if [ "$num_options" -gt 1 ]; then
+    echo "ERROR: Too Many Options"
+    echo "$usage"
+    exit
+fi
+
+if $showHelp; then
+    echo "$usage"
+    exit
+fi
+
+for i in "$@"; do
+    output=$(getent ahostsv4 "$i")
+    if [ -z "$output" ]; then
+        echo '***WARNING: could not find hostname "$i"'
+        echo ""
+    else
+        hosts+="$i "
+    fi
+done
+
+if $clean; then
+    cleanup
+    exit
+fi
+
+echo "***authenticating to:"
+for host in $hosts; do
+    echo "$host"
+done
+
+echo
+
+if $persistent; then
+    echo '***Setting up persistent SSH configuration between all nodes'
+    persistentSetup
+    echo $'\n*** Sucessfully set up ssh throughout the cluster!'
+
+else
+    echo '*** Setting up temporary SSH configuration between all nodes'
+    tempSetup
+    echo $'\n***Finished temporary setup. When you are done with your cluster'
+    echo $'   session, tear down the SSH connections with'
+    echo $'   ./clustersetup.sh -c '$hosts''
+fi
+
+echo