Monkey patching interferes with threading in Python 3.7

[benfrankel]

Environment

$ uname -a
Linux hostname 4.20.4-arch1-1-ARCH #1 SMP PREEMPT Wed Jan 23 00:12:22 UTC 2019 x86_64 GNU/Linux

$ python -V
Python 3.7.2

$ pip freeze
appdirs==1.4.3
asn1crypto==0.24.0
attrs==18.2.0
backcall==0.1.0
bleach==3.1.0
btrfsutil==1.1.0
CacheControl==0.12.5
certifi==2018.11.29
cffi==1.11.5
chardet==3.0.4
colorama==0.4.1
cryptography==2.4.2
decorator==4.3.2
defusedxml==0.5.0
distlib==0.2.8
distro==1.3.0
dnspython==1.16.0
entrypoints==0.3
eventlet==0.25.1
greenlet==0.4.15
html5lib==1.0.1
idna==2.8
ipykernel==4.9.0
ipython==7.2.0
ipython-genutils==0.1.0
ipywidgets==7.4.2
isc==2.0
jedi==0.13.2
Jinja2==2.10
jsonschema==2.6.0
jupyter-client==5.2.4
jupyter-console==6.0.0
jupyter-core==4.4.0
lensfun==0.3.2
lit==0.7.1.dev0
lockfile==0.12.2
louis==3.8.0
MarkupSafe==1.1.0
mistune==0.8.4
monotonic==1.5
msgpack==0.6.0
nbconvert==5.4.0
nbformat==4.4.0
notebook==5.7.4
numpy==1.16.0
onboard==1.4.1
packaging==19.0
pandocfilters==1.4.2
parso==0.3.2
pep517==0.5.0
pew==1.1.5
pexpect==4.6.0
pickleshare==0.7.5
pipenv==2018.11.15.dev0
ply==3.11
progress==1.4
prometheus-client==0.5.0
prompt-toolkit==2.0.7
psutil==5.5.0
ptyprocess==0.6.0
pycairo==1.18.0
pycparser==2.19
pyenchant==2.0.0
Pygments==2.3.1
PyGObject==3.30.4
pyOpenSSL==19.0.0
pyparsing==2.3.1
pyPEG2==2.15.2
PyQt5==5.11.3
PyQt5-sip==4.19.13
python-dateutil==2.7.5
pytoml==0.1.20
PyYAML==3.13
pyzmq==17.1.2
requests==2.21.0
retrying==1.3.3
Send2Trash==1.5.0
simplegeneric==0.8.1
sip==4.19.13
six==1.12.0
SQLAlchemy==1.2.17
terminado==0.8.1
testpath==0.4.2
tornado==5.1.1
traitlets==4.3.2
Unidecode==1.0.23
urllib3==1.24.1
virtualenv==16.0.0
virtualenv-clone==0.5.1
wcwidth==0.1.7
webencodings==0.5.1
widgetsnbextension==3.4.2

Steps to reproduce the error

>>> import eventlet
>>> eventlet.monkey_patch()
>>> import subprocess
>>> subprocess.Popen(['true'], preexec_fn=lambda: None)
Exception ignored in: <function _after_fork at 0x7f5f1b8c09d8>
Traceback (most recent call last):
  File "/usr/lib/python3.7/threading.py", line 1335, in _after_fork
    assert len(_active) == 1
AssertionError: 
<eventlet.green.subprocess.Popen object at 0x7f5f1a49f630>

Cause of the error

Eventlet replaces threading.current_thread with eventlet.green.threading.current_thread, which falls back to the original current_thread in some cases. The original current_thread is operating with the original _active global variable. The original _active global variable is not in use, as the entire threading module has been re-imported, so it contains outdated values. As a result, threading._after_fork is unable to find a thread in threading._active with the same id as threading.current_thread().

Workaround

This can be fixed by monkey patching the original current_thread to use the up-to-date _active global variable, as in:

>>> import eventlet
>>> eventlet.monkey_patch()
>>> import __original_module_threading
>>> import threading
>>> __original_module_threading.current_thread.__globals__['_active'] = threading._active
>>> import subprocess
>>> subprocess.Popen(['true'], preexec_fn=lambda: None)
<eventlet.green.subprocess.Popen object at 0x7ff9509600f0>

Special case

That workaround doesn't solve every case. Consider what happens when a greenthread forks:

>>> import eventlet
>>> eventlet.monkey_patch()
>>> import __original_module_threading
>>> import threading
>>> __original_module_threading.current_thread.__globals__['_active'] = threading._active
>>> import subprocess
>>> eventlet.spawn(lambda: subprocess.Popen(['true'], preexec_fn=lambda: None))
<eventlet.greenthread.GreenThread object at 0x7f1647f18630>
>>> eventlet.sleep()
Exception ignored in: <function _after_fork at 0x7f16493489d8>
Traceback (most recent call last):
  File "/usr/lib/python3.7/threading.py", line 1335, in _after_fork
    assert len(_active) == 1
AssertionError: 

Cause of the error for special case

Eventlet's implementation of current_thread uses its own thread-local active variable to track greenthreads. threading._after_fork doesn't know about that, so it gets its list of threads from threading._active via threading._enumerate, and its "current thread" from active via eventlet.green.threading.current_thread. This mismatch again causes the assertion error.

Workaround for the special case

