[pypy-commit] pypy default: Issue 1895: test and fix for the file's locks on multithreaded apps with fork()

[arigo]

Author: Armin Rigo <arigo at tunes.org>
Branch: 
Changeset: r74331:99a70ef5eaaf
Date: 2014-11-03 15:29 +0100
http://bitbucket.org/pypy/pypy/changeset/99a70ef5eaaf/

Log:	Issue 1895: test and fix for the file's locks on multithreaded apps
	with fork()

diff --git a/pypy/interpreter/executioncontext.py b/pypy/interpreter/executioncontext.py
--- a/pypy/interpreter/executioncontext.py
+++ b/pypy/interpreter/executioncontext.py
@@ -32,6 +32,17 @@
         self.compiler = space.createcompiler()
         self.profilefunc = None
         self.w_profilefuncarg = None
+        self.thread_disappeared = False   # might be set to True after os.fork()
+
+    @staticmethod
+    def _mark_thread_disappeared(space):
+        # Called in the child process after os.fork() by interp_posix.py.
+        # Marks all ExecutionContexts except the current one
+        # with 'thread_disappeared = True'.
+        me = space.getexecutioncontext()
+        for ec in space.threadlocals.getallvalues().values():
+            if ec is not me:
+                ec.thread_disappeared = True
 
     def gettopframe(self):
         return self.topframeref()
diff --git a/pypy/module/_file/interp_stream.py b/pypy/module/_file/interp_stream.py
--- a/pypy/module/_file/interp_stream.py
+++ b/pypy/module/_file/interp_stream.py
@@ -34,8 +34,12 @@
         # this function runs with the GIL acquired so there is no race
         # condition in the creation of the lock
         me = self.space.getexecutioncontext()   # used as thread ident
-        if self.slockowner is me:
-            return False    # already acquired by the current thread
+        if self.slockowner is not None:
+            if self.slockowner is me:
+                return False    # already acquired by the current thread
+            if self.slockowner.thread_disappeared:
+                self.slockowner = None
+                self.slock = None
         try:
             if self.slock is None:
                 self.slock = self.space.allocate_lock()
diff --git a/pypy/module/posix/interp_posix.py b/pypy/module/posix/interp_posix.py
--- a/pypy/module/posix/interp_posix.py
+++ b/pypy/module/posix/interp_posix.py
@@ -10,6 +10,7 @@
 
 from pypy.interpreter.gateway import unwrap_spec
 from pypy.interpreter.error import OperationError, wrap_oserror, wrap_oserror2
+from pypy.interpreter.executioncontext import ExecutionContext
 from pypy.module.sys.interp_encoding import getfilesystemencoding
 
 
@@ -721,6 +722,8 @@
     "NOT_RPYTHON"
     get_fork_hooks(where).append(hook)
 
+add_fork_hook('child', ExecutionContext._mark_thread_disappeared)
+
 @specialize.arg(0)
 def run_fork_hooks(where, space):
     for hook in get_fork_hooks(where):
diff --git a/pypy/module/test_lib_pypy/test_posix_extra.py b/pypy/module/test_lib_pypy/test_posix_extra.py
new file mode 100644
--- /dev/null
+++ b/pypy/module/test_lib_pypy/test_posix_extra.py
@@ -0,0 +1,40 @@
+import py
+import sys, os, subprocess
+
+
+CODE = """
+import sys, os, thread, time
+
+fd1, fd2 = os.pipe()
+f1 = os.fdopen(fd1, 'r', 0)
+f2 = os.fdopen(fd2, 'w', 0)
+
+def f():
+    print "thread started"
+    x = f1.read(1)
+    assert x == "X"
+    print "thread exit"
+
+thread.start_new_thread(f, ())
+time.sleep(0.5)
+if os.fork() == 0:   # in the child
+    time.sleep(0.5)
+    x = f1.read(1)
+    assert x == "Y"
+    print "ok!"
+    sys.exit()
+
+f2.write("X")   # in the parent
+f2.write("Y")   # in the parent
+time.sleep(1.0)
+"""
+
+
+def test_thread_fork_file_lock():
+    if not hasattr(os, 'fork'):
+        py.test.skip("requires 'fork'")
+    output = subprocess.check_output([sys.executable, '-u', '-c', CODE])
+    assert output.splitlines() == [
+        'thread started',
+        'thread exit',
+        'ok!']