move batch_jobs to workflow_jobs

Author: jedwards4b

config/cesm/config_files.xml

@@ -51,6 +51,15 @@
     <schema>$CIMEROOT/config/xml_schemas/config_batch.xsd</schema>
   </entry>
 
+  <entry id="WORKFLOW_SPEC_FILE">
+    <type>char</type>
+    <default_value>$CIMEROOT/config/$MODEL/machines/config_workflow.xml</default_value>
+    <group>case_last</group>
+    <file>env_case.xml</file>
+    <desc>file containing workflow (for documentation only - DO NOT EDIT)</desc>
+    <schema>$CIMEROOT/config/xml_schemas/config_workflow.xsd</schema>
+  </entry>
+
   <entry id="INPUTDATA_SPEC_FILE">
     <type>char</type>
     <default_value>$CIMEROOT/config/$MODEL/config_inputdata.xml</default_value>

config/cesm/machines/config_batch.xml

@@ -1,5 +1,5 @@
 <?xml version="1.0"?>
-<config_batch version="2.0">
+<config_batch version="2.1">
   <!--
      File:    config_batch.xml
      Purpose: abstract out the parts of run scripts that are different, and use this configuration to
@@ -541,25 +541,4 @@
       <queue walltimemax="00:60:00" nodemin="1" nodemax="50" default="true">default</queue>
     </queues>
   </batch_system>
-
-  <batch_jobs>
-    <!-- order matters, with no-batch jobs will be run in the order listed here -->
-    <job name="case.run">
-      <template>template.case.run</template>
-      <prereq>$BUILD_COMPLETE and not $TEST</prereq>
-    </job>
-    <job name="case.test">
-      <template>template.case.test</template>
-      <prereq>$BUILD_COMPLETE and $TEST</prereq>
-    </job>
-    <job name="case.st_archive">
-      <template>template.st_archive</template>
-      <task_count>1</task_count>
-      <walltime>0:20:00</walltime>
-      <!-- If DOUT_S is true and case.run (or case.test) exits successfully then run st_archive-->
-      <dependency>case.run or case.test</dependency>
-      <prereq>$DOUT_S</prereq>
-    </job>
-  </batch_jobs>
-
 </config_batch>

config/config_headers.xml

@@ -10,6 +10,13 @@
     </header>
   </file>
 
+  <file name="env_workflow.xml">
+    <header>
+      These variables may be changed anytime during a run, they
+      control jobs that will be submitted and their dependancies.
+    </header>
+  </file>
+
   <file name="env_case.xml">
     <header>
     These variables CANNOT BE CHANGED once a case has been created.

config/xml_schemas/config_batch.xsd

@@ -24,7 +24,7 @@
   <xs:element name="batch_mail_type_flag" type="xs:string"/>
   <xs:element name="batch_mail_type" type="xs:string"/>
   <xs:element name="batch_mail_default" type="xs:string"/>
-  <xs:element name="template" type="xs:NCName"/>
+  <xs:element name="template" type="xs:anyURI"/>
   <xs:element name="task_count" type="xs:integer"/>
   <xs:element name="walltime" type="xs:string"/>
   <xs:element name="dependency" type="xs:string"/>
@@ -37,7 +37,7 @@
     <xs:complexType>
       <xs:sequence>
         <xs:element ref="batch_system" maxOccurs="unbounded" />
-        <xs:element ref="batch_jobs"/>
+        <xs:element ref="batch_jobs" minOccurs="0" maxOccurs="1"/>
       </xs:sequence>
       <xs:attribute ref="version" use="required"/>
     </xs:complexType>

scripts/lib/CIME/XML/batch.py

