Avoid parsl executor contention (#496)

* avoid parsl executor contention

* better tests

* better docs

Co-Authored-By: Gregory Way <gregory.way@gmail.com>

---------

Co-authored-by: Gregory Way <gregory.way@gmail.com>

Author: d33bs

cytotable/utils.py

@@ -93,16 +93,18 @@ def _parsl_loaded() -> bool:
 def _default_parsl_config():
     """
     Return a default Parsl configuration for use with CytoTable.
+
+    Only a single executor (HighThroughputExecutor) is included by default.
+    Use of Parsl's ThreadPoolExecutor may result in unfreed memory within
+    certain systems because of Apache Arrow's memory allocators, so
+    _ensure_thread_executor adds it only for workflows -- such as Iceberg
+    image-crop export -- that specifically require it.
     """
     return Config(
         executors=[
             HighThroughputExecutor(
                 label="htex_default_for_cytotable",
             ),
-            ParslThreadPoolExecutor(
-                label=CYTOTABLE_THREAD_EXECUTOR_LABEL,
-                max_threads=4,
-            ),
         ]
     )
 

tests/test_iceberg.py

@@ -18,7 +18,7 @@
 import pytest
 from duckdb import IOException as DuckDBIOException
 from parsl.config import Config
-from parsl.executors import ThreadPoolExecutor
+from parsl.executors import HighThroughputExecutor, ThreadPoolExecutor
 from pyarrow import parquet
 
 from cytotable.exceptions import CytoTableException
@@ -82,13 +82,22 @@ def test_ensure_thread_executor_does_not_duplicate():
     assert len(matching) == 1
 
 
-def test_default_parsl_config_includes_thread_executor():
+def test_default_parsl_config_has_single_executor():
     """
-    _default_parsl_config always includes the cytotable thread executor.
+    _default_parsl_config only includes a single HighThroughputExecutor by
+    default; _ensure_thread_executor adds the cytotable thread executor
+    only for workflows that need it (e.g. Iceberg image-crop export).
     """
     cfg = _default_parsl_config()
-    labels = {e.label for e in cfg.executors}
-    assert CYTOTABLE_THREAD_EXECUTOR_LABEL in labels
+    assert len(cfg.executors) == 1
+    default_executor = cfg.executors[0]
+    assert isinstance(default_executor, HighThroughputExecutor)
+    assert default_executor.label == "htex_default_for_cytotable"
+
+    result = _ensure_thread_executor(cfg)
+    assert len(result.executors) == 2
+    assert result.executors[0] is default_executor
+    assert result.executors[1].label == CYTOTABLE_THREAD_EXECUTOR_LABEL
 
 
 def test_rewrite_join_sql_for_warehouse():