Repository: geofileops/geofileops
Pull request #631 (2025-02-13 16:07:53)

BUG: fix error with multiprocessing on windows
Issue introduced in #627 (unreleased change)

For performance reasons, by default multithreading is used on windows for small files... but because the tests use small files... multiprocessing is not covered in the tests.

In this PR:
- Make multithreading versus multiprocessing configurable so multiprocessing is properly covered in the CI tests
- Fix the issue


Patch:
diff --git a/geofileops/_compat.py b/geofileops/_compat.py
index 6e81db7..9ee5061 100644
--- a/geofileops/_compat.py
+++ b/geofileops/_compat.py
@@ -47,15 +47,32 @@ def _pyogrio_spatialite_version_info() -> dict[str, str]:
     return versions
 
 
+# Determine the versions of the runtime dependencies
+# gdal.__version__ includes a "dev-..." suffix for master/development versions. This
+# must be dropped for the version checks here.
+GDAL_BASE_VERSION = version.parse(gdal.__version__.split("-")[0]).base_version
+GDAL_GTE_38 = version.parse(GDAL_BASE_VERSION) >= version.parse("3.8")
+GDAL_ST_311 = version.parse(GDAL_BASE_VERSION) < version.parse("3.11")
+
+GEOPANDAS_GTE_10 = version.parse(gpd.__version__) >= version.parse("1.0")
+PANDAS_GTE_22 = version.parse(pd.__version__) >= version.parse("2.2")
+PYOGRIO_GTE_07 = version.parse(pyogrio.__version__) >= version.parse("0.7")
+SHAPELY_GTE_20 = version.parse(shapely.__version__) >= version.parse("2")
+
+sqlite3_spatialite_version_info = _sqlite_util.spatialite_version_info()
+sqlite3_spatialite_version = sqlite3_spatialite_version_info["spatialite_version"]
+SPATIALITE_GTE_51 = version.parse(sqlite3_spatialite_version) >= version.parse("5.1")
+
+
 # If running in the main process, check the spatialite versions of the dependencies
 if multiprocessing.parent_process() is None:
     pyogrio_spatialite_version_info = _pyogrio_spatialite_version_info()
-    sqlite3_spatialite_version_info = _sqlite_util.spatialite_version_info()
+
     gdal_spatialite_version_info = _ogr_util.spatialite_version_info()
 
     # Check that the spatialite versions are the same
     pyogrio_spatialite_version = pyogrio_spatialite_version_info["spatialite_version"]
-    sqlite3_spatialite_version = sqlite3_spatialite_version_info["spatialite_version"]
+
     gdal_spatialite_version = gdal_spatialite_version_info["spatialite_version"]
     if (
         pyogrio_spatialite_version != sqlite3_spatialite_version
@@ -66,18 +83,3 @@ if multiprocessing.parent_process() is None:
             f"{sqlite3_spatialite_version=} vs {gdal_spatialite_version=}",
             stacklevel=1,
         )
-
-# Determine the versions of the runtime dependencies
-# gdal.__version__ includes a "dev-..." suffix for master/development versions. This
-# must be dropped for the version checks here.
-GDAL_BASE_VERSION = version.parse(gdal.__version__.split("-")[0]).base_version
-
-GDAL_GTE_38 = version.parse(GDAL_BASE_VERSION) >= version.parse("3.8")
-GEOPANDAS_GTE_10 = version.parse(gpd.__version__) >= version.parse("1.0")
-PANDAS_GTE_22 = version.parse(pd.__version__) >= version.parse("2.2")
-PYOGRIO_GTE_07 = version.parse(pyogrio.__version__) >= version.parse("0.7")
-SHAPELY_GTE_20 = version.parse(shapely.__version__) >= version.parse("2")
-SPATIALITE_GTE_51 = version.parse(sqlite3_spatialite_version) >= version.parse("5.1")
-
-GDAL_GTE_38 = version.parse(GDAL_BASE_VERSION) >= version.parse("3.8")
-GDAL_ST_311 = version.parse(GDAL_BASE_VERSION) < version.parse("3.11")
diff --git a/geofileops/helpers/_configoptions_helper.py b/geofileops/helpers/_configoptions_helper.py
index 08c8af4..22a042a 100644
--- a/geofileops/helpers/_configoptions_helper.py
+++ b/geofileops/helpers/_configoptions_helper.py
@@ -55,6 +55,20 @@ class ConfigOptions:
         """
         return get_bool("GFO_REMOVE_TEMP_FILES", default=True)
 
+    @classproperty
+    def worker_type(cls) -> str:
+        """The type of workers to use for parallel processing.
+
+        Supported values (case insensitive):
+            - "thread": use threads when processing in parallel.
+            - "process": use processes when processing in parallel.
+            - "auto": determine the type automatically.
+
+        Returns:
+            str: the type of workers to use. Defaults to "auto".
+        """
+        return os.environ.get("GFO_WORKER_TYPE", default="auto").strip().lower()
+
 
 def get_bool(key: str, default: bool) -> bool:
     """Get the value for the environment variable ``key`` as a bool.
