Context Navigation

-                      rc2dbca8
+                      r4b362df
     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:
 …
         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
 …
         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))
 …
             allocated = { }         # Testbeds we can access
             started = { }           # Testbeds where a sub-experiment started
                                 # successfully
+                                    # successfully
             # Objects to parse the splitter output (defined above)
 …
             # 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)
 …
         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)
 …
                     # 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)

Note: See TracChangeset for help on using the changeset viewer.

Context Navigation

Changeset 4b362df

Legend:

fedd/federation/experiment_control.py

Download in other formats: