Index: fedd/federation/experiment_control.py
===================================================================
--- fedd/federation/experiment_control.py	(revision c2dbca87a0ebce6e350942d5a68fef9006e3867f)
+++ fedd/federation/experiment_control.py	(revision 728001e2221b78f3f0a0a56f19bac83d82049f42)
@@ -42,4 +42,30 @@
 
     class ssh_cmd_timeout(RuntimeError): pass
+
+    class list_log:
+	"""
+	Provide an interface that lets logger.StreamHandler s write to a list
+	of strings.
+	"""
+	def __init__(self, l=[]):
+	    """
+	    Link to an existing list or just create a log
+	    """
+	    self.ll = l
+	    self.lock = Lock()
+	def write(self, str):
+	    """
+	    Add the string to the log.  Lock for consistency.
+	    """
+	    self.lock.acquire()
+	    self.ll.append(str)
+	    self.lock.release()
+
+	def flush(self):
+	    """
+	    No-op that StreamHandlers expect
+	    """
+	    pass
+
     
     class thread_pool:
@@ -204,7 +230,5 @@
         self.randomize_experiments = False
 
-        self.scp_exec = "/usr/bin/scp"
         self.splitter = None
-        self.ssh_exec="/usr/bin/ssh"
         self.ssh_keygen = "/usr/bin/ssh-keygen"
         self.ssh_identity_file = None
@@ -468,304 +492,330 @@
         f.close()
 
-    def scp_file(self, file, user, host, dest=""):
-        """
-        scp a file to the remote host.  If debug is set the action is only
-        logged.
-        """
-
-        scp_cmd = [self.scp_exec, '-o', 'IdentitiesOnly yes', 
-                '-o', 'StrictHostKeyChecking yes', '-i', 
-                self.ssh_privkey_file, file, "%s@%s:%s" % (user, host, dest)]
-        rv = 0
-
-        try:
-            dnull = open("/dev/null", "w")
-        except IOError:
-            self.log.debug("[ssh_file]: failed to open /dev/null for redirect")
-            dnull = Null
-
-        self.log.debug("[scp_file]: %s" % " ".join(scp_cmd))
-        if not self.debug:
-            rv = call(scp_cmd, stdout=dnull, stderr=dnull)
-
-        return rv == 0
-
-    def ssh_cmd(self, user, host, cmd, wname=None, timeout=None):
-        """
-        Run a remote command on host as user.  If debug is set, the action is
-        only logged.
-        """
-        sh_str = ("%s -o 'IdentitiesOnly yes' -o " + \
-                "'StrictHostKeyChecking yes' -i %s %s@%s %s") % \
-                (self.ssh_exec, self.ssh_privkey_file, 
-                        user, host, cmd)
-
-        try:
-            dnull = open("/dev/null", "r")
-        except IOError:
-            self.log.debug("[ssh_cmd]: failed to open /dev/null for redirect")
-            dnull = Null
-
-        self.log.debug("[ssh_cmd]: %s" % sh_str)
-        if not self.debug:
-            if dnull:
-                sub = Popen(sh_str, shell=True, stdout=dnull, stderr=dnull)
-            else:
-                sub = Popen(sh_str, shell=True)
-	    if timeout:
-		i = 0
-		rv = sub.poll()
-		while i < timeout:
-		    if rv is not None: break
+    class emulab_segment:
+	def __init__(self, log=None, keyfile=None, debug=False):
+	    self.log = log or logging.getLogger(\
+		    'fedd.experiment_control.emulab_segment')
+	    self.ssh_privkey_file = keyfile
+	    self.debug = debug
+	    self.ssh_exec="/usr/bin/ssh"
+	    self.scp_exec = "/usr/bin/scp"
+	    self.ssh_cmd_timeout = experiment_control_local.ssh_cmd_timeout
+
+	def scp_file(self, file, user, host, dest=""):
+	    """
+	    scp a file to the remote host.  If debug is set the action is only
+	    logged.
+	    """
+
+	    scp_cmd = [self.scp_exec, '-o', 'IdentitiesOnly yes', 
+		    '-o', 'StrictHostKeyChecking yes', '-i', 
+		    self.ssh_privkey_file, file, 
+		    "%s@%s:%s" % (user, host, dest)]
+	    rv = 0
+
+	    try:
+		dnull = open("/dev/null", "w")
+	    except IOError:
+		self.log.debug("[ssh_file]: failed to open " + \
+			"/dev/null for redirect")
+		dnull = Null
+
+	    self.log.debug("[scp_file]: %s" % " ".join(scp_cmd))
+	    if not self.debug:
+		rv = call(scp_cmd, stdout=dnull, stderr=dnull)
+
+	    return rv == 0
+
+	def ssh_cmd(self, user, host, cmd, wname=None, timeout=None):
+	    """
+	    Run a remote command on host as user.  If debug is set, the action is
+	    only logged.
+	    """
+	    sh_str = ("%s -o 'IdentitiesOnly yes' -o " + \
+		    "'StrictHostKeyChecking yes' -i %s %s@%s %s") % \
+		    (self.ssh_exec, self.ssh_privkey_file, 
+			    user, host, cmd)
+
+	    try:
+		dnull = open("/dev/null", "r")
+	    except IOError:
+		self.log.debug("[ssh_cmd]: failed to open /dev/null " + \
+			"for redirect")
+		dnull = Null
+
+	    self.log.debug("[ssh_cmd]: %s" % sh_str)
+	    if not self.debug:
+		if dnull:
+		    sub = Popen(sh_str, shell=True, stdout=dnull, stderr=dnull)
+		else:
+		    sub = Popen(sh_str, shell=True)
+		if timeout:
+		    i = 0
+		    rv = sub.poll()
+		    while i < timeout:
+			if rv is not None: break
+			else:
+			    time.sleep(1)
+			    rv = sub.poll()
+			    i += 1
 		    else:
