From patchwork Wed May 20 20:52:10 2009 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Ryan Harper X-Patchwork-Id: 25086 Received: from vger.kernel.org (vger.kernel.org [209.132.176.167]) by demeter.kernel.org (8.14.2/8.14.2) with ESMTP id n4KKqG9Z005873 for ; Wed, 20 May 2009 20:52:17 GMT Received: (majordomo@vger.kernel.org) by vger.kernel.org via listexpand id S1754614AbZETUwM (ORCPT ); Wed, 20 May 2009 16:52:12 -0400 Received: (majordomo@vger.kernel.org) by vger.kernel.org id S1754214AbZETUwM (ORCPT ); Wed, 20 May 2009 16:52:12 -0400 Received: from e34.co.us.ibm.com ([32.97.110.152]:51331 "EHLO e34.co.us.ibm.com" rhost-flags-OK-OK-OK-OK) by vger.kernel.org with ESMTP id S1753229AbZETUwL (ORCPT ); Wed, 20 May 2009 16:52:11 -0400 Received: from d03relay02.boulder.ibm.com (d03relay02.boulder.ibm.com [9.17.195.227]) by e34.co.us.ibm.com (8.13.1/8.13.1) with ESMTP id n4KKnROL000440 for ; Wed, 20 May 2009 14:49:27 -0600 Received: from d03av01.boulder.ibm.com (d03av01.boulder.ibm.com [9.17.195.167]) by d03relay02.boulder.ibm.com (8.13.8/8.13.8/NCO v9.2) with ESMTP id n4KKqCAG249840 for ; Wed, 20 May 2009 14:52:12 -0600 Received: from d03av01.boulder.ibm.com (loopback [127.0.0.1]) by d03av01.boulder.ibm.com (8.12.11.20060308/8.13.3) with ESMTP id n4KKqBlX020277 for ; Wed, 20 May 2009 14:52:12 -0600 Received: from localhost.localdomain (frylock.austin.ibm.com [9.53.41.12]) by d03av01.boulder.ibm.com (8.12.11.20060308/8.12.11) with ESMTP id n4KKqBiE020267; Wed, 20 May 2009 14:52:11 -0600 Received: by localhost.localdomain (Postfix, from userid 1000) id E56901BF7C; Wed, 20 May 2009 15:52:10 -0500 (CDT) Date: Wed, 20 May 2009 15:52:10 -0500 From: Ryan Harper To: Michael Goldish Cc: KVM List Subject: Re: [RFC] KVM-Autotest: basic parallel test execution Message-ID: <20090520205210.GU6905@us.ibm.com> References: <360943614.47181242571294363.JavaMail.root@zmail05.collab.prod.int.phx2.redhat.com> <1294159996.47211242571742642.JavaMail.root@zmail05.collab.prod.int.phx2.redhat.com> Mime-Version: 1.0 Content-Disposition: inline In-Reply-To: <1294159996.47211242571742642.JavaMail.root@zmail05.collab.prod.int.phx2.redhat.com> User-Agent: Mutt/1.5.6+20040907i Sender: kvm-owner@vger.kernel.org Precedence: bulk List-ID: X-Mailing-List: kvm@vger.kernel.org * Michael Goldish [2009-05-17 09:50]: > Hi all, > > We've recently implemented a very simple form of parallel test > execution into KVM-Autotest and we'd like some feedback on it. This > suggestion allows the user to manually assign tests to hosts/queues. > It also takes care of assigning different MAC address ranges to > hosts/queues. By 'queues' I mean parallel execution pipelines. Each > host has one or more queues. The number of queues is defined by the > user, and should reflect the capabilities of the host. > > This implementation involves only minor modifications to the code > itself; most of the work is done in a new config file kvm_hosts.cfg, > which has the exact same format as kvm_tests.cfg. The new file > provides the framework with information about hosts/queues. The new > file is parsed after kvm_tests.cfg. The test sets (such as > 'nightly' and 'weekly'), previously defined at the end of > kvm_tests.cfg, should now be defined last, after kvm_hosts.cfg. > Test sets no longer select only the tests to execute, but also > where each test should be executed (i.e. on what host/queue). > > The final result of parsing the config files is a list of tests, each > with its own 'hostname' and 'queue' parameters. Each host executes > only the tests whose 'hostname' parameter matches the current host, > and puts tests with different 'queue' values in parallel pipelines > of execution. > > Ideally, the Autotest server should take care of assigning tests to > hosts automatically, but there are still a few technical difficulties > to be resolved before we can implement that. We're considering the > current suggestion as a temporary solution until a better one is > found. > > Basically, the advantages are: > - allows the user full control over what tests run, and where/how they run > - takes care of assigning MAC address ranges to different hosts/queues (required for TAP networking) > - can be used from the server or with the client, which makes it relevant also for users who don't have an Autotest server installed > - involves only minor code changes (except for the config files) > - is pretty much the simplest possible solution (and simple is good) > > Drawbacks: > - requires some initial work to be done by the user -- the user has to define exactly where each test should run > - test sets need to be modified when tests or hosts are added/removed, to include/exclude them I took a slightly different approach. The kvm_tests.cfg file already provides a dependency relationship between different tests. I modified the main loop in the control file to walk the entire list of jobs and pull out any jobs that don't have any dependencies (ie, install tests). And then run N jobs in parallel from that list until it is exhausted; then store the results. Then loop the over the remaining list of jobs again finding the jobs that can be run. On a larger multi core system, one might set the number of parallel jobs equal to the number of cores. I think this works well with using autoserv to farm out different kvm_tests.cfg to different machines. Attaching my stale patch just for comment. Needs to be updated since I sat on this for a while. There were a number of issues: - kvm_log is a shared resource, fixed it up so parallel jobs can both call it - vnc, redir and other network resources are shared, so, in kvm_tests.cfg file each job needs a parallel offset. - in kvm_tests.cfg file need to define additional vm and nic objects, one for each parallel threads. Advantages: - works a lot like the single threaded model does, and if threads=1 then it runs the same path - config files don't change significantly, just some additional VM objects at the top and some offset values - transparent to an autoserv setup, autoserv would just need to specify the kvm_tests.cfg file to each host. Disadvantages: - the main loop waits for each group of parallel jobs to complete before starting any more. If somehow an install is mixed with a reboot test, we'll wait around before starting more jobs - probably a few more here, but I don't have them on the top of my head. diff --git a/client/bin/job.py b/client/bin/job.py index 6ed5b36..d772488 100755 --- a/client/bin/job.py +++ b/client/bin/job.py @@ -751,21 +751,27 @@ class base_job(object): """Run tasks in parallel""" pids = [] + pid_to_task = {} + return_vals = {} old_log_filename = self.log_filename for i, task in enumerate(tasklist): assert isinstance(task, (tuple, list)) self.log_filename = old_log_filename + (".%d" % i) task_func = lambda: task[0](*task[1:]) - pids.append(parallel.fork_start(self.resultdir, task_func)) + pid = parallel.fork_start(self.resultdir, task_func) + pids.append(pid) + pid_to_task[pid] = i old_log_path = os.path.join(self.resultdir, old_log_filename) old_log = open(old_log_path, "a") exceptions = [] for i, pid in enumerate(pids): # wait for the task to finish + task = pid_to_task[pid] try: - parallel.fork_waitfor(self.resultdir, pid) + return_vals[task] = parallel.fork_waitfor(self.resultdir, pid) except Exception, e: + return_vals[task] = 1 exceptions.append(e) # copy the logs from the subtask into the main log new_log_path = old_log_path + (".%d" % i) @@ -784,6 +790,9 @@ class base_job(object): msg = "%d task(s) failed in job.parallel" % len(exceptions) raise error.JobError(msg) + # send back array True|False on success or failure + return map(lambda x: x == 0, return_vals.values()) + def quit(self): # XXX: should have a better name. diff --git a/client/bin/parallel.py b/client/bin/parallel.py index e8749db..e381c89 100644 --- a/client/bin/parallel.py +++ b/client/bin/parallel.py @@ -45,3 +45,4 @@ def fork_waitfor(tmp, pid): if (status != 0): raise error.TestError("test failed rc=%d" % (status)) + return status diff --git a/client/tests/kvm_runtest_2/control b/client/tests/kvm_runtest_2/control index cb27821..1251e94 100644 --- a/client/tests/kvm_runtest_2/control +++ b/client/tests/kvm_runtest_2/control @@ -1,4 +1,4 @@ -import sys, os +import sys, os, copy # enable modules import from current directory (tests/kvm_runtest_2) pwd = os.path.join(os.environ['AUTODIR'],'tests/kvm_runtest_2') @@ -82,24 +82,80 @@ list = kvm_config.config(filename).get_list() # ------------- # Run the tests # ------------- +threads = 1 status_dict = {} - -for dict in list: - if dict.get("skip") == "yes": - continue - dependencies_satisfied = True - for dep in dict.get("depend"): - for test_name in status_dict.keys(): - if not dep in test_name: - continue - if not status_dict[test_name]: +current_list = copy.deepcopy(list) + +while len(current_list) > 0: + parallel_jobs = [] + for dict in current_list: + name = dict['name'] + if dict.get("skip") == "yes": + continue + dependencies_satisfied = True + # it's ok to ignore deps that we won't run, but not ok to ignore + # deps that have failed + for dep in dict.get("depend"): + # if we're waiting for the job to be run, deps aren't met + if dep in map(lambda x: x['name'], parallel_jobs): dependencies_satisfied = False break - if dependencies_satisfied: - current_status = job.run_test("kvm_runtest_2", params=dict, tag=dict.get("shortname")) - else: - current_status = False - status_dict[dict.get("name")] = current_status + for test_name in status_dict.keys(): + if not dep in test_name: + continue + if not status_dict[test_name]: + dependencies_satisfied = False + break + + # add the job to parallel list if we can run it *now* + if dependencies_satisfied == True: + parallel_jobs.append(dict) + current_list = filter(lambda x: x['name'] != name, current_list) + + + print '**************WARK************* %d jobs to run in parallel' %(len(parallel_jobs)) + # now, we can run the current parallel jobs in sets of N tests where + # N is the number of parallel jobs allowed. + for g in range(0,len(parallel_jobs), threads): + group = parallel_jobs[g:g+threads] + names = map(lambda x: x['name'], group) + + # HACK - my python fu is not great enough to get around exec/eval to + # dynamically define functions which are needed for job.parallel + j = [] + k = [] + for i,dict in enumerate(group): + # reset vm and main_vm list into each dict + dict['vms'] = "vm%s" % i + dict['main_vm'] = "vm%s" % i + # FIXME: should be # of threads * index + dict['parallel_offset'] = 2*i + print dict + # create unique dict name and make a copy of dict + pdict_name = "dict_vm%s"%(i) + print pdict_name + exec "%s = copy.deepcopy(dict)" %(pdict_name) + tag = dict.get("shortname") + sdtag = 'parallel_%s' % i + defstr = 'def j%s():\n job.run_test("kvm_runtest_2", params=%s, tag="%s", subdir_tag="%s")' % (i,pdict_name, tag,sdtag) + # define dummy function for this job + exec defstr + eval("j.append(j%s)" % i) + k.append(dict.get("shortname")) + + print "***************** WARK ************* parallel: %s"%(j) + print "***************** WARK ************* parallel: %s"%(map(lambda x: [x], k)) + + # need to wrap each function in the array with [], then use * to unpack the list + # as variable arg list to job.parallel + current_status = job.parallel(*map(lambda x: [x], j)) + #current_status = job.parallel([j[0]], [j[1]]) + print "status: %s"%(current_status) + + # update status of each job from parallel run + for i,dict in enumerate(group): + status_dict[dict.get("name")] = current_status[i] + # create the html report in result dir reporter = os.path.join(pwd, 'make_html_report.py') diff --git a/client/tests/kvm_runtest_2/kvm_log.py b/client/tests/kvm_runtest_2/kvm_log.py index ff1dfe9..06dc47d 100644 --- a/client/tests/kvm_runtest_2/kvm_log.py +++ b/client/tests/kvm_runtest_2/kvm_log.py @@ -5,10 +5,9 @@ version = "20081205" import time import sys +import os -prefix_stack = [""] -log_level = 4 - +pids = {} def name_of_caller(): return sys._getframe(2).f_code.co_name @@ -28,24 +27,50 @@ def format_stack(min_depth, max_depth, stop_func): list.reverse() return "|".join(list) -def set_prefix(prefix): - global prefix_stack - prefix_stack.append(" %s" % prefix) +def get_prefix(pid): + global pids + if (pids.has_key(pid) and not pids[pid].has_key('prefix')) or not pids.has_key(pid): + pids[pid] = {} + default_prefix = " [%s]" % pid + pids[pid]['prefix'] = [default_prefix] + return pids[pid]['prefix'] + +def set_prefix(pid, prefix): + global pids + p = get_prefix(pid) + p.append(" %s" % prefix) + pids[pid]['prefix'] = p + +def restore_prefix(pid): + global pids + if pid in pids.keys(): + pids[pid]['prefix'].pop() -def restore_prefix(): - global prefix_stack - prefix_stack.pop() +def get_level(pid): + global pids + if (pids.has_key(pid) and not pids[pid].has_key('loglevel')) or not pids.has_key(pid): + pids[pid] = {} + pids[pid]['loglevel'] = 4 -def set_level(level): - global log_level - log_level = level + return pids[pid]['loglevel'] + +def set_level(pid, level): + global pids + l = get_level(pid) + l = level + pids[pid]['loglevel'] = l def log(level, message, prefix=True): + global pids + mypid = os.getpid() + prefix = get_prefix(mypid) + loglevel = get_level(mypid) + if message.endswith("\n"): message = message[:-1] - if level <= log_level: + if level <= loglevel: if prefix: - print "%s%s: %s" % (time.strftime("%Y%m%d-%H%M%S"), prefix_stack[-1], message) + print "%s%s: %s" % (time.strftime("%Y%m%d-%H%M%S"), prefix[-1], message) else: print message diff --git a/client/tests/kvm_runtest_2/kvm_runtest_2.py b/client/tests/kvm_runtest_2/kvm_runtest_2.py index c53877f..c6504ba 100644 --- a/client/tests/kvm_runtest_2/kvm_runtest_2.py +++ b/client/tests/kvm_runtest_2/kvm_runtest_2.py @@ -53,7 +53,10 @@ class kvm_runtest_2(test.test): resource.setrlimit(resource.RLIMIT_CORE, (-1, -1)) # Set the logging prefix - kvm_log.set_prefix(params.get("shortname")) + tag = params.get("shortname") + mypid = os.getpid() + print '[%s] ************************** WARK ******************* i am tag=%s\n' %(mypid, tag) + kvm_log.set_prefix(mypid, tag) # Report the parameters we've received and write them as keyvals kvm_log.debug("Test parameters:") @@ -65,6 +68,7 @@ class kvm_runtest_2(test.test): # Open the environment file env_filename = os.path.join(self.bindir, "env") + kvm_log.debug("using env file: %s" %(env_filename)) env = shelve.open(env_filename, writeback=True) kvm_log.debug("Contents of environment: %s" % str(env)) diff --git a/client/tests/kvm_runtest_2/kvm_vm.py b/client/tests/kvm_runtest_2/kvm_vm.py index fa68ce4..90ca21a 100644 --- a/client/tests/kvm_runtest_2/kvm_vm.py +++ b/client/tests/kvm_runtest_2/kvm_vm.py @@ -271,6 +271,9 @@ class VM: kvm_log.error("Actual MD5 sum differs from expected one") return False + # store parallel offset + self.port_offset = int(params.get("parallel_offset", 0)) + # Find available monitor filename while True: # The monitor filename should be unique @@ -281,7 +284,7 @@ class VM: # Handle port redirections redir_names = kvm_utils.get_sub_dict_names(params, "redirs") - host_ports = kvm_utils.find_free_ports(5000, 6000, len(redir_names)) + host_ports = kvm_utils.find_free_ports(self.port_offset+5000, 6000, len(redir_names)) self.redirs = {} for i in range(len(redir_names)): redir_params = kvm_utils.get_sub_dict(params, redir_names[i]) @@ -290,7 +293,7 @@ class VM: # Find available VNC port, if needed if params.get("display") == "vnc": - self.vnc_port = kvm_utils.find_free_port(5900, 6000) + self.vnc_port = kvm_utils.find_free_port(self.port_offset+5900, 6000) # Make qemu command qemu_command = self.make_qemu_command() @@ -298,7 +301,7 @@ class VM: # Is this VM supposed to accept incoming migrations? if for_migration: # Find available migration port - self.migration_port = kvm_utils.find_free_port(5200, 6000) + self.migration_port = kvm_utils.find_free_port(self.port_offset+5200, 6000) # Add -incoming option to the qemu command qemu_command += " -incoming tcp:0:%d" % self.migration_port