@@ -112,14 +112,15 @@ def get_batch_jobs(self):
         and the second a dict of qualifiers for the job
         """
         jobs = []
-        bnode = self.get_child("batch_jobs")
-        for jnode in self.get_children(root=bnode):
-            if self.name(jnode) == "job":
-                name = self.get(jnode, "name")
-                jdict = {}
-                for child in self.get_children(root=jnode):
-                    jdict[self.name(child)] = self.text(child)
-
-            jobs.append((name, jdict))
+        bnode = self.get_optional_child("batch_jobs")
+        if bnode:
+            for jnode in self.get_children(root=bnode):
+                if self.name(jnode) == "job":
+                    name = self.get(jnode, "name")
+                    jdict = {}
+                    for child in self.get_children(root=jnode):
+                        jdict[self.name(child)] = self.text(child)
+
+                    jobs.append((name, jdict))
 
         return jobs

scripts/lib/CIME/XML/env_batch.py

@@ -15,14 +15,15 @@
 
 class EnvBatch(EnvBase):
 
-    def __init__(self, case_root=None, infile="env_batch.xml"):
+    def __init__(self, case_root=None, infile="env_batch.xml", schema=None):
         """
         initialize an object interface to file env_batch.xml in the case directory
         """
         self._batchtype = None
         # This arbitrary setting should always be overwritten
         self._default_walltime = "00:20:00"
-        schema = os.path.join(get_cime_root(), "config", "xml_schemas", "env_batch.xsd")
+        if not schema:
+            schema = os.path.join(get_cime_root(), "config", "xml_schemas", "env_batch.xsd")
         super(EnvBatch,self).__init__(case_root, infile, schema=schema)
 
     # pylint: disable=arguments-differ
@@ -32,21 +33,6 @@ def set_value(self, item, value, subgroup=None, ignore_type=False):
         """
         val = None
 
-        if item == "JOB_WALLCLOCK_TIME":
-            #Most systems use %H:%M:%S format for wallclock but LSF
-            #uses %H:%M this code corrects the value passed in to be
-            #the correct format - if we find we have more exceptions
-            #than this we may need to generalize this further
-            walltime_format = self.get_value("walltime_format", subgroup=None)
-            if walltime_format is not None and walltime_format.count(":") != value.count(":"): # pylint: disable=maybe-no-member
-                if value.count(":") == 1:
-                    t_spec = "%H:%M"
-                elif value.count(":") == 2:
-                    t_spec = "%H:%M:%S"
-                else:
-                    expect(False, "could not interpret format for wallclock time {}".format(value))
-                value = format_time(walltime_format, t_spec, value)
-
         if item == "JOB_QUEUE":
             expect(value in self._get_all_queue_names() or ignore_type,
                    "Unknown Job Queue specified use --force to set")
@@ -214,14 +200,16 @@ def make_batch_script(self, input_template, job, case, outfile=None):
         # make sure batch script is exectuble
         os.chmod(output_name, os.stat(output_name).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
 
-    def set_job_defaults(self, batch_jobs, case):
+    def set_job_defaults(self, batch_jobs, case, env_workflow=None):
         if self._batchtype is None:
             self._batchtype = self.get_batch_system_type()
 
         if self._batchtype == "none":
             return
-
-        known_jobs = self.get_jobs()
+        if env_workflow:
+            known_jobs = env_workflow.get_jobs()
+        else:
+            known_jobs = self.get_jobs()
 
         for job, jsect in batch_jobs:
             if job not in known_jobs:
@@ -281,9 +269,12 @@ def set_job_defaults(self, batch_jobs, case):
                     walltime = specs[3]
 
                 walltime = self._default_walltime if walltime is None else walltime # last-chance fallback
-
-            self.set_value("JOB_QUEUE", queue, subgroup=job, ignore_type=specs is None)
-            self.set_value("JOB_WALLCLOCK_TIME", walltime, subgroup=job)
+            if env_workflow:
+                env_workflow.set_value("JOB_QUEUE", queue, subgroup=job, ignore_type=specs is None)
+                env_workflow.set_value("JOB_WALLCLOCK_TIME", walltime, subgroup=job)
+            else:
+                self.set_value("JOB_QUEUE", queue, subgroup=job, ignore_type=specs is None)
+                self.set_value("JOB_WALLCLOCK_TIME", walltime, subgroup=job)
             logger.debug("Job {} queue {} walltime {}".format(job, queue, walltime))
 
     def _match_attribs(self, attribs, case, queue):
@@ -426,8 +417,11 @@ def get_submit_args(self, case, job):
 
     def submit_jobs(self, case, no_batch=False, job=None, user_prereq=None, skip_pnl=False,
                     allow_fail=False, resubmit_immediate=False, mail_user=None, mail_type=None,
-                    batch_args=None, dry_run=False):
-        alljobs = self.get_jobs()
+                    batch_args=None, dry_run=False, runwithcylc=False):
+        if runwithcylc:
+            alljobs = [job]
+        else:
+            alljobs = self.get_jobs()
         startindex = 0
         jobs = []
         firstjob = job