-			time.sleep(1)
-			rv = sub.poll()
-			i += 1
+			self.log.debug("Process exceeded runtime: %s" % sh_str)
+			os.kill(sub.pid, signal.SIGKILL)
+			raise self.ssh_cmd_timeout();
+		    return rv == 0
 		else:
-		    self.log.debug("Process exceeded runtime: %s" % sh_str)
-		    os.kill(sub.pid, signal.SIGKILL)
-		    raise self.ssh_cmd_timeout();
-		return rv == 0
+		    return sub.wait() == 0
 	    else:
-		return sub.wait() == 0
-        else:
-            return True
-
-
-    def create_config_tree(self, src_dir, dest_dir, script):
-        """
-	Append commands to script that will create the directory hierarchy on
-	the remote federant.
-        """
-
-	if os.path.isdir(src_dir):
-	    print >>script, "mkdir -p %s" % dest_dir
-	    print >>script, "chmod 770 %s" % dest_dir
-
+		return True
+
+    class start_segment(emulab_segment):
+	def __init__(self, log=None, keyfile=None, debug=False):
+	    experiment_control_local.emulab_segment.__init__(self,
+		    log=log, keyfile=keyfile, debug=debug)
+
+	def create_config_tree(self, src_dir, dest_dir, script):
+	    """
+	    Append commands to script that will create the directory hierarchy
+	    on the remote federant.
+	    """
+
+	    if os.path.isdir(src_dir):
+		print >>script, "mkdir -p %s" % dest_dir
+		print >>script, "chmod 770 %s" % dest_dir
+
+		for f in os.listdir(src_dir):
+		    if os.path.isdir(f):
+			self.create_config_tree("%s/%s" % (src_dir, f), 
+				"%s/%s" % (dest_dir, f), script)
+	    else:
+		self.log.debug("[create_config_tree]: Not a directory: %s" \
+			% src_dir)
+
+	def ship_configs(self, host, user, src_dir, dest_dir):
+	    """
+	    Copy federant-specific configuration files to the federant.
+	    """
 	    for f in os.listdir(src_dir):
 		if os.path.isdir(f):
