Save all files in a task at the same time to avoid recomputing intermediate results

doc/quickstart/configure.rst

@@ -269,6 +269,38 @@ For example, Python's ``None`` is YAML's ``null``, Python's ``True`` is YAML's
    will be downloaded; otherwise, local data will be used.
 
 
+.. _config-logging:
+
+Logging configuration
+=====================
+
+These options can be specified under the ``"logging"`` section in the
+configuration file.
+
+Example:
+
+.. code:: yaml
+
+  logging:
+    log_progress_interval: 10s
+
+will log progress with Dask computations every 10 seconds instead of showing a
+progress bar. If ``max_parallel_tasks`` is not equal to 1, progress is always
+logged.
+
+Available options:
+
++-------------------------------+----------------------------------------+-----------------------------+----------------------------------------+
+| Option                        | Description                            | Type                        | Default value                          |
++===============================+========================================+=============================+========================================+
+| ``log_progress_interval``     | When running computations with Dask,   | :obj:`str` or :obj:`float`  | 0                                      |
+|                               | log progress every                     |                             |                                        |
+|                               | ``log_progress_interval`` instead of   |                             |                                        |
+|                               | showing a progress bar. The value can  |                             |                                        |
+|                               | be specified in the format accepted by |                             |                                        |
+|                               | :func:`dask.utils.parse_timedelta`.    |                             |                                        |
++-------------------------------+----------------------------------------+-----------------------------+----------------------------------------+
+
 .. _config-dask:
 
 Dask configuration

environment.yml

@@ -40,6 +40,7 @@ dependencies:
   - python-stratify >=0.3
   - pyyaml
   - requests
+  - rich
   - scipy >=1.6
   - shapely >=2.0.0
   - yamale

esmvalcore/_recipe/recipe.py

@@ -220,7 +220,10 @@ def _get_default_settings(dataset):
     settings["remove_supplementary_variables"] = {}
 
     # Configure saving cubes to file
-    settings["save"] = {"compress": session["compress_netcdf"]}
+    settings["save"] = {
+        "compress": session["compress_netcdf"],
+        "compute": False,
+    }
     if facets["short_name"] != facets["original_short_name"]:
         settings["save"]["alias"] = facets["short_name"]
 
@@ -381,6 +384,8 @@ def _get_downstream_settings(step, order, products):
         if key in remaining_steps:
             if all(p.settings.get(key, object()) == value for p in products):
                 settings[key] = value
+    # Set the compute argument to the save step.
+    settings["save"] = {"compute": some_product.settings["save"]["compute"]}
     return settings
 
 

esmvalcore/config/_config_validators.py

@@ -332,6 +332,7 @@ def validate_extra_facets_dir(value):
     "exit_on_warning": validate_bool,
     "extra_facets_dir": validate_extra_facets_dir,
     "log_level": validate_string,
+    "logging": validate_dict,
     "max_datasets": validate_int_positive_or_none,
     "max_parallel_tasks": validate_int_or_none,
     "max_years": validate_int_positive_or_none,

esmvalcore/config/configurations/defaults/logging.yml

@@ -0,0 +1,2 @@
+logging:
+  log_progress_interval: 0.

esmvalcore/preprocessor/__init__.py

@@ -9,6 +9,7 @@
 from pprint import pformat
 from typing import Any, Iterable
 
+from dask.delayed import Delayed
 from iris.cube import Cube
 
 from .._provenance import TrackedFile
@@ -25,6 +26,7 @@
 )
 from ._compare_with_refs import bias, distance_metric
 from ._cycles import amplitude
+from ._dask_progress import _compute_with_progress
 from ._derive import derive
 from ._detrend import detrend
 from ._io import (
@@ -428,6 +430,9 @@ def preprocess(
                 )
             )
 
+    if step == "save":
+        return result
+
     items = []
     for item in result:
         if isinstance(item, (PreprocessorFile, Cube, str, Path)):
@@ -536,22 +541,24 @@ def cubes(self):
     def cubes(self, value):
         self._cubes = value
 
-    def save(self):
+    def save(self) -> Delayed | None:
         """Save cubes to disk."""
-        preprocess(
+        return preprocess(
             self._cubes,
             "save",
             input_files=self._input_files,
             **self.settings["save"],
-        )
+        )[0]
 
-    def close(self):
+    def close(self) -> Delayed | None:
         """Close the file."""
+        result = None
         if self._cubes is not None:
             self._update_attributes()
-            self.save()
+            result = self.save()
             self._cubes = None
             self.save_provenance()
+        return result
 
     def _update_attributes(self):
         """Update product attributes from cube metadata."""
@@ -693,7 +700,7 @@ def _initialize_products(self, products):
         for product in products:
             product.initialize_provenance(self.activity)
 
-    def _run(self, _):
+    def _run(self, _) -> list[str]:
         """Run the preprocessor."""
         self._initialize_product_provenance()
 
@@ -703,6 +710,7 @@ def _run(self, _):
         blocks = get_step_blocks(steps, self.order)
 
         saved = set()
+        delayeds = []
         for block in blocks:
             logger.debug("Running block %s", block)
             if block[0] in MULTI_MODEL_FUNCTIONS:
@@ -718,14 +726,19 @@ def _run(self, _):
                             product.apply(step, self.debug)
                     if block == blocks[-1]:
                         product.cubes  # noqa: B018 pylint: disable=pointless-statement
-                        product.close()
+                        delayed = product.close()
+                        delayeds.append(delayed)
                         saved.add(product.filename)
 
         for product in self.products:
             if product.filename not in saved:
                 product.cubes  # noqa: B018 pylint: disable=pointless-statement
-                product.close()
+                delayed = product.close()
+                delayeds.append(delayed)
 
+        logger.info("Computing and saving data for task %s", self.name)
+        delayeds = [d for d in delayeds if d is not None]
+        _compute_with_progress(delayeds, description=self.name)
         metadata_files = write_metadata(
             self.products, self.write_ncl_interface
         )