@@ -439,15 +433,20 @@ def submit_jobs(self, case, no_batch=False, job=None, user_prereq=None, skip_pnl
             logger.debug( "Index {:d} job {} startindex {:d}".format(index, job, startindex))
             if index < startindex:
                 continue
-            try:
-                prereq = self.get_value('prereq', subgroup=job, resolved=False)
-                if prereq is None or job == firstjob or (dry_run and prereq == "$BUILD_COMPLETE"):
-                    prereq = True
-                else:
-                    prereq = case.get_resolved_value(prereq)
-                    prereq = eval(prereq)
-            except:
-                expect(False,"Unable to evaluate prereq expression '{}' for job '{}'".format(self.get_value('prereq',subgroup=job), job))
+            jobfile = get_batch_script_for_job(job)
+            if not os.path.isfile(jobfile):
+                logger.info("Jobfile {} not found in case, skipping job step".format(jobfile))
+                prereq = False
+            else:
+                try:
+                    prereq = self.get_value('prereq', subgroup=job, resolved=False)
+                    if prereq is None or job == firstjob or (dry_run and prereq == "$BUILD_COMPLETE"):
+                        prereq = True
+                    else:
+                        prereq = case.get_resolved_value(prereq)
+                        prereq = eval(prereq)
+                except:
+                    expect(False,"Unable to evaluate prereq expression '{}' for job '{}'".format(self.get_value('prereq',subgroup=job), job))
 
             if prereq:
                 jobs.append((job, self.get_value('dependency', subgroup=job)))
@@ -489,7 +488,7 @@ def submit_jobs(self, case, no_batch=False, job=None, user_prereq=None, skip_pnl
                                                  resubmit_immediate=resubmit_immediate,
                                                  dep_jobs=dep_jobs,
                                                  allow_fail=allow_fail,
-                                                 no_batch=no_batch,
+                                                 no_batch=no_batch or runwithcylc,
                                                  mail_user=mail_user,
                                                  mail_type=mail_type,
                                                  batch_args=batch_args,
@@ -874,6 +873,14 @@ def make_all_batch_files(self, case):
         machdir  = case.get_value("MACHDIR")
         logger.info("Creating batch scripts")
         for job in self.get_jobs():
-            input_batch_script = os.path.join(machdir,self.get_value('template', subgroup=job))
-            logger.info("Writing {} script from input template {}".format(job, input_batch_script))
-            self.make_batch_script(input_batch_script, job, case)
+            template = self.get_value('template', subgroup=job)
+            template = template.replace('$CASEROOT', self._caseroot)
+            if os.path.isabs(template):
+                input_batch_script = template
+            else:
+                input_batch_script = os.path.join(machdir,template)
+            if os.path.isfile(input_batch_script):
+                logger.info("Writing {} script from input template {}".format(job, input_batch_script))
+                self.make_batch_script(input_batch_script, job, case)
+            else:
+                logger.warning("Input template file {} for job {} does not exist or cannot be read.".format(input_batch_script, job))

scripts/lib/CIME/XML/generic_xml.py

@@ -332,7 +332,7 @@ def scan_optional_child(self, nodename, attributes=None, root=None):
         """
         nodes = self.scan_children(nodename, attributes=attributes, root=root)
 
-        expect(len(nodes) <= 1, "Multiple matches for nodename '{}' and attrs '{}' in file '{}'".format(nodename, attributes, self.filename))
+        expect(len(nodes) <= 1, "Multiple matches for nodename '{}' and attrs '{}' in file '{}', found {} matches".format(nodename, attributes, self.filename, len(nodes)))
         return nodes[0] if nodes else None
 
     def scan_children(self, nodename, attributes=None, root=None):

scripts/lib/CIME/case/case.py

@@ -22,6 +22,7 @@
 from CIME.XML.compsets              import Compsets
 from CIME.XML.grids                 import Grids
 from CIME.XML.batch                 import Batch
+from CIME.XML.workflow              import Workflow
 from CIME.XML.pio                   import PIO
 from CIME.XML.archive               import Archive
 from CIME.XML.env_test              import EnvTest
@@ -32,6 +33,7 @@
 from CIME.XML.env_run               import EnvRun
 from CIME.XML.env_archive           import EnvArchive
 from CIME.XML.env_batch             import EnvBatch
+from CIME.XML.env_workflow          import EnvWorkflow
 from CIME.XML.generic_xml           import GenericXML
 from CIME.user_mod_support          import apply_user_mods
 from CIME.aprun import get_aprun_cmd_for_case
@@ -201,6 +203,7 @@ def read_xml(self):
         self._env_entryid_files.append(EnvBuild(self._caseroot, components=components))
         self._env_entryid_files.append(EnvMachPes(self._caseroot, components=components))
         self._env_entryid_files.append(EnvBatch(self._caseroot))
+        self._env_entryid_files.append(EnvWorkflow(self._caseroot))
         if os.path.isfile(os.path.join(self._caseroot,"env_test.xml")):
             self._env_entryid_files.append(EnvTest(self._caseroot, components=components))
         self._env_generic_files = []
@@ -939,19 +942,28 @@ def configure(self, compset_name, grid_name, machine_name=None,
 
         batch_system_type = machobj.get_value("BATCH_SYSTEM")
         logger.info("Batch_system_type is {}".format(batch_system_type))
-        batch = Batch(batch_system=batch_system_type, machine=machine_name)
+        batch = Batch(batch_system=batch_system_type, machine=machine_name, files=files)
         bjobs = batch.get_batch_jobs()
+        # Workflow is optional in cime5.6 to support backward compatibility
+        # it was originally part of the config_batch/env_batch file.
+        workflow = None
+        if not bjobs:
+            workflow = Workflow(files=files)
+            bjobs = workflow.get_workflow_jobs()
+            env_workflow = self.get_env("workflow")
 
         env_batch.set_batch_system(batch, batch_system_type=batch_system_type)
-        env_batch.create_job_groups(bjobs, test)
+        if workflow:
+            env_workflow.create_job_groups(bjobs, test)
+        else:
+            env_batch.create_job_groups(bjobs, test)
+        env_batch.set_job_defaults(bjobs, self)
 
         if walltime:
             self.set_value("USER_REQUESTED_WALLTIME", walltime, subgroup=self.get_primary_job())
         if queue:
             self.set_value("USER_REQUESTED_QUEUE", queue, subgroup=self.get_primary_job())
 
-        env_batch.set_job_defaults(bjobs, self)
-
         # Make sure that parallel IO is not specified if total_tasks==1
         if self.total_tasks == 1:
             for compclass in self._component_classes:
@@ -1364,6 +1376,8 @@ def set_file(self, xmlfile):
                     new_env_file = EnvMachPes(infile=xmlfile, components=components)
                 elif ftype == "env_batch.xml":
                     new_env_file = EnvBatch(infile=xmlfile)
+                elif ftype == "env_workflow.xml":
+                    new_env_file = EnvWorkflow(infile=xmlfile)
                 elif ftype == "env_test.xml":
                     new_env_file = EnvTest(infile=xmlfile)
             if new_env_file is not None:

src/drivers/mct/cime_config/config_component.xml

@@ -2607,64 +2607,6 @@
     <desc>External script to be run after model completion</desc>
   </entry>
 
- <!-- Job Submission stuff -->
- <entry id="USER_REQUESTED_QUEUE">
-   <type>char</type>
-   <group>job_submission</group>
-   <file>env_batch.xml</file>
-   <desc>Store user override for queue</desc>
- </entry>
-
- <entry id="USER_REQUESTED_WALLTIME">
-   <type>char</type>
-   <group>job_submission</group>
-   <file>env_batch.xml</file>
-   <desc>Store user override for walltime</desc>
- </entry>
-
- <entry id="JOB_QUEUE">
-   <type>char</type>
-   <valid_values></valid_values>
-   <default_value></default_value>
-   <group>job_submission</group>
-   <file>env_batch.xml</file>
-   <desc>The machine queue in which to submit the job.  Default determined in config_machines.xml can be overwritten by testing</desc>
- </entry>
-
- <entry id="JOB_WALLCLOCK_TIME">
-   <type>char</type>
-   <valid_values></valid_values>
-   <default_value></default_value>
-   <group>job_submission</group>
-   <file>env_batch.xml</file>
-   <desc>The machine wallclock setting.  Default determined in config_machines.xml can be overwritten by testing</desc>
- </entry>
-
-  <entry id="BATCH_COMMAND_FLAGS">
-    <type>char</type>
-    <valid_values></valid_values>
-    <default_value></default_value>
-    <group>job_submission</group>
-    <file>env_batch.xml</file>
-    <desc>Override the batch submit command this job. Do not include executable or dependencies</desc>
-  </entry>
-
- <entry id="PROJECT">
-   <type>char</type>
-   <default_value></default_value>
-   <group>job_submission</group>
-   <file>env_batch.xml</file>
-   <desc>project for project-sensitive build and run paths, and job scripts</desc>
- </entry>
-
- <entry id="CHARGE_ACCOUNT">
-   <type>char</type>
-   <default_value></default_value>
-   <group>job_submission</group>
-   <file>env_batch.xml</file>
-   <desc>project to charge in scripts if different from PROJECT</desc>
- </entry>
-
  <entry id="MODEL_VERSION">
    <type>char</type>
    <default_value>unknown</default_value>