Repository: ungarj/mapchete (language: python)

Issue:
decide intelligently whether to use task graph processing
Processing using a task graph is only really useful if there are clear dependencies between task layers (preprocessing tasks -> tiles; tiles -> overview tiles, ...). In some cases tasks can be run independently from each other. In such cases using dask and "streaming" tasks to scheduler would provide a more stable process execution on a cluster.


Discussion / hints:

Fix patch:
diff --git a/doc/source/apidoc/mapchete.cli.progress_bar.rst b/doc/source/apidoc/mapchete.cli.progress_bar.rst
new file mode 100644
index 00000000..328a5d3f
--- /dev/null
+++ b/doc/source/apidoc/mapchete.cli.progress_bar.rst
@@ -0,0 +1,7 @@
+mapchete.cli.progress\_bar module
+=================================
+
+.. automodule:: mapchete.cli.progress_bar
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.cli.rst b/doc/source/apidoc/mapchete.cli.rst
index 1685ca23..f27eb18c 100644
--- a/doc/source/apidoc/mapchete.cli.rst
+++ b/doc/source/apidoc/mapchete.cli.rst
@@ -17,6 +17,7 @@ Submodules
 
    mapchete.cli.main
    mapchete.cli.options
+   mapchete.cli.progress_bar
 
 Module contents
 ---------------
diff --git a/doc/source/apidoc/mapchete.commands.observer.rst b/doc/source/apidoc/mapchete.commands.observer.rst
new file mode 100644
index 00000000..1d8c4932
--- /dev/null
+++ b/doc/source/apidoc/mapchete.commands.observer.rst
@@ -0,0 +1,7 @@
+mapchete.commands.observer module
+=================================
+
+.. automodule:: mapchete.commands.observer
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.commands.rst b/doc/source/apidoc/mapchete.commands.rst
index e0bbc757..8a8766e6 100644
--- a/doc/source/apidoc/mapchete.commands.rst
+++ b/doc/source/apidoc/mapchete.commands.rst
@@ -1,6 +1,14 @@
 mapchete.commands package
 =========================
 
+Submodules
+----------
+
+.. toctree::
+   :maxdepth: 4
+
+   mapchete.commands.observer
+
 Module contents
 ---------------
 
diff --git a/doc/source/apidoc/mapchete.io.settings.rst b/doc/source/apidoc/mapchete.config.base.rst
similarity index 56%
rename from doc/source/apidoc/mapchete.io.settings.rst
rename to doc/source/apidoc/mapchete.config.base.rst
index 7c9123c6..5e1ecc19 100644
--- a/doc/source/apidoc/mapchete.io.settings.rst
+++ b/doc/source/apidoc/mapchete.config.base.rst
@@ -1,7 +1,7 @@
-mapchete.io.settings module
+mapchete.config.base module
 ===========================
 
-.. automodule:: mapchete.io.settings
+.. automodule:: mapchete.config.base
    :members:
    :undoc-members:
    :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.config.models.rst b/doc/source/apidoc/mapchete.config.models.rst
new file mode 100644
index 00000000..b2d92c99
--- /dev/null
+++ b/doc/source/apidoc/mapchete.config.models.rst
@@ -0,0 +1,7 @@
+mapchete.config.models module
+=============================
+
+.. automodule:: mapchete.config.models
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.config.parse.rst b/doc/source/apidoc/mapchete.config.parse.rst
new file mode 100644
index 00000000..0c9b9f82
--- /dev/null
+++ b/doc/source/apidoc/mapchete.config.parse.rst
@@ -0,0 +1,7 @@
+mapchete.config.parse module
+============================
+
+.. automodule:: mapchete.config.parse
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.config.process_func.rst b/doc/source/apidoc/mapchete.config.process_func.rst
new file mode 100644
index 00000000..52f8877d
--- /dev/null
+++ b/doc/source/apidoc/mapchete.config.process_func.rst
@@ -0,0 +1,7 @@
+mapchete.config.process\_func module
+====================================
+
+.. automodule:: mapchete.config.process_func
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.config.rst b/doc/source/apidoc/mapchete.config.rst
index 1e0562d2..d36a26d2 100644
--- a/doc/source/apidoc/mapchete.config.rst
+++ b/doc/source/apidoc/mapchete.config.rst
@@ -1,5 +1,19 @@
-mapchete.config module
-======================
+mapchete.config package
+=======================
+
+Submodules
+----------
+
+.. toctree::
+   :maxdepth: 4
+
+   mapchete.config.base
+   mapchete.config.models
+   mapchete.config.parse
+   mapchete.config.process_func
+
+Module contents
+---------------
 
 .. automodule:: mapchete.config
    :members:
diff --git a/doc/source/apidoc/mapchete.enums.rst b/doc/source/apidoc/mapchete.enums.rst
new file mode 100644
index 00000000..d4e885db
--- /dev/null
+++ b/doc/source/apidoc/mapchete.enums.rst
@@ -0,0 +1,7 @@
+mapchete.enums module
+=====================
+
+.. automodule:: mapchete.enums
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.executor.base.rst b/doc/source/apidoc/mapchete.executor.base.rst
new file mode 100644
index 00000000..ddaa5d02
--- /dev/null
+++ b/doc/source/apidoc/mapchete.executor.base.rst
@@ -0,0 +1,7 @@
+mapchete.executor.base module
+=============================
+
+.. automodule:: mapchete.executor.base
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.executor.concurrent_futures.rst b/doc/source/apidoc/mapchete.executor.concurrent_futures.rst
new file mode 100644
index 00000000..6e7f8803
--- /dev/null
+++ b/doc/source/apidoc/mapchete.executor.concurrent_futures.rst
@@ -0,0 +1,7 @@
+mapchete.executor.concurrent\_futures module
+============================================
+
+.. automodule:: mapchete.executor.concurrent_futures
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.executor.dask.rst b/doc/source/apidoc/mapchete.executor.dask.rst
new file mode 100644
index 00000000..4f066e46
--- /dev/null
+++ b/doc/source/apidoc/mapchete.executor.dask.rst
@@ -0,0 +1,7 @@
+mapchete.executor.dask module
+=============================
+
+.. automodule:: mapchete.executor.dask
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.executor.future.rst b/doc/source/apidoc/mapchete.executor.future.rst
new file mode 100644
index 00000000..ee50802e
--- /dev/null
+++ b/doc/source/apidoc/mapchete.executor.future.rst
@@ -0,0 +1,7 @@
+mapchete.executor.future module
+===============================
+
+.. automodule:: mapchete.executor.future
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.executor.rst b/doc/source/apidoc/mapchete.executor.rst
new file mode 100644
index 00000000..5045ac27
--- /dev/null
+++ b/doc/source/apidoc/mapchete.executor.rst
@@ -0,0 +1,23 @@
+mapchete.executor package
+=========================
+
+Submodules
+----------
+
+.. toctree::
+   :maxdepth: 4
+
+   mapchete.executor.base
+   mapchete.executor.concurrent_futures
+   mapchete.executor.dask
+   mapchete.executor.future
+   mapchete.executor.sequential
+   mapchete.executor.types
+
+Module contents
+---------------
+
+.. automodule:: mapchete.executor
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.executor.sequential.rst b/doc/source/apidoc/mapchete.executor.sequential.rst
new file mode 100644
index 00000000..3f045de5
--- /dev/null
+++ b/doc/source/apidoc/mapchete.executor.sequential.rst
@@ -0,0 +1,7 @@
+mapchete.executor.sequential module
+===================================
+
+.. automodule:: mapchete.executor.sequential
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.executor.types.rst b/doc/source/apidoc/mapchete.executor.types.rst
new file mode 100644
index 00000000..750d63c3
--- /dev/null
+++ b/doc/source/apidoc/mapchete.executor.types.rst
@@ -0,0 +1,7 @@
+mapchete.executor.types module
+==============================
+
+.. automodule:: mapchete.executor.types
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.io.profiles.rst b/doc/source/apidoc/mapchete.io.profiles.rst
new file mode 100644
index 00000000..95acb120
--- /dev/null
+++ b/doc/source/apidoc/mapchete.io.profiles.rst
@@ -0,0 +1,7 @@
+mapchete.io.profiles module
+===========================
+
+.. automodule:: mapchete.io.profiles
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.io.rst b/doc/source/apidoc/mapchete.io.rst
index f8e6dfcd..1a85f75f 100644
--- a/doc/source/apidoc/mapchete.io.rst
+++ b/doc/source/apidoc/mapchete.io.rst
@@ -7,8 +7,8 @@ Submodules
 .. toctree::
    :maxdepth: 4
 
+   mapchete.io.profiles
    mapchete.io.raster
-   mapchete.io.settings
    mapchete.io.vector
 
 Module contents
diff --git a/doc/source/apidoc/mapchete.processing.base.rst b/doc/source/apidoc/mapchete.processing.base.rst
new file mode 100644
index 00000000..25df8328
--- /dev/null
+++ b/doc/source/apidoc/mapchete.processing.base.rst
@@ -0,0 +1,7 @@
+mapchete.processing.base module
+===============================
+
+.. automodule:: mapchete.processing.base
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.processing.execute.rst b/doc/source/apidoc/mapchete.processing.execute.rst
new file mode 100644
index 00000000..394b17fd
--- /dev/null
+++ b/doc/source/apidoc/mapchete.processing.execute.rst
@@ -0,0 +1,7 @@
+mapchete.processing.execute module
+==================================
+
+.. automodule:: mapchete.processing.execute
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.processing.mp.rst b/doc/source/apidoc/mapchete.processing.mp.rst
new file mode 100644
index 00000000..1c9bbc01
--- /dev/null
+++ b/doc/source/apidoc/mapchete.processing.mp.rst
@@ -0,0 +1,7 @@
+mapchete.processing.mp module
+=============================
+
+.. automodule:: mapchete.processing.mp
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.processing.profilers.memory.rst b/doc/source/apidoc/mapchete.processing.profilers.memory.rst
new file mode 100644
index 00000000..805f6100
--- /dev/null
+++ b/doc/source/apidoc/mapchete.processing.profilers.memory.rst
@@ -0,0 +1,7 @@
+mapchete.processing.profilers.memory module
+===========================================
+
+.. automodule:: mapchete.processing.profilers.memory
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.processing.profilers.requests.rst b/doc/source/apidoc/mapchete.processing.profilers.requests.rst
new file mode 100644
index 00000000..d21f7da4
--- /dev/null
+++ b/doc/source/apidoc/mapchete.processing.profilers.requests.rst
@@ -0,0 +1,7 @@
+mapchete.processing.profilers.requests module
+=============================================
+
+.. automodule:: mapchete.processing.profilers.requests
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.processing.profilers.rst b/doc/source/apidoc/mapchete.processing.profilers.rst
new file mode 100644
index 00000000..0725e1f4
--- /dev/null
+++ b/doc/source/apidoc/mapchete.processing.profilers.rst
@@ -0,0 +1,20 @@
+mapchete.processing.profilers package
+=====================================
+
+Submodules
+----------
+
+.. toctree::
+   :maxdepth: 4
+
+   mapchete.processing.profilers.memory
+   mapchete.processing.profilers.requests
+   mapchete.processing.profilers.time
+
+Module contents
+---------------
+
+.. automodule:: mapchete.processing.profilers
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.processing.profilers.time.rst b/doc/source/apidoc/mapchete.processing.profilers.time.rst
new file mode 100644
index 00000000..8285df82
--- /dev/null
+++ b/doc/source/apidoc/mapchete.processing.profilers.time.rst
@@ -0,0 +1,7 @@
+mapchete.processing.profilers.time module
+=========================================
+
+.. automodule:: mapchete.processing.profilers.time
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.processing.rst b/doc/source/apidoc/mapchete.processing.rst
new file mode 100644
index 00000000..5ae675a3
--- /dev/null
+++ b/doc/source/apidoc/mapchete.processing.rst
@@ -0,0 +1,30 @@
+mapchete.processing package
+===========================
+
+Subpackages
+-----------
+
+.. toctree::
+   :maxdepth: 4
+
+   mapchete.processing.profilers
+
+Submodules
+----------
+
+.. toctree::
+   :maxdepth: 4
+
+   mapchete.processing.base
+   mapchete.processing.execute
+   mapchete.processing.mp
+   mapchete.processing.tasks
+   mapchete.processing.types
+
+Module contents
+---------------
+
+.. automodule:: mapchete.processing
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.processing.tasks.rst b/doc/source/apidoc/mapchete.processing.tasks.rst
new file mode 100644
index 00000000..aa78ecd6
--- /dev/null
+++ b/doc/source/apidoc/mapchete.processing.tasks.rst
@@ -0,0 +1,7 @@
+mapchete.processing.tasks module
+================================
+
+.. automodule:: mapchete.processing.tasks
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.processing.types.rst b/doc/source/apidoc/mapchete.processing.types.rst
new file mode 100644
index 00000000..ce616374
--- /dev/null
+++ b/doc/source/apidoc/mapchete.processing.types.rst
@@ -0,0 +1,7 @@
+mapchete.processing.types module
+================================
+
+.. automodule:: mapchete.processing.types
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.protocols.rst b/doc/source/apidoc/mapchete.protocols.rst
new file mode 100644
index 00000000..dc813c6b
--- /dev/null
+++ b/doc/source/apidoc/mapchete.protocols.rst
@@ -0,0 +1,7 @@
+mapchete.protocols module
+=========================
+
+.. automodule:: mapchete.protocols
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.registered.rst b/doc/source/apidoc/mapchete.registered.rst
new file mode 100644
index 00000000..3fc2fef4
--- /dev/null
+++ b/doc/source/apidoc/mapchete.registered.rst
@@ -0,0 +1,7 @@
+mapchete.registered module
+==========================
+
+.. automodule:: mapchete.registered
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.rst b/doc/source/apidoc/mapchete.rst
index ed1b1f3c..df650439 100644
--- a/doc/source/apidoc/mapchete.rst
+++ b/doc/source/apidoc/mapchete.rst
@@ -10,9 +10,12 @@ Subpackages
    mapchete.cli
    mapchete.commands
    mapchete.commons
+   mapchete.config
+   mapchete.executor
    mapchete.formats
    mapchete.io
    mapchete.processes
+   mapchete.processing
 
 Submodules
 ----------
@@ -20,14 +23,18 @@ Submodules
 .. toctree::
    :maxdepth: 4
 
-   mapchete.config
+   mapchete.enums
    mapchete.errors
    mapchete.index
    mapchete.log
    mapchete.path
+   mapchete.protocols
+   mapchete.registered
+   mapchete.settings
    mapchete.stac
    mapchete.testing
    mapchete.tile
+   mapchete.timer
    mapchete.types
    mapchete.validate
 
diff --git a/doc/source/apidoc/mapchete.settings.rst b/doc/source/apidoc/mapchete.settings.rst
new file mode 100644
index 00000000..5901551d
--- /dev/null
+++ b/doc/source/apidoc/mapchete.settings.rst
@@ -0,0 +1,7 @@
+mapchete.settings module
+========================
+
+.. automodule:: mapchete.settings
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/apidoc/mapchete.timer.rst b/doc/source/apidoc/mapchete.timer.rst
new file mode 100644
index 00000000..53ff320b
--- /dev/null
+++ b/doc/source/apidoc/mapchete.timer.rst
@@ -0,0 +1,7 @@
+mapchete.timer module
+=====================
+
+.. automodule:: mapchete.timer
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/doc/source/conf.py b/doc/source/conf.py
index ada12f7d..aca04f13 100644
--- a/doc/source/conf.py
+++ b/doc/source/conf.py
@@ -67,7 +67,7 @@
 #
 # This is also used if you do content translation via gettext catalogs.
 # Usually you set "language" from the command line for these cases.
-language = None
+language = "en"
 
 # List of patterns, relative to source directory, that match files and
 # directories to ignore when looking for source files.
@@ -97,7 +97,7 @@
 # Add any paths that contain custom static files (such as style sheets) here,
 # relative to this directory. They are copied after the builtin static files,
 # so a file named "default.css" will overwrite the builtin "default.css".
-html_static_path = ["nstatic"]
+# html_static_path = ["nstatic"]
 
 
 # -- Options for HTMLHelp output ------------------------------------------
@@ -166,8 +166,11 @@
     "click",
     "click_plugins",
     "click_spinner",
+    "dask",
+    "distributed",
     "fiona",
     "flask",
+    "fsspec",
     "lxml",
     "matplotlib",
     "matplotlib.pyplot",
@@ -197,6 +200,7 @@
     "shapely.geos",
     "shapely.ops",
     "shapely.wkt",
+    "s3fs",
     "tilematrix",
     "tqdm",
 ]
diff --git a/mapchete/__init__.py b/mapchete/__init__.py
index 04795971..5e7bc0cc 100644
--- a/mapchete/__init__.py
+++ b/mapchete/__init__.py
@@ -8,7 +8,7 @@
 from mapchete.executor import Executor, MFuture
 from mapchete.formats import read_output_metadata
 from mapchete.path import MPath, fs_from_path
-from mapchete.processing import Job, Mapchete, MapcheteProcess
+from mapchete.processing import Mapchete, MapcheteProcess
 from mapchete.tile import count_tiles
 from mapchete.timer import Timer
 from mapchete.types import MPathLike
@@ -20,7 +20,6 @@
     "Timer",
     "Executor",
     "MFuture",
-    "Job",
 ]
 __version__ = "2023.11.0"
 
diff --git a/mapchete/cli/default/convert.py b/mapchete/cli/default/convert.py
index 327a1271..40ccc572 100644
--- a/mapchete/cli/default/convert.py
+++ b/mapchete/cli/default/convert.py
@@ -1,5 +1,4 @@
 import click
-import tilematrix
 import tqdm
 from rasterio.dtypes import dtype_ranges
 from rasterio.enums import Resampling
@@ -8,6 +7,8 @@
 import mapchete
 from mapchete import commands
 from mapchete.cli import options
+from mapchete.cli.progress_bar import PBar
+from mapchete.config import DaskSettings
 from mapchete.formats import available_output_formats
 
 OUTPUT_FORMATS = available_output_formats()
@@ -96,7 +97,6 @@ def _validate_bidx(ctx, param, bidx):
 @options.opt_no_pbar
 @options.opt_debug
 @options.opt_workers
-@options.opt_multi
 @options.opt_concurrency
 @options.opt_dask_no_task_graph
 @options.opt_logfile
@@ -118,41 +118,27 @@ def convert(
     **kwargs,
 ):
     with mapchete.Timer() as t:
