Resolve capture_output deadlock in multiprocessing environments

pyomo/common/dependencies.py

@@ -12,6 +12,7 @@
 import importlib.util
 import logging
 import sys
+import threading
 import warnings
 
 from collections.abc import Mapping
@@ -945,6 +946,11 @@ def __exit__(self, exc_type, exc_value, traceback):
 # Common optional dependencies used throughout Pyomo
 #
 
+#: lock for deconflicting access to capturing the process file
+#: descriptors.  This starts as a threading.Lock, unless the environment
+#: imports multiprocessing, in which case, it is upgraded to a
+#: multiprocessing lock.
+capture_output_lock = threading.Lock()
 yaml_load_args = {}
 
 
@@ -961,6 +967,18 @@ def _finalize_ctypes(module, available):
     import ctypes.util
 
 
+def _finalize_multiprocessing(module, available):
+    # Note: multiprocessing is very slow to import, but we need to make
+    # sure that the capture_output_lock Lock is created *before* the
+    # user spawns any subprocesses.  tee.capture_output will look here
+    # for the lock, which will start out as a "dummy" lock, and then
+    # will be updated to a multiprocessing.Lock when the first module
+    # triggers the multiprocessing import.
+
+    global capture_output_lock
+    capture_output_lock = module.Lock()
+
+
 def _finalize_scipy(module, available):
     if available:
         # Import key subpackages that we will want to assume are present
@@ -1081,12 +1099,25 @@ def _pyutilib_importer():
 
 
 with declare_modules_as_importable(globals()):
+    # Standard libraries that we will unconditionally import.  We are
+    # importing it here so that import timing is better reported from
+    # pyomo.environ.tests.test_environ (hence the imports are not
+    # necessarily alphebetical)
+    #
+    # Pickle is used by Pyomo and by multiprocessing
+    try:
+        import cPickle as pickle
+    except ImportError:
+        import pickle
+
     # Standard libraries that are slower to import and not strictly required
     # on all platforms / situations.
     ctypes, _ = attempt_import(
         'ctypes', deferred_submodules=['util'], callback=_finalize_ctypes
     )
-    multiprocessing, _ = attempt_import('multiprocessing')
+    multiprocessing, _ = attempt_import(
+        'multiprocessing', callback=_finalize_multiprocessing
+    )
     random, _ = attempt_import('random')
 
     # Necessary for minimum version checking for other optional dependencies
@@ -1127,8 +1158,3 @@ def _pyutilib_importer():
         deferred_submodules=['pyplot', 'pylab', 'backends'],
         catch_exceptions=(ImportError, RuntimeError),
     )
-
-try:
-    import cPickle as pickle
-except ImportError:
-    import pickle

pyomo/common/env.py

@@ -9,7 +9,7 @@
 
 import os
 
-from .dependencies import ctypes, multiprocessing
+from pyomo.common.dependencies import ctypes, multiprocessing
 
 
 def _as_bytes(val):

pyomo/common/tee.py

@@ -22,8 +22,10 @@
 import threading
 import time
 
+import pyomo.common.dependencies as dependencies
 from pyomo.common.errors import DeveloperError
 from pyomo.common.log import LoggingIntercept, LogStream
+from pyomo.common.shutdown import python_is_shutting_down
 
 _poll_interval = 0.0001
 _poll_rampup_limit = 0.099
@@ -35,6 +37,7 @@
 # ~(13.1 * #threads) seconds
 _poll_timeout = 1  # 14 rounds: 0.0001 * 2**14 == 1.6384
 _poll_timeout_deadlock = 100  # seconds
+_threading_deadlock = 200  # seconds; should be longer than _poll_timeout_deadlock
 _pipe_buffersize = 1 << 16  # 65536
 _noop = lambda: None
 _mswindows = sys.platform.startswith('win')