diff --git a/geofileops/helpers/_general_helper.py b/geofileops/helpers/_general_helper.py
new file mode 100644
index 0000000..101ad90
--- /dev/null
+++ b/geofileops/helpers/_general_helper.py
@@ -0,0 +1,20 @@
+import os
+
+from geofileops.helpers._configoptions_helper import ConfigOptions
+
+
+def use_threads(input_layer_featurecount: int) -> bool:
+    worker_type = ConfigOptions.worker_type
+    if worker_type == "thread":
+        return True
+    elif worker_type == "process":
+        return False
+
+    if os.name != "nt":
+        return False
+
+    # Processing in threads is 2x faster for small datasets on Windows
+    if input_layer_featurecount <= 100:
+        return True
+
+    return False
diff --git a/geofileops/util/_geoops_gpd.py b/geofileops/util/_geoops_gpd.py
index 055543f..fd3a201 100644
--- a/geofileops/util/_geoops_gpd.py
+++ b/geofileops/util/_geoops_gpd.py
@@ -31,7 +31,7 @@ from pygeoops import GeometryType, PrimitiveType
 import geofileops as gfo
 from geofileops import LayerInfo, fileops
 from geofileops._compat import GEOPANDAS_GTE_10, PANDAS_GTE_22
-from geofileops.helpers import _parameter_helper
+from geofileops.helpers import _general_helper, _parameter_helper
 from geofileops.helpers._configoptions_helper import ConfigOptions
 from geofileops.util import (
     _general_util,
@@ -782,7 +782,7 @@ def _apply_geooperation_to_layer(
     try:
         # Calculate the best number of parallel processes and batches for
         # the available resources
-        processing_params = _prepare_processing_params(
+        process_params = _prepare_processing_params(
             input_path=input_path,
             input_layer=input_layer,
             nb_parallel=nb_parallel,
@@ -790,19 +790,16 @@ def _apply_geooperation_to_layer(
             parallelization_config=parallelization_config,
             tmp_dir=tmp_dir,
         )
-        assert processing_params.batches is not None
+        assert process_params.batches is not None
 
+        # Start processing
         logger.info(
-            f"Start processing ({processing_params.nb_parallel} "
-            f"parallel workers, batch size: {processing_params.batchsize})"
-        )
-        # Processing in threads is 2x faster for small datasets (on Windows)
-        calculate_in_threads = (
-            True if processing_params.nb_rows_to_process <= 100 else False
+            f"Start processing ({process_params.nb_parallel} "
+            f"parallel workers, batch size: {process_params.batchsize})"
         )
         with _processing_util.PooledExecutorFactory(
-            threadpool=calculate_in_threads,
-            max_workers=processing_params.nb_parallel,
+            threadpool=_general_helper.use_threads(process_params.nb_rows_to_process),
+            max_workers=process_params.nb_parallel,
             initializer=_processing_util.initialize_worker(),
         ) as calculate_pool:
             # Prepare output filename
@@ -811,7 +808,7 @@ def _apply_geooperation_to_layer(
             batches: dict[int, dict] = {}
             future_to_batch_id = {}
 
-            for batch_id, batch_filter in enumerate(processing_params.batches):
+            for batch_id, batch_filter in enumerate(process_params.batches):
                 batches[batch_id] = {}
                 batches[batch_id]["layer"] = output_layer
 
@@ -851,13 +848,13 @@ def _apply_geooperation_to_layer(
             # can write to the same output file at the time...
             start_time = datetime.now()
             nb_done = 0
-            nb_batches = len(processing_params.batches)
+            nb_batches = len(process_params.batches)
             _general_util.report_progress(
                 start_time,
                 nb_done,
                 nb_todo=nb_batches,
                 operation=operation.value,
-                nb_parallel=processing_params.nb_parallel,
+                nb_parallel=process_params.nb_parallel,
             )
             for future in futures.as_completed(future_to_batch_id):
                 try:
@@ -908,7 +905,7 @@ def _apply_geooperation_to_layer(
                     nb_done,
                     nb_todo=nb_batches,
                     operation=operation.value,
-                    nb_parallel=processing_params.nb_parallel,
+                    nb_parallel=process_params.nb_parallel,
                 )
 
         # Round up and clean up
@@ -1650,10 +1647,8 @@ def _dissolve_polygons_pass(
     gfo.create_spatial_index(input_path, layer=input_layer, exist_ok=True)
 
     # Start calculation in parallel
-    # Processing in threads is 2x faster for small datasets (on Windows)
-    calculate_in_threads = True if input_layer.featurecount <= 100 else False
     with _processing_util.PooledExecutorFactory(
-        threadpool=calculate_in_threads,
+        threadpool=_general_helper.use_threads(input_layer.featurecount),
         max_workers=nb_parallel,
         initializer=_processing_util.initialize_worker(),
     ) as calculate_pool:
diff --git a/geofileops/util/_geoops_sql.py b/geofileops/util/_geoops_sql.py
index 8e5cece..9171590 100644
--- a/geofileops/util/_geoops_sql.py
+++ b/geofileops/util/_geoops_sql.py
@@ -24,7 +24,7 @@ import geofileops as gfo
 from geofileops import GeometryType, LayerInfo, PrimitiveType, fileops
 from geofileops._compat import SPATIALITE_GTE_51
 from geofileops.fileops import _append_to_nolock
-from geofileops.helpers import _parameter_helper
+from geofileops.helpers import _general_helper, _parameter_helper
 from geofileops.helpers._configoptions_helper import ConfigOptions
 from geofileops.util import (
     _general_util,
@@ -718,9 +718,8 @@ def _single_layer_vector_operation(
         tmp_output_path = tempdir / output_path.name
 
         # Processing in threads is 2x faster for small datasets (on Windows)
-        calculate_in_threads = True if input_layer.featurecount <= 100 else False
         with _processing_util.PooledExecutorFactory(
-            threadpool=calculate_in_threads,
+            threadpool=_general_helper.use_threads(input_layer.featurecount),
             max_workers=processing_params.nb_parallel,
             initializer=_processing_util.initialize_worker(),
         ) as calculate_pool:
@@ -3345,14 +3344,12 @@ def _two_layer_vector_operation(
 
         # Calculate
         # ---------
-        # Processing in threads is 2x faster for small datasets (on Windows)
-        calculate_in_threads = True if input1_layer.featurecount <= 100 else False
         logger.info(
             f"Start processing ({processing_params.nb_parallel} "
             f"parallel workers, batch size: {processing_params.batchsize})"
         )
         with _processing_util.PooledExecutorFactory(
-            threadpool=calculate_in_threads,
+            threadpool=_general_helper.use_threads(input1_layer.featurecount),
             max_workers=processing_params.nb_parallel,
             initializer=_processing_util.initialize_worker(),
         ) as calculate_pool:


Test patch:
diff --git a/tests/test_general_helper.py b/tests/test_general_helper.py
new file mode 100644
index 0000000..2816b64
--- /dev/null
+++ b/tests/test_general_helper.py
@@ -0,0 +1,27 @@
+import os
+
+import pytest
+
+from geofileops.helpers import _general_helper
+from geofileops.util import _general_util
+
+
+@pytest.mark.parametrize(
+    "worker_type, input_layer_featurecount",
+    [("process", 1), ("thread", 101), ("auto", 1), ("auto", 100), ("auto", 101)],
+)
+def test_use_threads(worker_type, input_layer_featurecount):
+    if worker_type == "process":
+        expected = False
+    elif worker_type == "thread":
+        expected = True
+    elif os.name == "nt":
+        if input_layer_featurecount <= 100:
+            expected = True
+        else:
+            expected = False
+    else:
+        expected = False
+
+    with _general_util.TempEnv({"GFO_WORKER_TYPE": worker_type}):
+        assert _general_helper.use_threads(input_layer_featurecount) == expected
diff --git a/tests/test_geofileops_singlelayer.py b/tests/test_geofileops_singlelayer.py
index 03ed628..0831a53 100644
--- a/tests/test_geofileops_singlelayer.py
+++ b/tests/test_geofileops_singlelayer.py
@@ -14,7 +14,7 @@ from shapely import MultiPolygon, Polygon
 
 from geofileops import GeometryType, fileops, geoops
 from geofileops._compat import SPATIALITE_GTE_51
-from geofileops.util import _geofileinfo, _geoops_sql
+from geofileops.util import _general_util, _geofileinfo, _geoops_sql
 from geofileops.util import _io_util as io_util
 from geofileops.util._geofileinfo import GeofileInfo
 from tests import test_helper
@@ -300,16 +300,18 @@ def test_buffer_force(tmp_path, geoops_module):
 
     # Run buffer
     output_path = tmp_path / f"{input_path.stem}-output{input_path.suffix}"
-    assert output_path.exists() is False
-
-    geoops.buffer(
-        input_path=input_path,
-        output_path=output_path,
-        distance=distance,
-        keep_empty_geoms=False,
-        nb_parallel=2,
-        batchsize=batchsize,
-    )
+    assert not output_path.exists()
+
+    # Use "process" worker type to test this as well
+    with _general_util.TempEnv({"GFO_WORKER_TYPE": "process"}):
+        geoops.buffer(
+            input_path=input_path,
+            output_path=output_path,
+            distance=distance,
+            keep_empty_geoms=False,
+            nb_parallel=2,
+            batchsize=batchsize,
+        )
 
     # Test buffer to existing output path
     assert output_path.exists()
diff --git a/tests/test_geofileops_twolayers.py b/tests/test_geofileops_twolayers.py
index dec909d..a1c3791 100644
--- a/tests/test_geofileops_twolayers.py
+++ b/tests/test_geofileops_twolayers.py
@@ -907,9 +907,12 @@ def test_intersection_input_no_index(tmp_path):
 
     # Now run test
     output_path = tmp_path / f"{input1_path.stem}_intersection_{input2_path.stem}.gpkg"
-    gfo.intersection(
-        input1_path=input1_path, input2_path=input2_path, output_path=output_path
-    )
+
+    # Use "process" worker type to test this as well
+    with gfo.TempEnv({"GFO_WORKER_TYPE": "process"}):
+        gfo.intersection(
+            input1_path=input1_path, input2_path=input2_path, output_path=output_path
+        )
 
     # Check if the tmp file is correctly created
     assert output_path.exists()
@@ -1460,17 +1463,20 @@ def test_join_nearest(tmp_path, suffix, epsg):
     output_path = tmp_path / f"{input1_path.stem}-output{suffix}"
     nb_nearest = 2
     input1_columns = ["OIDN", "UIDN", "HFDTLT", "fid"]
-    gfo.join_nearest(
-        input1_path=str(input1_path),
-        input1_columns=input1_columns,
-        input2_path=str(input2_path),
-        output_path=str(output_path),
-        nb_nearest=nb_nearest,
-        distance=1000,
-        expand=True,
-        batchsize=batchsize,
-        force=True,
-    )
+
+    # Use "process" worker type to test this as well
+    with gfo.TempEnv({"GFO_WORKER_TYPE": "process"}):
+        gfo.join_nearest(
+            input1_path=str(input1_path),
+            input1_columns=input1_columns,
+            input2_path=str(input2_path),
+            output_path=str(output_path),
+            nb_nearest=nb_nearest,
+            distance=1000,
+            expand=True,
+            batchsize=batchsize,
+            force=True,
+        )
 
     # Check if the output file is correctly created
     assert output_path.exists()
diff --git a/tests/test_geoops_dissolve.py b/tests/test_geoops_dissolve.py
index a7344f0..b22f3a1 100644
--- a/tests/test_geoops_dissolve.py
+++ b/tests/test_geoops_dissolve.py
@@ -755,15 +755,18 @@ def test_dissolve_polygons_aggcolumns_columns(tmp_path, suffix):
         ]
     }
     groupby_columns = ["GEWASgroep"]
-    gfo.dissolve(
-        input_path=input_path,
-        output_path=output_path,
-        groupby_columns=groupby_columns,
-        agg_columns=agg_columns,
-        explodecollections=False,
-        nb_parallel=2,
-        batchsize=batchsize,
-    )
+
+    # Force use of processes as workers
+    with gfo.TempEnv({"GFO_WORKER_TYPE": "process"}):
+        gfo.dissolve(
+            input_path=input_path,
+            output_path=output_path,
+            groupby_columns=groupby_columns,
+            agg_columns=agg_columns,
+            explodecollections=False,
+            nb_parallel=2,
+            batchsize=batchsize,
+        )
 
     # Now check if the tmp file is correctly created
     assert output_path.exists()