-		    self.create_config_tree("%s/%s" % (src_dir, f), 
-			    "%s/%s" % (dest_dir, f), script)
-	else:
-	    self.log.debug("[create_config_tree]: Not a directory: %s" \
-		    % src_dir)
-
-    def ship_configs(self, host, user, src_dir, dest_dir):
-        """
-        Copy federant-specific configuration files to the federant.
-        """
-        for f in os.listdir(src_dir):
-            if os.path.isdir(f):
-                if not self.ship_configs(host, user, "%s/%s" % (src_dir, f), 
-                        "%s/%s" % (dest_dir, f)):
-                    return False
-            else:
-                if not self.scp_file("%s/%s" % (src_dir, f), 
-                        user, host, dest_dir):
-                    return False
-        return True
-
-    def get_state(self, user, host, ssh_key, tb, pid, eid):
-        # command to test experiment state
-        expinfo_exec = "/usr/testbed/bin/expinfo"  
-        # Regular expressions to parse the expinfo response
-        state_re = re.compile("State:\s+(\w+)")
-        no_exp_re = re.compile("^No\s+such\s+experiment")
-        swapping_re = re.compile("^No\s+information\s+available.")
-        state = None    # Experiment state parsed from expinfo
-        # The expinfo ssh command.  Note the identity restriction to use only
-        # the identity provided in the pubkey given.
-        cmd = [self.ssh_exec, '-o', 'IdentitiesOnly yes', '-o', 
-                'StrictHostKeyChecking yes', '-i', 
-                ssh_key, "%s@%s" % (user, host), 
-                expinfo_exec, pid, eid]
-
-        dev_null = None
-        try:
-            dev_null = open("/dev/null", "a")
-        except IOError, e:
-            self.log.error("[get_state]: can't open /dev/null: %s" %e)
-
-        if self.debug:
-            state = 'swapped'
-            rv = 0
-        else:
-            status = Popen(cmd, stdout=PIPE, stderr=dev_null)
-            for line in status.stdout:
-                m = state_re.match(line)
-                if m: state = m.group(1)
-                else:
-		    for reg, st in ((no_exp_re, "none"),
-			    (swapping_re, "swapping")):
-			m = reg.match(line)
-			if m: state = st
-            rv = status.wait()
-
-        # If the experiment is not present the subcommand returns a non-zero
-        # return value.  If we successfully parsed a "none" outcome, ignore the
-        # return code.
-        if rv != 0 and state != 'none':
-            raise service_error(service_error.internal,
-                    "Cannot get status of segment %s:%s/%s" % (tb, pid, eid))
-        elif state not in ('active', 'swapped', 'swapping', 'none'):
-            raise service_error(service_error.internal,
-                    "Cannot get status of segment %s:%s/%s" % (tb, pid, eid))
-	else: return state
-
-
-    def start_segment(self, tb, eid, tbparams, tmpdir, timeout=0):
-        """
-        Start a sub-experiment on a federant.
-
-        Get the current state, modify or create as appropriate, ship data and
-        configs and start the experiment.  There are small ordering differences
-        based on the initial state of the sub-experiment.
-        """
-        # ops node in the federant
-        host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
-        user = tbparams[tb]['user']     # federant user
-        pid = tbparams[tb]['project']   # federant project
-        # XXX
-        base_confs = ( "hosts",)
-        tclfile = "%s.%s.tcl" % (eid, tb)   # sub-experiment description
-        # Configuration directories on the remote machine
-        proj_dir = "/proj/%s/exp/%s/tmp" % (pid, eid)
-        tarfiles_dir = "/proj/%s/tarfiles/%s" % (pid, eid)
-        rpms_dir = "/proj/%s/rpms/%s" % (pid, eid)
-
-	state = self.get_state(user, host, self.ssh_privkey_file, tb, pid, eid)
-
-        self.log.debug("[start_segment]: %s: %s" % (tb, state))
-        self.log.info("[start_segment]:transferring experiment to %s" % tb)
-
-        if not self.scp_file("%s/%s/%s" % (tmpdir, tb, tclfile), user, host):
-            return False
-        
-        if state == 'none':
-            # Create a null copy of the experiment so that we capture any logs
-            # there if the modify fails.  Emulab software discards the logs
-            # from a failed startexp
-            if not self.scp_file("%s/null.tcl" % tmpdir, user, host):
-                return False
-            self.log.info("[start_segment]: Creating %s on %s" % (eid, tb))
-	    timedout = False
+		    if not self.ship_configs(host, user, "%s/%s" % (src_dir, f), 
+			    "%s/%s" % (dest_dir, f)):
+			return False
+		else:
+		    if not self.scp_file("%s/%s" % (src_dir, f), 
+			    user, host, dest_dir):
+			return False
+	    return True
+
+	def get_state(self, user, host, tb, pid, eid):
+	    # command to test experiment state
+	    expinfo_exec = "/usr/testbed/bin/expinfo"  
+	    # Regular expressions to parse the expinfo response
+	    state_re = re.compile("State:\s+(\w+)")
+	    no_exp_re = re.compile("^No\s+such\s+experiment")
+	    swapping_re = re.compile("^No\s+information\s+available.")
+	    state = None    # Experiment state parsed from expinfo
+	    # The expinfo ssh command.  Note the identity restriction to use
+	    # only the identity provided in the pubkey given.
+	    cmd = [self.ssh_exec, '-o', 'IdentitiesOnly yes', '-o', 
+		    'StrictHostKeyChecking yes', '-i', 
+		    self.ssh_privkey_file, "%s@%s" % (user, host), 
+		    expinfo_exec, pid, eid]
+
+	    dev_null = None
+	    try:
+		dev_null = open("/dev/null", "a")
+	    except IOError, e:
+		self.log.error("[get_state]: can't open /dev/null: %s" %e)
+
+	    if self.debug:
+		state = 'swapped'
+		rv = 0
+	    else:
+		status = Popen(cmd, stdout=PIPE, stderr=dev_null)
+		for line in status.stdout:
+		    m = state_re.match(line)
+		    if m: state = m.group(1)
+		    else:
+			for reg, st in ((no_exp_re, "none"),
+				(swapping_re, "swapping")):
+			    m = reg.match(line)
+			    if m: state = st
+		rv = status.wait()
+
+	    # If the experiment is not present the subcommand returns a
+	    # non-zero return value.  If we successfully parsed a "none"
+	    # outcome, ignore the return code.
+	    if rv != 0 and state != 'none':
+		raise service_error(service_error.internal,
+			"Cannot get status of segment %s:%s/%s" % \
+				(tb, pid, eid))
+	    elif state not in ('active', 'swapped', 'swapping', 'none'):
+		raise service_error(service_error.internal,
+			"Cannot get status of segment %s:%s/%s" % \
+				(tb, pid, eid))
+	    else: return state
+
+
+	def __call__(self, tb, eid, tbparams, tmpdir, timeout=0):
+	    """
+	    Start a sub-experiment on a federant.
+
+	    Get the current state, modify or create as appropriate, ship data
+	    and configs and start the experiment.  There are small ordering
+	    differences based on the initial state of the sub-experiment.
+	    """
+	    # ops node in the federant
+	    host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
+	    user = tbparams[tb]['user']     # federant user
+	    pid = tbparams[tb]['project']   # federant project
+	    # XXX
+	    base_confs = ( "hosts",)
+	    tclfile = "%s.%s.tcl" % (eid, tb)   # sub-experiment description
+	    # Configuration directories on the remote machine
+	    proj_dir = "/proj/%s/exp/%s/tmp" % (pid, eid)
+	    tarfiles_dir = "/proj/%s/tarfiles/%s" % (pid, eid)
+	    rpms_dir = "/proj/%s/rpms/%s" % (pid, eid)
+
+	    state = self.get_state(user, host, tb, pid, eid)
+
+	    self.log.debug("[start_segment]: %s: %s" % (tb, state))
+	    self.log.info("[start_segment]:transferring experiment to %s" % tb)
+
+	    if not self.scp_file("%s/%s/%s" % \
+		    (tmpdir, tb, tclfile), user, host):
+		return False
+	    
+	    if state == 'none':
+		# Create a null copy of the experiment so that we capture any
+		# logs there if the modify fails.  Emulab software discards the
+		# logs from a failed startexp
+		if not self.scp_file("%s/null.tcl" % tmpdir, user, host):
+		    return False
+		self.log.info("[start_segment]: Creating %s on %s" % (eid, tb))
+		timedout = False
+		try:
+		    if not self.ssh_cmd(user, host,
+			    ("/usr/testbed/bin/startexp -i -f -w -p %s " + 
+			    "-e %s null.tcl") % (pid, eid), "startexp",
+			    timeout=60 * 10):
+			return False
+		except self.ssh_cmd_timeout:
+		    timedout = True
+
+		if timedout:
+		    state = self.get_state(user, host, self.ssh_privkey_file, 
+			    tb, eid, pid)
+		    if state != "swapped":
+			return False
+
+	    
+	    # Open up a temporary file to contain a script for setting up the
+	    # filespace for the new experiment.
+	    self.log.info("[start_segment]: creating script file")
+	    try:
+		sf, scriptname = tempfile.mkstemp()
+		scriptfile = os.fdopen(sf, 'w')
+	    except IOError:
+		return False
+
+	    scriptbase = os.path.basename(scriptname)
+
+	    # Script the filesystem changes
+	    print >>scriptfile, "/bin/rm -rf %s" % proj_dir
+	    # Clear and create the tarfiles and rpm directories
+	    for d in (tarfiles_dir, rpms_dir):
+		print >>scriptfile, "/bin/rm -rf %s/*" % d
+		print >>scriptfile, "mkdir -p %s" % d
+	    print >>scriptfile, 'mkdir -p %s' % proj_dir
+	    self.create_config_tree("%s/%s" % (tmpdir, tb),
+		    proj_dir, scriptfile)
+	    if os.path.isdir("%s/tarfiles" % tmpdir):
+		self.create_config_tree("%s/tarfiles" % tmpdir, tarfiles_dir,
+			scriptfile)
+	    if os.path.isdir("%s/rpms" % tmpdir):
+		self.create_config_tree("%s/rpms" % tmpdir, rpms_dir, 
+			scriptfile)
+	    print >>scriptfile, "rm -f %s" % scriptbase
+	    scriptfile.close()
+
+	    # Move the script to the remote machine
+	    # XXX: could collide tempfile names on the remote host
+	    if self.scp_file(scriptname, user, host, scriptbase):
+		os.remove(scriptname)
+	    else:
+		return False
+
+	    # Execute the script (and the script's last line deletes it)
+	    if not self.ssh_cmd(user, host, "sh -x %s" % scriptbase):
+		return False
+
+	    for f in base_confs:
+		if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
+			"%s/%s" % (proj_dir, f)):
+		    return False
+	    if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
+		    proj_dir):
+		return False
+	    if os.path.isdir("%s/tarfiles" % tmpdir):
+		if not self.ship_configs(host, user,
+			"%s/tarfiles" % tmpdir, tarfiles_dir):
+		    return False
+	    if os.path.isdir("%s/rpms" % tmpdir):
+		if not self.ship_configs(host, user,
+			"%s/rpms" % tmpdir, tarfiles_dir):
+		    return False
+	    # Stage the new configuration (active experiments will stay swapped
+	    # in now)
+	    self.log.info("[start_segment]: Modifying %s on %s" % (eid, tb))
 	    try:
 		if not self.ssh_cmd(user, host,
-			("/usr/testbed/bin/startexp -i -f -w -p %s " + 
-			"-e %s null.tcl") % (pid, eid), "startexp",
-			timeout=60 * 10):
+			"/usr/testbed/bin/modexp -r -s -w %s %s %s" % \
+				(pid, eid, tclfile),
+			"modexp", timeout= 60 * 10):
 		    return False
 	    except self.ssh_cmd_timeout:
-		timedout = True
-
-	    if timedout:
-		state = self.get_state(user, host, self.ssh_privkey_file, 
-			tb, eid, pid)
-		if state != "swapped":
-		    return False
-
-        
-        # Open up a temporary file to contain a script for setting up the
-        # filespace for the new experiment.
-        self.log.info("[start_segment]: creating script file")
-        try:
-            sf, scriptname = tempfile.mkstemp()
-            scriptfile = os.fdopen(sf, 'w')
-        except IOError:
-            return False
-
-	scriptbase = os.path.basename(scriptname)
-
-        # Script the filesystem changes
-        print >>scriptfile, "/bin/rm -rf %s" % proj_dir
-        # Clear and create the tarfiles and rpm directories
-        for d in (tarfiles_dir, rpms_dir):
-            print >>scriptfile, "/bin/rm -rf %s/*" % d
-            print >>scriptfile, "mkdir -p %s" % d
-        print >>scriptfile, 'mkdir -p %s' % proj_dir
-        self.create_config_tree("%s/%s" % (tmpdir, tb), proj_dir, scriptfile)
-        if os.path.isdir("%s/tarfiles" % tmpdir):
-	    self.create_config_tree("%s/tarfiles" % tmpdir, tarfiles_dir,
-		    scriptfile)
-        if os.path.isdir("%s/rpms" % tmpdir):
-            self.create_config_tree("%s/rpms" % tmpdir, rpms_dir, 
-		    scriptfile)
-        print >>scriptfile, "rm -f %s" % scriptbase
-        scriptfile.close()
-
-        # Move the script to the remote machine
-        # XXX: could collide tempfile names on the remote host
-        if self.scp_file(scriptname, user, host, scriptbase):
-            os.remove(scriptname)
-        else:
-            return False
-
-        # Execute the script (and the script's last line deletes it)
-        if not self.ssh_cmd(user, host, "sh -x %s" % scriptbase):
-            return False
-
-        for f in base_confs:
-            if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
-                    "%s/%s" % (proj_dir, f)):
-                return False
-        if not self.ship_configs(host, user, "%s/%s" % (tmpdir, tb),
-                proj_dir):
-            return False
-        if os.path.isdir("%s/tarfiles" % tmpdir):
-            if not self.ship_configs(host, user,
-                    "%s/tarfiles" % tmpdir, tarfiles_dir):
-                return False
-        if os.path.isdir("%s/rpms" % tmpdir):
-            if not self.ship_configs(host, user,
-                    "%s/rpms" % tmpdir, tarfiles_dir):
-                return False
-        # Stage the new configuration (active experiments will stay swapped in
-        # now)
-        self.log.info("[start_segment]: Modifying %s on %s" % (eid, tb))
-	try:
-	    if not self.ssh_cmd(user, host,
-		    "/usr/testbed/bin/modexp -r -s -w %s %s %s" % \
-			    (pid, eid, tclfile),
-		    "modexp", timeout= 60 * 10):
+		print "modexp timeout"
+		# There's really no way to see if this succeeded or failed, so
+		# if it hangs, assume the worst.
 		return False
-	except self.ssh_cmd_timeout:
-	    print "modexp timeout"
-	    # There's really no way to see if this succeeded or failed, so if
-	    # it hangs, assume the worst.
-	    return False
-        # Active experiments are still swapped, this swaps the others in.
-        if state != 'active':
-            self.log.info("[start_segment]: Swapping %s in on %s" % \
-                    (eid, tb))
-	    timedout = False
-	    try:
-		if not self.ssh_cmd(user, host,
-			"/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
-			"swapexp", timeout=10*60):
-		    return False
-	    except self.ssh_cmd_timeout:
-		timedout = True
-	   
-	    # If the command was terminated, but completed successfully, report
-	    # success.
-	    if timedout:
-		state = self.get_state(user, host, self.ssh_privkey_file,
-			tb, eid, pid)
-		self.log.debug("[start_segment]: swapin timed out (state)")
-		return state == 'active'
-	# Everything has gone OK.
-        return True
-
-    def stop_segment(self, tb, eid, tbparams):
-        """
-        Stop a sub experiment by calling swapexp on the federant
-        """
-        user = tbparams[tb]['user']
-        host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
-        pid = tbparams[tb]['project']
-
-        self.log.info("[stop_segment]: Stopping %s on %s" % (eid, tb))
-        return self.ssh_cmd(user, host,
-                "/usr/testbed/bin/swapexp -w %s %s out" % (pid, eid))
+	    # Active experiments are still swapped, this swaps the others in.
+	    if state != 'active':
+		self.log.info("[start_segment]: Swapping %s in on %s" % \
+			(eid, tb))
+		timedout = False
+		try:
+		    if not self.ssh_cmd(user, host,
+			    "/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
+			    "swapexp", timeout=10*60):
+			return False
+		except self.ssh_cmd_timeout:
+		    timedout = True
+	       
+		# If the command was terminated, but completed successfully,
+		# report success.
+		if timedout:
+		    state = self.get_state(user, host, self.ssh_privkey_file,
+			    tb, eid, pid)
+		    self.log.debug("[start_segment]: swapin timed out (state)")
+		    return state == 'active'
+	    # Everything has gone OK.
+	    return True
+
+    class stop_segment(emulab_segment):
+	def __init__(self, log=None, keyfile=None, debug=False):
+	    experiment_control_local.emulab_segment.__init__(self,
+		    log=log, keyfile=keyfile, debug=debug)
+
+	def __call__(self, tb, eid, tbparams):
+	    """
+	    Stop a sub experiment by calling swapexp on the federant
+	    """
+	    user = tbparams[tb]['user']
+	    host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
+	    pid = tbparams[tb]['project']
+
+	    self.log.info("[stop_segment]: Stopping %s on %s" % (eid, tb))
+	    return self.ssh_cmd(user, host,
+		    "/usr/testbed/bin/swapexp -w %s %s out" % (pid, eid))
 
         
@@ -1725,5 +1775,5 @@
             allocated = { }         # Testbeds we can access
             started = { }           # Testbeds where a sub-experiment started 
