Merge branch 'develop' into mesmith75-patch-2

Author: mesmith75

python/Ganga/GPIDev/Lib/Tasks/ITransform.py

@@ -214,11 +214,6 @@ def update(self):
             if task.transforms[trf_id].status != "completed":
                 return 0
 
-        # set the start time if not already set
-        if len(self.required_trfs) > 0 and self.units[0].start_time == 0:
-            for unit in self.units:
-                unit.start_time = time.time() + self.chain_delay * 60 - 1
-
         # report the info for this transform
         unit_status = { "new":0, "hold":0, "running":0, "completed":0, "bad":0, "recreating":0 }
         for unit in self.units:
@@ -234,6 +229,11 @@ def update(self):
         # current data
         self.createUnits()
 
+        # set the start time if not already set
+        if len(self.required_trfs) > 0 and self.units[0].start_time == 0:
+            for unit in self.units:
+                unit.start_time = time.time() + self.chain_delay * 60 - 1
+
         # loop over units and update them ((re)submits will be called here)
         old_status = self.status
         unit_status_list = []

python/Ganga/Lib/Splitters/ArgSplitter.py

@@ -5,11 +5,10 @@
 ###############################################################################
 
 import copy
-from Ganga.GPIDev.Adapters.ISplitter import ISplitter
+from Ganga.GPIDev.Adapters.ISplitter import ISplitter, SplittingError
 from Ganga.GPIDev.Base.Proxy import stripProxy
 from Ganga.GPIDev.Schema import Schema, Version, SimpleItem
 from Ganga.GPIDev.Lib.GangaList.GangaList import GangaList
-
 from Ganga.Utility.logging import getLogger
 logger = getLogger()
 
