From 9795972b4ecb53db364c8a3cb541246e5a823c35 Mon Sep 17 00:00:00 2001 From: Abigail Warden Date: Thu, 15 Sep 2022 11:12:02 -0500 Subject: [PATCH 1/4] support to submit hadd jobs to condor --- Tools/python/CRAB3ToolsSh.py | 6 +- Tools/python/CondorTools.py | 6 +- Tools/python/hadd.py | 105 ++++++++++++++++++++++++++++------- Tools/python/haddOnWorker.py | 25 +++++++++ Tools/scripts/mhadd | 3 + 5 files changed, 118 insertions(+), 27 deletions(-) create mode 100644 Tools/python/haddOnWorker.py diff --git a/Tools/python/CRAB3ToolsSh.py b/Tools/python/CRAB3ToolsSh.py index 5f2b0f8fa..6fc570ebe 100644 --- a/Tools/python/CRAB3ToolsSh.py +++ b/Tools/python/CRAB3ToolsSh.py @@ -106,10 +106,10 @@ def crab_hadd_files(working_dir, lpc_shortcut=False, **kwargs): return expected, files -def crab_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, chunk_size=900, pattern=None, lpc_shortcut=False, range_filter=None): +def crab_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, chunk_size=900, pattern=None, lpc_shortcut=False, range_filter=None, submit=False): working_dir, new_name, new_dir = crab_hadd_args(working_dir, new_name, new_dir) expected, files = crab_hadd_files(working_dir, lpc_shortcut, range_filter=range_filter) - result = HaddBatchResult('crab', working_dir, new_name, new_dir, expected, files) + result = HaddBatchResult('crab', working_dir, new_name, new_dir, expected, files, submit) print '%s: expecting %i files if all jobs succeeded' % (working_dir, expected) if pattern: @@ -157,7 +157,7 @@ def crab_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, ch if result.success and not new_name.startswith('root://'): os.chmod(new_name, 0644) else: - result.success = hadd(new_name, files) + result.success = hadd(new_name, files, submit) if automatic_splitting: n = norm_from_file(new_name) diff --git a/Tools/python/CondorTools.py b/Tools/python/CondorTools.py index 15675b255..4d1d4d661 100644 --- a/Tools/python/CondorTools.py +++ b/Tools/python/CondorTools.py @@ -313,10 +313,10 @@ def cs_hadd_files(working_dir, **kwargs): files = files[a:b:c] return expected, files -def cs_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, chunk_size=900, pattern=None, range_filter=None): +def cs_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, chunk_size=900, pattern=None, range_filter=None, submit=False): working_dir, new_name, new_dir = cs_hadd_args(working_dir, new_name, new_dir) expected, files = cs_hadd_files(working_dir, range_filter=range_filter) - result = HaddBatchResult('condor', working_dir, new_name, new_dir, expected, files) + result = HaddBatchResult('condor', working_dir, new_name, new_dir, expected, files, submit) print '%s: expecting %i files if all jobs succeeded' % (working_dir, expected) if pattern: @@ -351,7 +351,7 @@ def cs_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, chun if result.success and not new_name.startswith('root://'): os.chmod(new_name, 0644) else: - result.success = hadd(new_name, files) + result.success = hadd(new_name, files, submit) return result diff --git a/Tools/python/hadd.py b/Tools/python/hadd.py index 758130e54..f7eefe35e 100644 --- a/Tools/python/hadd.py +++ b/Tools/python/hadd.py @@ -1,11 +1,12 @@ #!/usr/bin/env python import os, subprocess, tempfile, re +import ROOT from datetime import datetime from JMTucker.Tools import colors, eos class HaddBatchResult(object): - def __init__(self, kind, working_dir, new_name, new_dir, expected, files): + def __init__(self, kind, working_dir, new_name, new_dir, expected, files, submit): self.success = True self.kind = kind self.working_dir = working_dir @@ -13,6 +14,7 @@ def __init__(self, kind, working_dir, new_name, new_dir, expected, files): self.new_dir = new_dir self.expected = expected self.files = files + self.submit = submit class HaddlogParser(object): target_re = re.compile(r'hadd Target file: (.*)') @@ -38,7 +40,59 @@ def __init__(self, fn): self.num_sources = len(self.sources) self.files = self.sources.values() -def hadd(output_fn, input_fns): +def haddcondor(dir_hadd_target, flist): + """If submit == True, hadd jobs are sent to condor. A condor + submission file is created for a given input dataset. These + are then submitted to condor, where haddOnWorker.py is called + in order to do the hadding. There is the option to hadd fewer + files together. Edit the path to the executable to work on + personal space. + """ + #splitting up the arguments into the hadd directory and input dataset + dir_hadd, target = dir_hadd_target.rsplit("/", 1) + + target = target.replace('.root','') + nTotal = len(flist) + nLoops = 0 + + while len(flist) > 0 : + strNLoops = str(nLoops) + + #nItemsToPop = 10 + nItemsToPop = 200 + + inputFiles = '' + + for i in xrange(nItemsToPop) : + if len(flist) > 0 : + fstr = flist.pop() + inputFiles = inputFiles + ' ' + fstr + + #prefixStr = dir_hadd+'/'+target+'-'+strNLoops + prefixStr = dir_hadd+'/'+target + + #condorFile = open(dir_hadd+'/submit_'+strNLoops, 'w') + condorFile = open(dir_hadd+'/submit_'+target, 'w') + condorFile.write('executable = /afs/hep.wisc.edu/home/acwarden/work/llp/mfv_1068p1/src/JMTucker/Tools/python/haddOnWorker.py\n') + + condorFile.write('universe = vanilla\n') + condorFile.write('log = '+prefixStr+'.log\n') + condorFile.write('output = '+prefixStr+'.out\n') + condorFile.write('error = '+prefixStr+'.err\n') + condorFile.write('arguments = '+prefixStr+'.root ' + inputFiles + '\n') + condorFile.write('getenv = True\n') + + + condorFile.write('queue\n') + condorFile.close() + + #args = ['condor_submit '+dir_hadd+'/submit_'+target+'-'+strNLoops] + args = ['condor_submit '+dir_hadd+'/submit_'+target] + p = subprocess.Popen(args =args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=True) + + return p + +def hadd(output_fn, input_fns, submit): """This is a simple wrapper around hadd that suppresses the stdout from hadd, only reporting a summary line of how many files were merged. Output to eos is supported, including for the log file for @@ -51,35 +105,44 @@ def hadd(output_fn, input_fns): l = len(input_fns) start = datetime.now() print 'hadding %i files to %s at %s' % (l, output_fn, start) - args = ['hadd', output_fn] + input_fns + + if submit : + p = haddcondor(output_fn, input_fns) + + else : + args = ['hadd', output_fn] + input_fns + p = subprocess.Popen(args=args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) - p = subprocess.Popen(args=args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) stdout, stderr = p.communicate() assert stderr is None - log_fn = output_fn + '.haddlog' - is_eos = '/store/' in output_fn # ugh - while eos.exists(log_fn) if is_eos else os.path.exists(log_fn): - log_fn += '.2' - - if is_eos: - fd, tmp_fn = tempfile.mkstemp() - os.fdopen(fd, 'wt').write(stdout) - eos.cp(tmp_fn, log_fn) # if the haddlog already exists the new one will silently go into the ether... - os.remove(tmp_fn) - else: - open(log_fn, 'wt').write(stdout) + ## output files for the non-submit case + if not submit : + log_fn = output_fn + '.haddlog' + is_eos = '/store/' in output_fn # ugh + while eos.exists(log_fn) if is_eos else os.path.exists(log_fn): + log_fn += '.2' + + if is_eos: + fd, tmp_fn = tempfile.mkstemp() + os.fdopen(fd, 'wt').write(stdout) + eos.cp(tmp_fn, log_fn) # if the haddlog already exists the new one will silently go into the ether... + os.remove(tmp_fn) + else: + open(log_fn, 'wt').write(stdout) if p.returncode != 0: print colors.error('PROBLEM hadding %s' % output_fn) #print p.stdout.read() return False - max_file_num = max(int(line.split(':')[0].split(' ')[-1]) for line in stdout.split('\n') if 'Source file' in line) - print '-> %i files merged in %s' % (max_file_num, datetime.now() - start) - if max_file_num != l: - print colors.error('PROBLEM hadding %s' % output_fn) - return False + #more output for non-submit case + if not submit : + max_file_num = max(int(line.split(':')[0].split(' ')[-1]) for line in stdout.split('\n') if 'Source file' in line) + print '-> %i files merged in %s' % (max_file_num, datetime.now() - start) + if max_file_num != l: + print colors.error('PROBLEM hadding %s' % output_fn) + return False return True diff --git a/Tools/python/haddOnWorker.py b/Tools/python/haddOnWorker.py new file mode 100644 index 000000000..fcb22ba50 --- /dev/null +++ b/Tools/python/haddOnWorker.py @@ -0,0 +1,25 @@ +#!/usr/bin/env python + +''' +Very simple script which basically acts like: +hadd [output] [in1 in2 ...] +This solely exists for use as a simple condor executable. +USAGE: +python haddOnWorker.py [output] [in1 in2 ...] +''' + +import sys, os +import socket + +hostname = socket.gethostname() +print "Host name: ", hostname +print sys.argv + +target = sys.argv[1] +flist = sys.argv[2:] + +fstr = ' '.join(flist) + +cmd = 'hadd ' + target + ' ' + fstr +print cmd +os.system(cmd) diff --git a/Tools/scripts/mhadd b/Tools/scripts/mhadd index 45e5b56ac..787e16ed3 100755 --- a/Tools/scripts/mhadd +++ b/Tools/scripts/mhadd @@ -32,6 +32,8 @@ parser.add_argument('-r', '--remove-originals', action='store_true', default=Fal help='Remove the original files.') parser.add_argument('-v', '--verbose', action='store_true', default=False, help='Print more info.') +parser.add_argument('-s', '--submit', action='store_true', default=False, + help='submits hadd jobs to condor') options = parser.parse_args() @@ -72,6 +74,7 @@ def hadd_(is_crab, *args, **kwargs): kwargs['chunk_size'] = options.chunk_size kwargs['pattern'] = options.pattern kwargs['range_filter'] = options.range + kwargs['submit'] = options.submit if is_crab: kwargs['lpc_shortcut'] = options.lpc_shortcut From 71a5a7bf3f48b27e312cd2ed9769d0e1bda381fd Mon Sep 17 00:00:00 2001 From: Abigail Warden Date: Wed, 21 Sep 2022 09:24:57 -0500 Subject: [PATCH 2/4] better output print statements for condor mhadd and simplified hadd_condor --- Tools/python/hadd.py | 66 ++++++++++++++++---------------------------- Tools/scripts/mhadd | 27 ++++++++++++++++-- 2 files changed, 48 insertions(+), 45 deletions(-) diff --git a/Tools/python/hadd.py b/Tools/python/hadd.py index f7eefe35e..658f643e6 100644 --- a/Tools/python/hadd.py +++ b/Tools/python/hadd.py @@ -44,51 +44,33 @@ def haddcondor(dir_hadd_target, flist): """If submit == True, hadd jobs are sent to condor. A condor submission file is created for a given input dataset. These are then submitted to condor, where haddOnWorker.py is called - in order to do the hadding. There is the option to hadd fewer - files together. Edit the path to the executable to work on - personal space. + in order to do the hadding. """ #splitting up the arguments into the hadd directory and input dataset dir_hadd, target = dir_hadd_target.rsplit("/", 1) - + dir_executable = os.path.dirname(os.path.abspath(__file__)) + target = target.replace('.root','') - nTotal = len(flist) - nLoops = 0 - - while len(flist) > 0 : - strNLoops = str(nLoops) - - #nItemsToPop = 10 - nItemsToPop = 200 - - inputFiles = '' - - for i in xrange(nItemsToPop) : - if len(flist) > 0 : - fstr = flist.pop() - inputFiles = inputFiles + ' ' + fstr - - #prefixStr = dir_hadd+'/'+target+'-'+strNLoops - prefixStr = dir_hadd+'/'+target - - #condorFile = open(dir_hadd+'/submit_'+strNLoops, 'w') - condorFile = open(dir_hadd+'/submit_'+target, 'w') - condorFile.write('executable = /afs/hep.wisc.edu/home/acwarden/work/llp/mfv_1068p1/src/JMTucker/Tools/python/haddOnWorker.py\n') - - condorFile.write('universe = vanilla\n') - condorFile.write('log = '+prefixStr+'.log\n') - condorFile.write('output = '+prefixStr+'.out\n') - condorFile.write('error = '+prefixStr+'.err\n') - condorFile.write('arguments = '+prefixStr+'.root ' + inputFiles + '\n') - condorFile.write('getenv = True\n') - - - condorFile.write('queue\n') - condorFile.close() - - #args = ['condor_submit '+dir_hadd+'/submit_'+target+'-'+strNLoops] - args = ['condor_submit '+dir_hadd+'/submit_'+target] - p = subprocess.Popen(args =args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=True) + inputFiles = " ".join(flist) + prefixStr = dir_hadd+'/'+target + + condorFile = open(dir_hadd+'/submit_'+target, 'w') + condorFile.write('universe = vanilla\n') + condorFile.write('executable = '+dir_executable+'/haddOnWorker.py\n') + condorFile.write('arguments = '+prefixStr+'.root ' + inputFiles + '\n') + condorFile.write('log = '+prefixStr+'.log\n') + condorFile.write('output = '+prefixStr+'.out\n') + condorFile.write('error = '+prefixStr+'.err\n') + condorFile.write('should_transfer_files = yes\n') + condorFile.write('getenv = True\n') + + condorFile.write('queue\n') + condorFile.close() + + args = ['condor_submit '+dir_hadd+'/submit_'+target] + p = subprocess.Popen(args =args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=True) + + print p.stdout.read() return p @@ -117,6 +99,7 @@ def hadd(output_fn, input_fns, submit): assert stderr is None ## output files for the non-submit case + if not submit : log_fn = output_fn + '.haddlog' is_eos = '/store/' in output_fn # ugh @@ -133,7 +116,6 @@ def hadd(output_fn, input_fns, submit): if p.returncode != 0: print colors.error('PROBLEM hadding %s' % output_fn) - #print p.stdout.read() return False #more output for non-submit case diff --git a/Tools/scripts/mhadd b/Tools/scripts/mhadd index 787e16ed3..571b0d9ea 100755 --- a/Tools/scripts/mhadd +++ b/Tools/scripts/mhadd @@ -53,6 +53,7 @@ if options.only_notdones: options.ignore_done = True options.overwrite = True + ######################################################################## wds = options.positional @@ -114,22 +115,42 @@ def hadd_(is_crab, *args, **kwargs): if options.overwrite: print colors.yellow('overwriting existing file %s' % new_name) rm(new_name) + else: log_fn = new_name + '.haddlog' + log_fn_submit = new_name.replace('.root', '') + '.log' + if on_eos: log_fn = eos.fusemount(log_fn) njobs = hadd_files(working_dir, **kwargs)[0] + if not exists(log_fn): log_njobs = 1 # assume 1 file copied - else: + + ## helpers to determine which log is present and which is not : submit log for jobs that were submitted to condor + ## and haddlog for those that were not + s_log = exists(log_fn_submit) and not exists(log_fn) + h_log = exists(log_fn) and not exists(log_fn_submit) + + if s_log: + if os.stat(new_name).st_size != 0 : + print colors.boldred('skipping existing file %s which has already been submitted to condor succesfully' % new_name) + #if the root file is created and it is zero, try to run again? + else: + rm(new_name) + + elif h_log: p = HaddlogParser(log_fn) log_njobs = p.num_sources - if log_njobs != njobs: - print colors.boldred('skipping existing file %s (for which num sources %i != njobs %i)' % (new_name, log_njobs, njobs)) + + if (log_njobs != njobs): + print colors.boldred('skipping existing file %s (for which num sources %i != njobs %i)' % (new_name, log_njobs, njobs)) + else: if options.verbose: print colors.yellow('skipping existing file %s (which looks complete by njobs and haddlog)' % new_name) return + result = hadd(*args, **kwargs) From 91cad6519e5dd7f671b7c90c6edae16542e7c586 Mon Sep 17 00:00:00 2001 From: Abigail Warden Date: Wed, 21 Sep 2022 10:38:16 -0500 Subject: [PATCH 3/4] more complete print statements for mhadd --- Tools/scripts/mhadd | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/Tools/scripts/mhadd b/Tools/scripts/mhadd index 571b0d9ea..f7bb2228c 100755 --- a/Tools/scripts/mhadd +++ b/Tools/scripts/mhadd @@ -145,8 +145,13 @@ def hadd_(is_crab, *args, **kwargs): if (log_njobs != njobs): print colors.boldred('skipping existing file %s (for which num sources %i != njobs %i)' % (new_name, log_njobs, njobs)) + else : + print colors.boldred('skipping existing file %s which has already been hadded succesfully' % new_name) else: + #assume that the last case to consider is the one in which files are copied over (which do not produce any log but the .root file does exist) + print colors.boldred('skipping existing file %s (for which num sources %i = njobs %i)' % (new_name, log_njobs, njobs)) + if options.verbose: print colors.yellow('skipping existing file %s (which looks complete by njobs and haddlog)' % new_name) return From c7285107b13f4a49c7972c99a9407d5f2b7d82b0 Mon Sep 17 00:00:00 2001 From: Abigail Warden Date: Tue, 4 Oct 2022 03:57:25 -0500 Subject: [PATCH 4/4] more robust hadding on condor to work on all machines; main author : Ang --- Tools/python/CRAB3ToolsSh.py | 10 ++-- Tools/python/CondorTools.py | 10 ++-- Tools/python/hadd.py | 86 ++++++++--------------------- Tools/python/haddOnWorker.py | 25 --------- Tools/python/hadd_on_condor.py | 98 ++++++++++++++++++++++++++++++++++ Tools/scripts/mhadd | 38 +++++-------- 6 files changed, 146 insertions(+), 121 deletions(-) delete mode 100644 Tools/python/haddOnWorker.py create mode 100644 Tools/python/hadd_on_condor.py diff --git a/Tools/python/CRAB3ToolsSh.py b/Tools/python/CRAB3ToolsSh.py index 6fc570ebe..e84be1f94 100644 --- a/Tools/python/CRAB3ToolsSh.py +++ b/Tools/python/CRAB3ToolsSh.py @@ -5,6 +5,7 @@ import re from JMTucker.Tools.CRAB3ToolsBase import * from JMTucker.Tools.hadd import HaddBatchResult, hadd +from JMTucker.Tools.hadd_on_condor import hadd_on_condor if crab_global_options.support_automatic_splitting: from JMTucker.Tools.Sample import fn_to_sample, norm_from_file from JMTucker.Tools import Samples, colors @@ -106,7 +107,7 @@ def crab_hadd_files(working_dir, lpc_shortcut=False, **kwargs): return expected, files -def crab_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, chunk_size=900, pattern=None, lpc_shortcut=False, range_filter=None, submit=False): +def crab_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, chunk_size=900, pattern=None, lpc_shortcut=False, range_filter=None, submit=False, cmssw_tar_path=''): working_dir, new_name, new_dir = crab_hadd_args(working_dir, new_name, new_dir) expected, files = crab_hadd_files(working_dir, lpc_shortcut, range_filter=range_filter) result = HaddBatchResult('crab', working_dir, new_name, new_dir, expected, files, submit) @@ -157,8 +158,11 @@ def crab_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, ch if result.success and not new_name.startswith('root://'): os.chmod(new_name, 0644) else: - result.success = hadd(new_name, files, submit) - + if not submit: + result.success = hadd(new_name, files) + else: + result.success = hadd_on_condor(new_name, working_dir, cmssw_tar_path, files) + if automatic_splitting: n = norm_from_file(new_name) sn, s = fn_to_sample(Samples, new_name) diff --git a/Tools/python/CondorTools.py b/Tools/python/CondorTools.py index 4d1d4d661..2cbadfa0a 100644 --- a/Tools/python/CondorTools.py +++ b/Tools/python/CondorTools.py @@ -9,6 +9,7 @@ from JMTucker.Tools.LumiJSONTools import fjr2ll from JMTucker.Tools.general import sub_popen, touch from JMTucker.Tools.hadd import HaddBatchResult, hadd +from JMTucker.Tools.hadd_on_condor import hadd_on_condor from JMTucker.Tools import colors class CSHelpersException(Exception): @@ -313,7 +314,7 @@ def cs_hadd_files(working_dir, **kwargs): files = files[a:b:c] return expected, files -def cs_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, chunk_size=900, pattern=None, range_filter=None, submit=False): +def cs_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, chunk_size=900, pattern=None, range_filter=None, submit=False, cmssw_tar_path=''): working_dir, new_name, new_dir = cs_hadd_args(working_dir, new_name, new_dir) expected, files = cs_hadd_files(working_dir, range_filter=range_filter) result = HaddBatchResult('condor', working_dir, new_name, new_dir, expected, files, submit) @@ -351,8 +352,11 @@ def cs_hadd(working_dir, new_name=None, new_dir=None, raise_on_empty=False, chun if result.success and not new_name.startswith('root://'): os.chmod(new_name, 0644) else: - result.success = hadd(new_name, files, submit) - + if not submit: + result.success = hadd(new_name, files) + else: + result.success = hadd_on_condor(new_name, working_dir, cmssw_tar_path, files) + return result def cs_report(wd, partial=False): diff --git a/Tools/python/hadd.py b/Tools/python/hadd.py index 658f643e6..5eb4ea573 100644 --- a/Tools/python/hadd.py +++ b/Tools/python/hadd.py @@ -40,41 +40,7 @@ def __init__(self, fn): self.num_sources = len(self.sources) self.files = self.sources.values() -def haddcondor(dir_hadd_target, flist): - """If submit == True, hadd jobs are sent to condor. A condor - submission file is created for a given input dataset. These - are then submitted to condor, where haddOnWorker.py is called - in order to do the hadding. - """ - #splitting up the arguments into the hadd directory and input dataset - dir_hadd, target = dir_hadd_target.rsplit("/", 1) - dir_executable = os.path.dirname(os.path.abspath(__file__)) - - target = target.replace('.root','') - inputFiles = " ".join(flist) - prefixStr = dir_hadd+'/'+target - - condorFile = open(dir_hadd+'/submit_'+target, 'w') - condorFile.write('universe = vanilla\n') - condorFile.write('executable = '+dir_executable+'/haddOnWorker.py\n') - condorFile.write('arguments = '+prefixStr+'.root ' + inputFiles + '\n') - condorFile.write('log = '+prefixStr+'.log\n') - condorFile.write('output = '+prefixStr+'.out\n') - condorFile.write('error = '+prefixStr+'.err\n') - condorFile.write('should_transfer_files = yes\n') - condorFile.write('getenv = True\n') - - condorFile.write('queue\n') - condorFile.close() - - args = ['condor_submit '+dir_hadd+'/submit_'+target] - p = subprocess.Popen(args =args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=True) - - print p.stdout.read() - - return p - -def hadd(output_fn, input_fns, submit): +def hadd(output_fn, input_fns): """This is a simple wrapper around hadd that suppresses the stdout from hadd, only reporting a summary line of how many files were merged. Output to eos is supported, including for the log file for @@ -83,48 +49,38 @@ def hadd(output_fn, input_fns, submit): other problems reported by hadd. If so, prints an error to stdout. Returns true if success. """ - l = len(input_fns) start = datetime.now() print 'hadding %i files to %s at %s' % (l, output_fn, start) - - if submit : - p = haddcondor(output_fn, input_fns) - - else : - args = ['hadd', output_fn] + input_fns - p = subprocess.Popen(args=args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + args = ['hadd', output_fn] + input_fns + p = subprocess.Popen(args=args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) stdout, stderr = p.communicate() assert stderr is None - - ## output files for the non-submit case - if not submit : - log_fn = output_fn + '.haddlog' - is_eos = '/store/' in output_fn # ugh - while eos.exists(log_fn) if is_eos else os.path.exists(log_fn): - log_fn += '.2' - - if is_eos: - fd, tmp_fn = tempfile.mkstemp() - os.fdopen(fd, 'wt').write(stdout) - eos.cp(tmp_fn, log_fn) # if the haddlog already exists the new one will silently go into the ether... - os.remove(tmp_fn) - else: - open(log_fn, 'wt').write(stdout) + log_fn = output_fn + '.haddlog' + is_eos = '/store/' in output_fn # ugh + while eos.exists(log_fn) if is_eos else os.path.exists(log_fn): + log_fn += '.2' + + if is_eos: + fd, tmp_fn = tempfile.mkstemp() + os.fdopen(fd, 'wt').write(stdout) + eos.cp(tmp_fn, log_fn) # if the haddlog already exists the new one will silently go into the ether... + os.remove(tmp_fn) + else: + open(log_fn, 'wt').write(stdout) if p.returncode != 0: print colors.error('PROBLEM hadding %s' % output_fn) + #print p.stdout.read() return False - #more output for non-submit case - if not submit : - max_file_num = max(int(line.split(':')[0].split(' ')[-1]) for line in stdout.split('\n') if 'Source file' in line) - print '-> %i files merged in %s' % (max_file_num, datetime.now() - start) - if max_file_num != l: - print colors.error('PROBLEM hadding %s' % output_fn) - return False + max_file_num = max(int(line.split(':')[0].split(' ')[-1]) for line in stdout.split('\n') if 'Source file' in line) + print '-> %i files merged in %s' % (max_file_num, datetime.now() - start) + if max_file_num != l: + print colors.error('PROBLEM hadding %s' % output_fn) + return False return True diff --git a/Tools/python/haddOnWorker.py b/Tools/python/haddOnWorker.py deleted file mode 100644 index fcb22ba50..000000000 --- a/Tools/python/haddOnWorker.py +++ /dev/null @@ -1,25 +0,0 @@ -#!/usr/bin/env python - -''' -Very simple script which basically acts like: -hadd [output] [in1 in2 ...] -This solely exists for use as a simple condor executable. -USAGE: -python haddOnWorker.py [output] [in1 in2 ...] -''' - -import sys, os -import socket - -hostname = socket.gethostname() -print "Host name: ", hostname -print sys.argv - -target = sys.argv[1] -flist = sys.argv[2:] - -fstr = ' '.join(flist) - -cmd = 'hadd ' + target + ' ' + fstr -print cmd -os.system(cmd) diff --git a/Tools/python/hadd_on_condor.py b/Tools/python/hadd_on_condor.py new file mode 100644 index 000000000..e4fd0e22d --- /dev/null +++ b/Tools/python/hadd_on_condor.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python + +import sys +import os +import subprocess +import stat +from JMTucker.Tools.CMSSWTools import cmssw_base + +def make_bash_file(cmssw_tar_path, bash_file_path): + cmssw_name = cmssw_base().rsplit("/")[-1] + bash_meat = '''#!/bin/bash +workdir=$(pwd) +echo "Starting job on " `date` #Date/time of start of job +echo "Running on: `uname -a`" #Condor job is running on this node +echo "System software: `cat /etc/redhat-release`" #Operating System on that node +export SCRAM_ARCH=__SCRAM_ARCH__ +source /cvmfs/cms.cern.ch/cmsset_default.sh +scram project CMSSW __CMSSW_VERSION__ 2>&1 > /dev/null +scramexit=$? +if [[ $scramexit -ne 0 ]]; then + echo scram exited with code $scramexit + exit $scramexit +fi +cd __CMSSW_VERSION__ +tar xf ${workdir}/input.tgz +cd src +eval `scram ru -sh` +scram b -j 2 2>&1 > /dev/null + +#python JMTucker/Tools/python/hadd.py $1 ${@:2} +cd ${workdir} +hadd.py $1 ${@:2} +''' \ +.replace('__SCRAM_ARCH__', os.environ['SCRAM_ARCH']) \ +.replace('__CMSSW_VERSION__', os.environ['CMSSW_VERSION']) + + bash_file = open(bash_file_path+'/bash_condor.sh', 'w') + bash_file.write(bash_meat) + bash_file.close() + subprocess.check_call(['chmod', '+x', bash_file_path+'/bash_condor.sh']) + return + + +def hadd_on_condor(new_name, job_dir, cmssw_tar_path, input_files): + output_file_path = os.path.dirname(os.path.abspath(new_name)) + new_nice_name = (new_name.rsplit("/")[-1]).replace('.root','') + job_dir = os.path.abspath(job_dir) + cmssw_tar_path = os.path.abspath(cmssw_tar_path) + is_eos = '/store/' in new_name + if not is_eos: + new_name = new_nice_name+'.root' + + hadd_job_dir = job_dir+'/mhadd/' + if not os.path.exists(hadd_job_dir): + os.mkdir(hadd_job_dir) + + make_bash_file(cmssw_tar_path, hadd_job_dir) + + #requirements necessary for wisconsin machines -- comment out if not needed + # the CentOS == 7 requires proxy with 2048 bit and will fail with usual 1024 + # voms-proxy-init -rfc -valid 144:00 -voms cms -bits 2048 + condorFile = open(hadd_job_dir+'/submit_'+new_nice_name, 'w') + condorFile.write('universe = vanilla\n') + condorFile.write('executable = '+hadd_job_dir+'/bash_condor.sh\n') + condorFile.write('arguments = '+new_name+' ' + ' '.join(input_files) + '\n') + condorFile.write('log = '+hadd_job_dir+'/hadd.log\n') + condorFile.write('output = '+hadd_job_dir+'/hadd.out\n') + condorFile.write('error = '+hadd_job_dir+'/hadd.err\n') + condorFile.write('requirements = TARGET.HAS_OSG_WN_CLIENT =?= TRUE\n') + condorFile.write('requirements = TARGET.OpSysMajorVer == 7\n') + condorFile.write('should_transfer_files = yes\n') + condorFile.write('when_to_transfer_output = ON_EXIT\n') + condorFile.write('transfer_input_files = '+cmssw_tar_path+'\n') + + if not is_eos: + condorFile.write('Transfer_Output_Files = '+new_nice_name+'.root, '+new_nice_name+'.root.haddlog'+'\n') + + condorFile.write('queue\n') + condorFile.close() + + submit_path = os.getcwd() + if not is_eos: + submit_path = output_file_path + args = ['condor_submit '+hadd_job_dir+'/submit_'+new_nice_name] + p = subprocess.Popen(args =args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=True, cwd=submit_path) + + print p.stdout.read() + + return p + +_all__ = [ + 'make_bash_file', + 'hadd_on_condor', + ] + +if __name__ == '__main__': + + hadd_on_condor(sys.argv[1], sys.argv[2], sys.argv[3:]) diff --git a/Tools/scripts/mhadd b/Tools/scripts/mhadd index f7bb2228c..45e95f71c 100755 --- a/Tools/scripts/mhadd +++ b/Tools/scripts/mhadd @@ -1,11 +1,12 @@ #!/usr/bin/env python -import sys, os, glob, argparse +import sys, os, glob, argparse, subprocess from JMTucker.Tools.CRAB3ToolsSh import is_crab_working_dir, crab_hadd_args, crab_hadd_files, crab_hadd from JMTucker.Tools.CondorTools import is_cs_dir, cs_hadd_args, cs_hadd_files, cs_hadd from JMTucker.Tools.general import touch as _touch from JMTucker.Tools.hadd import HaddlogParser from JMTucker.Tools import eos, colors +from JMTucker.Tools.CMSSWTools import cmssw_base, make_tarball parser = argparse.ArgumentParser(description = 'mhadd: hadd the files from condor/crab directories', usage = '%(prog)s [options] condor_or_crab_dirs') @@ -53,6 +54,8 @@ if options.only_notdones: options.ignore_done = True options.overwrite = True +if options.submit: + make_tarball('%s/input.tgz' % (os.getcwd()), include_python=True, include_interface=True) ######################################################################## @@ -64,7 +67,7 @@ def hadd_(is_crab, *args, **kwargs): found_one = True done = True - working_dir = args[0] + working_dir = os.path.abspath(args[0]) if not os.path.isfile(os.path.join(working_dir, 'mmon_done')): if options.ignore_done: done = False @@ -77,6 +80,9 @@ def hadd_(is_crab, *args, **kwargs): kwargs['range_filter'] = options.range kwargs['submit'] = options.submit + if options.submit: + kwargs['cmssw_tar_path'] = '%s/input.tgz' % (os.getcwd()) + if is_crab: kwargs['lpc_shortcut'] = options.lpc_shortcut hadd_args, hadd_files, hadd = crab_hadd_args, crab_hadd_files, crab_hadd @@ -127,35 +133,16 @@ def hadd_(is_crab, *args, **kwargs): if not exists(log_fn): log_njobs = 1 # assume 1 file copied - ## helpers to determine which log is present and which is not : submit log for jobs that were submitted to condor - ## and haddlog for those that were not - s_log = exists(log_fn_submit) and not exists(log_fn) - h_log = exists(log_fn) and not exists(log_fn_submit) - - if s_log: - if os.stat(new_name).st_size != 0 : - print colors.boldred('skipping existing file %s which has already been submitted to condor succesfully' % new_name) - #if the root file is created and it is zero, try to run again? - else: - rm(new_name) - - elif h_log: + else: p = HaddlogParser(log_fn) log_njobs = p.num_sources - - if (log_njobs != njobs): - print colors.boldred('skipping existing file %s (for which num sources %i != njobs %i)' % (new_name, log_njobs, njobs)) - else : - print colors.boldred('skipping existing file %s which has already been hadded succesfully' % new_name) - + if log_njobs != njobs: + print colors.boldred('skipping existing file %s (for which num sources %i != njobs %i)' % (new_name, log_njobs, njobs)) else: - #assume that the last case to consider is the one in which files are copied over (which do not produce any log but the .root file does exist) - print colors.boldred('skipping existing file %s (for which num sources %i = njobs %i)' % (new_name, log_njobs, njobs)) - if options.verbose: print colors.yellow('skipping existing file %s (which looks complete by njobs and haddlog)' % new_name) return - + result = hadd(*args, **kwargs) @@ -168,6 +155,7 @@ def hadd_(is_crab, *args, **kwargs): for x in wds: + x = os.path.abspath(x) is_crab = is_crab_working_dir(x) if is_crab or is_cs_dir(x): hadd_(is_crab, x)