-                                # successfully
+				    # successfully
 
             # Objects to parse the splitter output (defined above)
@@ -1837,5 +1887,7 @@
             # get the return value later
             thread_pool.wait_for_slot()
-            t  = self.pooled_thread(target=self.start_segment, 
+            t  = self.pooled_thread(\
+		    target=self.start_segment(log=self.log,
+			keyfile=self.ssh_privkey_file, debug=self.debug), 
                     args=(tb, eid, tbparams, tmpdir, 0), name=tb,
                     pdata=thread_pool, trace_file=self.trace_file)
@@ -1850,5 +1902,7 @@
 
         if len(failed) == 0:
-            if not self.start_segment(master, eid, tbparams, tmpdir):
+	    starter = self.start_segment(log=self.log, 
+		    keyfile=self.ssh_privkey_file, debug=self.debug)
+            if not starter(master, eid, tbparams, tmpdir):
                 failed.append(master)
 
@@ -1861,5 +1915,8 @@
                     # Create and start a thread to stop the segment
                     thread_pool.wait_for_slot()
-                    t  = self.pooled_thread(target=self.stop_segment, 
+                    t  = self.pooled_thread(\
+			    target=self.stop_segment(log=self.log,
+				keyfile=self.ssh_privkey_file,
+				debug=self.debug), 
                             args=(tb, eid, tbparams), name=tb,
                             pdata=thread_pool, trace_file=self.trace_file)