I'm not sure why eventlet is co-opting the threading library for greenthreads, since it seems that breaks any code that assumes it's working with real threads -- but anways, this particular breakage can be fixed by monkey patching threading._after_fork to simply not use eventlet.green.threading.current_thread. The purpose of threading._after_fork is to clean up left-over thread state in the child process after a fork, so it really doesn't care about greenthreads.

This workaround also requires removing _after_fork from __patched__. I don't know why this is necessary, but I also don't know why it's there in the first place (the commit that added it offers no explanation).

>>> import eventlet
>>> import eventlet.green.threading
>>> try: eventlet.green.threading.__patched__.remove('_after_fork')
... except ValueError: pass
... 
>>> import __original_module_threading
>>> import threading
>>> __original_module_threading.current_thread.__globals__['_active'] = threading._active
>>> threading._after_fork.__globals__['current_thread'] = __original_module_threading.current_thread
>>> import subprocess
>>> eventlet.spawn(lambda: subprocess.Popen(['true'], preexec_fn=lambda: None))
<eventlet.greenthread.GreenThread object at 0x7f267be12390>
>>> eventlet.sleep()

Yet another issue...

With the second workaround, there are no assertion errors. However, there is still some undesirable behavior due to eventlet's monkey patching. Since Python 3.7's threading library introduced a new import side-effect (... unfortunately), and eventlet re-imports threading twice in the process of monkey patching it, the side-effect is run 3 times. The side-effect in this case is os.register_at_fork(after_in_child=_after_fork), which registers _after_fork as an after fork hook. That means it's registered 3 times. So, 3 different versions of _after_fork are run after every fork.

Unfortunately, the Python documentation states that it's impossible to unregister a function, meaning the threading library now has an unrevertable import side-effect (... which is fun). Thus, I don't know of any simple workaround. I also currently don't know of any symptoms caused by the triple-registering issue.

[benfrankel]

Oh, and the preexec_fn is needed to reproduce the issue because after fork hooks are only run when preexec_fn is not None. I don't know why it's implemented that way.

[clayg]

@tipabu do you think this issue could be addressed in eventlet? Frankly forking with running greenthreads is terrifying (or even just after starting the hub at ALL). Maybe if we could define some sort of sane way to fork? Like maybe kill all running greenthreads abandoning them to the parent with the option to bootstrap a new hub in the child? I'm sure the threading module assert is doing exactly what it intends to and the running greenthreads are worth blowing up over. But without a solid use-case I'm not sure how to evaluate what the behavior SHOULD be.

[tipabu]

So I think all of these suggestions are definitely implementable in eventlet -- I even took a stab at addressing the 3x _after_fork functions (though it's still registered twice; I'm not sure how to avoid that). I came up with a patch like http://paste.openstack.org/show/793151/

Buuuut... running the existing tests, I see two new errors in patcher_test.py. The first, test_threading_current, bombs out at https://github.com/eventlet/eventlet/blob/v0.25.1/tests/isolated/patcher_threading_current.py#L23 with

Traceback (most recent call last):
  File ".../tests/isolated/patcher_threading_current.py", line 23, in <module>
    assert g == set(('t0', 't1', 't2')), repr(g)
AssertionError: {'MainThread'}

...which seems sensible enough: if you're going to monkey-patch threading to use greenthreads, it makes sense that threading.current_thread() should refer to the greenthread, not the main thread. I can fix this by backing out the additions from "Workaround for the special case" -- I've had success spinning up posix threads or full-on subprocesses which then fan out some greenthreads; going the other way, not so much. Maybe it's reasonable to just say that monkey-patched threading plus greenthreads invoking subprocesses don't really mix?

But even without that, I see Threading.test_tpool failing with

>           self.assertEqual(lines[1], "1", lines[1])
E           AssertionError: '21' != '1'

which also doesn't seem great -- it sure feels to me like the code at https://github.com/eventlet/eventlet/blob/v0.25.1/tests/patcher_test.py#L320-L327 ought to print 1 at the end of that 😕

Does that mean that the subprocess is sharing _active?? That seems like it'd be bad...

Maybe the better thing to do would be to embrace register_at_fork and add our own handler to fix up current_thread? Something like

diff --git a/eventlet/patcher.py b/eventlet/patcher.py
index b8f8c16..1661eb1 100644
--- a/eventlet/patcher.py
+++ b/eventlet/patcher.py
@@ -312,6 +312,27 @@ def monkey_patch(**on):
             for attr_name in deleted:
                 if hasattr(orig_mod, attr_name):
                     delattr(orig_mod, attr_name)
+
+            if name == 'threading':
+                def __fix_threading_active(
+                    _os=original('os'),
+                    _global_dict=original('threading').current_thread.__globals__,
+                    _patched=orig_mod
+                ):
+                    _prefork_active = [None]
+
+                    def __before_fork():
+                        _prefork_active[0] = _global_dict['_active']
+                        _global_dict['_active'] = _patched._active
+
+                    def __after_fork():
+                        _global_dict['_active'] = _prefork_active[0]
+
+                    _os.register_at_fork(
+                        before=__before_fork,
+                        after_in_parent=__after_fork,
+                        after_in_child=__after_fork)
+                __fix_threading_active()
     finally:
         imp.release_lock()
 

shows some promise... I'll try to play with it some more...