Merge branch 'tickets/DM-36831'

doc/changes/DM-36831.feature.md

@@ -0,0 +1,3 @@
+Always disable implicit threading (e.g. in OpenBLAS) by default in `pipetask run`, even when not using `-j`.
+
+The new `--enable-implicit-threading` can be used to turn it back on.

python/lsst/ctrl/mpexec/cli/opt/optionGroups.py

@@ -137,6 +137,7 @@ def __init__(self) -> None:
             ctrlMpExecOpts.graph_fixup_option(),
             ctrlMpExecOpts.mock_option(),
             ctrlMpExecOpts.summary_option(),
+            ctrlMpExecOpts.enable_implicit_threading_option(),
         ]
 
 

python/lsst/ctrl/mpexec/cli/opt/options.py

@@ -355,6 +355,16 @@
 )
 
 
+enable_implicit_threading_option = MWOptionDecorator(
+    "--enable-implicit-threading",
+    help=unwrap(
+        """Do not disable implicit threading use by third-party libraries (e.g. OpenBLAS).
+        Implicit threading is always disabled during execution with multiprocessing."""
+    ),
+    is_flag=True,
+)
+
+
 task_option = MWOptionDecorator(
     "-t",
     "--task",

python/lsst/ctrl/mpexec/cli/script/run.py

@@ -55,6 +55,7 @@ def run(  # type: ignore
     summary,
     mock,
     mock_configs,
+    enable_implicit_threading,
     **kwargs,
 ):
     """Implements the command line interface `pipetask run` subcommand, should
@@ -148,10 +149,14 @@ def run(  # type: ignore
         If `True` then run mock pipeline instead of real one.
     mock_configs : `list` [ `PipelineAction` ]
         A list of config overrides for mock tasks.
+    enable_implicit_threading : `bool`, optional
+        If `True`, do not disable implicit threading by third-party libraries.
+        Implicit threading is always disabled during actual quantum execution
+        if ``processes > 1``.
     kwargs : `dict` [`str`, `str`]
         Ignored; click commands may accept options for more than one script
         function and pass all the option kwargs to each of the script functions
-        which ingore these unused kwargs.
+        which ignore these unused kwargs.
     """
     args = SimpleNamespace(
         pdb=pdb,
@@ -180,6 +185,7 @@ def run(  # type: ignore
         summary=summary,
         mock=mock,
         mock_configs=mock_configs,
+        enable_implicit_threading=enable_implicit_threading,
     )
 
     f = CmdLineFwk()

python/lsst/ctrl/mpexec/cmdLineFwk.py

@@ -51,6 +51,7 @@
     buildExecutionButler,
 )
 from lsst.utils import doImportType
+from lsst.utils.threads import disable_implicit_threading
 
 from . import util
 from .dotTools import graph2dot, pipeline2dot
@@ -676,6 +677,9 @@ def runPipeline(
         if args.extend_run:
             args.skip_existing = True
 
+        if not args.enable_implicit_threading:
+            disable_implicit_threading()
+
         # make butler instance
         if butler is None:
             butler = _ButlerFactory.makeWriteButler(args, graph.iterTaskGraph())

tests/test_cmdLineFwk.py

@@ -492,6 +492,9 @@ def testSimpleQGraph(self):
         fwk.runPipeline(qgraph, taskFactory, args)
         self.assertEqual(taskFactory.countExec, self.nQuanta)
 
+        # test that we've disabled implicit threading
+        self.assertEqual(os.environ["OMP_NUM_THREADS"], "1")
+
     def testEmptyQGraph(self):
         """Test that making an empty QG produces the right error messages."""
         # We make QG generation fail by populating one input collection in the