-        job = commands.convert(
-            tiledir,
-            output,
-            *args,
-            as_iterator=True,
-            dask_compute_graph=not dask_no_task_graph,
-            msg_callback=tqdm.tqdm.write if verbose else None,
-            **kwargs,
-        )
-        if not len(job):
-            return
-        list(
-            tqdm.tqdm(
-                job,
-                unit="task",
-                disable=debug or no_pbar,
+        with PBar(
+            total=100, desc="tasks", disable=debug or no_pbar, print_messages=verbose
+        ) as pbar:
+            commands.convert(
+                tiledir,
+                output,
+                *args,
+                dask_settings=DaskSettings(process_graph=not dask_no_task_graph),
+                observers=[pbar],
+                **kwargs,
             )
-        )
         tqdm.tqdm.write(f"processing {tiledir} finished in {t}")
 
-    if vrt:
-        tqdm.tqdm.write("creating VRT(s)")
-        list(
-            tqdm.tqdm(
-                commands.index(
-                    output,
-                    *args,
-                    vrt=vrt,
-                    idx_out_dir=idx_out_dir,
-                    as_iterator=True,
-                    msg_callback=tqdm.tqdm.write if verbose else None,
-                    **kwargs,
-                ),
-                unit="tile",
-                disable=debug or no_pbar,
+        if vrt:
+            tqdm.tqdm.write("creating VRT(s)")
+            commands.index(
+                output,
+                *args,
+                vrt=vrt,
+                idx_out_dir=idx_out_dir,
+                observers=[pbar],
+                **kwargs,
             )
-        )
-        tqdm.tqdm.write(f"index(es) creation for {tiledir} finished")
+            tqdm.tqdm.write(f"index(es) creation for {tiledir} finished")
diff --git a/mapchete/cli/default/cp.py b/mapchete/cli/default/cp.py
index 6c8a4222..bd677b03 100644
--- a/mapchete/cli/default/cp.py
+++ b/mapchete/cli/default/cp.py
@@ -3,6 +3,7 @@
 
 from mapchete import commands
 from mapchete.cli import options
+from mapchete.cli.progress_bar import PBar
 
 
 def _cb_none_concurrency(ctx, param, value):
@@ -47,16 +48,13 @@ def cp(*args, debug=False, no_pbar=False, verbose=False, logfile=None, **kwargs)
                 f"'--{x} foo' is deprecated. You should use '--src-fs-opts {x}=foo' instead.",
             )
         kwargs.pop(x)
-    # copy
-    list(
-        tqdm.tqdm(
-            commands.cp(
-                *args,
-                as_iterator=True,
-                msg_callback=tqdm.tqdm.write if verbose else None,
-                **kwargs,
-            ),
-            unit="tile",
-            disable=debug or no_pbar,
+
+    with PBar(
+        total=100, desc="tiles", disable=debug or no_pbar, print_messages=verbose
+    ) as pbar:
+        # copy
+        commands.cp(
+            *args,
+            observers=[pbar],
+            **kwargs,
         )
-    )
diff --git a/mapchete/cli/default/execute.py b/mapchete/cli/default/execute.py
index f7fe3753..215a835f 100644
--- a/mapchete/cli/default/execute.py
+++ b/mapchete/cli/default/execute.py
@@ -4,6 +4,8 @@
 import mapchete
 from mapchete import commands
 from mapchete.cli import options
+from mapchete.cli.progress_bar import PBar
+from mapchete.config import DaskSettings
 
 
 @click.command(help="Execute a process.")
@@ -19,7 +21,6 @@
 @options.opt_overwrite
 @options.opt_concurrency
 @options.opt_workers
-@options.opt_multi
 @options.opt_dask_scheduler
 @options.opt_dask_no_task_graph
 @options.opt_input_file
@@ -41,6 +42,8 @@ def execute(
     input_file=None,
     dask_no_task_graph=False,
     profiling=False,
+    logfile=None,
+    dask_scheduler=None,
     **kwargs,
 ):
     # activate verbose if profiling is active
@@ -54,39 +57,32 @@ def execute(
     for mapchete_file in mapchete_files:
         tqdm.tqdm.write(f"preparing to process {mapchete_file}")
         with mapchete.Timer() as t:
-            job = commands.execute(
-                mapchete_file,
-                *args,
-                as_iterator=True,
-                msg_callback=tqdm.tqdm.write if verbose else None,
-                dask_compute_graph=not dask_no_task_graph,
-                profiling=profiling,
-                **kwargs,
-            )
-            list(
-                tqdm.tqdm(
-                    job,
-                    unit="task",
-                    disable=debug or no_pbar,
+            with PBar(
+                total=100,
+                desc="tasks",
+                disable=debug or no_pbar,
+                print_messages=verbose,
+            ) as pbar:
+                commands.execute(
+                    mapchete_file,
+                    *args,
+                    observers=[pbar],
+                    dask_settings=DaskSettings(
+                        process_graph=not dask_no_task_graph, scheduler=dask_scheduler
+                    ),
+                    profiling=profiling,
+                    **kwargs,
                 )
-            )
             tqdm.tqdm.write(f"processing {mapchete_file} finished in {t}")
 
-        if vrt:
-            tqdm.tqdm.write("creating VRT(s)")
-            list(
-                tqdm.tqdm(
-                    commands.index(
-                        mapchete_file,
-                        *args,
-                        vrt=vrt,
-                        idx_out_dir=idx_out_dir,
-                        as_iterator=True,
-                        msg_callback=tqdm.tqdm.write if verbose else None,
-                        **kwargs,
-                    ),
-                    unit="tile",
-                    disable=debug or no_pbar,
+            if vrt:
+                tqdm.tqdm.write("creating VRT(s)")
+                commands.index(
+                    mapchete_file,
+                    *args,
+                    vrt=vrt,
+                    idx_out_dir=idx_out_dir,
+                    observers=[pbar],
+                    **kwargs,
                 )
-            )
-            tqdm.tqdm.write(f"index(es) creation for {mapchete_file} finished")
+                tqdm.tqdm.write(f"index(es) creation for {mapchete_file} finished")
diff --git a/mapchete/cli/default/index.py b/mapchete/cli/default/index.py
index e2da6cdd..f037dd9b 100644
--- a/mapchete/cli/default/index.py
+++ b/mapchete/cli/default/index.py
@@ -3,12 +3,11 @@
 import logging
 
 import click
-import click_spinner
 import tqdm
 
-import mapchete
 from mapchete import commands
 from mapchete.cli import options
+from mapchete.cli.progress_bar import PBar
 
 # workaround for https://github.com/tqdm/tqdm/issues/481
 tqdm.monitor_interval = 0
@@ -52,16 +51,16 @@ def index(*args, debug=False, no_pbar=False, verbose=False, logfile=None, **kwar
                 f"'--{x} foo' is deprecated. You should use '--fs-opts {x}=foo' instead.",
             )
         kwargs.pop(x)
-    list(
-        tqdm.tqdm(
-            commands.index(
-                *args,
-                as_iterator=True,
-                msg_callback=tqdm.tqdm.write if verbose else None,
-                **kwargs,
-            ),
-            unit="tile",
-            disable=debug or no_pbar,
+
+    with PBar(
+        total=100, desc="tiles", disable=debug or no_pbar, print_messages=verbose
+    ) as pbar:
+        kwargs.update(some_input=kwargs.pop("tiledir"))
+        commands.index(
+            *args,
+            as_iterator=True,
+            msg_callback=tqdm.tqdm.write if verbose else None,
+            observers=[pbar],
+            **kwargs,
         )
-    )
     tqdm.tqdm.write(f"index(es) creation for {kwargs.get('tiledir')} finished")
diff --git a/mapchete/cli/default/rm.py b/mapchete/cli/default/rm.py
index 7571c84d..bdef8c9f 100644
--- a/mapchete/cli/default/rm.py
+++ b/mapchete/cli/default/rm.py
@@ -3,6 +3,8 @@
 
 from mapchete import commands
 from mapchete.cli import options
+from mapchete.cli.progress_bar import PBar
+from mapchete.commands._rm import existing_paths
 
 
 @click.command(help="Remove tiles from TileDirectory.")
@@ -12,13 +14,12 @@
 @options.opt_area_crs
 @options.opt_bounds
 @options.opt_bounds_crs
-@options.opt_multi
+@options.opt_workers
 @options.opt_verbose
 @options.opt_no_pbar
 @options.opt_debug
 @options.opt_logfile
 @options.opt_force
-@options.opt_fs_opts
 def rm(
     *args,
     force=False,
@@ -29,22 +30,20 @@ def rm(
     **kwargs,
 ):
     """Remove tiles from TileDirectory."""
-    tiles_to_delete = commands.rm(
+    tiles_to_delete = existing_paths(
         *args,
-        as_iterator=True,
-        msg_callback=tqdm.tqdm.write if verbose else None,
         **kwargs,
     )
     if len(tiles_to_delete):
         if force or click.confirm(
             f"Do you want to delete {len(tiles_to_delete)} tiles?", abort=True
         ):
-            list(
-                tqdm.tqdm(
-                    tiles_to_delete,
-                    unit="tile",
-                    disable=debug or no_pbar,
-                )
-            )
+            with PBar(
+                total=len(tiles_to_delete),
+                desc="tiles",
+                disable=debug or no_pbar,
+                print_messages=verbose,
+            ) as pbar:
+                commands.rm(paths=tiles_to_delete, observers=[pbar], **kwargs)
     else:  # pragma: no cover
         tqdm.tqdm.write("No tiles found to delete.")
diff --git a/mapchete/cli/options.py b/mapchete/cli/options.py
index 3943776a..0f77f9ff 100644
--- a/mapchete/cli/options.py
+++ b/mapchete/cli/options.py
@@ -81,7 +81,7 @@ def _cb_key_val(ctx, param, value):
     else:
         out = {}
         for pair in value:
-            if "=" not in pair:
+            if "=" not in pair:  # pragma: no cover
                 raise click.BadParameter(
                     "Invalid syntax for KEY=VAL arg: {}".format(pair)
                 )
@@ -204,12 +204,6 @@ def _cb_none_concurrency(ctx, param, value):
 opt_overwrite = click.option(
     "--overwrite", "-o", is_flag=True, help="Overwrite if tile(s) already exist(s)."
 )
-opt_multi = click.option(
-    "--multi",
-    "-m",
-    type=click.INT,
-    help="Number of workers when processing concurrently.",
-)
 opt_workers = click.option(
     "--workers",
     "-w",
diff --git a/mapchete/cli/progress_bar.py b/mapchete/cli/progress_bar.py
new file mode 100644
index 00000000..61a7d4ee
--- /dev/null
+++ b/mapchete/cli/progress_bar.py
@@ -0,0 +1,41 @@
+from typing import Optional
+
+from tqdm import tqdm
+
+from mapchete.protocols import ObserverProtocol
+from mapchete.types import Progress
+
+
+class PBar(ObserverProtocol):
+    """Custom progress bar which is used as an observer for commands."""
+
+    print_messages: bool = True
+    _pbar = tqdm
+
+    def __init__(self, *args, print_messages: bool = True, **kwargs):
+        self._args = args
+        self._kwargs = kwargs
+        self.print_messages = print_messages
+
+    def __enter__(self):
+        self._pbar = tqdm(*self._args, **self._kwargs)
+        return self
+
+    def __exit__(self, *args):
+        self._pbar.__exit__(*args)
+
+    def update(
+        self,
+        *args,
+        progress: Optional[Progress] = None,
+        message: Optional[str] = None,
+        **kwargs
+    ):
+        if progress:
+            if self._pbar.total is None or self._pbar.total != progress.total:
+                self._pbar.reset(progress.total)
+
+            self._pbar.update(progress.current - self._pbar.n)
+
+        if self.print_messages and message:
+            tqdm.write(message)
diff --git a/mapchete/commands/_convert.py b/mapchete/commands/_convert.py
index a33ec3a0..cc2cf98d 100644
--- a/mapchete/commands/_convert.py
+++ b/mapchete/commands/_convert.py
@@ -1,10 +1,10 @@
 import json
 import logging
 import os
-import warnings
+from contextlib import AbstractContextManager
 from multiprocessing import cpu_count
 from pprint import pformat
-from typing import Callable, List, Tuple, Union
+from typing import List, Optional, Tuple, Type, Union
 
 import tilematrix
 from rasterio.crs import CRS
@@ -12,9 +12,12 @@
 from shapely.geometry import box
 from shapely.geometry.base import BaseGeometry
 
-import mapchete
 from mapchete.commands._execute import execute
+from mapchete.commands.observer import ObserverProtocol, Observers
+from mapchete.config import DaskSettings
 from mapchete.config.parse import raw_conf, raw_conf_output_pyramid
+from mapchete.errors import JobCancelledError
+from mapchete.executor import Executor
 from mapchete.formats import (
     available_input_formats,
     available_output_formats,
@@ -23,6 +26,7 @@
 from mapchete.io import MPath, fiona_open, get_best_zoom_level, rasterio_open, read_json
 from mapchete.io.vector import reproject_geometry
 from mapchete.tile import BufferedTilePyramid
+from mapchete.types import MPathLike
 from mapchete.validate import validate_zooms
 
 logger = logging.getLogger(__name__)
@@ -30,8 +34,8 @@
 
 
 def convert(
-    tiledir: Union[str, dict, MPath],
-    output: Union[str, MPath],
+    tiledir: MPathLike,
+    output: MPathLike,
     zoom: Union[int, List[int]] = None,
     area: Union[BaseGeometry, str, dict] = None,
     area_crs: Union[CRS, str] = None,
@@ -42,13 +46,8 @@ def convert(
     tile: Tuple[int, int, int] = None,
     overwrite: bool = False,
     concurrency: str = "processes",
-    dask_scheduler: str = None,
-    dask_max_submitted_tasks: int = 500,
-    dask_chunksize: int = 100,
-    dask_client=None,
-    dask_compute_graph=True,
+    dask_settings: DaskSettings = DaskSettings(),
     workers: int = None,
-    multi: int = None,
     clip_geometry: str = None,
     bidx: List[int] = None,
     output_pyramid: str = None,
@@ -63,11 +62,14 @@ def convert(
     overviews: bool = False,
     overviews_resampling_method: str = "cubic_spline",
     cog: bool = False,
-    msg_callback: Callable = None,
-    as_iterator: bool = False,
     src_fs_opts: Union[dict, None] = None,
     dst_fs_opts: Union[dict, None] = None,
-) -> mapchete.Job:
+    executor_getter: AbstractContextManager = Executor,
+    observers: Optional[List[ObserverProtocol]] = None,
+    retry_on_exception: Tuple[Type[Exception], Type[Exception]] = Exception,
+    cancel_on_exception: Type[Exception] = JobCancelledError,
+    retries: int = 0,
+) -> None:
     """
     Convert mapchete outputs or other geodata.
 
@@ -144,38 +146,10 @@ def convert(
         Resampling method used for overviews. (default: cubic_spline)
     cog : bool
         Write a valid COG. This will automatically generate verviews. (GTiff only)
-    msg_callback : Callable
-        Optional callback function for process messages.
-    as_iterator : bool
-        Returns as generator but with a __len__() property.
-
-    Returns
-    -------
-    mapchete.Job instance either with already processed items or a generator with known length.
-
-    Examples
-    --------
-    >>> convert("foo", "bar")
-
-    This will run the whole conversion process.
-
-    >>> for i in convert("foo", "bar", as_iterator=True):
-    >>>     print(i)
-
-    This will return a generator where through iteration, tiles are copied.
-
-    >>> list(tqdm.tqdm(convert("foo", "bar", as_iterator=True)))
-
-    Usage within a process bar.
     """
 
-    def _empty_callback(*args, **kwargs):
-        pass
-
-    msg_callback = msg_callback or _empty_callback
-    if multi is not None:  # pragma: no cover
-        warnings.warn("The 'multi' parameter is deprecated and is now named 'workers'")
-    workers = workers or multi or cpu_count()
+    all_observers = Observers(observers)
+    workers = workers or cpu_count()
     creation_options = creation_options or {}
     bidx = [bidx] if isinstance(bidx, int) else bidx
     tiledir = MPath.from_inp(tiledir, storage_options=src_fs_opts)
@@ -297,11 +271,10 @@ def _empty_callback(*args, **kwargs):
             clip_geometry, dst_crs=out_pyramid.crs
         ).intersection(box(*inp_bounds))
         if clip_intersection.is_empty:
-            msg_callback(
-                "Process area is empty: clip bounds don't intersect with input bounds."
+            all_observers.notify(
+                message="Process area is empty: clip bounds don't intersect with input bounds."
             )
-            # this returns a Job with an empty iterator
-            return mapchete.Job(None, (), as_iterator=as_iterator)
+            return
     # add process bounds and output type
     mapchete_config.update(
         bounds=(clip_intersection.bounds if clip_geometry else inp_bounds),
@@ -323,14 +296,13 @@ def _empty_callback(*args, **kwargs):
         area=area,
         area_crs=area_crs,
         concurrency=concurrency,
-        dask_scheduler=dask_scheduler,
-        dask_max_submitted_tasks=dask_max_submitted_tasks,
-        dask_chunksize=dask_chunksize,
-        dask_client=dask_client,
-        dask_compute_graph=dask_compute_graph,
+        dask_settings=dask_settings,
         workers=workers,
-        as_iterator=as_iterator,
-        msg_callback=msg_callback,
+        executor_getter=executor_getter,
+        observers=observers,
+        retry_on_exception=retry_on_exception,
+        cancel_on_exception=cancel_on_exception,
+        retries=retries,
     )
 
 
diff --git a/mapchete/commands/_cp.py b/mapchete/commands/_cp.py
index 1d13ab8a..7a0040a9 100644
--- a/mapchete/commands/_cp.py
+++ b/mapchete/commands/_cp.py
@@ -1,17 +1,20 @@
 """Copy tiles between Tile Directories."""
 
 import logging
-import warnings
+from contextlib import AbstractContextManager
 from multiprocessing import cpu_count
-from typing import Callable, List, Tuple, Union
+from typing import List, Optional, Tuple, Union
 
 from rasterio.crs import CRS
 from shapely.geometry import Point
 from shapely.geometry.base import BaseGeometry
 
 import mapchete
+from mapchete.commands.observer import ObserverProtocol, Observers
+from mapchete.executor import Executor
 from mapchete.io import MPath, copy, tiles_exist
 from mapchete.io.vector import reproject_geometry
+from mapchete.types import Progress
 
 logger = logging.getLogger(__name__)
 
@@ -34,9 +37,9 @@ def cp(
     dask_client=None,
     src_fs_opts: Union[dict, None] = None,
     dst_fs_opts: Union[dict, None] = None,
-    msg_callback: Union[Callable, None] = None,
-    as_iterator: bool = False,
-) -> mapchete.Job:
+    executor_getter: AbstractContextManager = Executor,
+    observers: Optional[List[ObserverProtocol]] = None,
+):
     """
     Copy TileDirectory from source to destination.
 
@@ -75,38 +78,9 @@ def cp(
         Configuration options for source fsspec filesystem.
     dst_fs_opts : dict
         Configuration options for destination fsspec filesystem.
-    msg_callback : Callable
-        Optional callback function for process messages.
-    as_iterator : bool
-        Returns as generator but with a __len__() property.
-
-    Returns
-    -------
-    mapchete.Job instance either with already processed items or a generator with known length.
-
-    Examples
-    --------
-    >>> cp("foo", "bar", zoom=5)
-
-    This will run the whole copy process.
-
-    >>> for i in cp("foo", "bar", zoom=5, as_iterator=True):
-    >>>     print(i)
-
-    This will return a generator where through iteration, tiles are copied.
-
-    >>> list(tqdm.tqdm(cp("foo", "bar", zoom=5, as_iterator=True)))
-
-    Usage within a process bar.
     """
 
-    def _empty_callback(*_):
-        pass
-
-    msg_callback = msg_callback or _empty_callback
-    if multi is not None:  # pragma: no cover
-        warnings.warn("The 'multi' parameter is deprecated and is now named 'workers'")
-    workers = workers or multi or cpu_count()
+    workers = workers or cpu_count()
     src_fs_opts = src_fs_opts or {}
     dst_fs_opts = dst_fs_opts or {}
     if zoom is None:  # pragma: no cover
@@ -116,6 +90,7 @@ def _empty_callback(*_):
     dst_tiledir = MPath.from_inp(dst_tiledir, storage_options=dst_fs_opts)
     src_fs = src_tiledir.fs
     dst_fs = dst_tiledir.fs
+    all_observers = Observers(observers)
 
     # open source tile directory
     with mapchete.open(
@@ -137,7 +112,7 @@ def _empty_callback(*_):
         if not dst_fs.exists(dst_metadata):
             msg = f"copy {src_metadata} to {dst_metadata}"
             logger.debug(msg)
-            msg_callback(msg)
+            all_observers.notify(message=msg)
             copy(
                 src_metadata,
                 dst_metadata,
@@ -157,94 +132,74 @@ def _empty_callback(*_):
             fs_kwargs=dst_fs_opts,
             mode="readonly",
         ) as dst_mp:
-            return mapchete.Job(
-                _copy_tiles,
-                fargs=(
-                    msg_callback,
-                    src_mp,
-                    dst_mp,
-                    tp,
-                    workers,
-                    src_fs,
-                    dst_fs,
-                    point,
-                    point_crs,
-                    overwrite,
-                ),
-                executor_concurrency=concurrency,
-                executor_kwargs=dict(
-                    max_workers=workers,
-                    dask_scheduler=dask_scheduler,
-                    dask_client=dask_client,
-                ),
-                as_iterator=as_iterator,
-                tiles_tasks=1 if point else src_mp.count_tiles(),
-            )
-
-
-def _copy_tiles(
-    msg_callback,
-    src_mp,
-    dst_mp,
-    tp,
-    workers,
-    src_fs,
-    dst_fs,
-    point,
-    point_crs,
-    overwrite,
-    executor=None,
-):
-    for zoom in src_mp.config.init_zoom_levels:
-        msg_callback(f"copy tiles for zoom {zoom}...")
-
-        # materialize all tiles
-        if point:
-            point_geom = reproject_geometry(
-                Point(point), src_crs=point_crs or tp.crs, dst_crs=tp.crs
-            )
-            tiles = [tp.tile_from_xy(point_geom.x, point_geom.y, zoom)]
-        else:
-            aoi_geom = src_mp.config.area_at_zoom(zoom)
-            tiles = [
-                t
-                for t in tp.tiles_from_geom(aoi_geom, zoom)
-                # this is required to omit tiles touching the config area
-                if aoi_geom.intersection(t.bbox).area
-            ]
-
-        # check which source tiles exist
-        logger.debug("looking for existing source tiles...")
-        src_tiles_exist = dict(
-            tiles_exist(config=src_mp.config, output_tiles=tiles, workers=workers)
-        )
-
-        # check which destination tiles exist
-        logger.debug("looking for existing destination tiles...")
-        dst_tiles_exist = dict(
-            tiles_exist(config=dst_mp.config, output_tiles=tiles, workers=workers)
-        )
-
-        # copy
-        total_copied = 0
-        for future in executor.as_completed(
-            _copy_tile,
-            tiles,
-            fargs=(
-                src_mp,
-                dst_mp,
-                src_tiles_exist,
-                dst_tiles_exist,
-                src_fs,
-                dst_fs,
-                overwrite,
-            ),
-        ):
-            copied, msg = future.result()
-            total_copied += copied
-            yield msg
-
-        msg_callback(f"{total_copied} tiles copied")
+            with executor_getter(
+                concurrency=concurrency,
+                max_workers=workers,
+                dask_scheduler=dask_scheduler,
+                dask_client=dask_client,
+            ) as executor:
+                for zoom in src_mp.config.init_zoom_levels:
+                    all_observers.notify(message=f"copy tiles for zoom {zoom}...")
+
+                    # materialize all tiles
+                    if point:
+                        point_geom = reproject_geometry(
+                            Point(point), src_crs=point_crs or tp.crs, dst_crs=tp.crs
+                        )
+                        tiles = [tp.tile_from_xy(point_geom.x, point_geom.y, zoom)]
+                    else:
+                        aoi_geom = src_mp.config.area_at_zoom(zoom)
+                        tiles = [
+                            t
+                            for t in tp.tiles_from_geom(aoi_geom, zoom)
+                            # this is required to omit tiles touching the config area
+                            if aoi_geom.intersection(t.bbox).area
+                        ]
+
+                    all_observers.notify(progress=Progress(current=0, total=len(tiles)))
+
+                    # check which source tiles exist
+                    logger.debug("looking for existing source tiles...")
+                    src_tiles_exist = dict(
+                        tiles_exist(
+                            config=src_mp.config, output_tiles=tiles, workers=workers
+                        )
+                    )
+
+                    # check which destination tiles exist
+                    logger.debug("looking for existing destination tiles...")
+                    dst_tiles_exist = dict(
+                        tiles_exist(
+                            config=dst_mp.config, output_tiles=tiles, workers=workers
+                        )
+                    )
+
+                    # copy
+                    total_copied = 0
+                    for ii, future in enumerate(
+                        executor.as_completed(
+                            _copy_tile,
+                            tiles,
+                            fargs=(
+                                src_mp,
+                                dst_mp,
+                                src_tiles_exist,
+                                dst_tiles_exist,
+                                src_fs,
+                                dst_fs,
+                                overwrite,
+                            ),
+                        ),
+                        1,
+                    ):
+                        copied, message = future.result()
+                        total_copied += copied
+                        all_observers.notify(
+                            progress=Progress(current=ii, total=len(tiles)),
+                            message=message,
+                        )
+
+                    all_observers.notify(message=f"{total_copied} tiles copied")
 
 
 def _copy_tile(
diff --git a/mapchete/commands/_execute.py b/mapchete/commands/_execute.py
index 2183eea1..53a19496 100644
--- a/mapchete/commands/_execute.py
+++ b/mapchete/commands/_execute.py
@@ -1,53 +1,52 @@
 """Execute a process."""
-
 import logging
-import traceback
-import warnings
+from contextlib import AbstractContextManager
 from multiprocessing import cpu_count
-from typing import Callable, Iterator, List, Tuple, Union
+from typing import List, Optional, Tuple, Type, Union
 
 from rasterio.crs import CRS
 from shapely.geometry.base import BaseGeometry
 
 import mapchete
+from mapchete.commands.observer import ObserverProtocol, Observers
+from mapchete.config import DaskSettings
 from mapchete.config.parse import bounds_from_opts, raw_conf, raw_conf_process_pyramid
-from mapchete.enums import Concurrency, ProcessingMode
-from mapchete.processing.types import (
-    PreprocessingProcessInfo,
-    TaskResult,
-    TileProcessInfo,
-)
+from mapchete.enums import Concurrency, ProcessingMode, Status
+from mapchete.errors import JobCancelledError
+from mapchete.executor import Executor
+from mapchete.executor.types import Profiler
+from mapchete.pretty import pretty_bytes, pretty_seconds
+from mapchete.processing.profilers import preconfigured_profilers
+from mapchete.processing.profilers.time import measure_time
+from mapchete.processing.tasks import TaskInfo
+from mapchete.types import MPathLike, Progress
 
 logger = logging.getLogger(__name__)
 
 
 def execute(
-    mapchete_config: Union[str, dict],
-    zoom: Union[int, List[int]] = None,
-    area: Union[BaseGeometry, str, dict] = None,
-    area_crs: Union[CRS, str] = None,
-    bounds: Tuple[float] = None,
-    bounds_crs: Union[CRS, str] = None,
-    point: Tuple[float, float] = None,
-    point_crs: Tuple[float, float] = None,
-    tile: Tuple[int, int, int] = None,
+    mapchete_config: Union[dict, MPathLike],
+    zoom: Optional[Union[int, List[int]]] = None,
+    area: Optional[Union[BaseGeometry, str, dict]] = None,
+    area_crs: Optional[Union[CRS, str]] = None,
+    bounds: Optional[Tuple[float]] = None,
+    bounds_crs: Optional[Union[CRS, str]] = None,
+    point: Optional[Tuple[float, float]] = None,
+    point_crs: Optional[Tuple[float, float]] = None,
+    tile: Optional[Tuple[int, int, int]] = None,
     overwrite: bool = False,
     mode: ProcessingMode = ProcessingMode.CONTINUE,
     concurrency: Concurrency = Concurrency.processes,
     workers: int = None,
-    multi: int = None,
     multiprocessing_start_method: str = None,
-    dask_scheduler: str = None,
-    dask_max_submitted_tasks=1000,
-    dask_chunksize=100,
-    dask_client=None,
-    dask_compute_graph=True,
-    dask_propagate_results=True,
-    msg_callback: Callable = None,
-    as_iterator: bool = False,
+    dask_settings: DaskSettings = DaskSettings(),
+    executor_getter: AbstractContextManager = Executor,
     profiling: bool = False,
-    **kwargs,
-) -> mapchete.Job:
+    observers: Optional[List[ObserverProtocol]] = None,
+    retry_on_exception: Tuple[Type[Exception], Type[Exception]] = Exception,
+    cancel_on_exception: Type[Exception] = JobCancelledError,
+    retries: int = 0,
+):
     """
     Execute a Mapchete process.
 
@@ -83,158 +82,149 @@ def execute(
         depends on OS.
     concurrency : str
         Concurrency to be used. Could either be "processes", "threads" or "dask".
-    dask_scheduler : str
-        URL to dask scheduler if required.
-    dask_max_submitted_tasks : int
-        Make sure that not more tasks are submitted to dask scheduler at once. (default: 500)
-    dask_chunksize : int
-        Number of tasks submitted to the scheduler at once. (default: 100)
     dask_client : dask.distributed.Client
         Reusable Client instance if required. Otherwise a new client will be created.
-    dask_compute_graph : bool
-        Build and compute dask graph instead of submitting tasks as preprocessing & zoom tiles
-        batches. (default: True)
-    dask_propagate_results : bool
-        Propagate results between tasks. This helps to minimize read calls when building overviews
-        but can lead to a much higher memory consumption on the cluster. Only with effect if
-        dask_compute_graph is activated. (default: True)
-    msg_callback : Callable
-        Optional callback function for process messages.
-    as_iterator : bool
-        Returns as generator but with a __len__() property.
-
-    Returns
-    -------
-    mapchete.Job instance either with already processed items or a generator with known length.
-
-    Examples
-    --------
-    >>> execute("foo")
-
-    This will run the whole execute process.
-
-    >>> for i in execute("foo", as_iterator=True):
-    >>>     print(i)
-
-    This will return a generator where through iteration, tiles are copied.
-
-    >>> list(tqdm.tqdm(execute("foo", as_iterator=True)))
-
-    Usage within a process bar.
     """
-    mode = "overwrite" if overwrite else mode
-
-    def _empty_callback(_):
-        pass
-
-    print_task_details = msg_callback is not None
-    msg_callback = msg_callback or _empty_callback
-    if multi is not None:  # pragma: no cover
-        warnings.warn("The 'multi' parameter is deprecated and is now named 'workers'")
-    workers = workers or multi or cpu_count()
-
-    if tile:
-        tile = raw_conf_process_pyramid(raw_conf(mapchete_config)).tile(*tile)
-        bounds = tile.bounds
-        zoom = tile.zoom
-    else:
-        bounds = bounds_from_opts(
-            point=point,
-            point_crs=point_crs,
-            bounds=bounds,
-            bounds_crs=bounds_crs,
-            raw_conf=raw_conf(mapchete_config),
-        )
+    try:
+        mode = "overwrite" if overwrite else mode
+        all_observers = Observers(observers)
+
+        if not isinstance(retry_on_exception, tuple):
+            retry_on_exception = (retry_on_exception,)
+        workers = workers or cpu_count()
+
+        all_observers.notify(status=Status.parsing)
+
+        if tile:
+            tile = raw_conf_process_pyramid(raw_conf(mapchete_config)).tile(*tile)
+            bounds = tile.bounds
+            zoom = tile.zoom
+        else:
+            bounds = bounds_from_opts(
+                point=point,
+                point_crs=point_crs,
+                bounds=bounds,
+                bounds_crs=bounds_crs,
+                raw_conf=raw_conf(mapchete_config),
+            )
 
-    # be careful opening mapchete not as context manager
-    with mapchete.Timer() as t:
-        mp = mapchete.open(
+        # be careful opening mapchete not as context manager
+        with mapchete.open(
             mapchete_config,
             mode=mode,
             bounds=bounds,
             zoom=zoom,
             area=area,
             area_crs=area_crs,
-        )
-    logger.debug("initialized process in %s", t)
-    try:
-        preprocessing_tasks = mp.config.preprocessing_tasks_count()
-        tiles_tasks = 1 if tile else mp.count_tiles()
-        total_tasks = preprocessing_tasks + tiles_tasks
-        msg_callback(
-            f"processing {preprocessing_tasks} preprocessing tasks and {tiles_tasks} tile tasks on {workers} worker(s)"
-        )
-        # automatically use dask Executor if dask scheduler is defined
-        if dask_scheduler or dask_client or concurrency == "dask":
-            concurrency = "dask"
-        # use sequential Executor if only one tile or only one worker is defined
-        elif total_tasks == 1 or workers == 1:
-            logger.debug(
-                "using sequential Executor because there is only one %s",
-                "task" if total_tasks == 1 else "worker",
-            )
-            concurrency = None
-        stac_item_path = mp.config.output.stac_path
-        return mapchete.Job(
-            _process_everything,
-            fargs=(
-                msg_callback,
-                mp,
-            ),
-            fkwargs=dict(
-                tile=tile,
-                workers=workers,
-                zoom=None if tile else zoom,
-                print_task_details=print_task_details,
-                dask_max_submitted_tasks=dask_max_submitted_tasks,
-                dask_chunksize=dask_chunksize,
-                dask_compute_graph=dask_compute_graph,
-                dask_propagate_results=dask_propagate_results,
-                profiling=profiling,
-            ),
-            executor_concurrency=concurrency,
-            executor_kwargs=dict(
-                dask_scheduler=dask_scheduler,
-                dask_client=dask_client,
-                multiprocessing_start_method=multiprocessing_start_method,
-                max_workers=workers,
-            ),
-            as_iterator=as_iterator,
-            preprocessing_tasks=preprocessing_tasks,
-            tiles_tasks=tiles_tasks,
-            process_area=mp.config.init_area,
-            stac_item_path=stac_item_path,
-        )
-    # explicitly exit the mp object on failure
-    except Exception as exc:  # pragma: no cover
-        mp.__exit__(exc, repr(exc), traceback.format_exc())
-        raise
+        ) as mp:
+            attempt = 0
 
+            # the part below can be retried n times #
+            #########################################
 
-def _process_everything(
-    msg_callback,
-    mp,
-    print_task_details=True,
-    **kwargs,
-) -> Iterator[TaskResult]:
-    try:
-        for future in mp.compute(**kwargs):
-            result = TaskResult.from_future(future)
-            if print_task_details:
-                msg = f"task {result.id}: {result.process_msg}"
-                if result.profiling:  # pragma: no cover
-                    max_allocated = (
-                        result.profiling["memory"].max_allocated / 1024 / 1024
+            while retries + 1:
+                attempt += 1
+
+                # simulating that with every retry, probably less tasks have to be
+                # executed
+                if attempt > 1:
+                    retries_str = "retry" if retries == 1 else "retries"
+                    all_observers.notify(
+                        message=f"attempt {attempt}, {retries} {retries_str} left"
                     )
-                    head_requests = result.profiling["requests"].head_count
-                    get_requests = result.profiling["requests"].get_count
-                    requests = head_requests + get_requests
-                    transfer = result.profiling["requests"].get_bytes / 1024 / 1024
-                    msg += f" (max memory usage: {max_allocated:.2f}MB, {requests} GET and HEAD requests, {transfer:.2f}MB transferred)"
-                msg_callback(msg)
-            yield result
-        # explicitly exit the mp object on success
-        mp.__exit__(None, None, None)
-    except Exception as exc:  # pragma: no cover
-        mp.__exit__(exc, repr(exc), traceback.format_exc())
+
+                all_observers.notify(status=Status.initializing)
+
+                # determine tasks
+                tasks = mp.tasks(zoom=zoom, tile=tile)
+
+                if len(tasks) == 0:
+                    all_observers.notify(
+                        status=Status.done, message="no tasks to process"
+                    )
+                    return
+
+                all_observers.notify(
+                    message=f"processing {len(tasks)} tasks on {workers} worker(s)"
+                )
+
+                all_observers.notify(message="waiting for executor ...")
+                with executor_getter(
+                    concurrency=concurrency,
+                    dask_scheduler=dask_settings.scheduler,
+                    dask_client=dask_settings.client,
+                    multiprocessing_start_method=multiprocessing_start_method,
+                    max_workers=workers,
+                ) as executor:
+                    if profiling:
+                        for profiler in preconfigured_profilers:
+                            executor.add_profiler(profiler)
+                    else:
+                        executor.add_profiler(
+                            Profiler(name="time", decorator=measure_time)
+                        )
+                    all_observers.notify(
+                        status=Status.running,
+                        progress=Progress(total=len(tasks)),
+                        message=f"sending {len(tasks)} tasks to {executor} ...",
+                        executor=executor,
+                    )
+                    # TODO it would be nice to track the time it took sending tasks to the executor
+                    try:
+                        count = 0
+                        for task_info in mp.execute(
+                            executor=executor,
+                            tasks=tasks,
+                            dask_settings=dask_settings,
+                        ):
+                            count += 1
+
+                            msg = f"task {task_info.id}: {task_info.process_msg}"
+
+                            if task_info.profiling:
+                                msg += profiling_info(task_info)
+
+                            all_observers.notify(
+                                progress=Progress(total=len(tasks), current=count),
+                                task_result=task_info,
+                                message=msg,
+                            )
+
+                        all_observers.notify(status=Status.done)
+                        return
+
+                    except cancel_on_exception:
+                        # special exception indicating job was cancelled from the outside
+                        all_observers.notify(status=Status.cancelled)
+                        return
+
+                    except retry_on_exception as exception:
+                        all_observers.notify(status=Status.failed, exception=exception)
+
+                        if retries:
+                            retries -= 1
+                            all_observers.notify(status=Status.retrying)
+                        else:
+                            raise
+
+    except Exception as exception:
+        all_observers.notify(status=Status.failed, exception=exception)
         raise
+
+
+def profiling_info(task_info: TaskInfo) -> str:
+    profiling_info = []
+    if task_info.profiling.get("time"):
+        elapsed = task_info.profiling["time"].elapsed
+        profiling_info.append(f"time: {pretty_seconds(elapsed)}")
+    if task_info.profiling.get("memory"):  # pragma: no cover
+        max_allocated = task_info.profiling["memory"].max_allocated
+        profiling_info.append(f"max memory usage: {pretty_bytes(max_allocated)}")
+    if task_info.profiling.get("memory"):  # pragma: no cover
+        head_requests = task_info.profiling["requests"].head_count
+        get_requests = task_info.profiling["requests"].get_count
+        requests = head_requests + get_requests
+        transferred = task_info.profiling["requests"].get_bytes
+        profiling_info.append(f"{requests} GET and HEAD requests")
+        profiling_info.append(f"{pretty_bytes(transferred)} transferred")
+    return f" ({', '.join(profiling_info)})"
diff --git a/mapchete/commands/_index.py b/mapchete/commands/_index.py
index 8f625146..2191a86b 100644
--- a/mapchete/commands/_index.py
+++ b/mapchete/commands/_index.py
@@ -1,42 +1,45 @@
 """Create indexes of Tile Directories."""
 
 import logging
-from typing import Callable, List, Tuple, Union
+from typing import List, Optional, Tuple, Union
 
 from rasterio.crs import CRS
 from shapely.geometry.base import BaseGeometry
 
 import mapchete
+from mapchete.commands.observer import ObserverProtocol, Observers
+from mapchete.config import MapcheteConfig
+from mapchete.config.parse import bounds_from_opts, raw_conf, raw_conf_process_pyramid
 from mapchete.index import zoom_index_gen
 from mapchete.path import MPath
+from mapchete.types import MPathLike, Progress
 
 logger = logging.getLogger(__name__)
 
 
 def index(
-    tiledir: str,
-    idx_out_dir: str = None,
+    some_input: Union[MPathLike, dict, MapcheteConfig],
+    idx_out_dir: Optional[MPathLike] = None,
     geojson: bool = False,
     gpkg: bool = False,
     shp: bool = False,
     vrt: bool = False,
     txt: bool = False,
-    fieldname: str = "location",
-    basepath: str = None,
+    fieldname: Optional[str] = "location",
+    basepath: Optional[MPathLike] = None,
     for_gdal: bool = False,
-    zoom: Union[int, List[int]] = None,
-    area: Union[BaseGeometry, str, dict] = None,
-    area_crs: Union[CRS, str] = None,
-    bounds: Tuple[float] = None,
-    bounds_crs: Union[CRS, str] = None,
-    point: Tuple[float, float] = None,
-    point_crs: Tuple[float, float] = None,
-    tile: Tuple[int, int, int] = None,
-    fs_opts: dict = None,
-    msg_callback: Callable = None,
-    as_iterator: bool = False,
-    **_,
-) -> mapchete.Job:
+    zoom: Optional[Union[int, List[int]]] = None,
+    area: Optional[Union[BaseGeometry, str, dict]] = None,
+    area_crs: Optional[Union[CRS, str]] = None,
+    bounds: Optional[Tuple[float]] = None,
+    bounds_crs: Optional[Union[CRS, str]] = None,
+    point: Optional[Tuple[float, float]] = None,
+    point_crs: Optional[Tuple[float, float]] = None,
+    tile: Optional[Tuple[int, int, int]] = None,
+    fs_opts: Optional[dict] = None,
+    observers: Optional[List[ObserverProtocol]] = None,
+    **kwargs,
+):
     """
     Create one or more indexes from a TileDirectory.
 
@@ -81,29 +84,6 @@ def index(
         Zoom, row and column of tile to be processed (cannot be used with zoom)
     fs_opts : dict
         Configuration options for fsspec filesystem.
-    msg_callback : Callable
-        Optional callback function for process messages.
-    as_iterator : bool
-        Returns as generator but with a __len__() property.
-
-    Returns
-    -------
-    mapchete.Job instance either with already processed items or a generator with known length.
-
-    Examples
-    --------
-    >>> index("foo", vrt=True, zoom=5)
-
-    This will run the whole index process.
-
-    >>> for i in index("foo", vrt=True, zoom=5, as_iterator=True):
-    >>>     print(i)
-
-    This will return a generator where through iteration, tiles are copied.
-
-    >>> list(tqdm.tqdm(index("foo", vrt=True, zoom=5, as_iterator=True)))
-
-    Usage within a process bar.
     """
 
     if not any([geojson, gpkg, shp, txt, vrt]):
@@ -112,18 +92,31 @@ def index(
             """must be provided."""
         )
 
-    def _empty_callback(*_):
-        pass
+    all_observers = Observers(observers)
+
+    all_observers.notify(message=f"create index(es) for {some_input}")
+
+    if tile:
+        tile = raw_conf_process_pyramid(raw_conf(some_input)).tile(*tile)
+        bounds = tile.bounds
+        zoom = tile.zoom
+    else:
+        try:
+            bounds = bounds_from_opts(
+                point=point,
+                point_crs=point_crs,
+                bounds=bounds,
+                bounds_crs=bounds_crs,
+                raw_conf=raw_conf(some_input),
+            )
+        except IsADirectoryError:
+            pass
 
-    msg_callback = msg_callback or _empty_callback
-
-    msg_callback(f"create index(es) for {tiledir}")
-    # process single tile
     with mapchete.open(
-        tiledir,
+        some_input,
         mode="readonly",
         fs_kwargs=fs_opts,
-        zoom=tile[0] if tile else zoom,
+        zoom=zoom,
         point=point,
         point_crs=point_crs,
         bounds=bounds,
@@ -131,9 +124,10 @@ def _empty_callback(*_):
         area=area,
         area_crs=area_crs,
     ) as mp:
-        return mapchete.Job(
-            zoom_index_gen,
-            fkwargs=dict(
+        total = 1 if tile else mp.count_tiles()
+        all_observers.notify(progress=Progress(total=total))
+        for ii, tile in enumerate(
+            zoom_index_gen(
                 mp=mp,
                 zoom=None if tile else mp.config.init_zoom_levels,
                 tile=tile,
@@ -147,6 +141,8 @@ def _empty_callback(*_):
                 basepath=basepath,
                 for_gdal=for_gdal,
             ),
-            as_iterator=as_iterator,
-            tiles_tasks=1 if tile else mp.count_tiles(),
-        )
+            1,
+        ):
+            all_observers.notify(
+                progress=Progress(current=ii, total=total), message=f"{tile.id} indexed"
+            )
diff --git a/mapchete/commands/_rm.py b/mapchete/commands/_rm.py
index 4a772305..03ebe4cf 100644
--- a/mapchete/commands/_rm.py
+++ b/mapchete/commands/_rm.py
@@ -1,30 +1,32 @@
 """Remove tiles from Tile Directory."""
 
 import logging
-from typing import Callable, List, Tuple, Union
+from typing import List, Optional, Tuple, Union
 
 from rasterio.crs import CRS
 from shapely.geometry.base import BaseGeometry
 
 import mapchete
+from mapchete.commands.observer import ObserverProtocol, Observers
 from mapchete.io import tiles_exist
 from mapchete.path import MPath
+from mapchete.types import MPathLike, Progress
 
 logger = logging.getLogger(__name__)
 
 
 def rm(
-    tiledir: str,
+    tiledir: Optional[MPathLike] = None,
+    paths: List[MPath] = None,
     zoom: Union[int, List[int]] = None,
     area: Union[BaseGeometry, str, dict] = None,
     area_crs: Union[CRS, str] = None,
     bounds: Tuple[float] = None,
     bounds_crs: Union[CRS, str] = None,
-    multi: int = None,
+    workers: Optional[int] = None,
     fs_opts: dict = None,
-    msg_callback: Callable = None,
-    as_iterator: bool = False,
-) -> mapchete.Job:
+    observers: Optional[List[ObserverProtocol]] = None,
+):
     """
     Remove tiles from TileDirectory.
 
@@ -43,45 +45,68 @@ def rm(
         Override bounds or area provided in process configuration.
     bounds_crs : CRS or str
         CRS of area (default: process CRS).
-    multi : int
-        Number of threads used to check whether tiles exist.
     fs_opts : dict
         Configuration options for fsspec filesystem.
-    msg_callback : Callable
-        Optional callback function for process messages.
-    as_iterator : bool
-        Returns as generator but with a __len__() property.
-
-    Returns
-    -------
-    mapchete.Job instance either with already processed items or a generator with known length.
-
-    Examples
-    --------
-    >>> rm("foo", zoom=5)
-
-    This will run the whole rm process.
-
-    >>> for i in rm("foo", zoom=5, as_iterator=True):
-    >>>     print(i)
-
-    This will return a generator where through iteration, tiles are removed.
-
-    >>> list(tqdm.tqdm(rm("foo", zoom=5, as_iterator=True)))
+    """
+    all_observers = Observers(observers)
+
+    if tiledir:
+        if zoom is None:  # pragma: no cover
+            raise ValueError("zoom level(s) required")
+        tiledir = MPath.from_inp(tiledir, storage_options=fs_opts)
+        paths = existing_paths(
+            tiledir=tiledir,
+            zoom=zoom,
+            area=area,
+            area_crs=area_crs,
+            bounds=bounds,
+            bounds_crs=bounds_crs,
+            workers=workers,
+        )
+        fs = tiledir.fs
+    elif isinstance(paths, list):
+        fs = MPath.from_inp(paths[0]).fs
+    else:  # pragma: no cover
+        raise ValueError(
+            "either a tile directory or a list of paths has to be provided"
+        )
 
-    Usage within a process bar.
+    total = len(paths)
+    all_observers.notify(progress=Progress(total=total))
+    logger.debug("got %s path(s) on %s", len(paths), fs)
 
-    """
+    # s3fs enables multiple paths as input, so let's use this:
+    if "s3" in fs.protocol:
+        fs.rm(paths)
+        for ii, path in enumerate(paths, 1):
+            msg = f"deleted {path}"
+            logger.debug(msg)
+            all_observers.notify(
+                progress=Progress(current=ii, total=total), message=msg
+            )
 
-    def _empty_callback(*_):
-        pass
+    # otherwise, just iterate through the paths
+    else:
+        for ii, path in enumerate(paths, 1):
+            fs.rm(path)
+            msg = f"deleted {path}"
+            logger.debug(msg)
+            all_observers.notify(
+                progress=Progress(current=ii, total=total), message=msg
+            )
 
-    msg_callback = msg_callback or _empty_callback
-    if zoom is None:  # pragma: no cover
-        raise ValueError("zoom level(s) required")
+    all_observers.notify(message=f"{len(paths)} tiles deleted")
 
-    tiledir = MPath.from_inp(tiledir, storage_options=fs_opts)
 
+def existing_paths(
+    tiledir: Optional[MPathLike] = None,
+    zoom: Union[int, List[int]] = None,
+    area: Union[BaseGeometry, str, dict] = None,
+    area_crs: Union[CRS, str] = None,
+    bounds: Tuple[float] = None,
+    bounds_crs: Union[CRS, str] = None,
+    workers: Optional[int] = None,
+) -> dict:
     with mapchete.open(
         tiledir,
         zoom=zoom,
@@ -92,7 +117,6 @@ def _empty_callback(*_):
         mode="readonly",
     ) as mp:
         tp = mp.config.output_pyramid
-
         tiles = {}
         for zoom in mp.config.init_zoom_levels:
             tiles[zoom] = []
@@ -106,55 +130,13 @@ def _empty_callback(*_):
                     # this is required to omit tiles touching the config area
                     if mp.config.area_at_zoom(zoom).intersection(t.bbox).area
                 ],
-                workers=multi,
+                workers=workers,
             ):
                 if exists:
                     tiles[zoom].append(tile)
 
-        paths = [
+        return [
             mp.config.output_reader.get_path(tile)
             for zoom_tiles in tiles.values()
             for tile in zoom_tiles
         ]
-        return mapchete.Job(
-            _rm,
-            fargs=(
-                paths,
-                tiledir.fs,
-                msg_callback,
-            ),
-            as_iterator=as_iterator,
-            tiles_tasks=len(paths),
-        )
-
-
-def _rm(paths, fs, msg_callback, recursive=False, **_):
-    """
-    Remove one or multiple paths from file system.
-
-    Note: all paths have to be from the same file system!
-
-    Parameters
-    ----------
-    paths : str or list
-    fs : fsspec.FileSystem
-    """
-    logger.debug("got %s path(s) on %s", len(paths), fs)
-
-    # s3fs enables multiple paths as input, so let's use this:
-    if "s3" in fs.protocol:  # pragma: no cover
-        fs.rm(paths, recursive=recursive)
-        for path in paths:
-            msg = f"deleted {path}"
-            logger.debug(msg)
-            yield msg
-
-    # otherwise, just iterate through the paths
-    else:
-        for path in paths:
-            fs.rm(path, recursive=recursive)
-            msg = f"deleted {path}"
-            logger.debug(msg)
-            yield msg
-
-    msg_callback(f"{len(paths)} tiles deleted")
diff --git a/mapchete/commands/observer.py b/mapchete/commands/observer.py
new file mode 100644
index 00000000..5c520c7e
--- /dev/null
+++ b/mapchete/commands/observer.py
@@ -0,0 +1,14 @@
+from typing import List, Optional
+
+from mapchete.protocols import ObserverProtocol
+
+
+class Observers:
+    observers = List[ObserverProtocol]
+
+    def __init__(self, observers: Optional[List[ObserverProtocol]]):
+        self.observers = observers or []
+
+    def notify(self, *args, **kwargs) -> None:
+        for observer in self.observers:
+            observer.update(*args, **kwargs)
diff --git a/mapchete/commons/hillshade.py b/mapchete/commons/hillshade.py
index 3653bbbf..8bd0020c 100644
--- a/mapchete/commons/hillshade.py
+++ b/mapchete/commons/hillshade.py
@@ -34,6 +34,7 @@
 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 """
 import math
+import warnings
 from itertools import product
 
 import numpy as np
@@ -71,24 +72,27 @@ def calculate_slope_aspect(elevation, xres, yres, z=1.0, scale=1.0):
     -------
     slope shade : array
     """
-    z = float(z)
-    scale = float(scale)
-    height, width = elevation.shape[0] - 2, elevation.shape[1] - 2
-    w = [
-        z * elevation[row : (row + height), col : (col + width)]
-        for (row, col) in product(range(3), range(3))
-    ]
-    x = ((w[0] + w[3] + w[3] + w[6]) - (w[2] + w[5] + w[5] + w[8])) / (
-        8.0 * xres * scale
-    )
-    y = ((w[6] + w[7] + w[7] + w[8]) - (w[0] + w[1] + w[1] + w[2])) / (
-        8.0 * yres * scale
-    )
-    # in radians, from 0 to pi/2
-    slope = math.pi / 2 - np.arctan(np.sqrt(x * x + y * y))
-    # in radians counterclockwise, from -pi at north back to pi
-    aspect = np.arctan2(x, y)
-    return slope, aspect
+    with warnings.catch_warnings():
+        # this is to filter out division by zero warnings
+        warnings.simplefilter("ignore", category=RuntimeWarning)
+        z = float(z)
+        scale = float(scale)
+        height, width = elevation.shape[0] - 2, elevation.shape[1] - 2
+        w = [
+            z * elevation[row : (row + height), col : (col + width)]
+            for (row, col) in product(range(3), range(3))
+        ]
+        x = ((w[0] + w[3] + w[3] + w[6]) - (w[2] + w[5] + w[5] + w[8])) / (
+            8.0 * xres * scale
+        )
+        y = ((w[6] + w[7] + w[7] + w[8]) - (w[0] + w[1] + w[1] + w[2])) / (
+            8.0 * yres * scale
+        )
+        # in radians, from 0 to pi/2
+        slope = math.pi / 2 - np.arctan(np.sqrt(x * x + y * y))
+        # in radians counterclockwise, from -pi at north back to pi
+        aspect = np.arctan2(x, y)
+        return slope, aspect
 
 
 def hillshade(
diff --git a/mapchete/config/__init__.py b/mapchete/config/__init__.py
index 3bc39d63..95b171de 100644
--- a/mapchete/config/__init__.py
+++ b/mapchete/config/__init__.py
@@ -1,5 +1,5 @@
 from mapchete.config.base import MapcheteConfig, get_hash, snap_bounds, validate_values
-from mapchete.config.models import ProcessConfig, PyramidConfig
+from mapchete.config.models import DaskSettings, ProcessConfig, PyramidConfig
 from mapchete.executor import MULTIPROCESSING_DEFAULT_START_METHOD
 
 __all__ = [
@@ -10,4 +10,5 @@
     "validate_values",
     "ProcessConfig",
     "PyramidConfig",
+    "DaskSettings",
 ]
diff --git a/mapchete/config/base.py b/mapchete/config/base.py
index 9fed322e..491ac87c 100644
--- a/mapchete/config/base.py
+++ b/mapchete/config/base.py
@@ -256,6 +256,9 @@ def __init__(
         logger.debug("prepare output")
         self.output.prepare(process_area=self.area_at_zoom())
 
+    def __repr__(self):  # pragma: no cover
+        return f"<MapcheteConfig init_zoom_levels={self.init_zoom_levels}, init_bounds={self.init_bounds}>"
+
     def input_at_zoom(self, key=None, zoom=None):
         if zoom is None:  # pragma: no cover
             raise ValueError("zoom not provided")
@@ -474,7 +477,9 @@ def input(self):
     @cached_property
     def baselevels(self):
         """
-        Optional baselevels configuration.
+        Base levels are zoom levels which are processed but not generated by other zoom levels.
+
+        If base levels are not defined, all zoom levels will be processed.
 
         baselevels:
             min: <zoom>
@@ -516,6 +521,20 @@ def baselevels(self):
             ),
         )
 
+    @cached_property
+    def overview_levels(self) -> Union[ZoomLevels, None]:
+        if self.baselevels:
+            return self.zoom_levels.difference(self.baselevels["zooms"])
+        else:
+            return None
+
+    @cached_property
+    def processing_levels(self) -> ZoomLevels:
+        if self.overview_levels:
+            return self.zoom_levels.difference(self.overview_levels)
+        else:
+            return self.zoom_levels
+
     def get_process_func_params(self, zoom):
         """Return function kwargs."""
         return self.process.filter_parameters(
diff --git a/mapchete/config/models.py b/mapchete/config/models.py
index 822ae208..94835661 100644
--- a/mapchete/config/models.py
+++ b/mapchete/config/models.py
@@ -1,15 +1,16 @@
-from typing import List, Tuple, Union
+from typing import List, Optional, Type, Union
 
+from distributed import Client
 from pydantic import BaseModel, NonNegativeInt, field_validator
 from shapely.geometry.base import BaseGeometry
 
-from mapchete.types import Bounds, MPathLike, ZoomLevels
+from mapchete.types import Bounds, BoundsLike, MPathLike, ZoomLevels, ZoomLevelsLike
 
 
 class OutputConfigBase(BaseModel):
     format: str
-    metatiling: Union[int, None] = 1
-    pixelbuffer: Union[NonNegativeInt, None] = 0
+    metatiling: Optional[int] = 1
+    pixelbuffer: Optional[NonNegativeInt] = 0
 
     @field_validator("metatiling", mode="before")
     def _metatiling(cls, value: int) -> int:  # pragma: no cover
@@ -21,8 +22,8 @@ def _metatiling(cls, value: int) -> int:  # pragma: no cover
 
 class PyramidConfig(BaseModel):
     grid: Union[str, dict]
-    metatiling: Union[int, None] = 1
-    pixelbuffer: Union[NonNegativeInt, None] = 0
+    metatiling: Optional[int] = 1
+    pixelbuffer: Optional[NonNegativeInt] = 0
 
     @field_validator("metatiling", mode="before")
     def _metatiling(cls, value: int) -> int:  # pragma: no cover
@@ -35,13 +36,22 @@ def _metatiling(cls, value: int) -> int:  # pragma: no cover
 class ProcessConfig(BaseModel, arbitrary_types_allowed=True):
     pyramid: PyramidConfig
     output: dict
-    zoom_levels: Union[ZoomLevels, dict, list, int]
-    process: Union[MPathLike, List[str], None] = None
-    baselevels: Union[dict, None] = None
-    input: Union[dict, None] = None
-    config_dir: Union[MPathLike, None] = None
-    area: Union[MPathLike, BaseGeometry, None] = None
-    area_crs: Union[dict, str, None] = None
-    bounds: Union[Bounds, Tuple[float, float, float, float], None] = None
-    bounds_crs: Union[dict, str, None] = None
-    process_parameters: Union[dict, None] = None
+    zoom_levels: Union[ZoomLevels, ZoomLevelsLike]
+    process: Optional[Union[MPathLike, List[str]]] = None
+    baselevels: Optional[dict] = None
+    input: Optional[dict] = None
+    config_dir: Optional[MPathLike] = None
+    mapchete_file: Optional[MPathLike] = None
+    area: Optional[Union[MPathLike, BaseGeometry]] = None
+    area_crs: Optional[Union[dict, str]] = None
+    bounds: Optional[Union[Bounds, BoundsLike]] = None
+    bounds_crs: Optional[Union[dict, str]] = None
+    process_parameters: Optional[dict] = None
+
+
+class DaskSettings(BaseModel):
+    process_graph: bool = True
+    max_submitted_tasks: int = 500
+    chunksize: int = 100
+    scheduler: Optional[str] = None
+    client: Optional[Type[Client]] = None
diff --git a/mapchete/config/parse.py b/mapchete/config/parse.py
index c3f50336..8cfe652a 100644
--- a/mapchete/config/parse.py
+++ b/mapchete/config/parse.py
@@ -278,7 +278,9 @@ def bounds_from_opts(
         zoom_levels = get_zoom_levels(
             process_zoom_levels=raw_conf["zoom_levels"], init_zoom_levels=zoom
         )
-        return Bounds(*tp.tile_from_xy(x, y, max(zoom_levels)).bounds)
+        return Bounds(
+            *tp.without_pixelbuffer().tile_from_xy(x, y, max(zoom_levels)).bounds
+        )
     elif bounds:
         bounds = Bounds.from_inp(bounds)
         if bounds_crs:
diff --git a/mapchete/enums.py b/mapchete/enums.py
index 20617cd9..eca41439 100644
--- a/mapchete/enums.py
+++ b/mapchete/enums.py
@@ -13,3 +13,39 @@ class Concurrency(str, Enum):
     threads = "threads"
     processes = "processes"
     dask = "dask"
+
+
+class Status(str, Enum):
+    r"""
+    Status describin life cycle of a Job.
+
+           parsing --> cancelled|failed
+              |
+    /--> initializing --> cancelled|failed
+    |         |
+    |      running --> cancelled|failed
+     \    /     |
+      retrying  post_processing --> cancelled|failed
+                   |
+                  done
+    """
+
+    # (A) doing something
+    # (A.1) parsing configuration
+    parsing = "parsing"
+    # (A.2) determine job tasks
+    initializing = "initializing"
+    # (A.3.a) processing has begun
+    running = "running"
+    # (A.3.b) processing has failed and is being retried; jumping back to (B.2)
+    retrying = "retrying"
+    # (A.4)
+    post_processing = "post_processing"
+
+    # (B) final stage
+    # (B.1) job successfully finished
+    done = "done"
+    # (B.2) job was cancelled
+    cancelled = "cancelled"
+    # (B.3) job failed
+    failed = "failed"
diff --git a/mapchete/executor/__init__.py b/mapchete/executor/__init__.py
index 96cc99af..877e614b 100644
--- a/mapchete/executor/__init__.py
+++ b/mapchete/executor/__init__.py
@@ -1,3 +1,6 @@
+from typing import Optional
+
+from mapchete.enums import Concurrency
 from mapchete.executor.base import ExecutorBase
 from mapchete.executor.concurrent_futures import (
     MULTIPROCESSING_DEFAULT_START_METHOD,
@@ -15,14 +18,22 @@ class Executor:
     Executor factory for dask and concurrent.futures executor
     """
 
-    def __new__(cls, *args, concurrency=None, **kwargs) -> ExecutorBase:
-        if concurrency == "dask":
+    def __new__(
+        cls, *args, concurrency: Optional[Concurrency] = None, **kwargs
+    ) -> ExecutorBase:
+        concurrency = (
+            Concurrency.dask
+            if kwargs.get("dask_scheduler") or kwargs.get("dask_client")
+            else concurrency
+        )
+
+        if concurrency == Concurrency.dask:
             return DaskExecutor(*args, **kwargs)
 
-        elif concurrency is None or kwargs.get("max_workers") == 1:
+        elif concurrency in [None, Concurrency.none]:
             return SequentialExecutor(*args, **kwargs)
 
-        elif concurrency in ["processes", "threads"]:
+        elif concurrency in [Concurrency.processes, Concurrency.threads]:
             return ConcurrentFuturesExecutor(*args, concurrency=concurrency, **kwargs)
 
         else:  # pragma: no cover
diff --git a/mapchete/executor/base.py b/mapchete/executor/base.py
index d520275f..f1fdf9ba 100644
--- a/mapchete/executor/base.py
+++ b/mapchete/executor/base.py
@@ -124,7 +124,7 @@ def _finished_future(
         self.finished_futures.discard(future)
 
         # create minimal Future-like object with no references to the cluster
-        mfuture = MFuture.from_future(future, lazy=False, result=result)
+        mfuture = MFuture.from_future(future, lazy=True, result=result)
 
         # raise exception if future errored or was cancelled
         mfuture.raise_if_failed()
diff --git a/mapchete/executor/concurrent_futures.py b/mapchete/executor/concurrent_futures.py
index ba53390b..e83e6da9 100644
--- a/mapchete/executor/concurrent_futures.py
+++ b/mapchete/executor/concurrent_futures.py
@@ -73,6 +73,9 @@ def __init__(
         )
         super().__init__(*args, **kwargs)
 
+    def __str__(self) -> str:
+        return f"<ConcurrentFuturesExecutor max_workers={self.max_workers}, cls={self._executor_cls}>"
+
     def as_completed(
         self,
         func,
diff --git a/mapchete/executor/dask.py b/mapchete/executor/dask.py
index 64caad7d..9fa5c0e5 100644
--- a/mapchete/executor/dask.py
+++ b/mapchete/executor/dask.py
@@ -1,14 +1,17 @@
 import logging
 import os
 from functools import cached_property
-from typing import Any, Iterator, List
+from sys import getsizeof
+from typing import Any, Callable, Iterable, Iterator, List, Optional, Union
 
+from dask.delayed import Delayed, DelayedLeaf
 from dask.distributed import Client, LocalCluster, as_completed, wait
 
 from mapchete.errors import JobCancelledError
 from mapchete.executor.base import ExecutorBase
 from mapchete.executor.future import MFuture
 from mapchete.executor.types import Result
+from mapchete.pretty import pretty_bytes
 from mapchete.timer import Timer
 
 logger = logging.getLogger(__name__)
@@ -20,9 +23,9 @@ class DaskExecutor(ExecutorBase):
     def __init__(
         self,
         *args,
-        dask_scheduler=None,
-        dask_client=None,
-        max_workers=None,
+        dask_scheduler: Optional[str] = None,
+        dask_client: Optional[Client] = None,
+        max_workers: int = os.cpu_count(),
         **kwargs,
     ):
         self.cancel_signal = False
@@ -31,9 +34,7 @@ def __init__(
         if self._executor_client:  # pragma: no cover
             logger.debug("using existing dask client: %s", dask_client)
         else:
-            local_cluster_kwargs = dict(
-                n_workers=max_workers or os.cpu_count(), threads_per_worker=1
-            )
+            local_cluster_kwargs = dict(n_workers=max_workers, threads_per_worker=1)
             self._executor_cls = Client
             if dask_scheduler is None:
                 self._local_cluster = LocalCluster(**local_cluster_kwargs)
@@ -47,7 +48,16 @@ def __init__(
         self._submitted = 0
         super().__init__(*args, **kwargs)
 
-    def map(self, func, iterable, fargs=None, fkwargs=None) -> List[Any]:
+    def __str__(self) -> str:
+        return f"<DaskExecutor dashboard_link={self._executor.dashboard_link}>"
+
+    def map(
+        self,
+        func: Callable,
+        iterable: Iterable,
+        fargs: Optional[tuple] = None,
+        fkwargs: Optional[dict] = None,
+    ) -> List[Any]:
         fargs = fargs or []
         fkwargs = fkwargs or {}
 
@@ -72,13 +82,13 @@ def _as_completed(self, *args, **kwargs) -> Iterator[MFuture]:  # pragma: no cov
 
     def as_completed(
         self,
-        func,
-        iterable,
-        fargs=None,
-        fkwargs=None,
-        max_submitted_tasks=500,
-        item_skip_bool=False,
-        chunksize=100,
+        func: Callable,
+        iterable: Iterable,
+        fargs: Optional[tuple] = None,
+        fkwargs: Optional[dict] = None,
+        max_submitted_tasks: int = 500,
+        item_skip_bool: bool = False,
+        chunksize: int = 100,
         **kwargs,
     ) -> Iterator[MFuture]:
         """
@@ -196,11 +206,12 @@ def as_completed(
 
     def compute_task_graph(
         self,
-        dask_collection=None,
-        with_results=False,
-        raise_errors=False,
+        dask_collection: List[Union[Delayed, DelayedLeaf]],
+        with_results: bool = False,
+        raise_errors: bool = False,
     ) -> Iterator[MFuture]:
         # send to scheduler
+        logger.debug("task graph has %s", pretty_bytes(getsizeof(dask_collection)))
 
         with Timer() as t:
             futures = self._executor.compute(
@@ -227,7 +238,13 @@ def compute_task_graph(
                     raise JobCancelledError()
                 yield self._finished_future(future, result, _dask=True)
 
-    def _submit_chunk(self, chunk=None, func=None, fargs=None, fkwargs=None):
+    def _submit_chunk(
+        self,
+        chunk: List[Any],
+        func: Callable,
+        fargs: Optional[tuple] = None,
+        fkwargs: Optional[dict] = None,
+    ) -> None:
         if chunk:
             logger.debug("submit chunk of %s items to cluster", len(chunk))
             futures = self._executor.map(
diff --git a/mapchete/executor/future.py b/mapchete/executor/future.py
index 07930738..8a4e93b6 100644
--- a/mapchete/executor/future.py
+++ b/mapchete/executor/future.py
@@ -87,10 +87,7 @@ def from_future(
         else:
             name = str(future)
 
-        if hasattr(future, "profiling"):
-            profiling = future.profiling
-        else:
-            profiling = {}
+        profiling = future.profiling if hasattr(future, "profiling") else {}
 
         if lazy:
             # keep around Future for later and don't call Future.result()
@@ -101,8 +98,9 @@ def from_future(
                 status=status,
                 name=name,
                 profiling=profiling,
+                exception=future.exception(),
             )
-        else:
+        else:  # pragma: no cover
             # immediately fetch Future.result() or use provided result
             try:
                 result = result or future.result(timeout=timeout)
@@ -119,10 +117,6 @@ def from_future(
                 profiling=profiling,
             )
 
-    @staticmethod
-    def from_result(result: Any, profiling: Optional[dict] = None) -> MFuture:
-        return MFuture(result=result, profiling=profiling)
-
     @staticmethod
     def skip(skip_info: Optional[Any] = None, result: Optional[Any] = None) -> MFuture:
         return MFuture(result=result, skip_info=skip_info, skipped=True)
diff --git a/mapchete/executor/sequential.py b/mapchete/executor/sequential.py
index 281a6827..d5ee08dc 100644
--- a/mapchete/executor/sequential.py
+++ b/mapchete/executor/sequential.py
@@ -15,6 +15,9 @@ def __init__(self, *args, **kwargs):
         logger.debug("init SequentialExecutor")
         super().__init__(*args, **kwargs)
 
+    def __str__(self) -> str:
+        return "<SequentialExecutor>"
+
     def as_completed(
         self, func, iterable, fargs=None, fkwargs=None, item_skip_bool=False, **kwargs
     ) -> Iterator[MFuture]:
diff --git a/mapchete/formats/default/_fiona_base.py b/mapchete/formats/default/_fiona_base.py
index 220094ce..068f1c1f 100644
--- a/mapchete/formats/default/_fiona_base.py
+++ b/mapchete/formats/default/_fiona_base.py
@@ -221,7 +221,7 @@ def read(self, validity_check=True, no_neighbors=False, **kwargs):
             raise NotImplementedError()
         return self._from_cache(validity_check=validity_check)
 
-    def is_empty(self, validity_check=True):
+    def is_empty(self, validity_check=True):  # pragma: no cover
         """
         Check if there is data within this tile.
 
diff --git a/mapchete/formats/default/gtiff.py b/mapchete/formats/default/gtiff.py
index 27dc608c..fffc11c8 100644
--- a/mapchete/formats/default/gtiff.py
+++ b/mapchete/formats/default/gtiff.py
@@ -586,8 +586,6 @@ def write(self, process_tile, data):
                     from_bounds(
                         *out_tile.bounds,
                         transform=self.dst.transform,
-                        height=self.dst.height,
-                        width=self.dst.width,
                     )
                     .round_lengths(pixel_precision=0)
                     .round_offsets(pixel_precision=0)
diff --git a/mapchete/formats/default/raster_file.py b/mapchete/formats/default/raster_file.py
index 636ee1be..f9dd1798 100644
--- a/mapchete/formats/default/raster_file.py
+++ b/mapchete/formats/default/raster_file.py
@@ -277,12 +277,20 @@ def read(self, indexes=None, resampling="nearest", **kwargs):
         data : array
         """
         if self._memory_cache_active:
+            logger.debug(
+                "available preprocessing tasks results: %s",
+                self.preprocessing_tasks_results,
+            )
             self._in_memory_raster = (
                 self._in_memory_raster
                 or self.preprocessing_tasks_results.get(self.cache_task_key)
             )
             if self._in_memory_raster is None:  # pragma: no cover
-                raise RuntimeError("preprocessing tasks have not yet been run")
+                raise RuntimeError(
+                    "preprocessing tasks have not yet been run "
+                    f"(task key {self.cache_task_key} not found in "
+                    f"{list(self.preprocessing_tasks_results.keys())})"
+                )
             return resample_from_array(
                 in_raster=ma.stack(
                     [
diff --git a/mapchete/formats/default/vector_file.py b/mapchete/formats/default/vector_file.py
index 23b28f13..5e0a1661 100644
--- a/mapchete/formats/default/vector_file.py
+++ b/mapchete/formats/default/vector_file.py
@@ -259,7 +259,11 @@ def read(self, validity_check=True, clip_to_crs_bounds=False, **kwargs):
                 or self.preprocessing_tasks_results.get(self.cache_task_key)
             )
             if self._in_memory_features is None:  # pragma: no cover
-                raise RuntimeError("preprocessing tasks have not yet been run")
+                raise RuntimeError(
+                    "preprocessing tasks have not yet been run "
+                    f"(task key {self.cache_task_key} not found in "
+                    f"{list(self.preprocessing_tasks_results.keys())})"
+                )
         return (
             []
             if self.is_empty()
diff --git a/mapchete/index.py b/mapchete/index.py
index 07205186..221d92f7 100644
--- a/mapchete/index.py
+++ b/mapchete/index.py
@@ -98,7 +98,8 @@ def zoom_index_gen(
     """
     if tile and zoom:  # pragma: no cover
         raise ValueError("tile and zoom cannot be used at the same time")
-    zoom = tile[0] if tile else zoom
+
+    zoom = tile.zoom if tile else zoom
     for zoom in get_zoom_levels(process_zoom_levels=zoom):
         with ExitStack() as es:
             # get index writers for all enabled formats
diff --git a/mapchete/path.py b/mapchete/path.py
index 6935a947..cda2c11e 100644
--- a/mapchete/path.py
+++ b/mapchete/path.py
@@ -300,7 +300,7 @@ def read_text(self) -> str:
     def makedirs(self, exist_ok: bool = True) -> None:
         """Create all parent directories for path."""
         # create parent directories on local filesystems
-        if self.fs.protocol == "file":
+        if "file" in self.fs.protocol:
             # if path has no suffix, assume a file path and only create parent directories
             logger.debug("create directory %s", str(self))
             self.fs.makedirs(self, exist_ok=exist_ok)
diff --git a/mapchete/pretty.py b/mapchete/pretty.py
new file mode 100644
index 00000000..a381299d
--- /dev/null
+++ b/mapchete/pretty.py
@@ -0,0 +1,31 @@
+def pretty_bytes(count: float, round_value: int = 2) -> str:
+    """Return human readable bytes."""
+    for measurement in [
+        "bytes",
+        "KiB",
+        "MiB",
+        "GiB",
+        "TiB",
+        "PiB",
+        "EiB",
+        "ZiB",
+        "YiB",
+    ]:
+        out = f"{round(count, round_value)} {measurement}"
+        if count < 1024.0:
+            break
+        count /= 1024.0
+
+    return out
+
+
+def pretty_seconds(elapsed_seconds: float, round_value: int = 3):
+    """Return human readable seconds."""
+    minutes, seconds = divmod(elapsed_seconds, 60)
+    hours, minutes = divmod(minutes, 60)
+    if hours:
+        return f"{int(hours)}h {int(minutes)}m {int(seconds)}s"
+    elif minutes:
+        return f"{int(minutes)}m {int(seconds)}s"
+    else:
+        return f"{round(seconds, round_value)}s"
diff --git a/mapchete/processing/__init__.py b/mapchete/processing/__init__.py
index 92544870..e5cb0246 100644
--- a/mapchete/processing/__init__.py
+++ b/mapchete/processing/__init__.py
@@ -1,5 +1,4 @@
 from mapchete.processing.base import Mapchete
-from mapchete.processing.job import Job
 from mapchete.processing.mp import MapcheteProcess
 
-__all__ = ["Mapchete", "Job", "MapcheteProcess"]
+__all__ = ["Mapchete", "MapcheteProcess"]
diff --git a/mapchete/processing/base.py b/mapchete/processing/base.py
index 60771cf3..c55b296a 100644
--- a/mapchete/processing/base.py
+++ b/mapchete/processing/base.py
@@ -2,36 +2,36 @@
 
 import json
 import logging
-import multiprocessing
+import os
 import threading
+from contextlib import ExitStack
 from typing import Any, Iterator, List, Optional, Tuple, Union
 
 from cachetools import LRUCache
+from shapely.geometry import Polygon, base
+from shapely.ops import unary_union
 
-from mapchete.config import MapcheteConfig
-from mapchete.enums import Concurrency
+from mapchete.config import DaskSettings, MapcheteConfig
+from mapchete.enums import Concurrency, ProcessingMode
 from mapchete.errors import MapcheteNodataTile, ReprojectionFailed
 from mapchete.executor import (
     MULTIPROCESSING_DEFAULT_START_METHOD,
+    DaskExecutor,
+    Executor,
     ExecutorBase,
     MFuture,
 )
 from mapchete.executor.base import func_partial
 from mapchete.executor.types import Profiler
 from mapchete.path import batch_sort_property, tiles_exist
-from mapchete.processing.compute import (
-    PreprocessingProcessInfo,
-    TileProcessInfo,
-    _execute,
-    _execute_and_write,
-    _filter_skipable,
-    _preprocess,
-    _preprocess_task_wrapper,
-    _run_area,
-    _run_on_single_tile,
-    compute,
+from mapchete.processing.execute import batches, dask_graph, single_batch
+from mapchete.processing.tasks import (
+    TaskBatch,
+    TaskInfo,
+    Tasks,
+    TileTask,
+    TileTaskBatch,
 )
-from mapchete.processing.tasks import TaskBatch, TileTask, TileTaskBatch
 from mapchete.stac import tile_direcotry_item_to_dict, update_tile_directory_stac_item
 from mapchete.tile import BufferedTile, count_tiles
 from mapchete.timer import Timer
@@ -62,7 +62,11 @@ class Mapchete(object):
         process output data cached in memory
     """
 
-    def __init__(self, config: MapcheteConfig, with_cache: bool = False):
+    def __init__(
+        self,
+        config: MapcheteConfig,
+        with_cache: bool = False,
+    ):
         """
         Initialize Mapchete processing endpoint.
 
@@ -77,7 +81,9 @@ def __init__(self, config: MapcheteConfig, with_cache: bool = False):
         if not isinstance(config, MapcheteConfig):
             raise TypeError("config must be MapcheteConfig object")
         self.config = config
-        self.with_cache = True if self.config.mode == "memory" else with_cache
+        self.with_cache = (
+            True if self.config.mode == ProcessingMode.MEMORY else with_cache
+        )
         if self.with_cache:
             self.process_tile_cache = LRUCache(maxsize=512)
             self.current_processes = {}
@@ -116,7 +122,7 @@ def get_process_tiles(
             ):
                 yield tile
         else:
-            for i in self.config.zoom_levels.descending():
+            for i in self.config.init_zoom_levels.descending():
                 for tile in self.config.process_pyramid.tiles_from_geom(
                     self.config.area_at_zoom(i), zoom=i, batch_by=batch_by, exact=True
                 ):
@@ -143,7 +149,7 @@ def skip_tiles(
         """
         logger.debug("determine which tiles to skip...")
         # only check for existing output in "continue" mode
-        if self.config.mode == "continue":
+        if self.config.mode == ProcessingMode.CONTINUE:
             yield from tiles_exist(
                 config=self.config,
                 process_tiles=tiles,
@@ -151,7 +157,7 @@ def skip_tiles(
             )
         # otherwise don't skip tiles
         else:
-            if tiles_batches:
+            if tiles_batches:  # pragma: no cover
                 for batch in tiles_batches:
                     for tile in batch:
                         yield (tile, False)
@@ -159,259 +165,139 @@ def skip_tiles(
                 for tile in tiles:
                     yield (tile, False)
 
-    def task_batches(
+    def tasks(
         self,
         zoom: Optional[ZoomLevelsLike] = None,
         tile: Optional[TileLike] = None,
-        skip_output_check: bool = False,
-        propagate_results: bool = True,
         profilers: Optional[List[Profiler]] = None,
-    ) -> Iterator[Union[TaskBatch, TileTaskBatch]]:
+    ) -> Tasks:
         """
-        Generate task batches from preprocessing tasks and tile tasks.
+        Generate tasks from preprocessing tasks and tile tasks.
         """
-        yield from _task_batches(
-            self,
-            zoom=zoom,
-            tile=tile,
-            skip_output_check=skip_output_check,
-            propagate_results=propagate_results,
-            profilers=profilers,
+        return Tasks(
+            _task_batches(
+                self,
+                zoom=zoom,
+                tile=tile,
+                profilers=profilers,
+            ),
         )
 
-    def compute(
+    def preprocessing_tasks(
         self,
-        zoom: Optional[ZoomLevelsLike] = None,
-        tile: Optional[TileLike] = None,
-        executor: Optional[ExecutorBase] = None,
-        concurrency: Concurrency = Concurrency.processes,
-        workers: int = multiprocessing.cpu_count(),
-        multiprocessing_start_method: Optional[str] = None,
-        skip_output_check: bool = False,
-        dask_scheduler: Optional[str] = None,
-        dask_compute_graph: bool = True,
-        dask_propagate_results: bool = True,
-        dask_max_submitted_tasks: bool = 500,
-        raise_errors: bool = True,
-        with_results: bool = False,
-        **kwargs,
-    ) -> Iterator[MFuture]:
-        """Compute preprocessing tasks and tile tasks in one go."""
-        yield from compute(
-            self,
-            zoom_levels=(
-                self.config.zoom_levels if zoom is None else ZoomLevels.from_inp(zoom)
+        profilers: Optional[List[Profiler]] = None,
+    ) -> Tasks:
+        """
+        Generate tasks from preprocessing tasks and tile tasks.
+        """
+        return Tasks(
+            _preprocessing_task_batches(
+                self,
+                profilers=profilers,
             ),
-            tile=self.config.process_pyramid.tile(*tile) if tile else None,
-            executor=executor,
-            concurrency=concurrency,
-            workers=workers,
-            multiprocessing_start_method=multiprocessing_start_method,
-            skip_output_check=skip_output_check,
-            dask_scheduler=dask_scheduler,
-            dask_compute_graph=dask_compute_graph,
-            dask_propagate_results=dask_propagate_results,
-            dask_max_submitted_tasks=dask_max_submitted_tasks,
-            raise_errors=raise_errors,
-            with_results=with_results,
-            **kwargs,
         )
 
-    def batch_preprocessor(
+    def execute(
         self,
-        dask_scheduler: Optional[str] = None,
-        dask_max_submitted_tasks: int = 500,
-        dask_chunksize: int = 100,
-        workers: Optional[int] = None,
+        tasks: Optional[Tasks] = None,
+        zoom: Optional[ZoomLevelsLike] = None,
+        tile: Optional[TileLike] = None,
         executor: Optional[ExecutorBase] = None,
-    ) -> Iterator[MFuture]:
+        concurrency: Concurrency = Concurrency.processes,
+        workers: int = os.cpu_count(),
+        propagate_results: bool = False,
+        dask_settings: DaskSettings = DaskSettings(),
+        remember_preprocessing_results: bool = False,
+    ) -> Iterator[TaskInfo]:
         """
-        Run all required preprocessing steps and yield over results.
+        Execute all tasks on given executor and yield TaskInfo as they finish.
+        """
+        # determine tasks if not provided extra
+        # we have to do this before it can be decided which type of processing can be applied
+        tasks = self.tasks(zoom=zoom, tile=tile) if tasks is None else tasks
 
-        The task count can be determined by self.config.preprocessing_tasks_count().
+        if len(tasks) == 0:
+            return
 
-        Parameters
-        ----------
+        with ExitStack() as exit_stack:
+            # create a default executor if not available
+            if executor is None:
+                executor = exit_stack.enter_context(
+                    Executor(concurrency=concurrency, workers=workers),
+                )
 
-        dask_schedulter : str
-            URL to a dask scheduler if distributed execution is desired
-        dask_max_submitted_tasks : int
-            Make sure that not more tasks are submitted to dask scheduler at once. (default: 500)
-        dask_chunksize : int
-            Number of tasks submitted to the scheduler at once. (default: 100)
-        workers : int
-            number of workers to be used for local processing
-        executor : mapchete.Executor
-            optional executor class to be used for processing
+            # tasks have no dependencies with each other and can be executed in
+            # any arbitrary order
+            if self.config.preprocessing_tasks_count() == 0 and (
+                not self.config.baselevels or len(self.config.init_zoom_levels) == 1
+            ):
+                logger.debug("decided to process tasks in single batch")
+                yield from single_batch(
+                    executor,
+                    tasks,
+                    output_writer=self.config.output,
+                    write_in_parent_process=self.config.output.write_in_parent_process,
+                    propagate_results=propagate_results,
+                )
 
-        """
-        # process everything using executor and yield from results
-        yield from _preprocess(
-            self.config.preprocessing_tasks(),
-            process=self,
-            dask_scheduler=dask_scheduler,
-            dask_max_submitted_tasks=dask_max_submitted_tasks,
-            dask_chunksize=dask_chunksize,
-            workers=workers,
-            executor=executor,
-        )
+            # tasks are connected via a dependency graph and will be sent to the
+            # executor all at once
+            elif dask_settings.process_graph and hasattr(
+                executor, "compute_task_graph"
+            ):
+                logger.debug("decided to use dask graph processing")
+                for task_info in dask_graph(
+                    executor,
+                    tasks,
+                    output_writer=self.config.output,
+                    write_in_parent_process=self.config.output.write_in_parent_process,
+                    propagate_results=propagate_results,
+                ):
+                    # TODO: is this really necessary?
+                    if remember_preprocessing_results and task_info.tile is None:
+                        self.config.set_preprocessing_task_result(
+                            task_info.id, task_info.output
+                        )
 
-    def batch_preprocess(
-        self,
-        dask_scheduler: Optional[str] = None,
-        dask_max_submitted_tasks: int = 500,
-        dask_chunksize: int = 100,
-        workers: Optional[int] = None,
-        executor: Optional[ExecutorBase] = None,
-    ) -> None:
-        """
-        Run all required preprocessing steps.
+                    yield task_info
 
-        Parameters
-        ----------
+            # tasks are sorted into batches which have to be executed in a
+            # particular order
+            else:
+                logger.debug("decided to process tasks in batches")
+                for task_info in batches(
+                    executor,
+                    tasks,
+                    output_writer=self.config.output,
+                    write_in_parent_process=self.config.output.write_in_parent_process,
+                    propagate_results=propagate_results,
+                ):
+                    # TODO: is this really necessary?
+                    if remember_preprocessing_results and task_info.tile is None:
+                        self.config.set_preprocessing_task_result(
+                            task_info.id, task_info.output
+                        )
 
-        dask_schedulter : str
-            URL to a dask scheduler if distributed execution is desired
-        dask_max_submitted_tasks : int
-            Make sure that not more tasks are submitted to dask scheduler at once. (default: 500)
-        dask_chunksize : int
-            Number of tasks submitted to the scheduler at once. (default: 100)
-        workers : int
-            number of workers to be used for local processing
-        executor : mapchete.Executor
-            optional executor class to be used for processing
-        """
-        list(
-            self.batch_preprocessor(
-                dask_scheduler=dask_scheduler,
-                dask_max_submitted_tasks=dask_max_submitted_tasks,
-                dask_chunksize=dask_chunksize,
-                workers=workers,
-                executor=executor,
-            )
-        )
+                    yield task_info
 
-    def batch_process(
+    def execute_preprocessing_tasks(
         self,
-        zoom: Optional[ZoomLevelsLike] = None,
-        tile: Optional[TileLike] = None,
-        dask_scheduler: Optional[str] = None,
-        dask_max_submitted_tasks: int = 500,
-        dask_chunksize: int = 100,
-        workers: Optional[int] = None,
-        multiprocessing_start_method: str = MULTIPROCESSING_DEFAULT_START_METHOD,
-        skip_output_check: bool = False,
         executor: Optional[ExecutorBase] = None,
-    ) -> None:
-        """
-        Process a large batch of tiles.
-
-        Parameters
-        ----------
-        process : MapcheteProcess
-            process to be run
-        zoom : list or int
-            either single zoom level or list of minimum and maximum zoom level;
-            None processes all (default: None)
-        tile : tuple
-            zoom, row and column of tile to be processed (cannot be used with
-            zoom)
-        workers : int
-            number of workers (default: number of CPU cores)
-        dask_schedulter : str
-            URL to a dask scheduler if distributed execution is desired
-        dask_max_submitted_tasks : int
-            Make sure that not more tasks are submitted to dask scheduler at once. (default: 500)
-        dask_chunksize : int
-            Number of tasks submitted to the scheduler at once. (default: 100)
-        multiprocessing_start_method : str
-            "fork", "forkserver" or "spawn"
-            (default: "spawn")
-        skip_output_check : bool
-            skip checking whether process tiles already have existing output before
-            starting to process;
-        executor : mapchete.Executor
-            optional executor class to be used for processing
+    ) -> Iterator[MFuture]:
         """
-        list(
-            self.batch_processor(
-                zoom=zoom,
-                tile=tile,
-                dask_scheduler=dask_scheduler,
-                dask_max_submitted_tasks=dask_max_submitted_tasks,
-                dask_chunksize=dask_chunksize,
-                workers=workers,
-                multiprocessing_start_method=multiprocessing_start_method,
-                skip_output_check=skip_output_check,
-                executor=executor,
-            )
-        )
-
-    def batch_processor(
-        self,
-        zoom: Optional[ZoomLevelsLike] = None,
-        tile: Optional[TileLike] = None,
-        dask_scheduler: Optional[str] = None,
-        dask_max_submitted_tasks: int = 500,
-        dask_chunksize: int = 100,
-        workers: Optional[int] = None,
-        multiprocessing_start_method: str = MULTIPROCESSING_DEFAULT_START_METHOD,
-        skip_output_check: bool = False,
-        executor: Optional[ExecutorBase] = None,
-    ) -> Union[PreprocessingProcessInfo, TileProcessInfo]:
+        Run all required preprocessing steps.
         """
-        Process a large batch of tiles and yield report messages per tile.
+        # If preprocessing tasks already finished, don't run them again.
+        if self.config.preprocessing_tasks_finished:  # pragma: no cover
+            return
 
-        Parameters
-        ----------
-        zoom : list or int
-            either single zoom level or list of minimum and maximum zoom level;
-            None processes all (default: None)
-        tile : tuple
-            zoom, row and column of tile to be processed (cannot be used with
-            zoom)
-        dask_schedulter : str
-            URL to a dask scheduler if distributed execution is desired
-        dask_max_submitted_tasks : int
-            Make sure that not more tasks are submitted to dask scheduler at once. (default: 500)
-        dask_chunksize : int
-            Number of tasks submitted to the scheduler at once. (default: 100)
-        multiprocessing_start_method : str
-            "fork", "forkserver" or "spawn"
-            (default: "spawn")
-        skip_output_check : bool
-            skip checking whether process tiles already have existing output before
-            starting to process;
-        executor : mapchete.Executor
-            optional executor class to be used for processing
-        """
-        if zoom and tile:
-            raise ValueError("use either zoom or tile")
+        for task_info in self.execute(
+            tasks=self.preprocessing_tasks(),
+            executor=executor,
+        ):
+            self.config.set_preprocessing_task_result(task_info.id, task_info.output)
 
-        # run single tile
-        if tile:
-            yield _run_on_single_tile(
-                executor=executor,
-                dask_scheduler=dask_scheduler,
-                process=self,
-                tile=self.config.process_pyramid.tile(*tuple(tile)),
-            ).result()
-        # run area
-        else:
-            for future in _run_area(
-                process=self,
-                zoom_levels=self.config.zoom_levels
-                if zoom is None
-                else ZoomLevels.from_inp(zoom),
-                dask_scheduler=dask_scheduler,
-                dask_max_submitted_tasks=dask_max_submitted_tasks,
-                dask_chunksize=dask_chunksize,
-                workers=workers or multiprocessing.cpu_count(),
-                multiprocessing_start_method=multiprocessing_start_method,
-                skip_output_check=skip_output_check,
-                executor=executor,
-            ):
-                yield future.result()
+        self.config.preprocessing_tasks_finished = True
 
     def count_tasks(
         self,
@@ -476,7 +362,9 @@ def count_tiles(
             logger.debug("tiles counted in %s", t)
         return self._count_tiles_cache[(minzoom, maxzoom)]
 
-    def execute(self, process_tile: BufferedTile, raise_nodata: bool = False) -> Any:
+    def execute_tile(
+        self, process_tile: BufferedTile, raise_nodata: bool = False
+    ) -> Any:
         """
         Run Mapchete process on a tile.
 
@@ -494,7 +382,8 @@ def execute(self, process_tile: BufferedTile, raise_nodata: bool = False) -> Any
             process output
         """
         process_tile = validate_tile(process_tile, self.config.process_pyramid)
-        self.batch_preprocess()
+        # make sure preprocessing tasks are finished
+        self.execute_preprocessing_tasks()
         try:
             return self.config.output.streamline_output(
                 TileTask(tile=process_tile, config=self.config).execute()
@@ -520,11 +409,15 @@ def read(self, output_tile: TileLike) -> Any:
             process output
         """
         output_tile = validate_tile(output_tile, self.config.output_pyramid)
-        if self.config.mode not in ["readonly", "continue", "overwrite"]:
+        if self.config.mode not in [
+            ProcessingMode.READONLY,
+            ProcessingMode.CONTINUE,
+            ProcessingMode.OVERWRITE,
+        ]:
             raise ValueError("process mode must be readonly, continue or overwrite")
         return self.config.output.read(output_tile)
 
-    def write(self, process_tile: TileLike, data: Any):
+    def write(self, process_tile: TileLike, data: Any) -> TaskInfo:
         """
         Write data into output format.
 
@@ -536,15 +429,15 @@ def write(self, process_tile: TileLike, data: Any):
             data to be written
         """
         process_tile = validate_tile(process_tile, self.config.process_pyramid)
-        if self.config.mode not in ["continue", "overwrite"]:
+        if self.config.mode not in [ProcessingMode.CONTINUE, ProcessingMode.OVERWRITE]:
             raise ValueError("cannot write output in current process mode")
 
-        if self.config.mode == "continue" and (
+        if self.config.mode == ProcessingMode.CONTINUE and (
             self.config.output.tiles_exist(process_tile)
         ):
             message = "output exists, not overwritten"
             logger.debug((process_tile.id, message))
-            return TileProcessInfo(
+            return TaskInfo(
                 tile=process_tile,
                 processed=False,
                 process_msg=None,
@@ -554,7 +447,7 @@ def write(self, process_tile: TileLike, data: Any):
         elif data is None:
             message = "output empty, nothing written"
             logger.debug((process_tile.id, message))
-            return TileProcessInfo(
+            return TaskInfo(
                 tile=process_tile,
                 processed=False,
                 process_msg=None,
@@ -566,7 +459,7 @@ def write(self, process_tile: TileLike, data: Any):
                 self.config.output.write(process_tile=process_tile, data=data)
             message = "output written in %s" % t
             logger.debug((process_tile.id, message))
-            return TileProcessInfo(
+            return TaskInfo(
                 tile=process_tile,
                 processed=False,
                 process_msg=None,
@@ -609,7 +502,7 @@ def get_raw_output(self, tile: TileLike, _baselevel_readonly: bool = False) -> A
                 "reprojection between processes not yet implemented"
             )
 
-        if self.config.mode == "memory":
+        if self.config.mode == ProcessingMode.MEMORY:
             # Determine affected process Tile and check whether it is already
             # cached.
             process_tile = self.config.process_pyramid.intersecting(tile)[0]
@@ -630,17 +523,17 @@ def get_raw_output(self, tile: TileLike, _baselevel_readonly: bool = False) -> A
         else:
             output_tiles = self.config.output_pyramid.intersecting(tile)
 
-        if self.config.mode == "readonly" or _baselevel_readonly:
+        if self.config.mode == ProcessingMode.READONLY or _baselevel_readonly:
             if self.config.output.tiles_exist(process_tile):
                 return self._read_existing_output(tile, output_tiles)
             else:
                 return self.config.output.empty(tile)
-        elif self.config.mode == "continue" and not _baselevel_readonly:
+        elif self.config.mode == ProcessingMode.CONTINUE and not _baselevel_readonly:
             if self.config.output.tiles_exist(process_tile):
                 return self._read_existing_output(tile, output_tiles)
             else:
                 return self._process_and_overwrite_output(tile, process_tile)
-        elif self.config.mode == "overwrite" and not _baselevel_readonly:
+        elif self.config.mode == ProcessingMode.OVERWRITE and not _baselevel_readonly:
             return self._process_and_overwrite_output(tile, process_tile)
 
     def write_stac(self, indent: int = 4) -> None:
@@ -657,8 +550,8 @@ def write_stac(self, indent: int = 4) -> None:
             return
 
         if not self.config.output.use_stac or self.config.mode in [
-            "readonly",
-            "memory",
+            ProcessingMode.READONLY,
+            ProcessingMode.MEMORY,
         ]:
             return
         # read existing STAC file
@@ -694,8 +587,7 @@ def _process_and_overwrite_output(self, tile, process_tile):
         if self.with_cache:
             output = self._execute_using_cache(process_tile)
         else:
-            output = self.execute(process_tile)
-
+            output = self.execute_tile(process_tile)
         self.write(process_tile, output)
         return self._extract(in_tile=process_tile, in_data=output, out_tile=tile)
 
@@ -723,9 +615,12 @@ def _execute_using_cache(self, process_tile):
                 return self.process_tile_cache[process_tile.id]
             else:
                 try:
-                    output = self.execute(process_tile)
+                    output = self.execute_tile(process_tile)
                     self.process_tile_cache[process_tile.id] = output
-                    if self.config.mode in ["continue", "overwrite"]:
+                    if self.config.mode in [
+                        ProcessingMode.CONTINUE,
+                        ProcessingMode.OVERWRITE,
+                    ]:
                         self.write(process_tile, output)
                     return self.process_tile_cache[process_tile.id]
                 finally:
@@ -767,48 +662,118 @@ def __exit__(self, exc_type=None, exc_value=None, exc_traceback=None):
             self.process_lock = None
 
     def __repr__(self):  # pragma: no cover
-        return f"Mapchete <process_name={self.config.process.name}>"
+        return (
+            f"<Mapchete process_name={self.config.process.name}, config={self.config}>"
+        )
 
 
 def _task_batches(
     process: Mapchete,
     zoom: Optional[ZoomLevelsLike] = None,
     tile: Optional[TileLike] = None,
-    skip_output_check: bool = False,
-    propagate_results: bool = True,
     profilers: Optional[List[Profiler]] = None,
 ) -> Iterator[Union[TaskBatch, TileTaskBatch]]:
     """Create task batches for each processing stage."""
     profilers = profilers or []
+    if tile:
+        tile = process.config.process_pyramid.tile(*tile)
+
+    # first, materialize tile task batches to determine process AOI
+    tile_task_batches = _tile_task_batches(
+        process=process,
+        zoom=zoom,
+        tile=tile,
+        profilers=profilers,
+    )
+
+    # create processing AOI (i.e. processing area without overviews)
+    if process.config.preprocessing_tasks().values():
+        zoom_aois = []
+        for zoom in process.config.processing_levels:
+            for batch in tile_task_batches:
+                if batch.id == f"zoom-{zoom}":
+                    zoom_aois.append(batch.geometry)
+        process_aoi = unary_union(zoom_aois) if zoom_aois else Polygon()
+    else:
+        process_aoi = None
+
+    yield from _preprocessing_task_batches(
+        process=process, profilers=profilers, process_aoi=process_aoi
+    )
+
+    yield from tile_task_batches
+
+
+def _preprocessing_task_batches(
+    process: Mapchete,
+    profilers: Optional[List[Profiler]] = None,
+    process_aoi: Optional[base.BaseGeometry] = None,
+) -> Iterator[TaskBatch]:
     with Timer() as duration:
+        tasks = []
+        for task in process.config.preprocessing_tasks().values():
+            if isinstance(process_aoi, base.BaseGeometry) and task.has_geometry():
+                if task.geometry.intersects(process_aoi):
+                    tasks.append(task)
+            else:
+                tasks.append(task)
+
         # preprocessing tasks
-        yield TaskBatch(
+        preprocessing_batch = TaskBatch(
             id="preprocessing_tasks",
-            tasks=process.config.preprocessing_tasks().values(),
-            func=_preprocess_task_wrapper,
+            tasks=tasks,
             profilers=profilers,
         )
+        if len(preprocessing_batch):
+            yield preprocessing_batch
     logger.debug("preprocessing tasks batch generated in %s", duration)
+
+
+def _tile_task_batches(
+    process: Mapchete,
+    zoom: Optional[ZoomLevelsLike] = None,
+    tile: Optional[TileLike] = None,
+    profilers: Optional[List[Profiler]] = None,
+) -> List[TileTaskBatch]:
     with Timer() as duration:
+        batches = []
+
         if tile:
-            zoom_levels = ZoomLevels.from_inp(tile.zoom)
-            skip_output_check = True
-            tiles = {tile.zoom: [(tile, False)]}
+            # nothing to process here
+            if (
+                process.config.mode == ProcessingMode.CONTINUE
+                and process.config.output_reader.tiles_exist(tile)
+            ):
+                pass
+
+            # process this tile
+            else:
+                batches = [
+                    TileTaskBatch(
+                        id=f"zoom-{tile.zoom}",
+                        tasks=[
+                            TileTask(
+                                tile=tile,
+                                config=process.config,
+                            )
+                        ],
+                        profilers=profilers,
+                    )
+                ]
+
         else:
             zoom_levels = (
-                process.config.zoom_levels
+                process.config.init_zoom_levels
                 if zoom is None
                 else ZoomLevels.from_inp(zoom)
             )
-            tiles = {}
-
             # here we store the parents of tiles about to be processed so we can update overviews
             # also in "continue" mode in case there were updates at the baselevel
             overview_parents = set()
             for i, zoom in enumerate(zoom_levels.descending()):
-                tiles[zoom] = []
+                tile_tasks = []
 
-                for tile, skip, _ in _filter_skipable(
+                for tile in _filter_skipable(
                     process=process,
                     tiles_batches=process.get_process_tiles(
                         zoom,
@@ -816,52 +781,48 @@ def _task_batches(
                             process.config.output_reader.tile_path_schema
                         ),
                     ),
-                    target_set=(
+                    overview_tiles=(
                         overview_parents if process.config.baselevels and i else None
                     ),
-                    skip_output_check=skip_output_check,
                 ):
-                    tiles[zoom].append((tile, skip))
+                    # we don't need the current tile anymore
+                    overview_parents.discard(tile)
+
+                    # add tile task to batch
+                    tile_tasks.append(TileTask(tile=tile, config=process.config))
+
                     # in case of building overviews from baselevels, remember which parent
                     # tile needs to be updated later on
-                    if (
-                        not skip_output_check
-                        and process.config.baselevels
-                        and tile.zoom > 0
-                    ):
-                        # add parent tile
+                    if process.config.baselevels and tile.zoom > 0:
+                        # add parent tile to be reprocessed
                         overview_parents.add(tile.get_parent())
-                        # we don't need the current tile anymore
-                    overview_parents.discard(tile)
-
-        if process.config.output.write_in_parent_process:
-            func = func_partial(_execute, profilers=profilers)
-            fkwargs = dict(append_data=propagate_results)
-        else:
-            func = func_partial(_execute_and_write, profilers=profilers)
-            fkwargs = dict(
-                append_data=propagate_results, output_writer=process.config.output
-            )
 
-        # tile tasks
-        for zoom in zoom_levels.descending():
-            yield TileTaskBatch(
-                id=f"zoom_{zoom}",
-                tasks=(
-                    TileTask(
-                        tile=tile,
-                        config=process.config,
-                        skip=(
-                            process.config.mode == "continue"
-                            and process.config.output_reader.tiles_exist(tile)
-                        )
-                        if skip_output_check
-                        else skip,
+                batches.append(
+                    TileTaskBatch(
+                        id=f"zoom-{zoom}",
+                        tasks=tile_tasks,
+                        profilers=profilers,
                     )
-                    for tile, skip in tiles[zoom]
-                ),
-                func=func,
-                fkwargs=fkwargs,
-                profilers=profilers,
-            )
+                )
     logger.debug("tile task batches generated in %s", duration)
+    return batches
+
+
+def _filter_skipable(
+    process: Mapchete,
+    tiles_batches: Iterator[Iterator[BufferedTile]],
+    overview_tiles: Optional[set] = None,
+) -> Iterator[Tuple[BufferedTile, bool]]:
+    # we don't filter here
+    if process.config.mode == ProcessingMode.OVERWRITE:
+        for batch in tiles_batches:
+            yield from batch
+
+    # only yield tiles which don't yet exist or need to be reprocessed
+    else:
+        overview_tiles = overview_tiles or set()
+        for tile, skip in process.skip_tiles(tiles_batches=tiles_batches):
+            if skip and tile not in overview_tiles:
+                pass
+            else:
+                yield tile
diff --git a/mapchete/processing/compute.py b/mapchete/processing/compute.py
deleted file mode 100644
index 987b5f1d..00000000
--- a/mapchete/processing/compute.py
+++ /dev/null
@@ -1,709 +0,0 @@
-import logging
-import multiprocessing
-from contextlib import ExitStack
-from typing import Any, Iterator, Optional
-
-from mapchete.enums import Concurrency
-from mapchete.errors import MapcheteNodataTile
-from mapchete.executor import DaskExecutor, Executor, ExecutorBase
-from mapchete.executor.future import MFuture
-from mapchete.executor.types import Profiler
-from mapchete.path import batch_sort_property
-from mapchete.processing.profilers import preconfigured_profilers
-from mapchete.processing.tasks import TileTask, to_dask_collection
-from mapchete.processing.types import PreprocessingProcessInfo, TileProcessInfo
-from mapchete.tile import BufferedTile
-from mapchete.timer import Timer
-from mapchete.types import ZoomLevels, ZoomLevelsLike
-
-logger = logging.getLogger(__name__)
-
-
-# TODO: this function probably better goes to base
-def compute(
-    process,
-    zoom_levels: Optional[ZoomLevelsLike] = None,
-    tile: Optional[BufferedTile] = None,
-    executor: Optional[ExecutorBase] = None,
-    concurrency: Concurrency = Concurrency.processes,
-    workers: int = multiprocessing.cpu_count(),
-    multiprocessing_start_method: Optional[str] = None,
-    skip_output_check: bool = False,
-    dask_scheduler: Optional[str] = None,
-    dask_compute_graph: bool = True,
-    dask_propagate_results: bool = True,
-    dask_max_submitted_tasks: bool = 500,
-    raise_errors: bool = True,
-    with_results: bool = False,
-    profiling: bool = False,
-    **kwargs,
-) -> Iterator[MFuture]:
-    """Computes all tasks and yields progress."""
-    concurrency = "dask" if dask_scheduler else concurrency
-    num_processed = 0
-    with ExitStack() as exit_stack:
-        # create fresh Executor if it was not passed on
-        if executor is None:
-            executor = exit_stack.enter_context(
-                Executor(
-                    max_workers=workers,
-                    concurrency=concurrency,
-                    start_method=multiprocessing_start_method,
-                    dask_scheduler=dask_scheduler,
-                )
-            )
-        logger.info("run process on area")
-        duration = exit_stack.enter_context(Timer())
-
-        if tile:
-            zoom_levels = ZoomLevels.from_inp(tile.zoom)
-        elif zoom_levels is None:  # pragma: no cover
-            raise ValueError("either tile or zoom_levels has to be provided")
-
-        profilers = []
-        if profiling:
-            profilers = preconfigured_profilers
-            for profiler in profilers:
-                executor.add_profiler(profiler)
-
-        if dask_compute_graph and isinstance(executor, DaskExecutor):
-            # TODO optimize memory management, e.g. delete preprocessing tasks from input
-            # once the dask graph is ready.
-            # materialize all tasks including dependencies
-            with Timer() as t:
-                dask_collection = to_dask_collection(
-                    process.task_batches(
-                        zoom=zoom_levels,
-                        tile=tile,
-                        skip_output_check=skip_output_check,
-                        propagate_results=process.config.output.write_in_parent_process
-                        or dask_propagate_results,
-                        profilers=profilers,
-                    )
-                )
-            logger.debug(
-                "dask collection with %s tasks generated in %s", len(dask_collection), t
-            )
-            for num_processed, future in enumerate(
-                _compute_task_graph(
-                    dask_collection=dask_collection,
-                    executor=executor,
-                    with_results=with_results,
-                    write_in_parent_process=process.config.output.write_in_parent_process,
-                    raise_errors=raise_errors,
-                    output_writer=process.config.output,
-                ),
-                1,
-            ):
-                logger.debug("task %s finished: %s", num_processed, future)
-                if raise_errors:
-                    future.raise_if_failed()
-                yield future
-        else:
-            for num_processed, future in enumerate(
-                _compute_tasks(
-                    executor=executor,
-                    process=process,
-                    zoom_levels=zoom_levels,
-                    tile=tile,
-                    skip_output_check=skip_output_check,
-                    dask_max_submitted_tasks=dask_max_submitted_tasks,
-                    **kwargs,
-                ),
-                1,
-            ):
-                logger.debug("task %s finished: %s", num_processed, future)
-                future.raise_if_failed()
-                yield future
-
-    logger.info("computed %s tasks in %s", num_processed, duration)
-
-
-def _compute_task_graph(
-    dask_collection,
-    executor: DaskExecutor,
-    with_results: bool = False,
-    write_in_parent_process: bool = False,
-    raise_errors: bool = False,
-    output_writer: Optional[Any] = None,
-    **kwargs,
-) -> Iterator[MFuture]:
-    # send task graph to executor and yield as ready
-    for future in executor.compute_task_graph(
-        dask_collection, with_results=with_results, raise_errors=raise_errors
-    ):
-        if write_in_parent_process:
-            yield MFuture.from_result(
-                result=_write(
-                    process_info=future.result(),
-                    output_writer=output_writer,
-                    append_output=True,
-                ),
-                profiling=future.profiling,
-            )
-        else:
-            yield MFuture.from_future(future)
-
-
-def _compute_tasks(
-    executor=None,
-    process=None,
-    workers=None,
-    zoom_levels=None,
-    tile=None,
-    skip_output_check=False,
-    dask_max_submitted_tasks=None,
-    dask_chunksize=None,
-    **kwargs,
-) -> Iterator[MFuture]:
-    if not process.config.preprocessing_tasks_finished:
-        tasks = process.config.preprocessing_tasks()
-        logger.info(
-            "run preprocessing on %s tasks using %s workers", len(tasks), workers
-        )
-        # process all remaining tiles using todo list from before
-        for future in executor.as_completed(
-            func=_preprocess_task_wrapper,
-            iterable=tasks.values(),
-            fkwargs=dict(append_data=True),
-            max_submitted_tasks=dask_max_submitted_tasks,
-            chunksize=dask_chunksize,
-            **kwargs,
-        ):
-            future.raise_if_failed()
-            result = future.result()
-            process.config.set_preprocessing_task_result(result.task_key, result.data)
-            yield future
-
-    # run single tile
-    if tile:
-        logger.info("run process on single tile")
-        for future in executor.as_completed(
-            func=_execute_and_write,
-            iterable=[
-                TileTask(
-                    tile=tile,
-                    config=process.config,
-                    skip=(
-                        process.config.mode == "continue"
-                        and process.config.output_reader.tiles_exist(tile)
-                    ),
-                ),
-            ],
-            fkwargs=dict(output_writer=process.config.output),
-        ):
-            future.raise_if_failed()
-            yield future
-
-    else:
-        # for output drivers requiring writing data in parent process
-        if process.config.output.write_in_parent_process:
-            func = _execute
-            fkwargs = dict(append_data=True)
-            write_in_parent_process = True
-
-        # for output drivers which can write data in child processes
-        else:
-            func = _execute_and_write
-            fkwargs = dict(append_data=False, output_writer=process.config.output)
-            write_in_parent_process = False
-
-        # for outputs which have overviews
-        if process.config.baselevels:
-            _process_batches = _run_multi_overviews
-        # for outputs with no overviews
-        else:
-            _process_batches = _run_multi_no_overviews
-
-        for future in _process_batches(
-            zoom_levels=zoom_levels,
-            executor=executor,
-            func=func,
-            process=process,
-            skip_output_check=skip_output_check,
-            fkwargs=fkwargs,
-            write_in_parent_process=write_in_parent_process,
-            dask_max_submitted_tasks=dask_max_submitted_tasks,
-            dask_chunksize=dask_chunksize,
-            **kwargs,
-        ):
-            future.raise_if_failed()
-            yield future
-
-
-###########################
-# batch execution options #
-###########################
-
-
-def _run_on_single_tile(
-    executor=None,
-    process=None,
-    tile=None,
-    dask_scheduler=None,
-):
-    with ExitStack() as exit_stack:
-        # create fresh Executor if it was not passed on
-        if executor is None:
-            executor = exit_stack.enter_context(
-                Executor(
-                    concurrency="dask" if dask_scheduler else None,
-                    dask_scheduler=dask_scheduler,
-                )
-            )
-        logger.info("run process on single tile")
-        return next(
-            executor.as_completed(
-                func=_execute_and_write,
-                iterable=[
-                    TileTask(
-                        tile=tile,
-                        config=process.config,
-                        skip=(
-                            process.config.mode == "continue"
-                            and process.config.output_reader.tiles_exist(tile)
-                        ),
-                    ),
-                ],
-                fkwargs=dict(output_writer=process.config.output),
-            )
-        )
-
-
-def _run_area(
-    executor=None,
-    process=None,
-    zoom_levels=None,
-    dask_scheduler=None,
-    dask_max_submitted_tasks=None,
-    dask_chunksize=None,
-    workers=None,
-    multiprocessing_start_method=None,
-    skip_output_check=False,
-):
-    logger.info("run process on area")
-    # for output drivers requiring writing data in parent process
-    if process.config.output.write_in_parent_process:
-        for future in _run_multi(
-            executor=executor,
-            func=_execute,
-            zoom_levels=zoom_levels,
-            process=process,
-            dask_scheduler=dask_scheduler,
-            dask_max_submitted_tasks=dask_max_submitted_tasks,
-            dask_chunksize=dask_chunksize,
-            workers=workers,
-            multiprocessing_start_method=multiprocessing_start_method,
-            write_in_parent_process=True,
-            skip_output_check=skip_output_check,
-        ):
-            yield future
-
-    # for output drivers which can write data in child processes
-    else:
-        for future in _run_multi(
-            executor=executor,
-            func=_execute_and_write,
-            fkwargs=dict(output_writer=process.config.output),
-            zoom_levels=zoom_levels,
-            process=process,
-            dask_scheduler=dask_scheduler,
-            dask_max_submitted_tasks=dask_max_submitted_tasks,
-            dask_chunksize=dask_chunksize,
-            workers=workers,
-            multiprocessing_start_method=multiprocessing_start_method,
-            write_in_parent_process=False,
-            skip_output_check=skip_output_check,
-        ):
-            yield future
-
-
-def _filter_skipable(
-    process=None, tiles_batches=None, target_set=None, skip_output_check=False
-):
-    if skip_output_check:  # pragma: no cover
-        for batch in tiles_batches:
-            for tile in batch:
-                yield (tile, False, None)
-    else:
-        target_set = target_set or set()
-        for tile, skip in process.skip_tiles(tiles_batches=tiles_batches):
-            if skip and tile not in target_set:
-                yield (tile, True, "output already exists")
-            else:
-                yield (tile, False, None)
-
-
-def _run_multi(
-    executor=None,
-    func=None,
-    zoom_levels=None,
-    process=None,
-    dask_scheduler=None,
-    dask_max_submitted_tasks=None,
-    dask_chunksize=None,
-    workers=None,
-    multiprocessing_start_method=None,
-    write_in_parent_process=False,
-    fkwargs=None,
-    skip_output_check=False,
-):
-    total_tiles = process.count_tiles(zoom_levels.min, zoom_levels.max)
-    workers = min([workers, total_tiles])
-    num_processed = 0
-
-    # If an Executor is passed on, don't close after processing. If no Executor is passed on,
-    # create one and properly close it afterwards.
-    with ExitStack() as exit_stack:
-        # create fresh Executor if it was not passed on
-        if executor is None:
-            executor = exit_stack.enter_context(
-                Executor(
-                    max_workers=workers,
-                    concurrency="dask" if dask_scheduler else "processes",
-                    start_method=multiprocessing_start_method,
-                    dask_scheduler=dask_scheduler,
-                )
-            )
-        with Timer() as duration:
-            logger.info(
-                "run process on %s tiles using %s workers on executor %s",
-                total_tiles,
-                workers,
-                executor,
-            )
-            if process.config.baselevels:
-                f = _run_multi_overviews
-            else:
-                f = _run_multi_no_overviews
-            for num_processed, future in enumerate(
-                f(
-                    zoom_levels=zoom_levels,
-                    executor=executor,
-                    func=func,
-                    process=process,
-                    skip_output_check=skip_output_check,
-                    fkwargs=fkwargs,
-                    dask_chunksize=dask_chunksize,
-                    dask_max_submitted_tasks=dask_max_submitted_tasks,
-                    write_in_parent_process=write_in_parent_process,
-                ),
-                1,
-            ):
-                logger.debug("task %s finished: %s", num_processed, future)
-                yield future
-
-        logger.info("%s tile(s) iterated in %s", str(num_processed), duration)
-
-
-def _run_multi_overviews(
-    zoom_levels=None,
-    executor=None,
-    func=None,
-    process=None,
-    skip_output_check=None,
-    fkwargs=None,
-    dask_max_submitted_tasks=None,
-    dask_chunksize=None,
-    write_in_parent_process=None,
-) -> Iterator[MFuture]:
-    # here we store the parents of processed tiles so we can update overviews
-    # also in "continue" mode in case there were updates at the baselevel
-    overview_parents = set()
-    for i, zoom in enumerate(zoom_levels.descending()):
-        logger.debug("sending tasks to executor %s...", executor)
-        # get generator list of tiles, whether they are to be skipped and skip_info
-        # from _filter_skipable and pass on to executor
-        for future in executor.as_completed(
-            func=func,
-            iterable=(
-                (
-                    TileTask(
-                        tile=tile,
-                        config=process.config,
-                        skip=(
-                            process.mode == "continue"
-                            and process.config.output_reader.tiles_exist(tile)
-                        )
-                        if skip_output_check
-                        else False,
-                    ),
-                    skip,
-                    process_msg,
-                )
-                for tile, skip, process_msg in _filter_skipable(
-                    process=process,
-                    tiles_batches=process.get_process_tiles(
-                        zoom,
-                        batch_by=batch_sort_property(
-                            process.config.output_reader.tile_path_schema
-                        ),
-                    ),
-                    target_set=(
-                        overview_parents if process.config.baselevels and i else None
-                    ),
-                    skip_output_check=skip_output_check,
-                )
-            ),
-            fkwargs=fkwargs,
-            max_submitted_tasks=dask_max_submitted_tasks,
-            chunksize=dask_chunksize,
-            item_skip_bool=True,
-        ):
-            # tiles which were not processed
-            if future.skipped:
-                process_info = TileProcessInfo(
-                    tile=future.result().tile,
-                    processed=False,
-                    process_msg=future.skip_info,
-                    written=False,
-                    write_msg="nothing written",
-                )
-            # tiles which were processed
-            else:
-                # trigger output write for driver which require parent process for writing
-                # the code coverage below is omitted because we don't usually calculate overviews
-                # when writing single files
-                if write_in_parent_process:  # pragma: no cover
-                    process_info = _write(
-                        process_info=future.result(),
-                        output_writer=process.config.output,
-                    )
-
-                # output already has been written, so just use task process info
-                else:
-                    process_info = future.result()
-                    # in case of building overviews from baselevels, remember which parent
-                    # tile needs to be updated later on
-                    if (
-                        not skip_output_check
-                        and process.config.baselevels
-                        and process_info.processed
-                        and process_info.tile.zoom > 0
-                    ):
-                        overview_parents.add(process_info.tile.get_parent())
-            overview_parents.discard(process_info.tile)
-            yield MFuture.from_result(result=process_info)
-
-
-def _preprocess_task_wrapper(
-    task, append_data=True, **kwargs
-) -> PreprocessingProcessInfo:
-    data = task.execute(**kwargs)
-    return PreprocessingProcessInfo.from_inp(task.id, data, append_data=append_data)
-
-
-def _preprocess(
-    tasks,
-    process=None,
-    dask_scheduler=None,
-    dask_max_submitted_tasks=None,
-    dask_chunksize=None,
-    workers=None,
-    multiprocessing_start_method=None,
-    executor=None,
-) -> Iterator[MFuture]:
-    # If preprocessing tasks already finished, don't run them again.
-    if process.config.preprocessing_tasks_finished:  # pragma: no cover
-        return
-    num_processed = 0
-
-    # If an Executor is passed on, don't close after processing. If no Executor is passed on,
-    # create one and properly close it afterwards.
-    with ExitStack() as exit_stack:
-        # create fresh Executor if it was not passed on
-        if executor is None:
-            executor = exit_stack.enter_context(
-                Executor(
-                    max_workers=workers,
-                    concurrency="dask" if dask_scheduler else "processes",
-                    start_method=multiprocessing_start_method,
-                    dask_scheduler=dask_scheduler,
-                )
-            )
-        duration = exit_stack.enter_context(Timer())
-        logger.info(
-            "run preprocessing on %s tasks using %s workers", len(tasks), workers
-        )
-        for num_processed, future in enumerate(
-            executor.as_completed(
-                func=_preprocess_task_wrapper,
-                iterable=tasks.values(),
-                fkwargs=dict(append_data=True),
-                max_submitted_tasks=dask_max_submitted_tasks,
-                chunksize=dask_chunksize,
-            ),
-            1,
-        ):
-            result = future.result()
-            logger.debug(
-                "preprocessing task %s/%s %s processed successfully",
-                num_processed,
-                len(tasks),
-                result.task_key,
-            )
-            process.config.set_preprocessing_task_result(result.task_key, result.data)
-            yield future
-    process.config.preprocessing_tasks_finished = True
-
-    logger.info("%s task(s) iterated in %s", str(len(tasks)), duration)
-
-
-def _run_multi_no_overviews(
-    zoom_levels=None,
-    executor=None,
-    func=None,
-    process=None,
-    skip_output_check=None,
-    fkwargs=None,
-    dask_max_submitted_tasks=None,
-    dask_chunksize=None,
-    write_in_parent_process=None,
-) -> Iterator[MFuture]:
-    logger.debug("sending tasks to executor %s...", executor)
-    # get generator list of tiles, whether they are to be skipped and skip_info
-    # from _filter_skipable and pass on to executor
-    for future in executor.as_completed(
-        func=func,
-        iterable=(
-            (
-                TileTask(
-                    tile=tile,
-                    config=process.config,
-                    skip=(
-                        process.mode == "continue"
-                        and process.config.output_reader.tiles_exist(tile)
-                    )
-                    if skip_output_check
-                    else False,
-                ),
-                skip,
-                process_msg,
-            )
-            for tile, skip, process_msg in _filter_skipable(
-                process=process,
-                tiles_batches=(
-                    batch
-                    for zoom in zoom_levels.descending()
-                    for batch in process.get_process_tiles(
-                        zoom,
-                        batch_by=batch_sort_property(
-                            process.config.output_reader.tile_path_schema
-                        ),
-                    )
-                ),
-                target_set=None,
-                skip_output_check=skip_output_check,
-            )
-        ),
-        fkwargs=fkwargs,
-        max_submitted_tasks=dask_max_submitted_tasks,
-        chunksize=dask_chunksize,
-        item_skip_bool=True,
-    ):
-        # tiles which were not processed
-        if future.skipped:
-            process_info = TileProcessInfo(
-                tile=future.result().tile,
-                processed=False,
-                process_msg=future.skip_info,
-                written=False,
-                write_msg="nothing written",
-            )
-        # tiles which were processed
-        else:
-            # trigger output write for driver which require parent process for writing
-            if write_in_parent_process:
-                process_info = _write(
-                    process_info=future.result(),
-                    output_writer=process.config.output,
-                )
-
-            # output already has been written, so just use task process info
-            else:
-                process_info = future.result()
-        yield MFuture.from_result(result=process_info, profiling=future.profiling)
-
-
-###############################
-# execute and write functions #
-###############################
-
-
-def _execute(
-    tile_process=None, dependencies=None, append_data=True, **_
-) -> TileProcessInfo:
-    logger.debug(
-        (tile_process.tile.id, "running on %s" % multiprocessing.current_process().name)
-    )
-
-    # skip execution if overwrite is disabled and tile exists
-    if tile_process.skip:
-        logger.debug((tile_process.tile.id, "tile exists, skipping"))
-        return TileProcessInfo(
-            tile=tile_process.tile,
-            processed=False,
-            process_msg="output already exists",
-            written=False,
-            write_msg="nothing written",
-            data=None,
-        )
-
-    try:
-        output = tile_process.execute(dependencies=dependencies)
-    except MapcheteNodataTile:  # pragma: no cover
-        output = "empty"
-    processor_message = "processed successfully"
-    logger.debug((tile_process.tile.id, processor_message))
-    return TileProcessInfo(
-        tile=tile_process.tile,
-        processed=True,
-        process_msg=processor_message,
-        written=None,
-        write_msg=None,
-        data=output if append_data else None,
-    )
-
-
-def _write(
-    process_info=None, output_writer=None, append_data=False, **_
-) -> TileProcessInfo:
-    if process_info.processed:
-        try:
-            output_data = output_writer.streamline_output(process_info.data)
-        except MapcheteNodataTile:
-            output_data = None
-        if output_data is None:
-            message = "output empty, nothing written"
-            logger.debug((process_info.tile.id, message))
-            return TileProcessInfo(
-                tile=process_info.tile,
-                processed=process_info.processed,
-                process_msg=process_info.process_msg,
-                written=False,
-                write_msg=message,
-            )
-        with Timer() as duration:
-            output_writer.write(process_tile=process_info.tile, data=output_data)
-        message = "output written in %s" % duration
-        logger.debug((process_info.tile.id, message))
-        return TileProcessInfo(
-            tile=process_info.tile,
-            processed=process_info.processed,
-            process_msg=process_info.process_msg,
-            written=True,
-            write_msg=message,
-            data=output_data if append_data else None,
-        )
-
-    return process_info
-
-
-def _execute_and_write(
-    tile_process=None, output_writer=None, dependencies=None, append_data=False, **_
-) -> TileProcessInfo:
-    return _write(
-        process_info=_execute(tile_process=tile_process, dependencies=dependencies),
-        output_writer=output_writer,
-        append_data=append_data,
-    )
diff --git a/mapchete/processing/execute.py b/mapchete/processing/execute.py
new file mode 100644
index 00000000..e34498cc
--- /dev/null
+++ b/mapchete/processing/execute.py
@@ -0,0 +1,225 @@
+import logging
+from functools import partial
+from multiprocessing import current_process
+from typing import Callable, Iterator, Optional
+
+from mapchete.errors import MapcheteNodataTile
+from mapchete.executor import DaskExecutor, ExecutorBase
+from mapchete.formats.base import OutputDataWriter
+from mapchete.processing.tasks import Task, Tasks
+from mapchete.processing.types import TaskInfo, default_tile_task_id
+from mapchete.timer import Timer
+
+logger = logging.getLogger(__name__)
+
+
+def single_batch(
+    executor: ExecutorBase,
+    tasks: Tasks,
+    output_writer: Optional[OutputDataWriter] = None,
+    write_in_parent_process: bool = False,
+    propagate_results: bool = False,
+) -> Iterator[TaskInfo]:
+    """
+    Treat all tasks as from a single batch, i.e. they don't have dependencies.
+    """
+    task_wrapper = get_task_wrapper(
+        write_in_parent_process=write_in_parent_process,
+        output_writer=output_writer,
+        propagate_results=propagate_results,
+    )
+
+    for future in executor.as_completed(task_wrapper, tasks.to_batch()):
+        task_info = TaskInfo.from_future(future)
+
+        if write_in_parent_process:
+            task_info = write_wrapper(
+                task_info,
+                output_writer=output_writer,
+                append_data=propagate_results,
+            )
+
+        yield task_info
+
+
+def batches(
+    executor: ExecutorBase,
+    tasks: Tasks,
+    output_writer: Optional[OutputDataWriter] = None,
+    write_in_parent_process: bool = False,
+    propagate_results: bool = False,
+) -> Iterator[TaskInfo]:
+    """
+    Execute batches in sequential order but tasks within batches don't have any order.
+    """
+    preprocessing_tasks_results = {}
+
+    task_wrapper = get_task_wrapper(
+        write_in_parent_process=write_in_parent_process,
+        output_writer=output_writer,
+        propagate_results=propagate_results,
+    )
+
+    for batch in tasks.to_batches():
+        # preprocessing task results have to be appended to each tile task batch:
+        if batch.id != "preprocessing_tasks":
+            for task in batch:
+                for id, result in preprocessing_tasks_results.items():
+                    task.add_dependency(id, result)
+
+        for future in executor.as_completed(task_wrapper, batch):
+            task_info = TaskInfo.from_future(future)
+
+            if write_in_parent_process:
+                task_info = write_wrapper(
+                    task_info,
+                    output_writer=output_writer,
+                    append_data=propagate_results,
+                )
+
+            # remember preprocessing task result
+            if task_info.tile is None:
+                preprocessing_tasks_results[task_info.id] = task_info.output
+
+            yield task_info
+
+
+def dask_graph(
+    executor: DaskExecutor,
+    tasks: Tasks,
+    output_writer: Optional[OutputDataWriter] = None,
+    write_in_parent_process: bool = False,
+    propagate_results: bool = False,
+) -> Iterator[TaskInfo]:
+    """
+    Tasks share dependencies with each other.
+    """
+    task_wrapper = get_task_wrapper(
+        write_in_parent_process=write_in_parent_process,
+        output_writer=output_writer,
+        propagate_results=propagate_results,
+    )
+
+    for future in executor.compute_task_graph(
+        tasks.to_dask_graph(
+            preprocessing_task_wrapper=task_wrapper,
+            tile_task_wrapper=task_wrapper,
+        ),
+    ):
+        task_info = TaskInfo.from_future(future)
+        if write_in_parent_process:
+            yield write_wrapper(
+                task_info,
+                output_writer=output_writer,
+                append_data=propagate_results,
+            )
+        else:
+            yield task_info
+
+
+def get_task_wrapper(
+    write_in_parent_process: bool = False,
+    output_writer: Optional[OutputDataWriter] = None,
+    propagate_results: bool = False,
+) -> Callable:
+    """Return a partially initialized wrapper function for task."""
+    if write_in_parent_process:
+        return partial(execute_wrapper, append_data=True)
+    else:
+        return partial(
+            execute_and_write_wrapper,
+            output_writer=output_writer,
+            append_data=propagate_results,
+        )
+
+
+def execute_wrapper(
+    task: Task, dependencies: Optional[dict] = None, append_data: bool = True
+) -> TaskInfo:
+    """
+    Executes tasks and wraps output in a TaskInfo object.
+    """
+    logger.debug((task.id, "running on %s" % current_process().name))
+
+    try:
+        output = task.execute(dependencies=dependencies)
+    except MapcheteNodataTile:  # pragma: no cover
+        output = "empty"
+    processor_message = "processed successfully"
+
+    logger.debug((task.id, processor_message))
+
+    if task.tile:
+        return TaskInfo(
+            id=default_tile_task_id(task.tile),
+            tile=task.tile,
+            processed=True,
+            process_msg=processor_message,
+            written=None,
+            write_msg=None,
+            output=output if append_data else None,
+        )
+
+    else:
+        return TaskInfo(
+            id=task.id,
+            processed=True,
+            process_msg=processor_message,
+            written=None,
+            write_msg=None,
+            output=output if append_data else None,
+        )
+
+
+def write_wrapper(
+    task_info: TaskInfo, output_writer: OutputDataWriter, append_data: bool = False, **_
+) -> TaskInfo:
+    """Write output from previous step and return updated TileTaskInfo object."""
+    if task_info.processed and task_info.tile:
+        try:
+            output_data = output_writer.streamline_output(task_info.output)
+        except MapcheteNodataTile:
+            output_data = None
+        if output_data is None:
+            message = "output empty, nothing written"
+            logger.debug((task_info.tile.id, message))
+            return TaskInfo(
+                id=default_tile_task_id(task_info.tile),
+                tile=task_info.tile,
+                processed=task_info.processed,
+                process_msg=task_info.process_msg,
+                written=False,
+                write_msg=message,
+            )
+        with Timer() as duration:
+            output_writer.write(process_tile=task_info.tile, data=output_data)
+        message = "output written in %s" % duration
+        logger.debug((task_info.tile.id, message))
+        return TaskInfo(
+            id=default_tile_task_id(task_info.tile),
+            tile=task_info.tile,
+            processed=task_info.processed,
+            process_msg=task_info.process_msg,
+            written=True,
+            write_msg=message,
+            output=output_data if append_data else None,
+        )
+
+    return task_info
+
+
+def execute_and_write_wrapper(
+    task: Task,
+    output_writer: Optional[OutputDataWriter] = None,
+    dependencies: Optional[dict] = None,
+    append_data: bool = False,
+    **_
+) -> TaskInfo:
+    """
+    Execute tile task and write output in one step.
+    """
+    return write_wrapper(
+        task_info=execute_wrapper(task, dependencies=dependencies),
+        output_writer=output_writer,
+        append_data=append_data,
+    )
diff --git a/mapchete/processing/job.py b/mapchete/processing/job.py
deleted file mode 100644
index 1423fed9..00000000
--- a/mapchete/processing/job.py
+++ /dev/null
@@ -1,134 +0,0 @@
-"""Internal processing classes and functions."""
-
-import logging
-import os
-from enum import Enum
-from typing import Callable, Iterator, Optional
-
-from shapely.geometry import base, mapping
-
-from mapchete.enums import Concurrency
-from mapchete.executor import Executor, ExecutorBase
-from mapchete.processing.types import TaskResult
-from mapchete.types import Bounds, MPathLike
-
-FUTURE_TIMEOUT = float(os.environ.get("MP_FUTURE_TIMEOUT", 10))
-
-
-logger = logging.getLogger(__name__)
-
-
-class Status(Enum):
-    pending = "pending"
-    running = "running"
-    finished = "finished"
-    cancelled = "cancelled"
-
-
-class Job:
-    """
-    Wraps the output of a processing function into a generator with known length.
-
-    This class also exposes the internal Executor.cancel() function in order to cancel all remaining
-    tasks/futures.
-
-    Will move into the mapchete core package.
-    """
-
-    func: Callable[..., Iterator[TaskResult]]
-    fargs: tuple
-    fkwargs: dict
-    status: Status
-    executor: ExecutorBase
-    executor_concurrency: Concurrency
-    executor_kwargs: dict
-    tiles_tasks: int
-    bounds: Optional[Bounds]
-    stac_item_path: Optional[MPathLike]
-
-    def __init__(
-        self,
-        func: Callable[..., Iterator[TaskResult]],
-        fargs: tuple = None,
-        fkwargs: dict = None,
-        as_iterator: bool = False,
-        tiles_tasks: int = None,
-        preprocessing_tasks: int = None,
-        executor_concurrency: Concurrency = Concurrency.processes,
-        executor_kwargs: dict = None,
-        process_area: Optional[base.BaseGeometry] = None,
-        stac_item_path: Optional[MPathLike] = None,
-    ):
-        self.func = func
-        self.fargs = fargs or ()
-        self.fkwargs = fkwargs or {}
-        self.status = Status.pending
-        self.executor_concurrency = executor_concurrency
-        self.executor_kwargs = executor_kwargs or {}
-        self.tiles_tasks = tiles_tasks or 0
-        self.preprocessing_tasks = preprocessing_tasks or 0
-        self._total = self.preprocessing_tasks + self.tiles_tasks
-        self._as_iterator = as_iterator
-        self._process_area = process_area
-        self.bounds = Bounds(*process_area.bounds) if process_area is not None else None
-        self.stac_item_path = stac_item_path
-        if not as_iterator:
-            self._results: Iterator[TaskResult] = list(self._run())
-
-    @property
-    def __geo_interface__(self) -> dict:  # pragma: no cover
-        if self._process_area is not None:
-            return mapping(self._process_area)
-        else:
-            raise AttributeError(f"{self} has no geo information assigned")
-
-    def _run(self) -> Iterator[TaskResult]:
-        if self._total == 0:
-            return
-        logger.debug("opening executor for job %s", repr(self))
-
-        with Executor(
-            concurrency=self.executor_concurrency, **self.executor_kwargs
-        ) as self.executor:
-            self.status = Status.running
-            logger.debug("change of job status: %s", self)
-            yield from self.func(*self.fargs, executor=self.executor, **self.fkwargs)
-            self.status = Status.finished
-            logger.debug("change of job status: %s", self)
-
-    def set_executor_kwargs(self, executor_kwargs):
-        """
-        Overwrite default or previously set Executor creation kwargs.
-
-        This only has an effect if Job was initialized with 'as_iterator' and has not yet run.
-        """
-        self.executor_kwargs = executor_kwargs
-
-    def set_executor_concurrency(self, executor_concurrency):
-        """
-        Overwrite default or previously set Executor concurrency.
-
-        This only has an effect if Job was initialized with 'as_iterator' and has not yet run.
-        """
-        self.executor_concurrency = executor_concurrency
-
-    def cancel(self):
-        """Cancel all running and pending Job tasks."""
-        if self._as_iterator:
-            # requires client and futures
-            if self.executor is None:  # pragma: no cover
-                raise ValueError("nothing to cancel because no executor is running")
-            self.executor.cancel()
-            self.status = Status.cancelled
-
-    def __len__(self):
-        return self._total
-
-    def __iter__(self) -> Iterator[TaskResult]:
-        if self._as_iterator:
-            yield from self._run()
-        else:
-            return self._results
-
-    def __repr__(self):  # pragma: no cover
-        return f"<{self.status} Job with {self._total} tasks.>"
diff --git a/mapchete/processing/profilers/time.py b/mapchete/processing/profilers/time.py
index a1a48a2f..7b6fe6d7 100644
--- a/mapchete/processing/profilers/time.py
+++ b/mapchete/processing/profilers/time.py
@@ -9,9 +9,9 @@
 
 @dataclass
 class MeasuredTime:
-    start: int
-    end: int
-    elapsed: int
+    start: int = 0
+    end: int = 0
+    elapsed: int = 0
 
 
 def measure_time(add_to_return: bool = True) -> Callable:
diff --git a/mapchete/processing/tasks.py b/mapchete/processing/tasks.py
index 14746353..1b3fd25e 100644
--- a/mapchete/processing/tasks.py
+++ b/mapchete/processing/tasks.py
@@ -1,42 +1,64 @@
 import logging
-from typing import Callable, Iterator, List, Optional
+from abc import ABC
+from enum import Enum
+from functools import cached_property
+from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, Union
 from uuid import uuid4
 
-from shapely.geometry import box, mapping
+import numpy.ma as ma
+from dask.delayed import Delayed, DelayedLeaf, delayed
+from shapely.geometry import Polygon, base, box, mapping, shape
+from shapely.ops import unary_union
 
+from mapchete.config import MapcheteConfig
+from mapchete.config.process_func import ProcessFunc
 from mapchete.errors import (
     MapcheteNodataTile,
     MapcheteProcessOutputError,
     NoTaskGeometry,
 )
-from mapchete.executor.base import Profiler, func_partial
+from mapchete.executor.base import Profiler
 from mapchete.io import raster
 from mapchete.io._geometry_operations import to_shape
 from mapchete.io.vector import IndexedFeatures
+from mapchete.path import MPath
 from mapchete.processing.mp import MapcheteProcess
+from mapchete.processing.types import TaskInfo, default_tile_task_id
+from mapchete.tile import BufferedTile
 from mapchete.timer import Timer
+from mapchete.types import Bounds, BoundsLike, TileLike, ZoomLevels, ZoomLevelsLike
 from mapchete.validate import validate_bounds
 
 logger = logging.getLogger(__name__)
 
 
-class Task:
+class Task(ABC):
     """Generic processing task.
 
     Can optionally have spatial properties attached which helps building up dependencies
     between tasks.
     """
 
+    id: str
+    func: Callable
+    fargs: Tuple
+    fkwargs: Tuple
+    dependencies: Dict[str, TaskInfo]
+    result_key_name: str
+    geometry: Optional[Union[base.BaseGeometry, dict]] = None
+    bounds: Optional[Bounds] = None
+    tile: Optional[BufferedTile] = None
+
     def __init__(
         self,
-        id=None,
-        func=None,
-        fargs=None,
-        fkwargs=None,
-        geometry=None,
-        bounds=None,
-        dependencies=None,
-        result_key_name=None,
+        id: Optional[str] = None,
+        func: Optional[Callable] = None,
+        fargs: Optional[Tuple] = None,
+        fkwargs: Optional[Tuple] = None,
+        dependencies: Optional[Dict[str, TaskInfo]] = None,
+        result_key_name: Optional[str] = None,
+        geometry: Optional[Union[base.BaseGeometry, dict]] = None,
+        bounds: Optional[BoundsLike] = None,
     ):
         self.id = id or uuid4().hex
         self.func = func
@@ -48,7 +70,7 @@ def __init__(
             raise ValueError("only provide one of either 'geometry' or 'bounds'")
         elif geometry:
             self.geometry = to_shape(geometry)
-            self.bounds = self.geometry.bounds
+            self.bounds = validate_bounds(self.geometry.bounds)
         elif bounds:
             self.bounds = validate_bounds(bounds)
             self.geometry = box(*self.bounds)
@@ -58,7 +80,7 @@ def __init__(
     def __repr__(self):  # pragma: no cover
         return f"Task(id={self.id}, bounds={self.bounds})"
 
-    def to_dict(self):
+    def to_dict(self) -> dict:
         return {
             "id": self.id,
             "bounds": self.bounds,
@@ -66,7 +88,7 @@ def to_dict(self):
             "properties": {},
         }
 
-    def add_dependencies(self, dependencies):
+    def add_dependencies(self, dependencies: Dict[str, TaskInfo]) -> None:
         dependencies = dependencies or {}
         if not isinstance(dependencies, dict):
             raise TypeError(
@@ -74,25 +96,31 @@ def add_dependencies(self, dependencies):
             )
         self.dependencies.update(dependencies)
 
-    def execute(self, dependencies=None):
-        return func_partial(self.func, self.fargs, self.fkwargs, profilers=None)()
+    def execute(self, dependencies: Optional[Dict[str, TaskInfo]] = None) -> Any:
+        return self.func(*self.fargs, **self.fkwargs)
 
-    def has_geometry(self):
+    def has_geometry(self) -> bool:
         return self.geometry is not None
 
     @property
-    def __geo_interface__(self):
+    def __geo_interface__(self) -> mapping:
         if self.has_geometry():
             return mapping(self.geometry)
         else:
             raise NoTaskGeometry(f"{self} has no geo information assigned")
 
 
-def _execute_task_wrapper(task, **kwargs):
+def _execute_task_wrapper(task, **kwargs) -> Any:
     return task.execute(**kwargs)
 
 
-class TaskBatch:
+class TaskBatch(ABC):
+    tasks: IndexedFeatures
+    id: str
+    func: Callable
+    fkwargs: dict
+    profilers: List[Profiler]
+
     def __init__(
         self,
         tasks: Iterator[Task],
@@ -110,9 +138,21 @@ def __init__(
         self.bounds = self.tasks.bounds
         self.func = func or _execute_task_wrapper
         self.fkwargs = fkwargs or {}
+        self.profilers = profilers or []
+
+    @cached_property
+    def geometry(self) -> base.BaseGeometry:
+        if self.tasks:
+            return unary_union([shape(task) for task in self.tasks])
+        else:
+            return Polygon()
+
+    @property
+    def __geo_interface__(self) -> mapping:
+        return mapping(self.geometry)
 
     def __repr__(self):  # pragma: no cover
-        return f"TaskBatch(id={self.id}, bounds={self.bounds})"
+        return f"TaskBatch(id={self.id}, bounds={self.bounds}, tasks={len(self.tasks)})"
 
     def __iter__(self):
         return iter(self.tasks.values())
@@ -129,7 +169,7 @@ def keys(self):
     def values(self):
         return self.tasks.values()
 
-    def intersection(self, other):
+    def intersection(self, other: Union[Task, Tuple, List]):
         if isinstance(other, Task):
             return self.tasks.filter(other.bounds)
         elif isinstance(other, (tuple, list)):
@@ -139,35 +179,55 @@ def intersection(self, other):
                 f"intersection only works with other Task instances or bounds, not {other}"
             )
 
-    def _validate(self, item):
+    def _validate(self, item: Task):
         if isinstance(item, Task):
             return item
         else:
             raise TypeError("TaskBatch items must be Taskss, not %s", type(item))
 
 
+class InterpolateFrom(str, Enum):
+    lower = "lower"
+    higher = "higher"
+
+
+def _execute_tile_task_wrapper(task, **kwargs) -> Any:
+    return task.execute(**kwargs)
+
+
 class TileTask(Task):
     """
     Class to process on a specific process tile.
-
-    If skip is set to True, all attributes will be set to None.
     """
 
-    def __init__(self, tile=None, id=None, config=None, skip=False, dependencies=None):
+    config_zoom_levels: ZoomLevels
+    config_baselevels: ZoomLevels
+    process = Optional[ProcessFunc]
+    config_dir = Optional[MPath]
+    tile: BufferedTile
+    _dependencies: dict
+
+    def __init__(
+        self,
+        tile: TileLike,
+        id: Optional[str] = None,
+        config: Optional[MapcheteConfig] = None,
+        func: Optional[Callable] = None,
+        dependencies: Optional[dict] = None,
+    ):
         """Set attributes depending on baselevels or not."""
         self.tile = (
             config.process_pyramid.tile(*tile) if isinstance(tile, tuple) else tile
         )
-        _default_id = f"tile_task_z{self.tile.zoom}-({self.tile.zoom}-{self.tile.row}-{self.tile.col})"
+        _default_id = default_tile_task_id(tile)
         self.id = id or _default_id
-        self.skip = skip
-        self.config_zoom_levels = None if skip else config.zoom_levels
-        self.config_baselevels = None if skip else config.baselevels
-        self.process = None if skip else config.process
-        self.config_dir = None if skip else config.config_dir
+        self.func = func or _execute_tile_task_wrapper
+        self.config_zoom_levels = config.zoom_levels
+        self.config_baselevels = config.baselevels
+        self.process = config.process
+        self.config_dir = config.config_dir
         if (
-            skip
-            or self.tile.zoom not in self.config_zoom_levels
+            self.tile.zoom not in self.config_zoom_levels
             or self.tile.zoom in self.config_baselevels
         ):
             self.input, self.process_func_params, self.output_params = {}, {}, {}
@@ -175,13 +235,46 @@ def __init__(self, tile=None, id=None, config=None, skip=False, dependencies=Non
             self.input = config.get_inputs_for_tile(tile)
             self.process_func_params = config.get_process_func_params(tile.zoom)
             self.output_params = config.output_reader.output_params
-        self.mode = None if skip else config.mode
-        self.output_reader = (
-            None if skip or not config.baselevels else config.output_reader
-        )
+        self.mode = config.mode
+        self.output_reader = config.output_reader if config.baselevels else None
+        self._dependencies = dict()
         super().__init__(id=self.id, geometry=tile.bbox)
 
-    def execute(self, dependencies=None):
+    def __repr__(self):  # pragma: no cover
+        return f"TileTask(id={self.id}, tile={self.tile}, bounds={self.bounds})"
+
+    def add_dependency(self, task_key: str, result: Any, raise_error: bool = True):
+        """Append preprocessing task result to input."""
+        # if dependency has geo information, only add if it intersects with task!
+        try:
+            if not shape(result).intersects(shape(self)):
+                logger.debug("dependency does not intersect with task")
+                return
+        except AttributeError:
+            pass
+
+        if ":" in task_key:
+            inp_key, inp_task_key = task_key.split(":")[:2]
+        else:  # pragma: no cover
+            raise KeyError(
+                "preprocessing task cannot be assigned to an input "
+                f"because of a malformed task key: {task_key}"
+            )
+
+        input_keys = {inp.input_key for inp in self.input.values()}
+        if inp_key not in input_keys:  # pragma: no cover
+            if raise_error:
+                raise KeyError(
+                    f"task {inp_task_key} cannot be assigned to input with key {inp_key} "
+                    f"(available keys: {input_keys})"
+                )
+            else:
+                return
+
+        logger.debug("remember preprocessing task (%s) result for execution", task_key)
+        self._dependencies[task_key] = result
+
+    def execute(self, dependencies: Optional[dict] = None) -> Any:
         """
         Run the Mapchete process and return the result.
 
@@ -200,7 +293,9 @@ def execute(self, dependencies=None):
             process output
         """
         if self.mode not in ["memory", "continue", "overwrite"]:  # pragma: no cover
-            raise ValueError("process mode must be memory, continue or overwrite")
+            raise ValueError(
+                f"process mode must be memory, continue or overwrite, not {self.mode}"
+            )
 
         if self.tile.zoom not in self.config_zoom_levels:
             raise MapcheteNodataTile
@@ -213,7 +308,7 @@ def execute(self, dependencies=None):
             raise MapcheteProcessOutputError("process output is empty")
         return process_output
 
-    def _execute(self, dependencies=None):
+    def _execute(self, dependencies: Optional[Dict[str, TaskInfo]] = None) -> Any:
         # If baselevel is active and zoom is outside of baselevel,
         # interpolate from other zoom levels.
         if self.config_baselevels:
@@ -224,9 +319,13 @@ def _execute(self, dependencies=None):
         # Otherwise, execute from process file.
         try:
             with Timer() as duration:
+                if self._dependencies:
+                    dependencies.update(self._dependencies)
                 # append dependent preprocessing task results to input objects
                 if dependencies:
                     for task_key, task_result in dependencies.items():
+                        if isinstance(task_result, TaskInfo):
+                            task_result = task_result.output
                         if not task_key.startswith("tile_task"):
                             inp_key, task_key = task_key.split(":")[0], ":".join(
                                 task_key.split(":")[1:]
@@ -236,7 +335,7 @@ def _execute(self, dependencies=None):
                             for inp in self.input.values():
                                 if inp.input_key == inp_key:
                                     inp.set_preprocessing_task_result(
-                                        task_key=task_key, result=task_result.data
+                                        task_key=task_key, result=task_result
                                     )
                 # Actually run process.
                 process_data = self.process(
@@ -257,7 +356,11 @@ def _execute(self, dependencies=None):
 
         return process_data
 
-    def _interpolate_from_baselevel(self, baselevel=None, dependencies=None):
+    def _interpolate_from_baselevel(
+        self,
+        baselevel: InterpolateFrom,
+        dependencies: Optional[Dict[str, TaskInfo]] = None,
+    ) -> ma.MaskedArray:
         # This is a special tile derived from a pyramid which has the pixelbuffer setting
         # from the output pyramid but metatiling from the process pyramid. This is due to
         # performance reasons as for the usual case overview tiles do not need the
@@ -273,7 +376,7 @@ def _interpolate_from_baselevel(self, baselevel=None, dependencies=None):
 
         with Timer() as duration:
             # resample from parent tile
-            if baselevel == "higher":
+            if baselevel == InterpolateFrom.higher:
                 parent_tile = self.tile.get_parent()
                 process_data = raster.resample_from_array(
                     self.output_reader.read(parent_tile),
@@ -283,19 +386,17 @@ def _interpolate_from_baselevel(self, baselevel=None, dependencies=None):
                     nodata=self.output_reader.output_params["nodata"],
                 )
             # resample from children tiles
-            elif baselevel == "lower":
+            elif baselevel == InterpolateFrom.lower:
                 src_tiles = {}
-                for result in dependencies.values():
-                    process_info = result.output
+                for task_info in dependencies.values():
                     logger.debug("reading output from dependend tasks")
-                    process_tile = process_info.tile
                     for output_tile in self.output_reader.pyramid.intersecting(
-                        process_tile
+                        task_info.tile
                     ):
-                        if process_info.data is not None:
+                        if task_info.output is not None:
                             src_tiles[output_tile] = raster.extract_from_array(
-                                in_raster=process_info.data,
-                                in_affine=process_tile.affine,
+                                in_raster=task_info.output,
+                                in_affine=task_info.tile.affine,
                                 out_tile=output_tile,
                             )
                 if self.output_reader.pyramid.pixelbuffer:  # pragma: no cover
@@ -343,12 +444,22 @@ def __init__(
     ):
         self.id = id or uuid4().hex
         self.bounds = None, None, None, None
-        self.tasks = {tile: item for tile, item in self._validate(tasks)}
+        self.tasks = {item.tile: item for item in self._validate(tasks)}
         self._update_bounds()
-        self.func = func or _execute_task_wrapper
+        self.func = func or _execute_tile_task_wrapper
         self.fkwargs = fkwargs or {}
         self.profilers = profilers or []
 
+    @cached_property
+    def geometry(self) -> base.BaseGeometry:
+        if self.tasks:
+            return unary_union([task.bbox for task in self.tasks])
+        else:  # pragma: no cover
+            return Polygon()
+
+    def __repr__(self):  # pragma: no cover
+        return f"TileTaskBatch(id={self.id}, bounds={self.bounds}, tasks={len(self.tasks)})"
+
     def _update_bounds(self):
         for tile in self.tasks.keys():
             left, bottom, right, top = self.bounds
@@ -384,7 +495,7 @@ def intersection(self, other):
                 "intersections only works with other Task instances or bounds"
             )
 
-    def _validate(self, items):
+    def _validate(self, items: Iterator[TileTask]) -> Iterator[TileTask]:
         self._tp = None
         self._zoom = None
         for item in items:
@@ -400,41 +511,98 @@ def _validate(self, items):
                 self._zoom = item.tile.zoom
             elif item.tile.zoom != self._zoom:  # pragma: no cover
                 raise TypeError("all TileTasks must lie on the same zoom level")
-            yield item.tile, item
+            yield item
+
 
+class Tasks:
+    preprocessing_batches: List[TaskBatch]
+    tile_batches: List[TileTaskBatch]
+
+    def __init__(
+        self,
+        task_batches: Iterator[Union[TaskBatch, TileTaskBatch]],
+    ):
+        with Timer() as tt:
+            self.preprocessing_batches = []
+            self.tile_batches = []
+            for batch in task_batches:
+                if isinstance(batch, TileTaskBatch):
+                    self.tile_batches.append(batch)
+                else:
+                    self.preprocessing_batches.append(batch)
+        logger.debug("task batches materialized in %s", tt)
 
-def to_dask_collection(batches):
-    from dask.delayed import delayed
+    def __len__(self):
+        return sum([len(batch) for batch in self])
+
+    def __iter__(self) -> Union[TaskBatch, TileTaskBatch]:
+        for phase in (self.preprocessing_batches, self.tile_batches):
+            for batch in phase:
+                yield batch
+
+    @property
+    def preprocessing_tasks_count(self):
+        return sum([len(batch) for batch in self.preprocessing_batches])
+
+    @property
+    def tile_tasks_count(self):
+        return sum([len(batch) for batch in self.tile_batches])
+
+    def to_dask_graph(
+        self,
+        preprocessing_task_wrapper: Optional[Callable] = None,
+        tile_task_wrapper: Optional[Callable] = None,
+    ) -> List[Union[Delayed, DelayedLeaf]]:
+        """Return task graph to use with dask Executor."""
+        tasks = {}
+        previous_batch = None
+        for batch in self:
+            logger.debug("converting batch %s", batch)
+
+            if isinstance(batch, TileTaskBatch):
+                task_func = tile_task_wrapper or batch.func
+            else:
+                task_func = preprocessing_task_wrapper or batch.func
 
-    tasks = {}
-    previous_batch = None
-    for batch in batches:
-        logger.debug("converting batch %s", batch)
-        if previous_batch:
-            logger.debug("previous batch had %s tasks", len(previous_batch))
-        for task in batch.values():
             if previous_batch:
-                dependencies = {
-                    child.id: tasks[child]
-                    for child in previous_batch.intersection(task)
-                }
-                logger.debug(
-                    "found %s dependencies from last batch for task %s",
-                    len(dependencies),
+                logger.debug("previous batch had %s tasks", len(previous_batch))
+
+            for task in batch.values():
+                if previous_batch:
+                    dependencies = {
+                        child.id: tasks[child]
+                        for child in previous_batch.intersection(task)
+                    }
+                    logger.debug(
+                        "found %s dependencies from last batch for task %s",
+                        len(dependencies),
+                        task,
+                    )
+                else:
+                    dependencies = {}
+
+                tasks[task] = delayed(
+                    task_func,
+                    pure=True,
+                    name=f"{task.id}",
+                    traverse=len(dependencies) > 0,
+                )(
                     task,
+                    dependencies=dependencies,
+                    **batch.fkwargs,
+                    dask_key_name=f"{task.result_key_name}",
                 )
-            else:
-                dependencies = {}
-            tasks[task] = delayed(
-                batch.func,
-                pure=True,
-                name=f"{task.id}",
-                traverse=len(dependencies) > 0,
-            )(
-                task,
-                dependencies=dependencies,
-                **batch.fkwargs,
-                dask_key_name=f"{task.result_key_name}",
-            )
-        previous_batch = batch
-    return list(tasks.values())
+
+            previous_batch = batch
+
+        return list(tasks.values())
+
+    def to_batch(self) -> Iterator[Task]:
+        """Return all tasks as one batch."""
+        for batch in self.to_batches():
+            for task in batch:
+                yield task
+
+    def to_batches(self) -> Iterator[Iterator[Task]]:
+        """Return batches of tasks."""
+        return list(self)
diff --git a/mapchete/processing/types.py b/mapchete/processing/types.py
index 72641b80..02a5c813 100644
--- a/mapchete/processing/types.py
+++ b/mapchete/processing/types.py
@@ -8,65 +8,32 @@
 from mapchete.tile import BufferedTile
 
 
-@dataclass
-class TileProcessInfo:
-    tile: BufferedTile
-    processed: bool = False
-    process_msg: Optional[str] = None
-    written: bool = False
-    write_msg: Optional[str] = None
-    data: Optional[Any] = None
-    profiling: dict = field(default_factory=dict)
+def default_tile_task_id(tile: BufferedTile) -> str:
+    return f"tile_task_z{tile.zoom}-({tile.zoom}-{tile.row}-{tile.col})"
 
 
 @dataclass
-class PreprocessingProcessInfo:
-    task_key: str
+class TaskInfo:
+    id: str = None
     processed: bool = False
     process_msg: Optional[str] = None
     written: bool = False
     write_msg: Optional[str] = None
-    data: Optional[Any] = None
-    profiling: dict = field(default_factory=dict)
-
-    @staticmethod
-    def from_inp(
-        task_key: str, inp: Any, append_data: bool = True
-    ) -> PreprocessingProcessInfo:
-        if isinstance(inp, Result):
-            profiling = inp.profiling
-            data = inp.output
-        return PreprocessingProcessInfo(
-            task_key=task_key, data=data if append_data else None, profiling=profiling
-        )
-
-
-@dataclass
-class TaskResult:
-    id: str
-    processed: bool
-    process_msg: Optional[str]
-    profiling: dict = field(default_factory=dict)
-    result: Optional[Any] = None
+    output: Optional[Any] = None
     tile: Optional[BufferedTile] = None
+    profiling: dict = field(default_factory=dict)
 
     @staticmethod
-    def from_future(future: MFuture) -> TaskResult:
-        process_info = future.result()
-        if isinstance(process_info, PreprocessingProcessInfo):
-            return TaskResult(
-                id=process_info.task_key,
-                processed=process_info.processed,
-                process_msg=process_info.process_msg,
-                profiling=process_info.profiling or future.profiling,
-            )
-        elif isinstance(process_info, TileProcessInfo):
-            tile = process_info.tile
-            return TaskResult(
-                id=f"tile-{tile.zoom}-{tile.row}-{tile.col}",
-                processed=process_info.processed,
-                process_msg=process_info.process_msg,
-                profiling=process_info.profiling or future.profiling,
-            )
+    def from_future(future: MFuture) -> TaskInfo:
+        result = future.result()
+        if isinstance(result, TaskInfo):
+            task_info = result
+            if future.profiling:
+                task_info.profiling = future.profiling
         else:  # pragma: no cover
-            raise TypeError(f"unknown process info type: {type(process_info)}")
+            task_info = TaskInfo(
+                id=future.key if hasattr(future, "key") else future.name,
+                processed=True,
+                profiling=future.profiling,
+            )
+        return task_info
diff --git a/mapchete/protocols.py b/mapchete/protocols.py
new file mode 100644
index 00000000..27efd584
--- /dev/null
+++ b/mapchete/protocols.py
@@ -0,0 +1,8 @@
+from typing import Protocol
+
+
+class ObserverProtocol(Protocol):
+    """Protocol used for custom observer classes hooked up into commands."""
+
+    def update(self, *args, **kwargs) -> None:  # pragma: no cover
+        ...
diff --git a/mapchete/settings.py b/mapchete/settings.py
index 3578f6b9..116eb7fd 100644
--- a/mapchete/settings.py
+++ b/mapchete/settings.py
@@ -33,6 +33,13 @@ class GDALHTTPOptions(BaseSettings):
 
 
 class IORetrySettings(BaseSettings):
+    """Combine default retry settings with env variables.
+
+    MAPCHETE_IO_RETRY_TRIES
+    MAPCHETE_IO_RETRY_DELAY
+    MAPCHETE_IO_RETRY_BACKOFF
+    """
+
     tries: int = 3
     delay: float = 1.0
     backoff: float = 1.0
diff --git a/mapchete/tile.py b/mapchete/tile.py
index ee81fd53..b6a74760 100644
--- a/mapchete/tile.py
+++ b/mapchete/tile.py
@@ -1,4 +1,6 @@
 """Mapchtete handling tiles."""
+from __future__ import annotations
+
 import logging
 from functools import cached_property
 from itertools import product
@@ -192,6 +194,12 @@ def to_dict(self):
             pixelbuffer=self.pixelbuffer,
         )
 
+    def without_pixelbuffer(self) -> BufferedTilePyramid:
+        config_dict = self.to_dict()
+        config_dict.update(pixelbuffer=0)
+        return BufferedTilePyramid(**config_dict)
+
+    @staticmethod
     def from_dict(config_dict):
         """
         Initialize TilePyramid from configuration dictionary.
@@ -367,11 +375,7 @@ def __ne__(self, other):
         return not self.__eq__(other)
 
     def __repr__(self):
-        return "BufferedTile(%s, tile_pyramid=%s, pixelbuffer=%s)" % (
-            self.id,
-            self.tp,
-            self.pixelbuffer,
-        )
+        return f"BufferedTile(zoom={self.zoom}, row={self.row}, col={self.col})"
 
     def __hash__(self):
         return hash(repr(self))
@@ -467,7 +471,7 @@ def _count_tiles(tiles, geometry, minzoom, maxzoom):
 
 
 def _count_cells(pyramid, geometry, minzoom, maxzoom):
-    if geometry.is_empty:
+    if geometry.is_empty:  # pragma: no cover
         return 0
 
     # for the rasterization algorithm we need to keep the all_touched flag True
diff --git a/mapchete/timer.py b/mapchete/timer.py
index 57d257e4..51630897 100644
--- a/mapchete/timer.py
+++ b/mapchete/timer.py
@@ -1,5 +1,7 @@
 import time
 
+from mapchete.pretty import pretty_seconds
+
 
 class Timer:
     """
@@ -60,14 +62,7 @@ def __repr__(self):
         )
 
     def __str__(self):
-        minutes, seconds = divmod(self.elapsed, 60)
-        hours, minutes = divmod(minutes, 60)
-        if hours:
-            return "%sh %sm %ss" % (int(hours), int(minutes), int(seconds))
-        elif minutes:
-            return "%sm %ss" % (int(minutes), int(seconds))
-        else:
-            return "%ss" % round(seconds, self._str_round)
+        return pretty_seconds(self.elapsed, self._str_round)
 
     @property
     def elapsed(self):
diff --git a/mapchete/types.py b/mapchete/types.py
index 2109400b..d3b15560 100644
--- a/mapchete/types.py
+++ b/mapchete/types.py
@@ -3,6 +3,7 @@
 import os
 from typing import Iterable, List, Optional, Tuple, Union
 
+from pydantic import BaseModel
 from shapely.geometry import shape
 from shapely.geometry.base import BaseGeometry
 
@@ -10,7 +11,7 @@
 
 MPathLike = Union[str, os.PathLike]
 BoundsLike = Union[List[float], Tuple[float], dict, BaseGeometry]
-ZoomLevelsLike = Union[Iterable[int], int, dict]
+ZoomLevelsLike = Union[List[int], int, dict]
 TileLike = Union[BufferedTile, Tuple[int, int, int]]
 
 
@@ -175,7 +176,7 @@ class ZoomLevels(list):
 
     def __init__(
         self,
-        min: Union[Iterable[int], int],
+        min: Union[List[int], int],
         max: Optional[int] = None,
         descending: bool = False,
     ):
@@ -223,7 +224,7 @@ def __contains__(self, value):
         return value in list(self)
 
     def _set_attributes(
-        self, minlevel: Union[Iterable[int], int], maxlevel: Optional[int] = None
+        self, minlevel: Union[List[int], int], maxlevel: Optional[int] = None
     ) -> None:
         """This method is important when ZoomLevel instances are passed on to the ProcessConfig schema."""
         if hasattr(minlevel, "__iter__"):  # pragma: no cover
@@ -291,6 +292,13 @@ def intersection(self, other: ZoomLevelsLike) -> ZoomLevels:
             raise ValueError("ZoomLevels do not intersect")
         return ZoomLevels(min(intersection), max(intersection))
 
+    def difference(self, other: ZoomLevelsLike) -> ZoomLevels:
+        other = other if isinstance(other, ZoomLevels) else ZoomLevels(other)
+        difference = set(self).difference(set(other))
+        if len(difference) == 0:  # pragma: no cover
+            raise ValueError("ZoomLevels do not differ")
+        return ZoomLevels(min(difference), max(difference))
+
     def intersects(self, other: ZoomLevelsLike) -> bool:
         other = other if isinstance(other, ZoomLevels) else ZoomLevels(other)
         try:
@@ -300,3 +308,8 @@ def intersects(self, other: ZoomLevelsLike) -> bool:
 
     def descending(self) -> ZoomLevels:
         return ZoomLevels(min=self.min, max=self.max, descending=True)
+
+
+class Progress(BaseModel):
+    current: int = 0
+    total: Optional[int] = None
diff --git a/pyproject.toml b/pyproject.toml
index f2378a38..84568976 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -48,14 +48,13 @@ dependencies = [
 [project.optional-dependencies]
 complete = [
     "aiohttp",
-    "aiobotocore>=1.1.2,!=2.7.0",
+    "aiobotocore>=1.1.2",
     "boto3>=1.14.44",
     "dask",
     "distributed",
     "Flask<3.0.0",
     "Flask-RangeRequest",
-    "fsspec[http]",
-    "fsspec[s3]",
+    "fsspec[http,s3]>=2023.12.0",
     "geobuf",
     "lxml",
     "matplotlib",
@@ -85,7 +84,7 @@ profiling = [
     "tilebench"
 ]
 s3 = [
-    "aiobotocore>=1.1.2,!=2.7.0",
+    "aiobotocore>=1.1.2,!=2.7.0,!=2.8.0",
     "boto3>=1.14.44",
     "fsspec[s3]",
     "s3fs!=2023.9.0",
diff --git a/requirements.txt b/requirements.txt
index 98ff51cd..d65d56bd 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-aiobotocore>=1.1.2,!=2.7.0
+aiobotocore>=1.1.2
 aiohttp>=3.6.2
 aioitertools>=0.7.0
 boto3>=1.14.44
@@ -12,7 +12,7 @@ python-dateutil
 fiona>=1.8.13.post1
 Flask>=1.1.2,<=3.0.0
 Flask-RangeRequest>=0.0.2
-fsspec[s3]>=0.8.4
+fsspec[http,s3]>=2023.12.0
 geobuf>=1.1.1
 importlib-metadata>=1.6.0
 importlib-resources>=1.5.0


Test patch:
diff --git a/mapchete/testing.py b/mapchete/testing.py
index 061e7223..39a47c01 100644
--- a/mapchete/testing.py
+++ b/mapchete/testing.py
@@ -10,6 +10,7 @@
 
 import mapchete
 from mapchete.config.base import initialize_inputs, open_inputs
+from mapchete.executor import SequentialExecutor
 from mapchete.path import MPath
 from mapchete.tile import BufferedTilePyramid
 
@@ -139,17 +140,15 @@ def process_mp(self, tile=None, tile_zoom=None, batch_preprocess=True):
             input=mp.config.get_inputs_for_tile(tile),
         )
 
-    def mp(self, batch_preprocess=True):
+    def mp(self, batch_preprocess=True, bounds=None, zoom=None):
         """
         Return Mapchete object from mapchete.open().
         """
-        from mapchete.executor import SequentialExecutor
-
         with SequentialExecutor() as executor:
             if not self._mp:
-                self._mp = mapchete.open(self.dict)
+                self._mp = mapchete.open(self.dict, bounds=bounds, zoom=zoom)
                 if batch_preprocess:
-                    self._mp.batch_preprocess(executor=executor)
+                    self._mp.execute_preprocessing_tasks(executor=executor)
 
             return self._mp
 
diff --git a/pytest.ini b/pytest.ini
index 8337fea5..e865bb36 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -5,4 +5,4 @@ markers =
 env =
     D:AK1=foo
     D:AK2=bar
-addopts = --durations 10 --verbose
\ No newline at end of file
+addopts = --durations 20 --verbose --nf --cov=mapchete
\ No newline at end of file
diff --git a/test/test_cli.py b/test/test_cli.py
index 3bdf3bbf..d047aef4 100644
--- a/test/test_cli.py
+++ b/test/test_cli.py
@@ -511,7 +511,7 @@ def test_convert_single_gtiff_overviews(cleantopo_br_tif, mp_tmpdir):
             "--overviews",
             "--overviews-resampling-method",
             "bilinear",
-            "--multi",
+            "--workers",
             "1",
             "--concurrency",
             "none",
@@ -673,7 +673,7 @@ def test_convert_zoom(cleantopo_br_tif, mp_tmpdir, zoom, tiles):
 def test_convert_mapchete(cleantopo_br, mp_tmpdir):
     # prepare data
     with mapchete.open(cleantopo_br.path) as mp:
-        mp.batch_process(zoom=[1, 4])
+        list(mp.execute(zoom=[1, 4]))
     run_cli(
         [
             "convert",
@@ -705,7 +705,7 @@ def test_convert_mapchete(cleantopo_br, mp_tmpdir):
 def test_convert_tiledir(cleantopo_br, mp_tmpdir):
     # prepare data
     with mapchete.open(cleantopo_br.path) as mp:
-        mp.batch_process(zoom=[1, 4])
+        list(mp.execute(zoom=[1, 4]))
     run_cli(
         [
             "convert",
@@ -748,16 +748,22 @@ def test_convert_geojson(landpoly, mp_tmpdir):
             "geodetic",
             "--zoom",
             "4",
+            "--bounds",
+            "-101.25",
+            "78.75",
+            "-90.0",
+            "90.0",
             "--concurrency",
             "none",
         ]
     )
-    for (zoom, row, col), control in zip([(4, 0, 7), (4, 1, 7)], [9, 32]):
-        out_file = mp_tmpdir / zoom / row / col + ".geojson"
-        with fiona_open(out_file, "r") as src:
-            assert len(src) == control
-            for f in src:
-                assert shape(f["geometry"]).is_valid
+    zoom, row, col = (4, 0, 7)
+    control = 9
+    out_file = mp_tmpdir / zoom / row / col + ".geojson"
+    with fiona_open(out_file, "r") as src:
+        assert len(src) == control
+        for f in src:
+            assert shape(f["geometry"]).is_valid
 
 
 def test_convert_geobuf(landpoly, mp_tmpdir):
@@ -900,7 +906,7 @@ def test_convert_errors(s2_band_jp2, mp_tmpdir, s2_band, cleantopo_br, landpoly)
 
     # prepare data for tiledir input
     with mapchete.open(cleantopo_br.path) as mp:
-        mp.batch_process(zoom=[1, 4])
+        mp.execute(zoom=[1, 4])
     tiledir_path = cleantopo_br.dict["config_dir"] / cleantopo_br.dict["output"]["path"]
 
     # zoom level required
@@ -1163,7 +1169,7 @@ def test_index_geojson_wkt_area(mp_tmpdir, cleantopo_br, wkt_geom):
 
     with mapchete.open(cleantopo_br.dict) as mp:
         files = os.listdir(mp.config.output.path)
-        assert len(files) == 7
+        assert len(files) == 8  # was 7 before doing the observer pattern
         assert "3.geojson" in files
 
 
@@ -1426,29 +1432,6 @@ def test_rm(cleantopo_br):
     assert not out_path.exists()
 
 
-def test_rm_storage_option_errors(cleantopo_br):
-    out_path = os.path.join(TESTDATA_DIR, cleantopo_br.dict["output"]["path"])
-    run_cli(
-        [
-            "rm",
-            out_path,
-            "-z",
-            "5",
-            "-b",
-            "169.19251592399996",
-            "-90",
-            "180",
-            "-80.18582802550002",
-            "-f",
-            "--fs-opts",
-            "invalid_opt",
-        ],
-        output_contains="Error: Invalid value for '--fs-opts': Invalid syntax for KEY=VAL arg: invalid_opt",
-        expected_exit_code=2,
-        raise_exc=False,
-    )
-
-
 def test_fs_opt_extractor():
     kwargs = options._cb_key_val(
         None,
diff --git a/test/test_commands.py b/test/test_commands.py
index cccfc0c1..39525f4e 100644
--- a/test/test_commands.py
+++ b/test/test_commands.py
@@ -9,137 +9,227 @@
 
 import mapchete
 from mapchete.commands import convert, cp, execute, index, rm
-from mapchete.executor import ConcurrentFuturesExecutor, SequentialExecutor
+from mapchete.config import DaskSettings
+from mapchete.enums import Status
+from mapchete.errors import JobCancelledError
 from mapchete.io import fiona_open, rasterio_open
-from mapchete.processing.job import Status
-from mapchete.processing.types import TaskResult
+from mapchete.processing.types import TaskInfo
+from mapchete.protocols import ObserverProtocol
+from mapchete.tile import BufferedTilePyramid
 
 SCRIPTDIR = os.path.dirname(os.path.realpath(__file__))
 TESTDATA_DIR = os.path.join(SCRIPTDIR, "testdata")
 
 
+class TaskCounter(ObserverProtocol):
+    tasks = 0
+
+    def update(self, *args, progress=None, **kwargs):
+        if progress:
+            self.tasks = progress.current
+
+
 def test_cp(mp_tmpdir, cleantopo_br, wkt_geom):
     # generate TileDirectory
     with mapchete.open(
         cleantopo_br.dict, bounds=[169.19251592399996, -90, 180, -80.18582802550002]
     ) as mp:
-        mp.batch_process(zoom=5)
+        list(mp.execute(zoom=5))
     out_path = os.path.join(TESTDATA_DIR, cleantopo_br.dict["output"]["path"])
     # copy tiles and subset by bounds
-    tiles = cp(
+    task_counter = TaskCounter()
+    cp(
         out_path,
         os.path.join(mp_tmpdir, "bounds"),
         zoom=5,
         bounds=[169.19251592399996, -90, 180, -80.18582802550002],
+        observers=[task_counter],
     )
-    assert len(tiles)
+    assert task_counter.tasks
 
     # copy all tiles
-    tiles = cp(
+    task_counter = TaskCounter()
+    cp(out_path, os.path.join(mp_tmpdir, "all"), zoom=5, observers=[task_counter])
+    assert task_counter.tasks
+
+    # copy tiles and subset by area
+    task_counter = TaskCounter()
+    cp(
         out_path,
-        os.path.join(mp_tmpdir, "all"),
+        os.path.join(mp_tmpdir, "area"),
         zoom=5,
+        area=wkt_geom,
+        observers=[task_counter],
     )
-    assert len(tiles)
-
-    # copy tiles and subset by area
-    tiles = cp(out_path, os.path.join(mp_tmpdir, "area"), zoom=5, area=wkt_geom)
-    assert len(tiles)
+    assert task_counter.tasks
 
     # copy local tiles without using threads
-    tiles = cp(out_path, os.path.join(mp_tmpdir, "nothreads"), zoom=5, multi=1)
-    assert len(tiles)
+    task_counter = TaskCounter()
+    cp(
+        out_path,
+        os.path.join(mp_tmpdir, "nothreads"),
+        zoom=5,
+        multi=1,
+        observers=[task_counter],
+    )
+    assert task_counter.tasks
 
 
 @pytest.mark.integration
 def test_cp_http(mp_tmpdir, http_tiledir):
     # copy tiles and subset by bounds
-    tiles = cp(
-        http_tiledir, os.path.join(mp_tmpdir, "http"), zoom=1, bounds=[3, 1, 4, 2]
+    task_counter = TaskCounter()
+    cp(
+        http_tiledir,
+        os.path.join(mp_tmpdir, "http"),
+        zoom=1,
+        bounds=[3, 1, 4, 2],
+        observers=[task_counter],
     )
-    assert len(tiles)
+    assert task_counter.tasks
 
 
-def test_rm(mp_tmpdir, cleantopo_br):
+def test_rm(cleantopo_br):
     # generate TileDirectory
     with mapchete.open(
         cleantopo_br.dict, bounds=[169.19251592399996, -90, 180, -80.18582802550002]
     ) as mp:
-        mp.batch_process(zoom=5)
+        list(mp.execute(zoom=5))
     out_path = os.path.join(TESTDATA_DIR, cleantopo_br.dict["output"]["path"])
 
     # remove tiles
-    tiles = rm(out_path, zoom=5)
-    assert len(tiles) > 0
+    task_counter = TaskCounter()
+    rm(out_path, zoom=5, observers=[task_counter])
+    assert task_counter.tasks
 
     # remove tiles but this time they should already have been removed
-    tiles = rm(out_path, zoom=5)
-    assert len(tiles) == 0
+    task_counter = TaskCounter()
+    rm(out_path, zoom=5, observers=[task_counter])
+    assert task_counter.tasks == 0
+
 
+def test_rm_path_list(mp_tmpdir):
+    out_path = mp_tmpdir / "some_file.txt"
+    with out_path.open("w") as dst:
+        dst.write("foo")
+
+    assert out_path.exists()
+    rm(paths=[out_path])
+    assert not out_path.exists()
+
+
+@pytest.mark.integration
+def test_rm_path_list_s3(s3_testdata_dir):
+    out_path = s3_testdata_dir / "some_file.txt"
+    with out_path.open("w") as dst:
+        dst.write("foo")
+
+    assert out_path.exists()
+    rm(paths=[out_path])
+    assert not out_path.exists()
+
+
+@pytest.mark.parametrize(
+    "concurrency,process_graph",
+    [
+        ("threads", None),
+        ("dask", True),
+        ("dask", False),
+        ("processes", None),
+        (None, None),
+    ],
+)
+def test_execute(
+    cleantopo_br_metatiling_1, cleantopo_br_tif, concurrency, process_graph
+):
+    execute_kwargs = dict(concurrency=concurrency)
+    if concurrency == "dask":
+        execute_kwargs.update(dask_settings=DaskSettings(process_graph=process_graph))
 
-def test_execute(mp_tmpdir, cleantopo_br_metatiling_1, cleantopo_br_tif):
     zoom = 5
     tp = TilePyramid("geodetic")
     with rasterio_open(cleantopo_br_tif) as src:
         tiles = list(tp.tiles_from_bounds(src.bounds, zoom))
-    job = execute(cleantopo_br_metatiling_1.dict, zoom=zoom)
-    for t in job:
-        assert t
-    assert len(tiles) == len(job)
+    execute(cleantopo_br_metatiling_1.dict, zoom=zoom, **execute_kwargs)
     mp = cleantopo_br_metatiling_1.mp()
     for t in tiles:
         with rasterio_open(mp.config.output.get_path(t)) as src:
             assert not src.read(masked=True).mask.all()
 
 
-def test_execute_set_executor(mp_tmpdir, cleantopo_br_metatiling_1, cleantopo_br_tif):
-    zoom = 5
-    tp = TilePyramid("geodetic")
-    with rasterio_open(cleantopo_br_tif) as src:
-        tiles = list(tp.tiles_from_bounds(src.bounds, zoom))
-    job = execute(cleantopo_br_metatiling_1.dict, zoom=zoom, as_iterator=True)
+def test_execute_retry(example_mapchete):
+    zoom = 10
+    retries = 2
 
-    # invalid concurrency
-    job.set_executor_concurrency("invalid")
-    with pytest.raises(ValueError):
-        next(iter(job))
-
-    # reset to concurrent.futures
-    job.set_executor_concurrency("processes")
-    next(iter(job))
-    assert isinstance(job.executor, ConcurrentFuturesExecutor)
-
-    # with max_workers: 1 it should fall back to a sequential executor
-    job.set_executor_kwargs({"max_workers": 1})
-    next(iter(job))
-    assert isinstance(job.executor, SequentialExecutor)
-
-    # make sure everything runs smoothly from here on
-    for t in job:
-        assert t
-    assert len(tiles) == len(job)
-    mp = cleantopo_br_metatiling_1.mp()
-    for t in tiles:
-        with rasterio_open(mp.config.output.get_path(t)) as src:
-            assert not src.read(masked=True).mask.all()
+    class ExceptionRaiser:
+        """Makes the job fail during progress."""
 
+        def update(*args, progress=None, **kwargs):
+            if progress and progress.current > 2:
+                raise RuntimeError("This job just raised an exception!")
 
-def test_execute_cancel(mp_tmpdir, cleantopo_br_metatiling_1, cleantopo_br_tif):
+    class RetryCounter:
+        """Count retry attempts."""
+
+        retries = 0
+
+        def update(self, *args, status=None, **kwargs):
+            if status and status == Status.retrying:
+                self.retries += 1
+
+    exception_raiser = ExceptionRaiser()
+    retry_counter = RetryCounter()
+
+    # this job should fail
+    with pytest.raises(RuntimeError):
+        execute(
+            example_mapchete.dict,
+            zoom=zoom,
+            retries=retries,
+            observers=[exception_raiser, retry_counter],
+            concurrency=None,
+        )
+
+    # make sure job has been retried
+    assert retry_counter.retries == retries
+
+
+def test_execute_cancel(cleantopo_br_metatiling_1):
     zoom = 5
-    job = execute(cleantopo_br_metatiling_1.dict, zoom=zoom, as_iterator=True)
-    for i, t in enumerate(job):
-        job.cancel()
-        break
-    assert i == 0
-    assert job.status == Status.cancelled
+
+    class CancelObserver:
+        """Cancels job when running."""
+
+        def update(*args, progress=None, **kwargs):
+            if progress and progress.current > 0:
+                raise JobCancelledError
+
+    class StatusObserver:
+        """Observes job state."""
+
+        status = None
+
+        def update(self, *args, status=None, **kwargs):
+            if status:
+                self.status = status
+
+    state_observer = StatusObserver()
+    execute(
+        cleantopo_br_metatiling_1.dict,
+        zoom=zoom,
+        observers=[CancelObserver(), state_observer],
+        concurrency=None,
+    )
+    assert state_observer.status == Status.cancelled
 
 
 def test_execute_tile(mp_tmpdir, cleantopo_br_metatiling_1):
     tile = (5, 30, 63)
 
-    job = execute(cleantopo_br_metatiling_1.dict, tile=tile)
+    task_counter = TaskCounter()
+    execute(cleantopo_br_metatiling_1.dict, tile=tile, observers=[task_counter])
 
-    assert len(job) == 1
+    assert task_counter.tasks == 1
 
     mp = cleantopo_br_metatiling_1.mp()
     with rasterio_open(
@@ -152,60 +242,84 @@ def test_execute_point(mp_tmpdir, example_mapchete, dummy2_tif):
     """Using bounds from WKT."""
     with rasterio_open(dummy2_tif) as src:
         g = box(*src.bounds)
-    job = execute(example_mapchete.dict, point=[g.centroid.x, g.centroid.y], zoom=10)
-    assert len(job) == 1
+
+    task_counter = TaskCounter()
+    execute(
+        example_mapchete.dict,
+        point=[g.centroid.x, g.centroid.y],
+        zoom=10,
+        observers=[task_counter],
+    )
+    assert task_counter.tasks == 1
 
 
 @pytest.mark.parametrize(
-    "concurrency",
+    "concurrency,process_graph",
     [
-        "threads",
-        "dask",
-        "processes",
-        None,
+        ("threads", None),
+        ("dask", True),
+        ("dask", False),
+        ("processes", None),
+        (None, None),
     ],
 )
-def test_execute_preprocessing_tasks(concurrency, preprocess_cache_raster_vector):
-    job = execute(preprocess_cache_raster_vector.path, concurrency=concurrency)
-    assert len(job)
+def test_execute_preprocessing_tasks(
+    concurrency, preprocess_cache_raster_vector, process_graph
+):
+    execute_kwargs = dict(concurrency=concurrency)
+    if concurrency == "dask":
+        execute_kwargs.update(dask_settings=DaskSettings(process_graph=process_graph))
+
+    task_counter = TaskCounter()
+    execute(
+        preprocess_cache_raster_vector.path, observers=[task_counter], **execute_kwargs
+    )
+    assert task_counter.tasks
 
 
 @pytest.mark.parametrize(
-    "concurrency,dask_compute_graph",
+    "concurrency,process_graph",
     [
-        # ("threads", False),
+        # ("threads", False),  # profiling does not work with threads
         ("dask", False),
         ("dask", True),
         ("processes", False),
         (None, False),
     ],
 )
-def test_execute_profiling(cleantopo_br_metatiling_1, concurrency, dask_compute_graph):
+def test_execute_profiling(cleantopo_br_metatiling_1, concurrency, process_graph):
+    execute_kwargs = dict(concurrency=concurrency)
+    if concurrency == "dask":
+        execute_kwargs.update(dask_settings=DaskSettings(process_graph=process_graph))
+
     zoom = 5
-    for task_result in execute(
+
+    class TaskResultObserver(ObserverProtocol):
+        def update(self, *args, task_result=None, **kwargs):
+            if task_result:
+                assert isinstance(task_result, TaskInfo)
+                assert task_result.profiling
+                for profiler in ["time", "memory"]:
+                    assert profiler in task_result.profiling
+
+                assert task_result.profiling["time"].elapsed > 0
+
+                assert task_result.profiling["memory"].max_allocated > 0
+                assert task_result.profiling["memory"].total_allocated > 0
+                assert task_result.profiling["memory"].allocations > 0
+
+    execute(
         cleantopo_br_metatiling_1.dict,
         zoom=zoom,
-        as_iterator=True,
         profiling=True,
-        concurrency=concurrency,
-        dask_compute_graph=dask_compute_graph,
-    ):
-        assert isinstance(task_result, TaskResult)
-        assert task_result.profiling
-        for profiler in ["time", "memory"]:
-            assert profiler in task_result.profiling
-
-        assert task_result.profiling["time"].elapsed > 0
-
-        assert task_result.profiling["memory"].max_allocated > 0
-        assert task_result.profiling["memory"].total_allocated > 0
-        assert task_result.profiling["memory"].allocations > 0
+        observers=[TaskResultObserver()],
+        **execute_kwargs
+    )
 
 
 def test_convert_geodetic(cleantopo_br_tif, mp_tmpdir):
     """Automatic geodetic tile pyramid creation of raster files."""
-    job = convert(cleantopo_br_tif, mp_tmpdir, output_pyramid="geodetic")
-    assert len(job)
+    convert(cleantopo_br_tif, mp_tmpdir, output_pyramid="geodetic")
     for zoom, row, col in [(4, 15, 31), (3, 7, 15), (2, 3, 7), (1, 1, 3)]:
         out_file = os.path.join(*[mp_tmpdir, str(zoom), str(row), str(col) + ".tif"])
         with rasterio_open(out_file, "r") as src:
@@ -217,8 +331,7 @@ def test_convert_geodetic(cleantopo_br_tif, mp_tmpdir):
 
 def test_convert_mercator(cleantopo_br_tif, mp_tmpdir):
     """Automatic mercator tile pyramid creation of raster files."""
-    job = convert(cleantopo_br_tif, mp_tmpdir, output_pyramid="mercator")
-    assert len(job)
+    convert(cleantopo_br_tif, mp_tmpdir, output_pyramid="mercator")
     for zoom, row, col in [(4, 15, 15), (3, 7, 7)]:
         out_file = os.path.join(*[mp_tmpdir, str(zoom), str(row), str(col) + ".tif"])
         with rasterio_open(out_file, "r") as src:
@@ -230,8 +343,7 @@ def test_convert_mercator(cleantopo_br_tif, mp_tmpdir):
 
 def test_convert_custom_grid(s2_band, mp_tmpdir, custom_grid_json):
     """Automatic mercator tile pyramid creation of raster files."""
-    job = convert(s2_band, mp_tmpdir, output_pyramid=custom_grid_json)
-    assert len(job)
+    convert(s2_band, mp_tmpdir, output_pyramid=custom_grid_json)
     for zoom, row, col in [(0, 5298, 631)]:
         out_file = mp_tmpdir / zoom / row / col + ".tif"
         with rasterio_open(out_file, "r") as src:
@@ -243,10 +355,8 @@ def test_convert_custom_grid(s2_band, mp_tmpdir, custom_grid_json):
 
 def test_convert_png(cleantopo_br_tif, mp_tmpdir):
     """Automatic PNG tile pyramid creation of raster files."""
-    job = convert(
-        cleantopo_br_tif, mp_tmpdir, output_pyramid="mercator", output_format="PNG"
-    )
-    assert len(job)
+    convert(cleantopo_br_tif, mp_tmpdir, output_pyramid="mercator", output_format="PNG")
+
     for zoom, row, col in [(4, 15, 15), (3, 7, 7)]:
         out_file = os.path.join(*[mp_tmpdir, str(zoom), str(row), str(col) + ".png"])
         with warnings.catch_warnings():
@@ -261,10 +371,7 @@ def test_convert_png(cleantopo_br_tif, mp_tmpdir):
 def test_convert_bidx(cleantopo_br_tif, mp_tmpdir):
     """Automatic geodetic tile pyramid creation of raster files."""
     single_gtiff = os.path.join(mp_tmpdir, "single_out_bidx.tif")
-    job = convert(
-        cleantopo_br_tif, single_gtiff, output_pyramid="geodetic", zoom=3, bidx=1
-    )
-    assert len(job)
+    convert(cleantopo_br_tif, single_gtiff, output_pyramid="geodetic", zoom=3, bidx=1)
     with rasterio_open(single_gtiff, "r") as src:
         assert src.meta["driver"] == "GTiff"
         assert src.meta["dtype"] == "uint16"
@@ -276,8 +383,7 @@ def test_convert_bidx(cleantopo_br_tif, mp_tmpdir):
 def test_convert_single_gtiff(cleantopo_br_tif, mp_tmpdir):
     """Automatic geodetic tile pyramid creation of raster files."""
     single_gtiff = os.path.join(mp_tmpdir, "single_out.tif")
-    job = convert(cleantopo_br_tif, single_gtiff, output_pyramid="geodetic", zoom=3)
-    assert len(job)
+    convert(cleantopo_br_tif, single_gtiff, output_pyramid="geodetic", zoom=3)
     with rasterio_open(single_gtiff, "r") as src:
         assert src.meta["driver"] == "GTiff"
         assert src.meta["dtype"] == "uint16"
@@ -289,10 +395,7 @@ def test_convert_single_gtiff(cleantopo_br_tif, mp_tmpdir):
 def test_convert_single_gtiff_cog(cleantopo_br_tif, mp_tmpdir):
     """Automatic geodetic tile pyramid creation of raster files."""
     single_gtiff = os.path.join(mp_tmpdir, "single_out_cog.tif")
-    job = convert(
-        cleantopo_br_tif, single_gtiff, output_pyramid="geodetic", zoom=5, cog=True
-    )
-    assert len(job)
+    convert(cleantopo_br_tif, single_gtiff, output_pyramid="geodetic", zoom=5, cog=True)
     with rasterio_open(single_gtiff, "r") as src:
         assert src.meta["driver"] == "GTiff"
         assert src.meta["dtype"] == "uint16"
@@ -304,7 +407,7 @@ def test_convert_single_gtiff_cog(cleantopo_br_tif, mp_tmpdir):
 def test_convert_single_gtiff_cog_dask(cleantopo_br_tif, mp_tmpdir):
     """Automatic geodetic tile pyramid creation of raster files."""
     single_gtiff = os.path.join(mp_tmpdir, "single_out_cog.tif")
-    job = convert(
+    convert(
         cleantopo_br_tif,
         single_gtiff,
         output_pyramid="geodetic",
@@ -312,7 +415,6 @@ def test_convert_single_gtiff_cog_dask(cleantopo_br_tif, mp_tmpdir):
         cog=True,
         concurrency="dask",
     )
-    assert len(job)
     with rasterio_open(single_gtiff, "r") as src:
         assert src.meta["driver"] == "GTiff"
         assert src.meta["dtype"] == "uint16"
@@ -324,7 +426,7 @@ def test_convert_single_gtiff_cog_dask(cleantopo_br_tif, mp_tmpdir):
 def test_convert_single_gtiff_overviews(cleantopo_br_tif, mp_tmpdir):
     """Automatic geodetic tile pyramid creation of raster files."""
     single_gtiff = os.path.join(mp_tmpdir, "single_out.tif")
-    job = convert(
+    convert(
         cleantopo_br_tif,
         single_gtiff,
         output_pyramid="geodetic",
@@ -333,7 +435,6 @@ def test_convert_single_gtiff_overviews(cleantopo_br_tif, mp_tmpdir):
         overviews_resampling_method="bilinear",
         concurrency=None,
     )
-    assert len(job)
     with rasterio_open(single_gtiff, "r") as src:
         assert src.meta["driver"] == "GTiff"
         assert src.meta["dtype"] == "uint16"
@@ -346,10 +447,9 @@ def test_convert_single_gtiff_overviews(cleantopo_br_tif, mp_tmpdir):
 def test_convert_remote_single_gtiff(http_raster, mp_tmpdir):
     """Automatic geodetic tile pyramid creation of raster files."""
     single_gtiff = os.path.join(mp_tmpdir, "single_out.tif")
-    job = convert(
+    convert(
         http_raster, single_gtiff, output_pyramid="geodetic", zoom=1, concurrency=None
     )
-    assert len(job)
     with rasterio_open(single_gtiff, "r") as src:
         assert src.meta["driver"] == "GTiff"
         assert src.meta["dtype"] == "uint16"
@@ -359,10 +459,9 @@ def test_convert_remote_single_gtiff(http_raster, mp_tmpdir):
 
 def test_convert_dtype(cleantopo_br_tif, mp_tmpdir):
     """Automatic tile pyramid creation using dtype scale."""
-    job = convert(
+    convert(
         cleantopo_br_tif, mp_tmpdir, output_pyramid="mercator", output_dtype="uint8"
     )
-    assert len(job)
     for zoom, row, col in [(4, 15, 15), (3, 7, 7)]:
         out_file = os.path.join(*[mp_tmpdir, str(zoom), str(row), str(col) + ".tif"])
         with rasterio_open(out_file, "r") as src:
@@ -374,14 +473,13 @@ def test_convert_dtype(cleantopo_br_tif, mp_tmpdir):
 
 def test_convert_scale_ratio(cleantopo_br_tif, mp_tmpdir):
     """Automatic tile pyramid creation cropping data."""
-    job = convert(
+    convert(
         cleantopo_br_tif,
         mp_tmpdir,
         output_pyramid="mercator",
         output_dtype="uint8",
         scale_ratio=0.003,
     )
-    assert len(job)
     for zoom, row, col in [(4, 15, 15), (3, 7, 7)]:
         out_file = os.path.join(*[mp_tmpdir, str(zoom), str(row), str(col) + ".tif"])
         with rasterio_open(out_file, "r") as src:
@@ -394,14 +492,13 @@ def test_convert_scale_ratio(cleantopo_br_tif, mp_tmpdir):
 
 def test_convert_scale_offset(cleantopo_br_tif, mp_tmpdir):
     """Automatic tile pyramid creation cropping data."""
-    job = convert(
+    convert(
         cleantopo_br_tif,
         mp_tmpdir,
         output_pyramid="mercator",
         output_dtype="uint8",
         scale_offset=1,
     )
-    assert len(job)
     for zoom, row, col in [(4, 15, 15), (3, 7, 7)]:
         out_file = os.path.join(*[mp_tmpdir, str(zoom), str(row), str(col) + ".tif"])
         with rasterio_open(out_file, "r") as src:
@@ -414,16 +511,20 @@ def test_convert_scale_offset(cleantopo_br_tif, mp_tmpdir):
 
 def test_convert_clip(cleantopo_br_tif, mp_tmpdir, landpoly):
     """Automatic tile pyramid creation cropping data."""
-    job = convert(
-        cleantopo_br_tif, mp_tmpdir, output_pyramid="geodetic", clip_geometry=landpoly
+    task_counter = TaskCounter()
+    convert(
+        cleantopo_br_tif,
+        mp_tmpdir,
+        output_pyramid="geodetic",
+        clip_geometry=landpoly,
+        observers=[task_counter],
     )
-    assert len(job) == 0
+    assert task_counter.tasks == 0
 
 
 def test_convert_zoom(cleantopo_br_tif, mp_tmpdir):
     """Automatic tile pyramid creation using a specific zoom."""
-    job = convert(cleantopo_br_tif, mp_tmpdir, output_pyramid="mercator", zoom=3)
-    assert len(job)
+    convert(cleantopo_br_tif, mp_tmpdir, output_pyramid="mercator", zoom=3)
     for zoom, row, col in [(4, 15, 15), (2, 3, 0)]:
         out_file = os.path.join(*[mp_tmpdir, str(zoom), str(row), str(col) + ".tif"])
         assert not os.path.isfile(out_file)
@@ -431,8 +532,7 @@ def test_convert_zoom(cleantopo_br_tif, mp_tmpdir):
 
 def test_convert_zoom_minmax(cleantopo_br_tif, mp_tmpdir):
     """Automatic tile pyramid creation using min max zoom."""
-    job = convert(cleantopo_br_tif, mp_tmpdir, output_pyramid="mercator", zoom=[3, 4])
-    assert len(job)
+    convert(cleantopo_br_tif, mp_tmpdir, output_pyramid="mercator", zoom=[3, 4])
     for zoom, row, col in [(2, 3, 0)]:
         out_file = os.path.join(*[mp_tmpdir, str(zoom), str(row), str(col) + ".tif"])
         assert not os.path.isfile(out_file)
@@ -440,8 +540,7 @@ def test_convert_zoom_minmax(cleantopo_br_tif, mp_tmpdir):
 
 def test_convert_zoom_maxmin(cleantopo_br_tif, mp_tmpdir):
     """Automatic tile pyramid creation using max min zoom."""
-    job = convert(cleantopo_br_tif, mp_tmpdir, output_pyramid="mercator", zoom=[4, 3])
-    assert len(job)
+    convert(cleantopo_br_tif, mp_tmpdir, output_pyramid="mercator", zoom=[4, 3])
     for zoom, row, col in [(2, 3, 0)]:
         out_file = os.path.join(*[mp_tmpdir, str(zoom), str(row), str(col) + ".tif"])
         assert not os.path.isfile(out_file)
@@ -449,17 +548,15 @@ def test_convert_zoom_maxmin(cleantopo_br_tif, mp_tmpdir):
 
 def test_convert_mapchete(cleantopo_br, mp_tmpdir):
     # prepare data
-    job = execute(cleantopo_br.path, zoom=[1, 3])
-    assert len(job)
+    execute(cleantopo_br.path, zoom=[1, 3])
 
-    job = convert(
+    convert(
         cleantopo_br.path,
         mp_tmpdir,
         output_pyramid="geodetic",
         output_metatiling=1,
         zoom=[1, 3],
     )
-    assert len(job)
     for zoom, row, col in [(3, 7, 15), (2, 3, 7), (1, 1, 3)]:
         out_file = os.path.join(*[mp_tmpdir, str(zoom), str(row), str(col) + ".tif"])
         with rasterio_open(out_file, "r") as src:
@@ -470,10 +567,11 @@ def test_convert_mapchete(cleantopo_br, mp_tmpdir):
 
 
 def test_convert_tiledir(cleantopo_br, mp_tmpdir):
+    bounds = BufferedTilePyramid("geodetic").tile(4, 15, 31).bounds
     # prepare data
     with mapchete.open(cleantopo_br.dict) as mp:
-        mp.batch_process(zoom=[1, 4])
-    job = convert(
+        list(mp.execute(zoom=[1, 4]))
+    convert(
         os.path.join(
             cleantopo_br.dict["config_dir"], cleantopo_br.dict["output"]["path"]
         ),
@@ -481,8 +579,8 @@ def test_convert_tiledir(cleantopo_br, mp_tmpdir):
         output_pyramid="geodetic",
         output_metatiling=1,
         zoom=[1, 4],
+        bounds=bounds,
     )
-    assert len(job)
     for zoom, row, col in [(4, 15, 31), (3, 7, 15), (2, 3, 7), (1, 1, 3)]:
         out_file = os.path.join(*[mp_tmpdir, str(zoom), str(row), str(col) + ".tif"])
         with rasterio_open(out_file, "r") as src:
@@ -495,8 +593,7 @@ def test_convert_tiledir(cleantopo_br, mp_tmpdir):
 def test_convert_gcps(gcps_tif, mp_tmpdir):
     """Automatic geodetic tile pyramid creation of raster files."""
     out_file = os.path.join(mp_tmpdir, "gcps_out.tif")
-    job = convert(gcps_tif, out_file, output_pyramid="geodetic", zoom=8)
-    assert len(job)
+    convert(gcps_tif, out_file, output_pyramid="geodetic", zoom=8)
     with rasterio_open(out_file, "r") as src:
         assert src.meta["driver"] == "GTiff"
         assert src.meta["dtype"] == "uint16"
@@ -505,8 +602,7 @@ def test_convert_gcps(gcps_tif, mp_tmpdir):
 
 
 def test_convert_geojson(landpoly, mp_tmpdir):
-    job = convert(landpoly, mp_tmpdir, output_pyramid="geodetic", zoom=4)
-    assert len(job)
+    convert(landpoly, mp_tmpdir, output_pyramid="geodetic", zoom=4)
     for (zoom, row, col), control in zip([(4, 0, 7), (4, 1, 7)], [9, 32]):
         out_file = os.path.join(
             *[mp_tmpdir, str(zoom), str(row), str(col) + ".geojson"]
@@ -520,14 +616,13 @@ def test_convert_geojson(landpoly, mp_tmpdir):
 def test_convert_geobuf(landpoly, mp_tmpdir):
     # convert to geobuf
     geobuf_outdir = os.path.join(mp_tmpdir, "geobuf")
-    job = convert(
+    convert(
         landpoly,
         geobuf_outdir,
         output_pyramid="geodetic",
         zoom=4,
         output_format="Geobuf",
     )
-    assert len(job)
     for (zoom, row, col), control in zip([(4, 0, 7), (4, 1, 7)], [9, 32]):
         out_file = os.path.join(
             *[geobuf_outdir, str(zoom), str(row), str(col) + ".pbf"]
@@ -551,7 +646,7 @@ def test_convert_errors(s2_band_jp2, mp_tmpdir, s2_band, cleantopo_br, landpoly)
 
     # prepare data for tiledir input
     with mapchete.open(cleantopo_br.dict) as mp:
-        mp.batch_process(zoom=[1, 4])
+        list(mp.execute(zoom=[1, 4]))
     tiledir_path = os.path.join(
         cleantopo_br.dict["config_dir"], cleantopo_br.dict["output"]["path"]
     )
@@ -581,12 +676,10 @@ def test_convert_errors(s2_band_jp2, mp_tmpdir, s2_band, cleantopo_br, landpoly)
 
 def test_index_geojson(mp_tmpdir, cleantopo_br):
     # execute process at zoom 3
-    job = execute(cleantopo_br.dict, zoom=3)
-    assert len(job)
+    execute(cleantopo_br.dict, zoom=3)
 
     # generate index for zoom 3
-    job = index(cleantopo_br.dict, zoom=3, geojson=True)
-    assert len(job)
+    index(cleantopo_br.dict, zoom=3, geojson=True)
 
     with mapchete.open(cleantopo_br.dict) as mp:
         files = os.listdir(mp.config.output.path)
@@ -600,17 +693,15 @@ def test_index_geojson(mp_tmpdir, cleantopo_br):
 
 def test_index_geojson_fieldname(mp_tmpdir, cleantopo_br):
     # execute process at zoom 3
-    job = execute(cleantopo_br.dict, zoom=3)
-    assert len(job)
+    execute(cleantopo_br.dict, zoom=3)
 
     # index and rename "location" to "new_fieldname"
-    job = index(
+    index(
         cleantopo_br.dict,
         zoom=3,
         geojson=True,
         fieldname="new_fieldname",
     )
-    assert len(job)
     with mapchete.open(cleantopo_br.dict) as mp:
         files = os.listdir(mp.config.output.path)
         assert "3.geojson" in files
@@ -622,13 +713,11 @@ def test_index_geojson_fieldname(mp_tmpdir, cleantopo_br):
 
 def test_index_geojson_basepath(mp_tmpdir, cleantopo_br):
     # execute process at zoom 3
-    job = execute(cleantopo_br.dict, zoom=3)
-    assert len(job)
+    execute(cleantopo_br.dict, zoom=3)
 
     basepath = "http://localhost"
     # index and rename "location" to "new_fieldname"
-    job = index(cleantopo_br.dict, zoom=3, geojson=True, basepath=basepath)
-    assert len(job)
+    index(cleantopo_br.dict, zoom=3, geojson=True, basepath=basepath)
 
     with mapchete.open(cleantopo_br.dict) as mp:
         files = os.listdir(mp.config.output.path)
@@ -641,15 +730,11 @@ def test_index_geojson_basepath(mp_tmpdir, cleantopo_br):
 
 def test_index_geojson_for_gdal(mp_tmpdir, cleantopo_br):
     # execute process at zoom 3
-    job = execute(cleantopo_br.dict, zoom=3)
-    assert len(job)
+    execute(cleantopo_br.dict, zoom=3)
 
     basepath = "http://localhost"
     # index and rename "location" to "new_fieldname"
-    job = index(
-        cleantopo_br.dict, zoom=3, geojson=True, basepath=basepath, for_gdal=True
-    )
-    assert len(job)
+    index(cleantopo_br.dict, zoom=3, geojson=True, basepath=basepath, for_gdal=True)
 
     with mapchete.open(cleantopo_br.dict) as mp:
         files = os.listdir(mp.config.output.path)
@@ -662,12 +747,10 @@ def test_index_geojson_for_gdal(mp_tmpdir, cleantopo_br):
 
 def test_index_geojson_tile(mp_tmpdir, cleantopo_tl):
     # execute process at zoom 3
-    job = execute(cleantopo_tl.dict, zoom=3)
-    assert len(job)
+    execute(cleantopo_tl.dict, zoom=3)
 
     # generate index
-    job = index(cleantopo_tl.dict, tile=(3, 0, 0), geojson=True)
-    assert len(job)
+    index(cleantopo_tl.dict, tile=(3, 0, 0), geojson=True)
 
     with mapchete.open(cleantopo_tl.dict) as mp:
         files = os.listdir(mp.config.output.path)
@@ -679,12 +762,10 @@ def test_index_geojson_tile(mp_tmpdir, cleantopo_tl):
 
 def test_index_geojson_wkt_area(mp_tmpdir, cleantopo_tl, wkt_geom_tl):
     # execute process at zoom 3
-    job = execute(cleantopo_tl.dict, area=wkt_geom_tl)
-    assert len(job)
+    execute(cleantopo_tl.dict, area=wkt_geom_tl)
 
     # generate index for zoom 3
-    job = index(cleantopo_tl.dict, geojson=True, area=wkt_geom_tl)
-    assert len(job)
+    index(cleantopo_tl.dict, geojson=True, area=wkt_geom_tl)
 
     with mapchete.open(cleantopo_tl.dict) as mp:
         files = os.listdir(mp.config.output.path)
@@ -694,12 +775,11 @@ def test_index_geojson_wkt_area(mp_tmpdir, cleantopo_tl, wkt_geom_tl):
 
 def test_index_gpkg(mp_tmpdir, cleantopo_br):
     # execute process
-    job = execute(cleantopo_br.dict, zoom=5)
-    assert len(job)
+    execute(cleantopo_br.dict, zoom=5)
 
     # generate index
-    job = index(cleantopo_br.dict, zoom=5, gpkg=True)
-    assert len(job)
+    index(cleantopo_br.dict, zoom=5, gpkg=True)
+
     with mapchete.open(cleantopo_br.dict) as mp:
         files = os.listdir(mp.config.output.path)
         assert "5.gpkg" in files
@@ -709,8 +789,8 @@ def test_index_gpkg(mp_tmpdir, cleantopo_br):
         assert len(list(src)) == 1
 
     # write again and assert there is no new entry because there is already one
-    job = index(cleantopo_br.dict, zoom=5, gpkg=True)
-    assert len(job)
+    index(cleantopo_br.dict, zoom=5, gpkg=True)
+
     with mapchete.open(cleantopo_br.dict) as mp:
         files = os.listdir(mp.config.output.path)
         assert "5.gpkg" in files
@@ -722,12 +802,11 @@ def test_index_gpkg(mp_tmpdir, cleantopo_br):
 
 def test_index_shp(mp_tmpdir, cleantopo_br):
     # execute process
-    job = execute(cleantopo_br.dict, zoom=5)
-    assert len(job)
+    execute(cleantopo_br.dict, zoom=5)
 
     # generate index
-    job = index(cleantopo_br.dict, zoom=5, shp=True)
-    assert len(job)
+    index(cleantopo_br.dict, zoom=5, shp=True)
+
     with mapchete.open(cleantopo_br.dict) as mp:
         files = os.listdir(mp.config.output.path)
         assert "5.shp" in files
@@ -737,8 +816,8 @@ def test_index_shp(mp_tmpdir, cleantopo_br):
         assert len(list(src)) == 1
 
     # write again and assert there is no new entry because there is already one
-    job = index(cleantopo_br.dict, zoom=5, shp=True)
-    assert len(job)
+    index(cleantopo_br.dict, zoom=5, shp=True)
+
     with mapchete.open(cleantopo_br.dict) as mp:
         files = os.listdir(mp.config.output.path)
         assert "5.shp" in files
@@ -750,12 +829,11 @@ def test_index_shp(mp_tmpdir, cleantopo_br):
 
 def test_index_text(cleantopo_br):
     # execute process
-    job = execute(cleantopo_br.dict, zoom=5)
-    assert len(job)
+    execute(cleantopo_br.dict, zoom=5)
 
     # generate index
-    job = index(cleantopo_br.dict, zoom=5, txt=True)
-    assert len(job)
+    index(cleantopo_br.dict, zoom=5, txt=True)
+
     with mapchete.open(cleantopo_br.dict) as mp:
         files = os.listdir(mp.config.output.path)
         assert "5.txt" in files
@@ -766,8 +844,8 @@ def test_index_text(cleantopo_br):
             assert l.endswith("7.tif\n")
 
     # write again and assert there is no new entry because there is already one
-    job = index(cleantopo_br.dict, zoom=5, txt=True)
-    assert len(job)
+    index(cleantopo_br.dict, zoom=5, txt=True)
+
     with mapchete.open(cleantopo_br.dict) as mp:
         files = os.listdir(mp.config.output.path)
         assert "5.txt" in files
diff --git a/test/test_config.py b/test/test_config.py
index 92af1bbf..d8cf9ac9 100644
--- a/test/test_config.py
+++ b/test/test_config.py
@@ -476,9 +476,8 @@ def test_init_overrides_config(example_mapchete):
 
 def test_custom_process(example_custom_process_mapchete):
     with mapchete.open(example_custom_process_mapchete.dict) as mp:
-        assert (
-            mp.execute(example_custom_process_mapchete.first_process_tile()) is not None
-        )
+        tile = example_custom_process_mapchete.first_process_tile()
+        assert mp.execute_tile(tile) is not None
 
 
 # pytest-env must be installed
diff --git a/test/test_errors.py b/test/test_errors.py
index c5ae2441..ebe88ba9 100644
--- a/test/test_errors.py
+++ b/test/test_errors.py
@@ -31,11 +31,11 @@ def test_execute(example_mapchete):
     # in readonly mode
     with mapchete.open(example_mapchete.dict, mode="readonly") as mp:
         with pytest.raises(AttributeError):
-            mp.execute(next(mp.get_process_tiles()))
+            mp.execute_tile(next(mp.get_process_tiles()))
     # wrong tile type
     with mapchete.open(example_mapchete.dict) as mp:
         with pytest.raises(TypeError):
-            mp.execute("invalid")
+            mp.execute_tile("invalid")
 
 
 def test_read(example_mapchete):
@@ -278,7 +278,7 @@ def test_process_exception(mp_tmpdir, cleantopo_br, process_error_py):
     config.update(process=process_error_py)
     with mapchete.open(config) as mp:
         with pytest.raises(AssertionError):
-            mp.execute((5, 0, 0))
+            mp.execute_tile((5, 0, 0))
 
 
 def test_output_error(mp_tmpdir, cleantopo_br, output_error_py):
@@ -287,7 +287,7 @@ def test_output_error(mp_tmpdir, cleantopo_br, output_error_py):
     config.update(process=output_error_py)
     with mapchete.open(config) as mp:
         with pytest.raises(errors.MapcheteProcessOutputError):
-            mp.execute((5, 0, 0))
+            mp.execute_tile((5, 0, 0))
 
 
 def _raise_error(i):
diff --git a/test/test_executor.py b/test/test_executor.py
index 6bb2360a..2bf86aec 100644
--- a/test/test_executor.py
+++ b/test/test_executor.py
@@ -5,6 +5,7 @@
 import pytest
 
 import mapchete
+from mapchete.config import DaskSettings
 from mapchete.errors import MapcheteTaskFailed
 from mapchete.executor import MFuture
 from mapchete.executor.base import Profiler, Result, run_func_with_profilers
@@ -132,28 +133,22 @@ def test_process_exception_tile(mp_tmpdir, cleantopo_br, process_error_py):
     config.update(process=process_error_py)
     with mapchete.open(config) as mp:
         with pytest.raises(MapcheteTaskFailed):
-            list(mp.compute(tile=(5, 0, 0), concurrency="processes"))
+            list(mp.execute(tile=(5, 0, 0), concurrency="processes"))
 
 
-def test_process_exception_tile_dask(mp_tmpdir, cleantopo_br, process_error_py):
+@pytest.mark.parametrize("process_graph", [True, False])
+def test_process_exception_tile_dask(cleantopo_br, process_error_py, process_graph):
     """Assert process exception is raised."""
     config = cleantopo_br.dict
     config.update(process=process_error_py)
     with mapchete.open(config) as mp:
         with pytest.raises(MapcheteTaskFailed):
             list(
-                mp.compute(tile=(5, 0, 0), concurrency="dask", dask_compute_graph=True)
-            )
-
-
-def test_process_exception_tile_dask_nograph(mp_tmpdir, cleantopo_br, process_error_py):
-    """Assert process exception is raised."""
-    config = cleantopo_br.dict
-    config.update(process=process_error_py)
-    with mapchete.open(config) as mp:
-        with pytest.raises(MapcheteTaskFailed):
-            list(
-                mp.compute(tile=(5, 0, 0), concurrency="dask", dask_compute_graph=False)
+                mp.execute(
+                    tile=(5, 0, 0),
+                    concurrency="dask",
+                    dask_settings=DaskSettings(process_graph=process_graph),
+                )
             )
 
 
@@ -163,25 +158,23 @@ def test_process_exception_zoom(mp_tmpdir, cleantopo_br, process_error_py):
     config.update(process=process_error_py)
     with mapchete.open(config) as mp:
         with pytest.raises(MapcheteTaskFailed):
-            list(mp.compute(zoom=5, concurrency="processes"))
+            list(mp.execute(zoom=5, concurrency="processes"))
 
 
-def test_process_exception_zoom_dask(mp_tmpdir, cleantopo_br, process_error_py):
+@pytest.mark.parametrize("process_graph", [True, False])
+def test_process_exception_zoom_dask(cleantopo_br, process_error_py, process_graph):
     """Assert process exception is raised."""
     config = cleantopo_br.dict
     config.update(process=process_error_py)
     with mapchete.open(config) as mp:
         with pytest.raises(MapcheteTaskFailed):
-            list(mp.compute(zoom=5, concurrency="dask", dask_compute_graph=True))
-
-
-def test_process_exception_zoom_dask_nograph(mp_tmpdir, cleantopo_br, process_error_py):
-    """Assert process exception is raised."""
-    config = cleantopo_br.dict
-    config.update(process=process_error_py)
-    with mapchete.open(config) as mp:
-        with pytest.raises(MapcheteTaskFailed):
-            list(mp.compute(zoom=5, concurrency="dask", dask_compute_graph=False))
+            list(
+                mp.execute(
+                    zoom=5,
+                    concurrency="dask",
+                    dask_settings=DaskSettings(process_graph=process_graph),
+                )
+            )
 
 
 def test_dask_cancellederror(dask_executor, items=10):
@@ -227,15 +220,20 @@ def test_profile_wrapper(request, path_fixture):
 )
 def test_profile_wrapper_requests(request, path_fixture):
     path = request.getfixturevalue(path_fixture)
-    result = run_func_with_profilers(
-        read_raster_no_crs,
-        path,
-        profilers=[
-            Profiler(name="time", decorator=measure_time),
-            Profiler(name="requests", decorator=measure_requests),
-            Profiler(name="memory", decorator=measure_memory),
-        ],
-    )
+
+    # setting this is important, otherwise GDAL will cache the file
+    # and thus measure_requests will not be able to count requests
+    # if the file has already been opened in a prior tests
+    with path.rio_env(opts=dict(CPL_VSIL_CURL_NON_CACHED=path.as_gdal_str())):
+        result = run_func_with_profilers(
+            read_raster_no_crs,
+            path,
+            profilers=[
+                Profiler(name="time", decorator=measure_time),
+                Profiler(name="requests", decorator=measure_requests),
+                Profiler(name="memory", decorator=measure_memory),
+            ],
+        )
     assert isinstance(result, Result)
     assert isinstance(result.output, ma.MaskedArray)
     assert isinstance(result.profiling, dict)
diff --git a/test/test_formats.py b/test/test_formats.py
index 1fe86a12..208666d2 100644
--- a/test/test_formats.py
+++ b/test/test_formats.py
@@ -253,7 +253,7 @@ def test_load_metadata_datetime_list(driver_output_params_dict):
 
 def test_tile_path_schema(tile_path_schema):
     mp = tile_path_schema.mp()
-    mp.batch_process()
+    list(mp.execute())
     tile = tile_path_schema.first_process_tile()
     control = [str(tile.zoom), str(tile.col), str(tile.row) + ".tif"]
     assert mp.config.output_reader.get_path(tile).elements[-3:] == control
@@ -261,7 +261,7 @@ def test_tile_path_schema(tile_path_schema):
 
 def test_tile_path_schema_metadata_json(tile_path_schema):
     mp = tile_path_schema.mp()
-    mp.batch_process()
+    list(mp.execute())
     tile = tile_path_schema.first_process_tile()
     output_metadata = read_output_metadata(
         mp.config.output_reader.path / "metadata.json"
@@ -278,7 +278,7 @@ def test_tile_path_schema_metadata_json(tile_path_schema):
 
 def test_tile_path_schema_stac_json(tile_path_schema):
     mp = tile_path_schema.mp()
-    mp.batch_process()
+    list(mp.execute())
     mp.write_stac()
     stac_json = json.loads((mp.config.output_reader.path / "out.json").read_text())
     template = stac_json.get("asset_templates")["bands"]["href"]
diff --git a/test/test_formats_flatgeobuf.py b/test/test_formats_flatgeobuf.py
index 28f800f5..41c26282 100644
--- a/test/test_formats_flatgeobuf.py
+++ b/test/test_formats_flatgeobuf.py
@@ -7,38 +7,34 @@
 from mapchete.tile import BufferedTile
 
 
-def test_input_data_read(mp_tmpdir, flatgeobuf, landpoly_3857):
+def test_input_data_read(flatgeobuf, landpoly_3857):
     """Check FlatGeobuf as input data."""
+    tile = flatgeobuf.first_process_tile()
     with mapchete.open(flatgeobuf.dict) as mp:
-        for tile in mp.get_process_tiles():
-            assert isinstance(tile, BufferedTile)
-            input_tile = formats.default.geojson.InputTile(tile, mp)
-            assert isinstance(input_tile.read(), list)
-            for feature in input_tile.read():
-                assert "geometry" in feature
-                assert "properties" in feature
+        assert isinstance(tile, BufferedTile)
+        input_tile = formats.default.geojson.InputTile(tile, mp)
+        assert isinstance(input_tile.read(), list)
+        for feature in input_tile.read():
+            assert "geometry" in feature
+            assert "properties" in feature
 
     # reprojected FlatGeobuf
     config = flatgeobuf.dict
     config["input"].update(file1=landpoly_3857)
     # first, write tiles
     with mapchete.open(config, mode="overwrite") as mp:
-        for tile in mp.get_process_tiles(4):
-            assert isinstance(tile, BufferedTile)
-            output = mp.get_raw_output(tile)
-            mp.write(tile, output)
+        assert isinstance(tile, BufferedTile)
+        output = mp.get_raw_output(tile)
+        mp.write(tile, output)
     # then, read output
     with mapchete.open(config, mode="readonly") as mp:
         any_data = False
-        for tile in mp.get_process_tiles(4):
-            with mp.config.output.open(tile, mp) as input_tile:
-                if input_tile.is_empty():
-                    continue
-                any_data = True
-                assert isinstance(input_tile.read(), list)
-                for feature in input_tile.read():
-                    assert "geometry" in feature
-                    assert "properties" in feature
+        with mp.config.output.open(tile, mp) as input_tile:
+            any_data = True
+            assert isinstance(input_tile.read(), list)
+            for feature in input_tile.read():
+                assert "geometry" in feature
+                assert "properties" in feature
         assert any_data
 
 
@@ -94,7 +90,7 @@ def test_s3_output_data_rw(flatgeobuf_s3):
         # write empty
         mp.write(tile, None)
         # write data
-        raw_output = mp.execute(tile)
+        raw_output = mp.execute_tile(tile)
         assert isinstance(raw_output, list)
         assert len(raw_output)
         mp.write(tile, raw_output)
diff --git a/test/test_formats_geobuf.py b/test/test_formats_geobuf.py
index 0051ec2f..592dbc59 100644
--- a/test/test_formats_geobuf.py
+++ b/test/test_formats_geobuf.py
@@ -10,34 +10,30 @@
 
 def test_input_data_read(mp_tmpdir, geobuf, landpoly_3857):
     """Check Geobuf as input data."""
+    tile = geobuf.first_process_tile()
     with mapchete.open(geobuf.dict) as mp:
-        for tile in mp.get_process_tiles():
-            assert isinstance(tile, BufferedTile)
-            input_tile = formats.default.geojson.InputTile(tile, mp)
-            assert isinstance(input_tile.read(), list)
-            for feature in input_tile.read():
-                assert isinstance(feature, dict)
+        assert isinstance(tile, BufferedTile)
+        input_tile = formats.default.geojson.InputTile(tile, mp)
+        assert isinstance(input_tile.read(), list)
+        for feature in input_tile.read():
+            assert isinstance(feature, dict)
 
     # reprojected Geobuf
     config = geobuf.dict
     config["input"].update(file1=landpoly_3857)
     # first, write tiles
     with mapchete.open(config, mode="overwrite") as mp:
-        for tile in mp.get_process_tiles(4):
-            assert isinstance(tile, BufferedTile)
-            output = mp.get_raw_output(tile)
-            mp.write(tile, output)
+        assert isinstance(tile, BufferedTile)
+        output = mp.get_raw_output(tile)
+        mp.write(tile, output)
     # then, read output
     with mapchete.open(config, mode="readonly") as mp:
         any_data = False
-        for tile in mp.get_process_tiles(4):
-            with mp.config.output.open(tile, mp) as input_tile:
-                if input_tile.is_empty():
-                    continue
-                any_data = True
-                assert isinstance(input_tile.read(), list)
-                for feature in input_tile.read():
-                    assert isinstance(feature, dict)
+        with mp.config.output.open(tile, mp) as input_tile:
+            any_data = True
+            assert isinstance(input_tile.read(), list)
+            for feature in input_tile.read():
+                assert isinstance(feature, dict)
         assert any_data
 
 
@@ -117,7 +113,7 @@ def test_s3_output_data_rw(geobuf_s3):
         # write empty
         mp.write(tile, None)
         # write data
-        raw_output = mp.execute(tile)
+        raw_output = mp.execute_tile(tile)
         mp.write(tile, raw_output)
         # read data
         read_output = mp.get_raw_output(tile)
diff --git a/test/test_formats_geojson.py b/test/test_formats_geojson.py
index 84bb09a0..93604703 100644
--- a/test/test_formats_geojson.py
+++ b/test/test_formats_geojson.py
@@ -7,38 +7,34 @@
 from mapchete.tile import BufferedTile
 
 
-def test_input_data_read(mp_tmpdir, geojson, landpoly_3857):
+def test_input_data_read(geojson, landpoly_3857):
     """Check GeoJSON as input data."""
+    tile = geojson.first_process_tile()
     with mapchete.open(geojson.dict) as mp:
-        for tile in mp.get_process_tiles():
-            assert isinstance(tile, BufferedTile)
-            input_tile = formats.default.geojson.InputTile(tile, mp)
-            assert isinstance(input_tile.read(), list)
-            for feature in input_tile.read():
-                assert "geometry" in feature
-                assert "properties" in feature
+        assert isinstance(tile, BufferedTile)
+        input_tile = formats.default.geojson.InputTile(tile, mp)
+        assert isinstance(input_tile.read(), list)
+        for feature in input_tile.read():
+            assert "geometry" in feature
+            assert "properties" in feature
 
     # reprojected GeoJSON
     config = geojson.dict
     config["input"].update(file1=landpoly_3857)
     # first, write tiles
     with mapchete.open(config, mode="overwrite") as mp:
-        for tile in mp.get_process_tiles(4):
-            assert isinstance(tile, BufferedTile)
-            output = mp.get_raw_output(tile)
-            mp.write(tile, output)
+        assert isinstance(tile, BufferedTile)
+        output = mp.get_raw_output(tile)
+        mp.write(tile, output)
     # then, read output
     with mapchete.open(config, mode="readonly") as mp:
         any_data = False
-        for tile in mp.get_process_tiles(4):
-            with mp.config.output.open(tile, mp) as input_tile:
-                if input_tile.is_empty():
-                    continue
-                any_data = True
-                assert isinstance(input_tile.read(), list)
-                for feature in input_tile.read():
-                    assert "geometry" in feature
-                    assert "properties" in feature
+        with mp.config.output.open(tile, mp) as input_tile:
+            any_data = True
+            assert isinstance(input_tile.read(), list)
+            for feature in input_tile.read():
+                assert "geometry" in feature
+                assert "properties" in feature
         assert any_data
 
 
@@ -115,7 +111,7 @@ def test_s3_output_data_rw(geojson_s3):
         # write empty
         mp.write(tile, None)
         # write data
-        raw_output = mp.execute(tile)
+        raw_output = mp.execute_tile(tile)
         mp.write(tile, raw_output)
         # read data
         read_output = mp.get_raw_output(tile)
diff --git a/test/test_formats_geotiff.py b/test/test_formats_geotiff.py
index 77f331cd..7864e9ad 100644
--- a/test/test_formats_geotiff.py
+++ b/test/test_formats_geotiff.py
@@ -170,7 +170,7 @@ def test_write_geotiff_tags(mp_tmpdir, cleantopo_br, write_rasterfile_tags_py):
     conf.update(process=write_rasterfile_tags_py)
     with mapchete.open(conf) as mp:
         for tile in mp.get_process_tiles():
-            data, tags = mp.execute(tile)
+            data, tags = mp.execute_tile(tile)
             assert data.any()
             assert isinstance(tags, dict)
             mp.write(process_tile=tile, data=(data, tags))
@@ -195,7 +195,7 @@ def test_s3_write_output_data(gtiff_s3, s3_example_tile):
         # check if tile exists
         assert not mp.config.output.tiles_exist(process_tile)
         # write
-        mp.batch_process(tile=process_tile.id)
+        list(mp.execute(tile=process_tile.id))
         # check if tile exists
         assert mp.config.output.tiles_exist(process_tile)
         # read again, this time with data
@@ -215,7 +215,7 @@ def test_output_single_gtiff(output_single_gtiff):
         # check if tile exists
         assert not mp.config.output.tiles_exist(process_tile)
         # write
-        mp.batch_process(workers=2)
+        list(mp.execute(workers=2))
         # check if tile exists
         assert mp.config.output.tiles_exist(process_tile)
         # read again, this time with data
@@ -239,7 +239,7 @@ def test_output_single_gtiff(output_single_gtiff):
         process_tile = mp.config.process_pyramid.tile(*tile_id)
         assert not mp.config.output.tiles_exist(process_tile)
         # write
-        mp.batch_process(tile=process_tile.id)
+        list(mp.execute(tile=process_tile.id))
         # check if tile exists
         assert mp.config.output.tiles_exist(process_tile)
         assert mp.config.output.tiles_exist(
@@ -279,7 +279,7 @@ def test_output_single_gtiff_pixelbuffer(output_single_gtiff):
         # check if tile exists
         assert not mp.config.output.tiles_exist(process_tile)
         # write
-        mp.batch_process(tile=process_tile.id)
+        list(mp.execute(tile=process_tile.id))
         # check if tile exists
         assert mp.config.output.tiles_exist(process_tile)
         # read again, this time with data
@@ -299,7 +299,7 @@ def test_output_single_gtiff_compression(output_single_gtiff):
         process_tile = mp.config.process_pyramid.tile(*tile_id)
         assert "compress" in mp.config.output.profile()
         assert mp.config.output.profile()["compress"] == "deflate"
-        mp.batch_process(tile=process_tile.id)
+        list(mp.execute(tile=process_tile.id))
 
     with rasterio_open(mp.config.output.path) as src:
         assert src.profile["compress"] == "deflate"
@@ -319,7 +319,7 @@ def test_output_single_gtiff_overviews(output_single_gtiff):
     ) as mp:
         tile_id = (5, 3, 7)
         process_tile = mp.config.process_pyramid.tile(*tile_id)
-        mp.batch_process(tile=process_tile.id)
+        list(mp.execute(tile=process_tile.id))
 
     with rasterio_open(mp.config.output.path) as src:
         assert src.overviews(1)
@@ -343,7 +343,7 @@ def test_output_single_gtiff_s3(output_single_gtiff_s3):
         # check if tile exists
         assert not mp.config.output.tiles_exist(process_tile)
         # write
-        mp.batch_process(workers=2)
+        list(mp.execute(workers=2))
         # check if tile exists
         assert mp.config.output.tiles_exist(process_tile)
         # read again, this time with data
@@ -379,7 +379,7 @@ def test_output_single_gtiff_s3_tempfile(output_single_gtiff_s3):
         # check if tile exists
         assert not mp.config.output.tiles_exist(process_tile)
         # write
-        mp.batch_process(workers=2)
+        list(mp.execute(workers=2))
         # check if tile exists
         assert mp.config.output.tiles_exist(process_tile)
         # read again, this time with data
@@ -409,7 +409,7 @@ def test_output_single_gtiff_cog(output_single_gtiff_cog):
         # check if tile exists
         assert not mp.config.output.tiles_exist(process_tile)
         # write
-        mp.batch_process(workers=2)
+        list(mp.execute(workers=2))
         # check if tile exists
         assert mp.config.output.tiles_exist(process_tile)
         # read again, this time with data
@@ -445,7 +445,7 @@ def test_output_single_gtiff_cog_tempfile(output_single_gtiff_cog):
         # check if tile exists
         assert not mp.config.output.tiles_exist(process_tile)
         # write
-        mp.batch_process(workers=2)
+        list(mp.execute(workers=2))
         # check if tile exists
         assert mp.config.output.tiles_exist(process_tile)
         # read again, this time with data
@@ -477,7 +477,7 @@ def test_output_single_gtiff_cog_s3(output_single_gtiff_cog_s3):
         # check if tile exists
         assert not mp.config.output.tiles_exist(process_tile)
         # write
-        mp.batch_process(workers=2)
+        list(mp.execute(workers=2))
         # check if tile exists
         assert mp.config.output.tiles_exist(process_tile)
         # read again, this time with data
diff --git a/test/test_formats_tiledir_input.py b/test/test_formats_tiledir_input.py
index 3e9f3883..58d5ee3a 100644
--- a/test/test_formats_tiledir_input.py
+++ b/test/test_formats_tiledir_input.py
@@ -32,27 +32,22 @@ def test_parse_bounds(geojson_tiledir):
         assert ip.bbox(out_crs="3857")
 
 
-def test_read_vector_data(mp_tmpdir, geojson, geojson_tiledir):
+@pytest.mark.parametrize("metatiling", [2, 4, 8])
+def test_read_vector_data(geojson, geojson_tiledir, metatiling):
     """Read vector data."""
+    tile = geojson.first_process_tile()
     # prepare data
     with mapchete.open(geojson.dict) as mp:
         bounds = mp.config.bounds_at_zoom()
-        mp.batch_process(zoom=4)
+        list(mp.execute(tile=tile))
     # read data
     config = geojson_tiledir.dict.copy()
     config["input"]["file1"]["path"] = mp.config.output.path
-    for metatiling in [2, 4, 8]:
-        _run_tiledir_process_vector(config, metatiling, bounds)
-
-
-def _run_tiledir_process_vector(conf_dict, metatiling, bounds):
-    conf = deepcopy(conf_dict)
-    conf["pyramid"].update(metatiling=metatiling)
+    config["pyramid"].update(metatiling=metatiling)
     features = []
-    with mapchete.open(conf, mode="overwrite", bounds=bounds) as mp:
-        for tile in mp.get_process_tiles(4):
-            input_tile = next(iter(mp.config.input.values())).open(tile)
-            features.extend(input_tile.read())
+    with mapchete.open(config, mode="overwrite", bounds=bounds) as mp:
+        input_tile = next(iter(mp.config.input.values())).open(tile)
+        features.extend(input_tile.read())
     assert features
 
 
@@ -61,7 +56,7 @@ def test_read_raster_data(mp_tmpdir, cleantopo_br, cleantopo_br_tiledir):
     # prepare data
     with mapchete.open(cleantopo_br.dict) as mp:
         bounds = mp.config.bounds_at_zoom()
-        mp.batch_process(zoom=4)
+        list(mp.execute(zoom=4))
     config = cleantopo_br_tiledir.dict.copy()
     config["input"]["file1"]["path"] = mp.config.output.path
     for metatiling in [1, 2, 4, 8]:
@@ -96,7 +91,7 @@ def test_read_reprojected_raster_data(
     zoom = 4
     # prepare data
     with mapchete.open(cleantopo_br.dict) as mp:
-        mp.batch_process(zoom=zoom)
+        list(mp.execute(zoom=zoom))
 
     config = cleantopo_br_tiledir_mercator.dict.copy()
     config["input"]["file1"] = mp.config.output.path
@@ -160,7 +155,7 @@ def test_read_from_dir(mp_tmpdir, cleantopo_br, cleantopo_br_tiledir):
     # prepare data
     with mapchete.open(cleantopo_br.dict) as mp:
         bounds = mp.config.bounds_at_zoom()
-        mp.batch_process(zoom=4)
+        list(mp.execute(zoom=4))
     config = dict(cleantopo_br_tiledir.dict, input=dict(file1=mp.config.output.path))
     _run_tiledir_process_raster(config, 4, bounds)
 
@@ -168,7 +163,7 @@ def test_read_from_dir(mp_tmpdir, cleantopo_br, cleantopo_br_tiledir):
 def test_read_indexes_shape(cleantopo_br_tiledir):
     mp = cleantopo_br_tiledir.mp()
     # create local TileDirectory
-    list(mp.compute())
+    list(mp.execute())
     input_data = InputData({"path": mp.config.output_reader.path})
     input_tile = input_data.open(cleantopo_br_tiledir.first_process_tile())
 
diff --git a/test/test_index.py b/test/test_index.py
index a2e389ce..93f8e771 100644
--- a/test/test_index.py
+++ b/test/test_index.py
@@ -43,7 +43,7 @@ def gen_indexes_and_check():
 
     with mapchete.open(gtiff_s3.dict) as mp:
         # write output data
-        mp.batch_process(zoom=zoom)
+        list(mp.execute(zoom=zoom))
 
         # generate indexes and check
         gen_indexes_and_check()
@@ -58,7 +58,7 @@ def test_vrt(mp_tmpdir, cleantopo_br):
         dict(cleantopo_br.dict, zoom_levels=dict(min=0, max=zoom))
     ) as mp:
         # generate output
-        mp.batch_process(zoom=zoom)
+        list(mp.execute(zoom=zoom))
 
         # generate index
         list(
@@ -114,7 +114,7 @@ def test_vrt(mp_tmpdir, cleantopo_br):
         dict(cleantopo_br.dict, zoom_levels=dict(min=0, max=zoom))
     ) as mp:
         # generate output
-        mp.batch_process(zoom=zoom)
+        list(mp.execute(zoom=zoom))
 
         # generate index
         list(
@@ -128,12 +128,12 @@ def test_vrt(mp_tmpdir, cleantopo_br):
 
 
 def test_vrt_mercator(cleantopo_br_mercator):
-    zoom = 8
+    zoom = 2
     with mapchete.open(
         dict(cleantopo_br_mercator.dict, zoom_levels=dict(min=0, max=zoom))
     ) as mp:
         # generate output
-        mp.batch_process(zoom=zoom)
+        list(mp.execute(zoom=zoom))
 
         # generate index
         list(
@@ -192,7 +192,7 @@ def test_vrt_mercator(cleantopo_br_mercator):
         dict(cleantopo_br_mercator.dict, zoom_levels=dict(min=0, max=zoom))
     ) as mp:
         # generate output
-        mp.batch_process(zoom=zoom)
+        list(mp.execute(zoom=zoom))
 
         # generate index
         list(
diff --git a/test/test_io.py b/test/test_io.py
index 295f0355..8e6e1330 100644
--- a/test/test_io.py
+++ b/test/test_io.py
@@ -196,7 +196,7 @@ def test_tiles_exist_local(example_mapchete):
         bounds=bounds,
     ) as mp:
         # generate tile directory
-        mp.batch_process(zoom=zoom)
+        list(mp.execute(zoom=zoom))
         process_tiles = list(mp.config.process_pyramid.tiles_from_bounds(bounds, zoom))
         output_tiles = list(mp.config.output_pyramid.tiles_from_bounds(bounds, zoom))
 
@@ -258,7 +258,7 @@ def test_tiles_exist_s3(gtiff_s3):
         mode="overwrite",
     ) as mp:
         # generate tile directory
-        mp.batch_process(zoom=zoom)
+        list(mp.execute(zoom=zoom))
         process_tiles = list(mp.config.process_pyramid.tiles_from_bounds(bounds, zoom))
         output_tiles = list(mp.config.output_pyramid.tiles_from_bounds(bounds, zoom))
 
diff --git a/test/test_io_raster.py b/test/test_io_raster.py
index ddbacb4c..47ab3592 100644
--- a/test/test_io_raster.py
+++ b/test/test_io_raster.py
@@ -84,7 +84,7 @@ def test_read_raster_window_input_list(cleantopo_br):
     conf = dict(**cleantopo_br.dict)
     conf["output"].update(metatiling=1)
     with mapchete.open(conf) as mp:
-        mp.batch_process(process_zoom)
+        list(mp.execute(zoom=process_zoom))
         tiles = [
             (tile, mp.config.output.get_path(tile))
             for tile in mp.config.output_pyramid.tiles_from_bounds(
@@ -935,7 +935,7 @@ def test_output_s3_single_gtiff_error(output_s3_single_gtiff_error):
     # the process file will raise an exception on purpose
     with pytest.raises(AssertionError):
         with output_s3_single_gtiff_error.mp() as mp:
-            mp.execute(output_s3_single_gtiff_error.first_process_tile())
+            mp.execute_tile(output_s3_single_gtiff_error.first_process_tile())
     # make sure no output has been written
     assert not path_exists(mp.config.output.path)
 
diff --git a/test/test_preprocessing.py b/test/test_preprocessing.py
index 144e3d98..48ff53a9 100644
--- a/test/test_preprocessing.py
+++ b/test/test_preprocessing.py
@@ -42,10 +42,12 @@ def test_add_preprocessing_task(example_mapchete):
         assert mp.count_tasks() == tasks + 1
 
 
-def test_run_preprocessing_task(example_mapchete):
-    with mapchete.open(example_mapchete.dict) as mp:
+@pytest.mark.parametrize("concurrency", ["threads", "processes", "dask", None])
+def test_run_preprocessing_task(cleantopo_br, concurrency):
+    zoom = 5
+    with mapchete.open(cleantopo_br.dict) as mp:
         # get input object
-        inp = mp.config.input_at_zoom("file1", 10)
+        inp = mp.config.input_at_zoom("file1", zoom)
         with pytest.raises(KeyError):
             inp.get_preprocessing_task_result("test_task")
         # add a preprocessing task and make sure it is registered
@@ -54,24 +56,25 @@ def test_run_preprocessing_task(example_mapchete):
         )
         with pytest.raises(ValueError):
             inp.get_preprocessing_task_result("test_task")
-        mp.batch_preprocess()
+        list(
+            mp.execute(
+                concurrency=concurrency, zoom=zoom, remember_preprocessing_results=True
+            )
+        )
         assert inp.get_preprocessing_task_result("test_task") == "foofoobar"
 
 
-def test_run_preprocessing_tasks(example_mapchete):
-    with mapchete.open(example_mapchete.dict) as mp:
-        inp1 = mp.config.input_at_zoom("file1", 10)
+@pytest.mark.parametrize("concurrency", ["threads", "processes", "dask", None])
+def test_run_preprocessing_tasks(cleantopo_br, concurrency):
+    zoom = 3
+    with mapchete.open(cleantopo_br.dict) as mp:
+        inp1 = mp.config.input_at_zoom("file1", zoom)
         inp1.add_preprocessing_task(
             _trivial_func, key="test_task", fargs="foo", fkwargs={"kwarg": "foo"}
         )
         inp1.add_preprocessing_task(
             _trivial_func, key="test_other_task", fargs="bar", fkwargs={"kwarg": "foo"}
         )
-        inp2 = mp.config.input_at_zoom("file2", 10)
-        inp2.add_preprocessing_task(
-            _trivial_func, key="test_task", fargs="foo", fkwargs={"kwarg": "foo"}
-        )
-
         # total number of tasks should be 3
         assert (
             sum(
@@ -80,20 +83,22 @@ def test_run_preprocessing_tasks(example_mapchete):
                     for tasks in mp.config.preprocessing_tasks_per_input().values()
                 ]
             )
-            == 3
+            == 2
         )
 
-        assert mp.config.preprocessing_tasks_count() == 3
+        assert mp.config.preprocessing_tasks_count() == 2
 
-        mp.batch_preprocess()
+        list(
+            mp.execute(
+                concurrency=concurrency, zoom=zoom, remember_preprocessing_results=True
+            )
+        )
         assert mp.config.preprocessing_task_finished(f"{inp1.input_key}:test_task")
         assert inp1.get_preprocessing_task_result("test_task") == "foofoobar"
         assert mp.config.preprocessing_task_finished(
             f"{inp1.input_key}:test_other_task"
         )
         assert inp1.get_preprocessing_task_result("test_other_task") == "barfoobar"
-        assert mp.config.preprocessing_task_finished(f"{inp2.input_key}:test_task")
-        assert inp2.get_preprocessing_task_result("test_task") == "foofoobar"
 
 
 def test_preprocess_cache_raster_vector(preprocess_cache_raster_vector):
@@ -132,7 +137,7 @@ def test_preprocessing_tasks_dependencies(preprocess_cache_memory):
             for task in input_data.preprocessing_tasks.values():
                 assert isinstance(task, Task)
                 assert task.has_geometry()
-        list(mp.compute())
+        list(mp.execute())
 
         out_path = mp.config.output_reader.path
         total_tifs = len(
@@ -154,7 +159,7 @@ def test_preprocessing_tasks_dependencies_dask(preprocess_cache_memory):
                 assert isinstance(task, Task)
                 assert task.has_geometry()
 
-        list(mp.compute(concurrency="dask"))
+        list(mp.execute(concurrency="dask"))
 
         out_path = mp.config.output_reader.path
         total_tifs = [
@@ -166,34 +171,10 @@ def test_preprocessing_tasks_dependencies_dask(preprocess_cache_memory):
         assert len(total_tifs) == 9
 
 
-def test_preprocessing_tasks_dependencies_single_tile(preprocess_cache_memory):
-    with preprocess_cache_memory.mp(batch_preprocess=False) as mp:
-        for i in ["clip", "inp"]:
-            input_data = mp.config.input_at_zoom(key=i, zoom=5)
-            for task in input_data.preprocessing_tasks.values():
-                assert isinstance(task, Task)
-                assert task.has_geometry()
-
-        tile = (5, 31, 63)
-        list(mp.compute(tile=tile))
-
-        out_path = mp.config.output_reader.get_path(
-            mp.config.process_pyramid.tile(*tile)
-        )
-        with rasterio_open(out_path) as src:
-            assert not src.read(masked=True).mask.all()
-
-        out_path = mp.config.output_reader.path
-        total_tifs = [
-            f"{directory[0]}/{file}"
-            for directory in fs_from_path(out_path).walk(out_path)
-            for file in directory[2]
-            if file.endswith(".tif")
-        ]
-        assert len(total_tifs) == 1
-
-
-def test_preprocessing_tasks_dependencies_single_tile_dask(preprocess_cache_memory):
+@pytest.mark.parametrize("concurrency", ["threads", "processes", "dask", None])
+def test_preprocessing_tasks_dependencies_single_tile(
+    preprocess_cache_memory, concurrency
+):
     with preprocess_cache_memory.mp(batch_preprocess=False) as mp:
         for i in ["clip", "inp"]:
             input_data = mp.config.input_at_zoom(key=i, zoom=5)
@@ -202,7 +183,10 @@ def test_preprocessing_tasks_dependencies_single_tile_dask(preprocess_cache_memo
                 assert task.has_geometry()
 
         tile = (5, 31, 63)
-        list(mp.compute(concurrency="dask", tile=tile))
+        for task_info in mp.execute(
+            tile=tile, concurrency=concurrency, remember_preprocessing_results=True
+        ):
+            pass
 
         out_path = mp.config.output_reader.get_path(
             mp.config.process_pyramid.tile(*tile)
diff --git a/test/test_mapchete.py b/test/test_processing_base.py
similarity index 78%
rename from test/test_mapchete.py
rename to test/test_processing_base.py
index 0c2c7bf5..a37ff0cc 100644
--- a/test/test_mapchete.py
+++ b/test/test_processing_base.py
@@ -20,17 +20,18 @@
 from shapely.ops import unary_union
 
 import mapchete
+from mapchete.config import DaskSettings
 from mapchete.errors import MapcheteProcessOutputError
 from mapchete.io import fs_from_path, rasterio_open
 from mapchete.io.raster import _shift_required, create_mosaic
-from mapchete.processing.types import PreprocessingProcessInfo, TileProcessInfo
+from mapchete.processing.types import TaskInfo
 from mapchete.tile import BufferedTilePyramid, count_tiles
 
 
 def test_empty_execute(cleantopo_br):
     """Execute process outside of defined zoom levels."""
     with mapchete.open(cleantopo_br.dict) as mp:
-        assert mp.execute((6, 0, 0)).mask.all()
+        assert mp.execute_tile((6, 0, 0)).mask.all()
 
 
 def test_read_existing_output(cleantopo_tl):
@@ -157,7 +158,7 @@ def test_baselevels(baselevels):
     """Baselevel interpolation."""
     with mapchete.open(baselevels.dict, mode="continue") as mp:
         # process data before getting baselevels
-        mp.batch_process()
+        list(mp.execute(concurrency=None))
 
         # get tile from lower zoom level
         for tile in mp.get_process_tiles(4):
@@ -184,7 +185,7 @@ def test_baselevels_dask(baselevels, dask_executor):
     """Baselevel interpolation."""
     with mapchete.open(baselevels.dict, mode="continue") as mp:
         # process data before getting baselevels
-        mp.batch_process(executor=dask_executor)
+        list(mp.execute(executor=dask_executor))
 
         # get tile from lower zoom level
         for tile in mp.get_process_tiles(4):
@@ -212,7 +213,7 @@ def test_baselevels_custom_nodata(baselevels_custom_nodata):
     fill_value = -32768.0
     with mapchete.open(baselevels_custom_nodata.dict, mode="continue") as mp:
         # process data before getting baselevels
-        mp.batch_process()
+        list(mp.execute())
 
         # get tile from lower zoom level
         for tile in mp.get_process_tiles(4):
@@ -256,22 +257,24 @@ def test_baselevels_custom_nodata(baselevels_custom_nodata):
 
 
 @pytest.mark.parametrize(
-    "concurrency,dask_compute_graph",
+    "concurrency,process_graph",
     [
-        ("threads", False),
-        ("dask", False),
+        ("threads", None),
         ("dask", True),
-        ("processes", False),
-        (None, False),
+        ("dask", False),
+        ("processes", None),
+        (None, None),
     ],
 )
-def test_update_overviews(overviews, concurrency, dask_compute_graph, dask_executor):
+def test_update_overviews(overviews, concurrency, process_graph, dask_executor):
     """Baselevel interpolation."""
-    compute_kwargs = (
-        {"executor": dask_executor, "dask_compute_graph": dask_compute_graph}
-        if concurrency == "dask"
-        else {"concurrency": concurrency}
-    )
+    if concurrency == "dask":
+        execute_kwargs = dict(
+            executor=dask_executor,
+            dask_settings=DaskSettings(process_graph=process_graph),
+        )
+    else:
+        execute_kwargs = dict(concurrency=concurrency)
     # process everything and make sure output was written
     with overviews.mp() as mp:
         baselevel_tile = overviews.first_process_tile()
@@ -284,7 +287,7 @@ def test_update_overviews(overviews, concurrency, dask_compute_graph, dask_execu
         )
 
         # process baselevel (zoom 7)
-        list(mp.compute(**compute_kwargs, zoom=7))
+        list(mp.execute(**execute_kwargs, zoom=7))
 
     # make sure baselevel tile has content
     with rasterio_open(baselevel_tile_path) as src:
@@ -296,7 +299,7 @@ def test_update_overviews(overviews, concurrency, dask_compute_graph, dask_execu
 
     with overviews.mp() as mp:
         # process overviews
-        list(mp.compute(**compute_kwargs, zoom=[0, 6]))
+        list(mp.execute(**execute_kwargs, zoom=[0, 6]))
     assert not baselevel_tile_path.exists()
 
     # read overview tile which is half empty
@@ -308,7 +311,7 @@ def test_update_overviews(overviews, concurrency, dask_compute_graph, dask_execu
     # the overview tiles
     with overviews.mp() as mp:
         # process data before getting baselevels
-        list(mp.compute(concurrency=concurrency))
+        list(mp.execute(concurrency=concurrency))
 
     assert baselevel_tile_path.exists()
     with rasterio_open(
@@ -352,7 +355,7 @@ def test_baselevels_output_buffer(baselevels_output_buffer):
     # (171.46155, -87.27184, 174.45159, -84.31281)
     with mapchete.open(baselevels_output_buffer.dict) as mp:
         # process all
-        mp.batch_process()
+        list(mp.execute())
         # read tile 6/62/125.tif
         with rasterio_open(
             mp.config.output.output_params["path"] / 6 / 62 / 125 + ".tif"
@@ -395,7 +398,7 @@ def test_processing(mp_tmpdir, cleantopo_br, cleantopo_tl):
             for zoom in range(6):
                 tiles = []
                 for tile in mp.get_process_tiles(zoom):
-                    output = mp.execute(tile)
+                    output = mp.execute_tile(tile)
                     tiles.append((tile, output))
                     assert isinstance(output, ma.MaskedArray)
                     assert output.shape == output.shape
@@ -436,7 +439,7 @@ def test_pickleability(cleantopo_tl):
 
 def _worker(mp, tile):
     """Multiprocessing worker processing a tile."""
-    return tile, mp.execute(tile)
+    return tile, mp.execute_tile(tile)
 
 
 def test_write_empty(cleantopo_tl):
@@ -464,7 +467,7 @@ def test_process_template(dummy1_tif, mp_tmpdir):
     process_tile = next(mp.get_process_tiles(zoom=4))
     # Mapchete throws a RuntimeError if process output is empty
     with pytest.raises(MapcheteProcessOutputError):
-        mp.execute(process_tile)
+        mp.execute_tile(process_tile)
 
 
 @pytest.mark.parametrize("minzoom", range(0, 14))
@@ -513,25 +516,11 @@ def test_count_tiles_large_init_zoom(geometrycollection, zoom):
     assert raster_count == tile_count == tiles
 
 
-def test_batch_process(cleantopo_tl):
-    """Test batch_process function."""
-    with mapchete.open(cleantopo_tl.dict) as mp:
-        # invalid parameters errors
-        with pytest.raises(ValueError):
-            mp.batch_process(zoom=1, tile=(1, 0, 0))
-        # process single tile
-        mp.batch_process(tile=(2, 0, 0))
-        # process using multiprocessing
-        mp.batch_process(zoom=2, workers=2)
-        # process without multiprocessing
-        mp.batch_process(zoom=2, workers=1)
-
-
 def test_skip_tiles(cleantopo_tl):
     """Test batch_process function."""
     zoom = 2
     with mapchete.open(cleantopo_tl.dict, mode="continue") as mp:
-        mp.batch_process(zoom=zoom)
+        list(mp.execute(zoom=zoom))
         for tile, skip in mp.skip_tiles(tiles=mp.get_process_tiles(zoom=zoom)):
             assert skip
 
@@ -544,7 +533,7 @@ def test_custom_grid(custom_grid):
     """Cutom grid processing."""
     # process and save
     with mapchete.open(custom_grid.dict) as mp:
-        mp.batch_process()
+        list(mp.execute())
     # read written output
     with mapchete.open(custom_grid.dict) as mp:
         for tile in mp.get_process_tiles(5):
@@ -558,7 +547,7 @@ def test_execute_kwargs(example_mapchete, execute_kwargs_py):
     config = example_mapchete.dict
     config.update(process=execute_kwargs_py)
     with mapchete.open(config) as mp:
-        mp.execute((7, 61, 129))
+        mp.execute_tile((7, 61, 129))
 
 
 @pytest.mark.parametrize("pixelbuffer", [0, 5, 10])
@@ -638,60 +627,65 @@ def test_bufferedtiles():
     assert a.get_neighbors() != a.get_neighbors(connectedness=4)
 
 
-@pytest.mark.parametrize("concurrency", ["processes", "dask", "threads", None])
-def test_compute(preprocess_cache_memory, concurrency, dask_executor):
-    compute_kwargs = (
-        {"executor": dask_executor}
-        if concurrency == "dask"
-        else {"concurrency": concurrency}
-    )
-    with preprocess_cache_memory.mp(batch_preprocess=False) as mp:
-        preprocessing_tasks = 0
-        tile_tasks = 0
-        for future in mp.compute(**compute_kwargs, dask_compute_graph=False):
-            result = future.result()
-            if isinstance(result, PreprocessingProcessInfo):
-                preprocessing_tasks += 1
-            else:
-                assert isinstance(result, TileProcessInfo)
-                tile_tasks += 1
-                assert result.data is None
-    assert tile_tasks == 20
-    assert preprocessing_tasks == 2
-
+@pytest.mark.parametrize(
+    "concurrency,process_graph",
+    [
+        ("threads", None),
+        ("dask", True),
+        ("dask", False),
+        ("processes", None),
+        (None, None),
+    ],
+)
+def test_execute(preprocess_cache_memory, concurrency, process_graph, dask_executor):
+    if concurrency == "dask":
+        execute_kwargs = dict(
+            executor=dask_executor,
+            dask_settings=DaskSettings(process_graph=process_graph),
+        )
+    else:
+        execute_kwargs = dict(concurrency=concurrency)
 
-def test_compute_dask_graph(preprocess_cache_memory, dask_executor):
     with preprocess_cache_memory.mp(batch_preprocess=False) as mp:
         preprocessing_tasks = 0
         tile_tasks = 0
-        for future in mp.compute(executor=dask_executor, dask_compute_graph=True):
-            result = future.result()
-            if isinstance(result, PreprocessingProcessInfo):
-                assert result.data is not None
+        for task_info in mp.execute(**execute_kwargs):
+            assert isinstance(task_info, TaskInfo)
+            if task_info.tile is None:
                 preprocessing_tasks += 1
             else:
-                assert isinstance(result, TileProcessInfo)
                 tile_tasks += 1
-                if result.written:
-                    assert result.data is not None
-                else:
-                    assert result.data is None
+                assert task_info.output is None
     assert tile_tasks == 20
     assert preprocessing_tasks == 2
 
 
-@pytest.mark.parametrize("concurrency", ["processes", "dask", "threads", None])
-def test_compute_continue(red_raster, green_raster, dask_executor, concurrency):
-    compute_kwargs = (
-        {"executor": dask_executor, "dask_compute_graph": True}
-        if concurrency == "dask"
-        else {"concurrency": concurrency}
-    )
+@pytest.mark.parametrize(
+    "concurrency,process_graph",
+    [
+        ("threads", None),
+        ("dask", True),
+        ("dask", False),
+        ("processes", None),
+        (None, None),
+    ],
+)
+def test_execute_continue(
+    red_raster, green_raster, dask_executor, concurrency, process_graph
+):
+    if concurrency == "dask":
+        execute_kwargs = dict(
+            executor=dask_executor,
+            dask_settings=DaskSettings(process_graph=process_graph),
+        )
+    else:
+        execute_kwargs = dict(concurrency=concurrency)
+
     zoom = 3
 
     # run red_raster on tile 1, 0, 0
     with red_raster.mp() as mp_red:
-        list(mp_red.compute(tile=(zoom, 0, 0), **compute_kwargs))
+        list(mp_red.execute(tile=(zoom, 0, 0), **execute_kwargs))
     fs_red = fs_from_path(mp_red.config.output.path)
     assert len(fs_red.glob(f"{mp_red.config.output.path}/*/*/*.tif")) == 1
     with rasterio_open(f"{mp_red.config.output.path}/{zoom}/0/0.tif") as src:
@@ -709,7 +703,7 @@ def test_compute_continue(red_raster, green_raster, dask_executor, concurrency):
             str(mp_green.config.output.path / f"{zoom}/0/0.tif"),
         )
         # run green_raster on zoom 1
-        list(mp_green.compute(zoom=[0, zoom], **compute_kwargs))
+        list(mp_green.execute(zoom=[0, zoom], **execute_kwargs))
 
     # assert red tile is still there and other tiles were written and are green
     assert len(
@@ -752,101 +746,81 @@ def test_compute_continue(red_raster, green_raster, dask_executor, concurrency):
                 )
 
 
-def test_compute_dask_without_results(baselevels, dask_executor):
+@pytest.mark.parametrize(
+    "concurrency,process_graph",
+    [
+        ("threads", None),
+        ("dask", True),
+        ("dask", False),
+        ("processes", None),
+        (None, None),
+    ],
+)
+def test_execute_without_results(baselevels, dask_executor, concurrency, process_graph):
+    if concurrency == "dask":
+        execute_kwargs = dict(
+            executor=dask_executor,
+            dask_settings=DaskSettings(process_graph=process_graph),
+        )
+    else:
+        execute_kwargs = dict(concurrency=concurrency)
+
     # make sure task results are appended to tasks
     with baselevels.mp() as mp:
         tile_tasks = 0
-        for future in mp.compute(
-            executor=dask_executor, dask_compute_graph=True, dask_propagate_results=True
-        ):
-            result = future.result()
-            assert result.data is not None
+        for task_info in mp.execute(**execute_kwargs, propagate_results=True):
+            assert task_info.output is not None
             tile_tasks += 1
     assert tile_tasks == 6
 
     # make sure task results are None
     with baselevels.mp() as mp:
         tile_tasks = 0
-        for future in mp.compute(
-            concurrency="dask", dask_compute_graph=True, dask_propagate_results=False
-        ):
-            result = future.result()
-            assert result.data is None
+        for task_info in mp.execute(**execute_kwargs, propagate_results=False):
+            assert task_info.output is None
             tile_tasks += 1
     assert tile_tasks == 6
 
 
-def test_compute_dask_graph_single_file(
-    preprocess_cache_memory_single_file, dask_executor
+@pytest.mark.parametrize(
+    "concurrency,process_graph",
+    [
+        ("threads", None),
+        ("dask", True),
+        ("dask", False),
+        ("processes", None),
+        (None, None),
+    ],
+)
+def test_execute_single_file(
+    preprocess_cache_memory_single_file, concurrency, process_graph, dask_executor
 ):
-    with preprocess_cache_memory_single_file.mp(batch_preprocess=False) as mp:
-        preprocessing_tasks = 0
-        tile_tasks = 0
-        for future in mp.compute(executor=dask_executor, dask_compute_graph=True):
-            result = future.result()
-            if isinstance(result, PreprocessingProcessInfo):
-                assert result.data is not None
-                preprocessing_tasks += 1
-            else:
-                assert isinstance(result, TileProcessInfo)
-                tile_tasks += 1
-                assert result.data is None
-    assert tile_tasks == 12
-    assert preprocessing_tasks == 2
-    with rasterio_open(mp.config.output.path) as src:
-        assert not src.read(masked=True).mask.all()
-
+    """Baselevel interpolation."""
+    if concurrency == "dask":
+        execute_kwargs = dict(
+            executor=dask_executor,
+            dask_settings=DaskSettings(process_graph=process_graph),
+        )
+    else:
+        execute_kwargs = dict(concurrency=concurrency)
 
-def test_compute_dask_single_file(preprocess_cache_memory_single_file, dask_executor):
     with preprocess_cache_memory_single_file.mp(batch_preprocess=False) as mp:
         preprocessing_tasks = 0
         tile_tasks = 0
-        for future in mp.compute(executor=dask_executor, dask_compute_graph=False):
-            result = future.result()
-            if isinstance(result, PreprocessingProcessInfo):
+        for task_info in mp.execute(**execute_kwargs):
+            assert isinstance(task_info, TaskInfo)
+            if task_info.tile is None:
+                assert task_info.output is not None
                 preprocessing_tasks += 1
             else:
-                assert isinstance(result, TileProcessInfo)
                 tile_tasks += 1
-                assert result.data is None
+                assert task_info.output is None
     assert tile_tasks == 12
     assert preprocessing_tasks == 2
     with rasterio_open(mp.config.output.path) as src:
         assert not src.read(masked=True).mask.all()
 
 
-def test_compute_threads_single_file(preprocess_cache_memory_single_file):
-    with preprocess_cache_memory_single_file.mp(batch_preprocess=False) as mp:
-        preprocessing_tasks = 0
-        tile_tasks = 0
-        for future in mp.compute(concurrency="threads", dask_compute_graph=False):
-            result = future.result()
-            if isinstance(result, PreprocessingProcessInfo):
-                preprocessing_tasks += 1
-            else:
-                assert isinstance(result, TileProcessInfo)
-                tile_tasks += 1
-                assert result.data is None
-    assert tile_tasks == 12
-    assert preprocessing_tasks == 2
-
-
-def test_compute_processes_single_file(preprocess_cache_memory_single_file):
-    with preprocess_cache_memory_single_file.mp(batch_preprocess=False) as mp:
-        preprocessing_tasks = 0
-        tile_tasks = 0
-        for future in mp.compute(concurrency="processes", dask_compute_graph=False):
-            result = future.result()
-            if isinstance(result, PreprocessingProcessInfo):
-                preprocessing_tasks += 1
-            else:
-                assert isinstance(result, TileProcessInfo)
-                tile_tasks += 1
-                assert result.data is None
-    assert tile_tasks == 12
-    assert preprocessing_tasks == 2
-
-
 def test_write_stac(stac_metadata):
     mp = stac_metadata.mp()
     out_path = stac_metadata.mp().config.output.stac_path
@@ -861,24 +835,47 @@ def test_write_stac(stac_metadata):
     assert item
 
 
-# @pytest.mark.parametrize("concurrency", ["processes", "dask", "threads", None])
-@pytest.mark.parametrize("concurrency", [None])
-def test_compute_request_count(preprocess_cache_memory, concurrency, dask_executor):
-    compute_kwargs = (
-        {"executor": dask_executor}
-        if concurrency == "dask"
-        else {"concurrency": concurrency}
-    )
+@pytest.mark.parametrize(
+    "concurrency,process_graph",
+    [
+        # ("threads", None),  --> request count does not work on threads
+        ("dask", True),
+        ("dask", False),
+        ("processes", None),
+        (None, None),
+    ],
+)
+def test_execute_request_count(
+    preprocess_cache_memory, concurrency, process_graph, dask_executor
+):
+    if concurrency == "dask":
+        execute_kwargs = dict(
+            executor=dask_executor,
+            dask_settings=DaskSettings(process_graph=process_graph),
+        )
+    else:
+        execute_kwargs = dict(concurrency=concurrency)
     with preprocess_cache_memory.mp(batch_preprocess=False) as mp:
         preprocessing_tasks = 0
         tile_tasks = 0
-        for future in mp.compute(**compute_kwargs, dask_compute_graph=False):
-            result = future.result()
-            if isinstance(result, PreprocessingProcessInfo):
+        for task_info in mp.execute(**execute_kwargs):
+            if task_info.tile is None:
                 preprocessing_tasks += 1
             else:
-                assert isinstance(result, TileProcessInfo)
                 tile_tasks += 1
-                assert result.data is None
+                assert task_info.output is None
     assert tile_tasks == 20
     assert preprocessing_tasks == 2
+
+
+def test_execute_no_tasks(preprocess_cache_memory):
+    """If a process is initialized out of bounds, there should be no tasks."""
+    some_bounds_outside_process = (-180, 80, -170, 90)
+    with preprocess_cache_memory.mp(
+        batch_preprocess=False, bounds=some_bounds_outside_process, zoom=5
+    ) as mp:
+        process_tiles = [tile for tile in mp.get_process_tiles(zoom=5)]
+        tasks = mp.tasks()
+        assert tasks.tile_tasks_count == len(process_tiles) == 0
+        assert tasks.preprocessing_tasks_count == 0
+        assert len(tasks) == 0
diff --git a/test/test_tasks.py b/test/test_processing_tasks.py
similarity index 59%
rename from test/test_tasks.py
rename to test/test_processing_tasks.py
index eb2568ab..d9ea55e9 100644
--- a/test/test_tasks.py
+++ b/test/test_processing_tasks.py
@@ -2,15 +2,13 @@
 
 import pytest
 from shapely.geometry import shape
+from shapely.ops import unary_union
 
 from mapchete.errors import NoTaskGeometry
-from mapchete.processing.tasks import (
-    Task,
-    TaskBatch,
-    TileTask,
-    TileTaskBatch,
-    to_dask_collection,
-)
+from mapchete.executor import Executor
+from mapchete.processing.tasks import Task, TaskBatch, Tasks, TileTask, TileTaskBatch
+from mapchete.testing import ProcessFixture
+from mapchete.tile import BufferedTilePyramid
 
 
 def test_task_geo_interface():
@@ -121,9 +119,9 @@ def test_task_batches_to_dask_graph(dem_to_hillshade):
                 for process_tile in dem_to_hillshade.mp().get_process_tiles(zoom=zoom)
             )
         )
-        for zoom in reversed(dem_to_hillshade.mp().config.zoom_levels)
+        for zoom in dem_to_hillshade.mp().config.zoom_levels.descending()
     )
-    collection = to_dask_collection((preprocessing_batch, *zoom_batches))
+    collection = Tasks((preprocessing_batch, *zoom_batches)).to_dask_graph()
     import dask
 
     dask.compute(collection)
@@ -152,3 +150,78 @@ def test_task_batches_mixed_geometries():
     other_task = Task(bounds=(3, 4, 5, 6))
     assert len(batch.intersection(other_task)) == 10
     assert len(batch.intersection((3, 4, 5, 6))) == 10
+
+
+def test_task_batch_geometry():
+    batch = TaskBatch(
+        [Task(func=str, bounds=(0, 1, 2, 3)), Task(func=str, bounds=(2, 3, 4, 5))]
+    )
+    assert shape(batch).bounds == (0, 1, 4, 5)
+
+    # return empty geometry
+    assert shape(TaskBatch([])).is_empty
+
+
+def task_batches_generator(process: ProcessFixture, preprocessing_tasks_count=10):
+    if preprocessing_tasks_count:
+        input_key = list(process.mp().config.inputs.keys())[0]
+        yield TaskBatch(
+            (
+                Task(
+                    func=str,
+                    fargs=(i,),
+                    bounds=process.mp().config.bounds,
+                    id=f"{input_key}:{i}",
+                )
+                for i in range(preprocessing_tasks_count)
+            )
+        )
+    for zoom in process.mp().config.zoom_levels.descending():
+        yield TileTaskBatch(
+            (
+                TileTask(tile=process_tile, config=process.mp().config)
+                for process_tile in process.mp().get_process_tiles(zoom=zoom)
+            ),
+            id=f"zoom-{zoom}",
+        )
+
+
+def test_task_batches_as_dask_graph(dem_to_hillshade):
+    task_batches = Tasks(task_batches_generator(dem_to_hillshade))
+    assert len(task_batches.preprocessing_batches) == 1
+    assert len(task_batches.tile_batches) == len(
+        dem_to_hillshade.mp().config.zoom_levels
+    )
+    graph = task_batches.to_dask_graph()
+    assert graph
+
+    import dask
+
+    dask.compute(graph)
+
+
+def test_task_batches_as_layered_batches(dem_to_hillshade):
+    task_batches = Tasks(task_batches_generator(dem_to_hillshade))
+    assert len(task_batches.preprocessing_batches) == 1
+    assert len(task_batches.tile_batches) == len(
+        dem_to_hillshade.mp().config.zoom_levels
+    )
+    batches = [list(batch) for batch in task_batches.to_batches()]
+    assert batches
+
+    for batch in batches:
+        assert batch
+        for tile_task in batch:
+            assert isinstance(tile_task, Task)
+
+
+def test_task_batches_as_single_batch(dem_to_hillshade):
+    task_batches = Tasks(task_batches_generator(dem_to_hillshade))
+    assert len(task_batches.preprocessing_batches) == 1
+    assert len(task_batches.tile_batches) == len(
+        dem_to_hillshade.mp().config.zoom_levels
+    )
+    batch = list(task_batches.to_batch())
+    assert batch
+    for tile_task in batch:
+        assert isinstance(tile_task, Task)
diff --git a/test/test_processing_types.py b/test/test_processing_types.py
new file mode 100644
index 00000000..aca4755e
--- /dev/null
+++ b/test/test_processing_types.py
@@ -0,0 +1,31 @@
+import pickle
+
+import numpy.ma as ma
+
+from mapchete.processing.profilers.memory import MeasuredMemory
+from mapchete.processing.profilers.requests import MeasuredRequests
+from mapchete.processing.profilers.time import MeasuredTime
+from mapchete.processing.types import TaskInfo
+from mapchete.tile import BufferedTilePyramid
+
+task_info_kwargs = dict(
+    id="foo",
+    output=ma.ones((1, 1, 1)),
+    processed=True,
+    process_msg="foo",
+    written=True,
+    write_msg="bar",
+    profiling={
+        "memory": MeasuredMemory(),
+        "requests": MeasuredRequests(),
+        "time": MeasuredTime(),
+    },
+    tile=BufferedTilePyramid("geodetic").tile(5, 5, 5),
+)
+
+
+def test_task_info_pickle():
+    task_info = TaskInfo(**task_info_kwargs)
+    # pickle and unpickle
+    reloaded = pickle.loads(pickle.dumps(task_info))
+    assert reloaded == task_info
diff --git a/test/test_stac.py b/test/test_stac.py
index 1f081b7b..003bf14e 100644
--- a/test/test_stac.py
+++ b/test/test_stac.py
@@ -266,12 +266,12 @@ def test_zoom_levels_from_item_errors():
     version.parse(rasterio.__gdal_version__) < version.parse("3.3.0"),
     reason="required STACTA driver is only available in GDAL>=3.3.0",
 )
-def test_create_prototype_file(example_mapchete):
+def test_create_prototype_file(cleantopo_br):
     # create sparse tiledirectory with no tiles at row/col 0/0
-    execute(example_mapchete.dict, zoom=[10, 11])
+    execute(cleantopo_br.dict, zoom=[3], concurrency=None)
 
     # read STACTA with rasterio and expect an exception
-    stac_path = example_mapchete.mp().config.output.stac_path
+    stac_path = cleantopo_br.mp().config.output.stac_path
     assert stac_path.exists()
 
     with pytest.raises(RasterioIOError):
@@ -279,7 +279,7 @@ def test_create_prototype_file(example_mapchete):
             pass
 
     # create prototype file and assert reading is possible
-    create_prototype_files(example_mapchete.mp())
+    create_prototype_files(cleantopo_br.mp())
     with rasterio_open(stac_path):
         pass