diff --git a/openquake/baselib/parallel.py b/openquake/baselib/parallel.py
index 9c91aa5cf6a1..e17d4879065a 100644
--- a/openquake/baselib/parallel.py
+++ b/openquake/baselib/parallel.py
@@ -1109,14 +1109,23 @@ def multispawn(func, allargs, nprocs=num_cores, logfinish=True,
func(*args)
return
tot = len(allargs)
+ allargs = allargs[::-1] # so that the first argument is submitted first
procs = {} # sentinel -> process
n = 1
while allargs:
- args = allargs.pop(0)
- name = names.pop(0) if names else None
+ args = allargs.pop()
+ name = names.pop() if names else None
proc = mp_context.Process(target=func, args=args, name=name)
proc.start()
procs[proc.sentinel] = proc
+ while len(procs) >= nprocs: # wait for something to finish
+ for finished in wait(procs):
+ procs[finished].join()
+ del procs[finished]
+ if logfinish:
+ logging.info('Finished job %s [%d of %d]', name, n, tot)
+ n += 1
+
while procs:
for finished in wait(procs):
name = procs[finished].name or ''
diff --git a/openquake/calculators/base.py b/openquake/calculators/base.py
index 62c1ee8c1684..d6d2877ee583 100644
--- a/openquake/calculators/base.py
+++ b/openquake/calculators/base.py
@@ -303,7 +303,7 @@ def check_precalc(self, precalc_mode):
(calc_mode, ok_mode, precalc_mode))
def run(self, pre_execute=True, concurrent_tasks=None, remove=False,
- shutdown=False, **kw):
+ shutdown=False, name='JOB', **kw):
"""
Run the calculation and return the exported outputs.
@@ -373,7 +373,9 @@ def run(self, pre_execute=True, concurrent_tasks=None, remove=False,
# removing in preclassical with multiFaultSources
# would break --hc which is reading the temp file
os.remove(self.datastore.tempname)
- return getattr(self, 'exported', {})
+ exportdic = getattr(self, 'exported', {})
+ exportdic[name] = name
+ return exportdic
def core_task(*args):
"""
diff --git a/openquake/engine/engine.py b/openquake/engine/engine.py
index 1f1e045ab851..5ff49d291c5b 100644
--- a/openquake/engine/engine.py
+++ b/openquake/engine/engine.py
@@ -218,7 +218,7 @@ def run_calc(log):
workflow = oqparam.calculation_mode == 'workflow'
set_concurrent_tasks_default(calc, 1 if workflow else 2)
t0 = time.time()
- calc.run(shutdown=True)
+ calc.run(shutdown=True, name=getattr(log, 'name', 'JOB{log.calc_id}'))
logging.info('Exposing the outputs to the database')
expose_outputs(calc.datastore)
calc.datastore.close()
@@ -338,12 +338,11 @@ def _run(jobctxs, job_id, nodes, sbatch, concurrent_jobs, notify_to):
w.WorkerMaster(job_id).send_jobs()
print('oq engine --show-log %d to see the progress' % job_id)
elif concurrent_jobs > 1:
- args = [(job,) for job in jobctxs]
- names = []
+ allargs = []
for job in jobctxs:
- name = f"{job.params['mosaic_model']}{job.calc_id}"
- names.append(name)
- parallel.multispawn(run_calc, args, concurrent_jobs, names=names)
+ job.name = f"{job.params['mosaic_model']}{job.calc_id}"
+ allargs.append((job,))
+ parallel.multispawn(run_calc, allargs, concurrent_jobs)
else:
for jobctx in jobctxs:
run_calc(jobctx)
@@ -535,9 +534,10 @@ def validate(self):
oqs.append(oq)
if 'classical' in oq.calculation_mode:
for oq in oqs[1:]:
+ ini = oq.inputs['job_ini']
if oq.retperiods != oqs[0].retperiods:
- raise NameError(
- f'Expected return_periods = {oqs[0].retperiods}, '
+ raise InvalidFile(
+ f'{ini}: expected return_periods = {oqs[0].retperiods}, '
f'got {oq.retperiods}')
if 'risk' in oq.calculation_mode or 'damage' in oq.calculation_mode:
for oq in oqs[1:]:
diff --git a/openquake/engine/tests/global_hazard_test.py b/openquake/engine/tests/global_hazard_test.py
new file mode 100644
index 000000000000..95e0c2f263fb
--- /dev/null
+++ b/openquake/engine/tests/global_hazard_test.py
@@ -0,0 +1,33 @@
+# -*- coding: utf-8 -*-
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+#
+# Copyright (C) 2024-2025, GEM Foundation
+#
+# OpenQuake is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# OpenQuake is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with OpenQuake. If not, see .
+
+import os
+from openquake.qa_tests_data import mosaic
+from openquake.commonlib.datastore import read
+from openquake.calculators.export import export
+from openquake.engine import global_hazard
+
+MOSAIC_DIR = os.path.dirname(mosaic.__file__)
+
+
+# TODO: enable the test
+def _test():
+ worflow_id = global_hazard.main(MOSAIC_DIR, 'job_vs30.ini')
+ dstore = read(worflow_id)
+ [fname] = export(('hmaps', 'csv'), dstore)
+ breakpoint()
diff --git a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/gmmLT.xml b/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/gmmLT.xml
deleted file mode 100644
index a8a5aa379845..000000000000
--- a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/gmmLT.xml
+++ /dev/null
@@ -1,77 +0,0 @@
-
-
-
-
-
- AbrahamsonEtAl2014
- 0.20000
-
-
- BooreEtAl2014
- 0.20000
-
-
- CampbellBozorgnia2014
- 0.20000
-
-
- ChiouYoungs2014
- 0.20000
-
-
- CauzziEtAl2014
- 0.20000
-
-
-
-
- NGAEastUSGSSeedFrankel
- 0.12500
-
-
- NGAEastUSGSSeedHA15
- 0.12500
-
-
- NGAEastUSGSSeedSP15
- 0.12500
-
-
- NGAEastUSGSSeedYA15
- 0.12500
-
-
- NGAEastUSGSSammons1
- 0.12500
-
-
- NGAEastUSGSSammons2
- 0.12500
-
-
- NGAEastUSGSSammons3
- 0.12500
-
-
- NGAEastUSGSSammons4
- 0.12500
-
-
-
-
- AbrahamsonGulerce2020SSlab
- 0.33000
-
-
- ParkerEtAl2020SSlab
- 0.33000
-
-
- KuehnEtAl2020SSlab
- 0.34000
-
-
-
-
-
-
diff --git a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/job_dos_vs30_760_small.ini b/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/job_dos_vs30_760_small.ini
deleted file mode 100644
index 9471c6b3cc17..000000000000
--- a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/job_dos_vs30_760_small.ini
+++ /dev/null
@@ -1,48 +0,0 @@
-[general]
-description = AELO Year 3 dos sites, CEA
-calculation_mode = classical
-random_seed = 23
-mosaic_model = CEA
-ps_grid_spacing = 0
-
-[geometry]
-site_model_file = sites_CEA_dos.csv
-sites = 76.953324 43.234055
-override_vs30 = 760
-
-[logic_tree]
-number_of_logic_tree_samples = 0
-
-[erf]
-rupture_mesh_spacing = 2.0
-width_of_mfd_bin = 0.1
-area_source_discretization = 10
-complex_fault_mesh_spacing = 10
-pointsource_distance = 200.0
-
-[calculation]
-source_model_logic_tree_file = ../../../../../../../HazardMosaic/test_models/cea24/in/ssmLT_little_test.xml
-gsim_logic_tree_file = ../gmmLT.xml
-investigation_time = 1
-intensity_measure_types_and_levels = {"PGA": logscale(0.005, 3.00, 25),
- "SA(0.2)": logscale(0.005, 9.00, 25),
- "SA(1.0)": logscale(0.005, 3.00, 25)}
-truncation_level = 3.0
-disagg_by_src = true
-maximum_distance = {'default': 200.}
-horiz_comp_to_geom_mean = true
-use_rates = true
-minimum_magnitude = 6.6
-
-[disaggregation]
-distance_bin_width = 20
-num_epsilon_bins = 10
-mag_bin_width = 0.1
-epsilon_star = true
-postproc_func = compute_rtgm.main
-asce_version = ASCE7-22
-
-[output]
-hazard_maps = true
-uniform_hazard_spectra = true
-poes = 0.000404 0.001025 0.002105 0.004453 0.013767
diff --git a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/sites_CEA_dos.csv b/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/sites_CEA_dos.csv
deleted file mode 100644
index 62c36eecdb8e..000000000000
--- a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/sites_CEA_dos.csv
+++ /dev/null
@@ -1,2 +0,0 @@
-custom_site_id,lon,lat,vs30,z1pt0,z2pt5,vs30measured
-10,76.953324,43.234055,800,31.070148646647844,0.5722405600686729,1