@@ -19,7 +18,8 @@ class ArgSplitter(ISplitter):
     """
     Split job by changing the args attribute of the application.
 
-    This splitter only applies to the applications which have args attribute (e.g. Executable, Root).
+    This splitter only applies to the applications which have args attribute (e.g. Executable, Root), or those
+    with extraArgs (GaudiExec). If an application has both, args takes precedence.
     It is a special case of the GenericSplitter.
 
     This splitter allows the creation of a series of subjobs where
@@ -58,7 +58,13 @@ def split(self, job):
             j = self.createSubjob(job,['application'])
             # Add new arguments to subjob
             app = copy.deepcopy(job.application)
-            app.args = arg
+            if hasattr(app, 'args'):
+                app.args = arg
+            elif hasattr(app, 'extraArgs'):
+                app.extraArgs = arg
+            else:
+                raise SplittingError('Application has neither args or extraArgs in its schema') 
+                    
             j.application = app
             logger.debug('Arguments for split job is: ' + str(arg))
             subjobs.append(stripProxy(j))

python/GangaAtlas/Lib/ATLASDataset/DQ2Dataset.py

@@ -6,7 +6,7 @@
 ###############################################################################
 # A DQ2 dataset
 
-import sys, os, re, urllib, commands, imp, threading, time, fnmatch
+import sys, os, re, urllib, commands, imp, threading, time, fnmatch, getpass
 
 from Ganga.GPIDev.Lib.Dataset import Dataset
 from Ganga.GPIDev.Schema import *
@@ -2023,25 +2023,27 @@ def addTask(self, cmd):
 logger = getLogger()
 
 # New for DQ2 client 2.3.0
-from Ganga.GPIDev.Credentials_old import GridProxy
-gridProxy = GridProxy()
-if not gridProxy.isValid():
-    gridProxy.create()
+#from Ganga.GPIDev.Credentials_old import GridProxy
+#gridProxy = GridProxy()
+#if not gridProxy.isValid():
+#    gridProxy.create()
 
-username = gridProxy.identity(safe=True)
+#username = gridProxy.identity(safe=True)
 # Note: Allow missing nickname as if we can't create a proxy for some reason, we still want to start Ganga
-nickname = getNickname(allowMissingNickname=True)
-if nickname:
-    username = nickname
-os.environ['RUCIO_ACCOUNT'] = username
+#nickname = getNickname(allowMissingNickname=True)
+#if nickname:
+#    username = nickname
+os.environ['RUCIO_ACCOUNT'] = getpass.getuser()
 logger.debug("Using RUCIO_ACCOUNT = %s " %(os.environ['RUCIO_ACCOUNT'])) 
 
 # Again, if we don't have a valid proxy, don't attempt to create DQ2 object as it will just fail
-if gridProxy.isValid():
-    from dq2.clientapi.DQ2 import DQ2
-    dq2=DQ2(force_backend='rucio')
-else:
-    dq2 = None
+#if gridProxy.isValid():
+#    from dq2.clientapi.DQ2 import DQ2
+#    dq2=DQ2(force_backend='rucio')
+#else:
+#    dq2 = None
+
+dq2 = None
 
 from threading import Lock
 dq2_lock = Lock()

python/GangaAtlas/Lib/Athena/Athena.py

@@ -1534,6 +1534,9 @@ class AthenaSplitterJob(ISplitter):
         'numfiles_subjob'     : SimpleItem(defvalue=0,sequence=0, doc="Number of files per subjob"),
         'match_subjobs_files' : SimpleItem(defvalue=False,sequence=0, doc="Match the number of subjobs to the number of inputfiles"),
         'split_per_dataset'   : SimpleItem(defvalue=False,sequence=0, doc="Match the number of subjobs to the number of datasets"),
+        'events_per_subjob'   : SimpleItem(defvalue=-1,sequence=0, doc='Number of Events to process per subjob. Must be used with numsubjobs.'
+                                                                       'The total events processed with be events_per_subjob * numsubjobs. Please'
+                                                                       'make sure this covers the number of events required'),
         'output_loc_to_input' : SimpleItem(defvalue={}, doc='Dictionary that lists the input files that should go to a '
                                                             'particular output dir, e.g. { "/out/dir": ["/in/file1", "/in/file2"].'
                                                             'Input files must match what is given to ATLASLocalDataset }')
@@ -1554,11 +1557,12 @@ def split(self,job):
         # Preparation
         inputnames=[]
         inputguids=[]
-        if job.inputdata:
+        if job.inputdata and (job.inputdata._name == 'ATLASLocalDataset'):
 
-            if (job.inputdata._name == 'ATLASLocalDataset'):
-                inputnames = []
-                outputnames = []
+            # Special case for events_per_subjob as input data is ignored
+            inputnames = []
+            outputnames = []
+            if self.events_per_subjob < 0:
                 numfiles = len(job.inputdata.get_dataset_filenames())
                 if self.numfiles_subjob > 0:
                     self.numsubjobs = int( math.ceil( numfiles / float(self.numfiles_subjob) ) )
@@ -1631,50 +1635,21 @@ def split(self,job):
 
                     for j in xrange(numfiles):
                         inputnames[j % self.numsubjobs].append(job.inputdata.get_dataset_filenames()[j])
+            else:
+                # for splitting on events, all data is passed to every subjob and skip events/max events
+                # is set appropriately
+                if self.numfiles_subjob > 0 or self.match_subjobs_files or self.split_per_dataset:
+                    raise ApplicationConfigurationError("Cannot use events_per_subjob with numfiles_subjob, match_subjobs_files, split_per_dataset")
 
-            if job.inputdata._name == 'DQ2Dataset':
-                # Splitting per dataset
-                if self.split_per_dataset:
-                    contents = job.inputdata.get_contents(overlap=False)
-                    datasets = job.inputdata.dataset
-                    self.numsubjobs = len(datasets)
-                    for dataset in datasets:
-                        content = contents[dataset]
-                        content.sort(lambda x,y:cmp(x[1],y[1]))
-                        inputnames.append( [ lfn for guid, lfn in content ] )
-                        inputguids.append( [ guid for guid, lfn in content ] )
-                else:
-                    # Splitting per file
-                    content = []
-                    input_files = []
-                    input_guids = []
-                    names = None
-                    # Get list of filenames and guids
-                    contents = job.inputdata.get_contents()
-                    if self.match_subjobs_files:
-                        self.numsubjobs = len(contents)
-                    elif self.numfiles_subjob>0:
-                        numjobs = len(contents) / int(self.numfiles_subjob)
-                        if (len(contents) % self.numfiles_subjob)>0:
-                            numjobs += 1
-                        self.numsubjobs = numjobs
-                        logger.info('Submitting %s subjobs',numjobs)
-
-                    # Fill dummy values
-                    for i in xrange(self.numsubjobs):    
-                        inputnames.append([])
-                        inputguids.append([])
-                    input_files = [ lfn  for guid, lfn in contents ]
-                    input_guids = [ guid for guid, lfn in contents ]
+                if self.numsubjobs < 1:
+                    raise ApplicationConfigurationError("Please specify the number of subjobs if using events_per_subjob")
 
-                    # Splitting
-                    for j in xrange(len(input_files)):
-                        inputnames[j % self.numsubjobs].append(input_files[j])
-                        inputguids[j % self.numsubjobs].append(input_guids[j])
+                logger.warning("Splitting by number of events. All data will be passed to all subjobs and the total number of events to be"
+                               "processed will be numsubjobs * events_per_subjob (%d * %d = %d in this case)" %
+                               (self.numsubjobs, self.events_per_subjob, self.numsubjobs * self.events_per_subjob))
 
-        if job.backend._name == 'LCG' and job.backend.middleware=='GLITE' and self.numsubjobs>config['MaxJobsAthenaSplitterJobLCG']:
-            printout = 'Job submission failed ! AthenaSplitterJob.numsubjobs>%s - glite WMS does not like bulk jobs with more than approximately 100 subjobs - use less subjobs or use job.backend.middleware=="EDG"  ' %config['MaxJobsAthenaSplitterJobLCG']
-            raise ApplicationConfigurationError(printout)
+                for j in xrange(self.numsubjobs):
+                    inputnames.append(job.inputdata.get_dataset_filenames())
 
         # Do the splitting
         for i in range(self.numsubjobs):
@@ -1683,93 +1658,25 @@ def split(self,job):
             j.inputdata=job.inputdata
             if job.inputdata:
                 j.inputdata.names=inputnames[i]
-                if job.inputdata._name == 'DQ2Dataset':
-                    j.inputdata.guids=inputguids[i]
-                    j.inputdata.number_of_files = len(inputguids[i])
-                    if self.split_per_dataset:
-                        j.inputdata.dataset=job.inputdata.dataset[i]
+
             j.outputdata=job.outputdata
 
             # Set the output location if we have mapping
             if self.output_loc_to_input and isinstance(job.outputdata, ATLASOutputDataset):
                 j.outputdata.location = outputnames[i]
 
             j.application = job.application
+            if self.events_per_subjob > 0:
+                j.application.max_events = self.events_per_subjob
+                j.application.skip_events = self.events_per_subjob * i
+
             j.backend=job.backend
             j.inputsandbox=job.inputsandbox
             j.outputsandbox=job.outputsandbox
 
             subjobs.append(j)
         return subjobs
 
-class ATLASTier3Splitter(ISplitter):
-    """Splitter for ATLASTier3Dataset"""
-    
-    _name = "ATLASTier3Splitter"
-    _schema = Schema(Version(1,0), {
-        'numjobs'              : SimpleItem(defvalue=0,sequence=0, doc="Number of subjobs"),
-        'numfiles'             : SimpleItem(defvalue=0,sequence=0, doc="Number of files per subjob")
-        } )
-
-    _GUIPrefs = [ { 'attribute' : 'numjobs',           'widget' : 'Int' },
-                  { 'attribute' : 'numfiles',          'widget' : 'Int' },
-                  ]
-
-    ### Splitting based on numsubjobs
-    def split(self,job):
-        from Ganga.GPIDev.Lib.Job import Job
-        subjobs = []
-        logger.debug("ATLASTier3Splitter split called")
-        
-        if not job.inputdata:
-            raise ApplicationConfigurationError("ATLASTier3Splitter requires ATLASTier3Dataset")
-        if job.inputdata._name != 'ATLASTier3Dataset':
-            raise ApplicationConfigurationError("ATLASTier3Splitter requires ATLASTier3Dataset")
-        if self.numjobs and self.numfiles:
-            logger.warning('You specified numjobs and numfiles. Setting numjobs = 0 to continue.')
-            self.numjobs = 0
-            #raise ApplicationConfigurationError(None, "ATLASTier3Splitter: specify numjobs or numfiles, but not both.")
-       
-        if job.inputdata.pfnListFile.name:
-            logger.info('Loading file names from %s'%job.inputdata.pfnListFile.name)
-            pfnListFile = open(job.inputdata.pfnListFile.name)
-            job.inputdata.names = [name.strip() for name in pfnListFile]
-            pfnListFile.close()
- 
-        allnames = list(job.inputdata.names)
-
-        # default behaviour is 20 subjobs
-        if not self.numjobs and not self.numfiles:
-            self.numjobs = min(20,len(allnames))
-
-        # limit numfiles and numjobs
-        self.numjobs = min(self.numjobs,len(allnames))
-        self.numfiles = min(self.numfiles,len(allnames))
-
-        # calculate numfiles and numjobs
-        if self.numfiles:
-            (self.numjobs,r) = divmod(len(allnames),self.numfiles)
-            if r: self.numjobs += 1
-        elif self.numjobs:
-            (self.numfiles,r) = divmod(len(allnames),self.numjobs)
-            if r: self.numfiles += 1
-
-        # Do the splitting
-        allnames.reverse()
-        for i in range(self.numjobs):
-            j = Job()
-            j.inputdata=job.inputdata
-            j.inputdata.names=[]
-            while allnames and len(j.inputdata.names) < self.numfiles:
-                j.inputdata.names.append(allnames.pop())
-            j.outputdata = job.outputdata
-            j.application = job.application
-            j.backend = job.backend
-            j.inputsandbox = job.inputsandbox
-            j.outputsandbox = job.outputsandbox
-            subjobs.append(j)
-        
-        return subjobs
 
 from Ganga.GPIDev.Adapters.IMerger import IMerger
 from commands import getstatusoutput    

python/GangaAtlas/Lib/Athena/AthenaLocalRTHandler.py

@@ -280,6 +280,17 @@ def prepare(self,app,appsubconfig,appmasterconfig,jobmasterconfig):
         # set EOS env setting
         environment['EOS_COMMAND_PATH'] = config['PathToEOSBinary']
 
+        # Max/skip event settings
+        if app.skip_events > 0:
+            environment['ATHENA_SKIP_EVENTS'] = str(app.skip_events)
+
+        if app.max_events > 0:
+            environment['ATHENA_MAX_EVENTS'] = str(app.max_events)
+
+        # Flag if we're in ByteStream Data
+        if app.atlas_run_config.has_key('input') and app.atlas_run_config['input'].has_key('inBS'):
+            environment['USE_BYTESTREAM'] = 'True'
+
         # flag for single output dir
         if (config['SingleDirForLocalOutput'] or config['NoSubDirsAtAllForLocalOutput']) and job._getParent():
             environment['SINGLE_OUTPUT_DIR'] = jid
@@ -312,21 +323,6 @@ def prepare(self,app,appsubconfig,appmasterconfig,jobmasterconfig):
             environment['DQ2_OUTPUT_SPACE_TOKENS']= ':'.join(configDQ2['DQ2_OUTPUT_SPACE_TOKENS'])
             environment['DQ2_BACKUP_OUTPUT_LOCATIONS']= ':'.join(configDQ2['DQ2_BACKUP_OUTPUT_LOCATIONS'])
 
-        # CN: extra condition for TNTSplitter
-        if job._getRoot().splitter and job._getRoot().splitter._name == 'TNTJobSplitter':
-            # set up dq2 environment
-            datasetname = job.inputdata.dataset
-            environment['DATASETNAME']= ':'.join(datasetname) 
-            environment['DATASETLOCATION'] = ':'.join(job.inputdata.get_locations())
-            environment['DQ2_URL_SERVER']=configDQ2['DQ2_URL_SERVER']
-            environment['DQ2_URL_SERVER_SSL']=configDQ2['DQ2_URL_SERVER_SSL']
-            #environment['DATASETTYPE']=job.inputdata.type
-            # At present, DQ2 download is the only thing that works
-            environment['DATASETTYPE']="DQ2_DOWNLOAD"
-            if job.inputdata.accessprotocol:
-                 environment['DQ2_LOCAL_PROTOCOL'] = job.inputdata.accessprotocol
-            if job.inputsandbox: inputbox += job.inputsandbox   
-
         # Fix DATASETNAME env variable for DQ2_COPY mode
         if job.inputdata and job.inputdata._name in [ 'DQ2Dataset' ] and job.inputdata.type in [ 'DQ2_LOCAL', 'DQ2_COPY', 'FILE_STAGER' ]:
             if job.inputdata.dataset:
@@ -340,11 +336,6 @@ def prepare(self,app,appsubconfig,appmasterconfig,jobmasterconfig):
                     raise ApplicationConfigurationError(printout )
 
 
-        if job.inputdata and job.inputdata._name == 'ATLASTier3Dataset':
-            environment['DATASETTYPE'] = 'TIER3'
-
-
-            
         # USE_POOLFILECATALOG_FAILOVER of Local/ATLASLocalDataset
         if job.inputdata and job.inputdata._name == 'ATLASLocalDataset':
             if job.inputdata.use_poolfilecatalog_failover:
@@ -407,16 +398,9 @@ def master_prepare( self, app, appconfig ):
             jobid = "%d" % job.id
 
         # Generate output dataset name
-        if job.outputdata:
-            if job.outputdata._name=='DQ2OutputDataset':
-                dq2_datasetname = job.outputdata.datasetname
-                dq2_isGroupDS = job.outputdata.isGroupDS
-                dq2_groupname = job.outputdata.groupname
-            else:
-                dq2_datasetname = ''
-                dq2_isGroupDS = False
-                dq2_groupname = ''
-            self.output_datasetname, self.output_lfn = dq2outputdatasetname(dq2_datasetname, jobid, dq2_isGroupDS, dq2_groupname)
+        dq2_datasetname = ''
+        dq2_isGroupDS = False
+        dq2_groupname = ''
 
         # Expand Athena jobOptions
         if not app.option_file and not app.command_line:

python/GangaAtlas/Lib/Athena/DQ2JobSplitter.py

@@ -19,8 +19,8 @@
 from GangaAtlas.Lib.ATLASDataset.DQ2Dataset import *
 from Ganga.Utility.Config import getConfig, makeConfig, ConfigError
 
-from Ganga.GPIDev.Credentials_old import GridProxy
-gridProxy = GridProxy()
+#from Ganga.GPIDev.Credentials_old import GridProxy
+#gridProxy = GridProxy()
 
 logger = getLogger()