Enable parallel metp jobs and fix race condition with the gfscleanup …

…job (#2907)

This brings in a change to EMC_verif-global that offsets the start time
of parallel instances when running metp jobs. This prevents Python from
attempting to create the same directory in multiple instances.
Simultaneously, this also fixes an issue with the `gfscleanup` job
potentially running before the `metp` jobs.
Resolves #2906
Resolves #2899

parm/archive/enkf.yaml.j2

@@ -27,7 +27,7 @@ enkf:
            {% else %}
               {% set steps = ["eobs", "eupd"] %}
               {% for mem in range(1, nmem_ens + 1) %}
-                 {% do steps.append("eomg_mem{{ '%03d' % mem }}") %}
+                 {% do steps.append("eomg_mem" ~ '%03d' % mem) %}
               {% endfor %}
            {% endif %}
         {% endif %}

parm/config/gfs/config.resources

@@ -988,8 +988,8 @@ case ${step} in
     threads_per_task=1
     walltime_gdas="03:00:00"
     walltime_gfs="06:00:00"
-    ntasks=1
-    tasks_per_node=1
+    ntasks=4
+    tasks_per_node=4
     export memory="80G"
     ;;
 

workflow/rocoto/gfs_tasks.py

@@ -855,8 +855,8 @@ def fcst(self):
         try:
             task = fcst_map[self.app_config.mode]()
         except KeyError:
-            raise NotImplementedError(f'{self.app_config.mode} is not a valid type.\n' +
-                                      'Currently supported forecast types are:\n' +
+            raise NotImplementedError(f'{self.app_config.mode} is not a valid type.\n'
+                                      f'Currently supported forecast types are:\n'
                                       f'{" | ".join(fcst_map.keys())}')
 
         return task
@@ -2330,6 +2330,10 @@ def cleanup(self):
                     dep_dict = {'type': 'task', 'name': f'{self.run}npoess_pgrb2_0p5deg'}
                     deps.append(rocoto.add_dependency(dep_dict))
 
+        if self.app_config.do_metp and self.run in ['gfs']:
+            dep_dict = {'type': 'metatask', 'name': f'{self.run}metp'}
+            deps.append(rocoto.add_dependency(dep_dict))
+
         dependencies = rocoto.create_dependency(dep_condition='and', dep=deps)
 
         resources = self.get_resource('cleanup')