@@ -297,8 +300,6 @@ class capture_output:
 
     """
 
-    startup_shutdown = threading.Lock()
-
     def __init__(self, output=None, capture_fd=False):
         self.output = output
         self.output_stream = None
@@ -339,13 +340,12 @@ def _exit_context_stack(self, et, ev, tb):
                 cm.__exit__(et, ev, tb)
             except:
                 _stack = self.context_stack
-                FAIL.append(
-                    f"{sys.exc_info()[0].__name__}: {sys.exc_info()[1]} ({len(_stack)+1}: {cm}@{id(cm):x})"
-                )
+                _et, _e, _tb = sys.exc_info()
+                FAIL.append(f"{_et.__name__}: {_e} ({len(_stack)+1}: {cm}@{id(cm):x})")
         return FAIL
 
     def __enter__(self):
-        if not capture_output.startup_shutdown.acquire(timeout=_poll_timeout_deadlock):
+        if not dependencies.capture_output_lock.acquire(timeout=_threading_deadlock):
             # This situation *shouldn't* happen.  If it does, it is
             # unlikely that the user can fix it (or even debug it).
             # Instead they should report it back to us.
@@ -358,20 +358,22 @@ def __enter__(self):
             #     was trying to start up / run (so the other solver held
             #     the lock, but the GC interrupted that thread and
             #     wouldn't let go).
-            raise DeveloperError("Deadlock starting capture_output")
+            if not python_is_shutting_down():
+                raise DeveloperError("Deadlock starting capture_output")
         try:
             return self._enter_impl()
         finally:
-            capture_output.startup_shutdown.release()
+            dependencies.capture_output_lock.release()
 
     def __exit__(self, et, ev, tb):
-        if not capture_output.startup_shutdown.acquire(timeout=_poll_timeout_deadlock):
+        if not dependencies.capture_output_lock.acquire(timeout=_threading_deadlock):
             # See comments & breadcrumbs in __enter__() above.
-            raise DeveloperError("Deadlock closing capture_output")
+            if not python_is_shutting_down():
+                raise DeveloperError("Deadlock closing capture_output")
         try:
             return self._exit_impl(et, ev, tb)
         finally:
-            capture_output.startup_shutdown.release()
+            dependencies.capture_output_lock.release()
 
     def _enter_impl(self):
         self.old = (sys.stdout, sys.stderr)

pyomo/common/tests/test_multithread.py

@@ -8,11 +8,16 @@
 # ____________________________________________________________________________________
 
 import threading
-import pyomo.common.unittest as unittest
-from pyomo.common.multithread import *
 from threading import Thread
+from multiprocessing.dummy import Pool as ThreadPool
+
+import pyomo.common.unittest as unittest
+
+from pyomo.common.multithread import MultiThreadWrapper, MultiThreadWrapperWithMain
 from pyomo.opt.base.solvers import check_available_solvers
 
+import pyomo.environ as pyo
+
 
 class Dummy:
     """asdfg"""
@@ -103,10 +108,6 @@ def thread_func():
     )
     def test_solve(self):
         # Based on the minimal example in https://github.com/Pyomo/pyomo/issues/2475
-        import pyomo.environ as pyo
-        from pyomo.opt import SolverFactory
-        from multiprocessing.dummy import Pool as ThreadPool
-
         model = pyo.ConcreteModel()
         model.nVars = pyo.Param(initialize=4)
         model.N = pyo.RangeSet(model.nVars)
@@ -115,7 +116,7 @@ def test_solve(self):
         model.cuts = pyo.ConstraintList()
 
         def test(model):
-            opt = SolverFactory('glpk')
+            opt = pyo.SolverFactory('glpk')
             opt.solve(model)
 
             # Iterate, adding a cut to exclude the previously found solution

pyomo/common/tests/test_tee.py

@@ -21,6 +21,7 @@
 from pyomo.common.errors import DeveloperError
 from pyomo.common.log import LoggingIntercept, LogStream
 from pyomo.common.tempfiles import TempfileManager
+import pyomo.common.dependencies as deps
 import pyomo.common.tee as tee
 import pyomo.common.unittest as unittest
 
@@ -581,28 +582,40 @@ def test_capture_output_stack_error(self):
             logging.getLogger('pyomo.common.tee').handlers.clear()
 
     def test_atomic_deadlock(self):
-        save_poll = tee._poll_timeout_deadlock
-        tee._poll_timeout_deadlock = 0.01
+        save_poll = tee._threading_deadlock
+        tee._threading_deadlock = 0.01
+
+        # Ensure multiprocessing is loaded:
+        deps.multiprocessing.Lock
 
         co = tee.capture_output()
         try:
-            tee.capture_output.startup_shutdown.acquire()
+            deps.capture_output_lock.acquire()
             with self.assertRaisesRegex(
                 DeveloperError, "Deadlock starting capture_output"
             ):
                 with tee.capture_output():
                     pass
-            tee.capture_output.startup_shutdown.release()
+            deps.capture_output_lock.release()
 
             with self.assertRaisesRegex(
                 DeveloperError, "Deadlock closing capture_output"
             ):
                 with co:
-                    tee.capture_output.startup_shutdown.acquire()
+                    deps.capture_output_lock.acquire()
         finally:
-            tee._poll_timeout_deadlock = save_poll
-            if tee.capture_output.startup_shutdown.locked():
-                tee.capture_output.startup_shutdown.release()
+            tee._threading_deadlock = save_poll
+            # We would like to just test if out Lock was acquired and
+            # then release it if necessary.  Unfortunately,
+            # multiprocessing.Lock doesn't support locked(), so we will
+            # just catch and eat the error for releasing an unlocked
+            # lock.
+            #
+            ## if deps.capture_output_lock.locked():
+            try:
+                deps.capture_output_lock.release()
+            except ValueError:
+                pass
             co.reset()
 
     def test_capture_output_invalid_ostream(self):

pyomo/common/tests/test_unittest.py

@@ -8,12 +8,13 @@
 # ____________________________________________________________________________________
 
 import datetime
-import multiprocessing
 import os
 import pickle
 import time
 
 import pyomo.common.unittest as unittest
+import pyomo.common.dependencies as deps
+from pyomo.common.dependencies import multiprocessing
 from pyomo.common.log import LoggingIntercept
 from pyomo.common.tee import capture_output
 from pyomo.common.tempfiles import TempfileManager
@@ -171,8 +172,28 @@ def test_assertStructuredAlmostEqual_numericvalue(self):
 
     def test_timeout_fcn_call(self):
         self.assertEqual(short_sleep(), 42)
-        with self.assertRaisesRegex(TimeoutError, 'test timed out after 0.01 seconds'):
-            long_sleep()
+        with LoggingIntercept() as LOG:
+            with self.assertRaisesRegex(
+                TimeoutError, 'test timed out after 0.01 seconds'
+            ):
+                long_sleep()
+            self.assertEqual(LOG.getvalue(), "")
+            deps.capture_output_lock.acquire()
+            save = unittest._timeout_terminate_timeout
+            unittest._timeout_terminate_timeout = 0.01
+            try:
+                with self.assertRaisesRegex(
+                    TimeoutError, 'test timed out after 0.01 seconds'
+                ):
+                    long_sleep()
+            finally:
+                unittest._timeout_terminate_timeout = save
+                deps.capture_output_lock.release()
+            self.assertEqual(
+                LOG.getvalue(),
+                "Failed to acquire capture_output_lock Lock before "
+                "terminating subprocess on timeout: process deadlock is likely.\n",
+            )
         with self.assertRaisesRegex(
             NameError, r"name 'foo' is not defined\s+Original traceback:"
         ):

pyomo/common/unittest.py

@@ -29,6 +29,7 @@
 # specifically later
 from unittest import *
 import unittest as _unittest
+import pyomo.common.dependencies as deps
 
 from pyomo.common.collections import Mapping, Sequence
 from pyomo.common.dependencies import attempt_import, check_min_version, multiprocessing
@@ -44,6 +45,10 @@
 # (and then enforce a strict dependence on pytest)
 pytest, pytest_available = attempt_import('pytest')
 
+#: A time limit for acquiring the capture_output_lock lock
+#: before terminating a subprocess
+_timeout_terminate_timeout = 2  # seconds
+
 
 def _defaultFormatter(msg, default):
     return msg or default
@@ -487,7 +492,27 @@ def test_timer(*args, **kwargs):
                 if pipe_recv.poll(seconds):
                     resultType, result, stdout = pipe_recv.recv()
                 else:
-                    test_proc.terminate()
+                    # Note: because we are using capture_output within
+                    # the _runner handler, we can trigger a deadlock
+                    # when we call terminate() while the _runner's
+                    # capture_output holds the capture_output_lock lock
+                    # (terminate() bypasses all __exit__ handlers!).  To
+                    # avoid that, we will grab the lock here before
+                    # terminating the subprocess.
+                    locked = deps.capture_output_lock.acquire(
+                        timeout=_timeout_terminate_timeout
+                    )
+                    if not locked:
+                        logging.getLogger(__name__).error(
+                            "Failed to acquire capture_output_lock "
+                            "Lock before terminating subprocess on timeout: "
+                            "process deadlock is likely."
+                        )
+                    try:
+                        test_proc.terminate()
+                    finally:
+                        if locked:
+                            deps.capture_output_lock.release()
                     raise timeout_raises(
                         "test timed out after %s seconds" % (seconds,)
                     ) from None

pyomo/contrib/solver/solvers/gams.py

@@ -16,7 +16,6 @@
 import sys
 import struct
 import re
-import pathlib
 
 from pyomo.common.dependencies import attempt_import
 from pyomo.common.errors import InfeasibleConstraintException