enable multithreading for PythonLanguage

- add support for starting threads using the builtin _thread module

Author: cosminbasca

graalpython/com.oracle.graal.python.test/src/tests/test_thread.py

@@ -0,0 +1,307 @@
+# Copyright (c) 2018, Oracle and/or its affiliates. All rights reserved.
+# DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+#
+# The Universal Permissive License (UPL), Version 1.0
+#
+# Subject to the condition set forth below, permission is hereby granted to any
+# person obtaining a copy of this software, associated documentation and/or
+# data (collectively the "Software"), free of charge and under any and all
+# copyright rights in the Software, and any and all patent rights owned or
+# freely licensable by each licensor hereunder covering either (i) the
+# unmodified Software as contributed to or provided by such licensor, or (ii)
+# the Larger Works (as defined below), to deal in both
+#
+# (a) the Software, and
+#
+# (b) any piece of software and/or hardware listed in the lrgrwrks.txt file if
+# one is included with the Software each a "Larger Work" to which the Software
+# is contributed by such licensors),
+#
+# without restriction, including without limitation the rights to copy, create
+# derivative works of, display, perform, and distribute the Software and make,
+# use, sell, offer for sale, import, export, have made, and have sold the
+# Software and the Larger Work(s), and to sublicense the foregoing rights on
+# either these or other terms.
+#
+# This license is subject to the following condition:
+#
+# The above copyright notice and either this complete permission notice or at a
+# minimum a reference to the UPL must be included in all copies or substantial
+# portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+import os
+import random
+import unittest
+from test import support
+
+thread = support.import_module('_thread')
+import time
+import weakref
+
+from test import lock_tests
+
+NUMTASKS = 10
+NUMTRIPS = 3
+POLL_SLEEP = 0.010 # seconds = 10 ms
+
+_print_mutex = thread.allocate_lock()
+
+def verbose_print(arg):
+    """Helper function for printing out debugging output."""
+    if support.verbose:
+        with _print_mutex:
+            print(arg)
+
+
+class BasicThreadTest(unittest.TestCase):
+
+    def setUp(self):
+        self.done_mutex = thread.allocate_lock()
+        self.done_mutex.acquire()
+        self.running_mutex = thread.allocate_lock()
+        self.random_mutex = thread.allocate_lock()
+        self.created = 0
+        self.running = 0
+        self.next_ident = 0
+
+        key = support.threading_setup()
+        self.addCleanup(support.threading_cleanup, *key)
+
+
+class ThreadRunningTests(BasicThreadTest):
+
+    def newtask(self):
+        with self.running_mutex:
+            self.next_ident += 1
+            verbose_print("creating task %s" % self.next_ident)
+            thread.start_new_thread(self.task, (self.next_ident,))
+            self.created += 1
+            self.running += 1
+
+    def task(self, ident):
+        with self.random_mutex:
+            delay = random.random() / 10000.0
+        verbose_print("task %s will run for %sus" % (ident, round(delay*1e6)))
+        time.sleep(delay)
+        verbose_print("task %s done" % ident)
+        with self.running_mutex:
+            self.running -= 1
+            if self.created == NUMTASKS and self.running == 0:
+                self.done_mutex.release()
+
+    def test_starting_threads(self):
+        # Basic test for thread creation.
+        for i in range(NUMTASKS):
+            self.newtask()
+        verbose_print("waiting for tasks to complete...")
+        self.done_mutex.acquire()
+        verbose_print("all tasks done")
+
+    def test_stack_size(self):
+        # Various stack size tests.
+        self.assertEqual(thread.stack_size(), 0, "initial stack size is not 0")
+
+        thread.stack_size(0)
+        self.assertEqual(thread.stack_size(), 0, "stack_size not reset to default")
+
+    @unittest.skipIf(os.name not in ("nt", "posix"), 'test meant for nt and posix')
+    def test_nt_and_posix_stack_size(self):
+        try:
+            thread.stack_size(4096)
+        except ValueError:
+            verbose_print("caught expected ValueError setting "
+                          "stack_size(4096)")
+        except thread.error:
+            self.skipTest("platform does not support changing thread stack "
+                          "size")
+
+        fail_msg = "stack_size(%d) failed - should succeed"
+        for tss in (262144, 0x100000, 0):
+            thread.stack_size(tss)
+            self.assertEqual(thread.stack_size(), tss, fail_msg % tss)
+            verbose_print("successfully set stack_size(%d)" % tss)
+
+        for tss in (262144, 0x100000):
+            verbose_print("trying stack_size = (%d)" % tss)
+            self.next_ident = 0
+            self.created = 0
+            for i in range(NUMTASKS):
+                self.newtask()
+
+            verbose_print("waiting for all tasks to complete")
+            self.done_mutex.acquire()
+            verbose_print("all tasks done")
+
+        thread.stack_size(0)
+
+    def test__count(self):
+        # Test the _count() function.
+        orig = thread._count()
+        mut = thread.allocate_lock()
+        mut.acquire()
+        started = []
+        def task():
+            started.append(None)
+            mut.acquire()
+            mut.release()
+        thread.start_new_thread(task, ())
+        while not started:
+            time.sleep(POLL_SLEEP)
+        self.assertEqual(thread._count(), orig + 1)
+        # Allow the task to finish.
+        mut.release()
+        # The only reliable way to be sure that the thread ended from the
+        # interpreter's point of view is to wait for the function object to be
+        # destroyed.
+        done = []
+        wr = weakref.ref(task, lambda _: done.append(None))
+        del task
+        while not done:
+            time.sleep(POLL_SLEEP)
+        self.assertEqual(thread._count(), orig)
+
+    def test_save_exception_state_on_error(self):
+        # See issue #14474
+        def task():
+            started.release()
+            raise SyntaxError
+        def mywrite(self, *args):
+            try:
+                raise ValueError
+            except ValueError:
+                pass
+            real_write(self, *args)
+        c = thread._count()
+        started = thread.allocate_lock()
+        with support.captured_output("stderr") as stderr:
+            real_write = stderr.write
+            stderr.write = mywrite
+            started.acquire()
+            thread.start_new_thread(task, ())
+            started.acquire()
+            while thread._count() > c:
+                time.sleep(POLL_SLEEP)
+        self.assertIn("Traceback", stderr.getvalue())
+
+
+class Barrier:
+    def __init__(self, num_threads):
+        self.num_threads = num_threads
+        self.waiting = 0
+        self.checkin_mutex  = thread.allocate_lock()
+        self.checkout_mutex = thread.allocate_lock()
+        self.checkout_mutex.acquire()
+
+    def enter(self):
+        self.checkin_mutex.acquire()
+        self.waiting = self.waiting + 1
+        if self.waiting == self.num_threads:
+            self.waiting = self.num_threads - 1
+            self.checkout_mutex.release()
+            return
+        self.checkin_mutex.release()
+
+        self.checkout_mutex.acquire()
+        self.waiting = self.waiting - 1
+        if self.waiting == 0:
+            self.checkin_mutex.release()
+            return
+        self.checkout_mutex.release()
+
+
+class BarrierTest(BasicThreadTest):
+
+    def test_barrier(self):
+        self.bar = Barrier(NUMTASKS)
+        self.running = NUMTASKS
+        for i in range(NUMTASKS):
+            thread.start_new_thread(self.task2, (i,))
+        verbose_print("waiting for tasks to end")
+        self.done_mutex.acquire()
+        verbose_print("tasks done")
+
+    def task2(self, ident):
+        for i in range(NUMTRIPS):
+            if ident == 0:
+                # give it a good chance to enter the next
+                # barrier before the others are all out
+                # of the current one
+                delay = 0
+            else:
+                with self.random_mutex:
+                    delay = random.random() / 10000.0
+            verbose_print("task %s will run for %sus" %
+                          (ident, round(delay * 1e6)))
+            time.sleep(delay)
+            verbose_print("task %s entering %s" % (ident, i))
+            self.bar.enter()
+            verbose_print("task %s leaving barrier" % ident)
+        with self.running_mutex:
+            self.running -= 1
+            # Must release mutex before releasing done, else the main thread can
+            # exit and set mutex to None as part of global teardown; then
+            # mutex.release() raises AttributeError.
+            finished = self.running == 0
+        if finished:
+            self.done_mutex.release()
+
+class LockTests(lock_tests.LockTests):
+    locktype = thread.allocate_lock
+
+
+class TestForkInThread(unittest.TestCase):
+    def setUp(self):
+        self.read_fd, self.write_fd = os.pipe()
+
+    @unittest.skipUnless(hasattr(os, 'fork'), 'need os.fork')
+    @support.reap_threads
+    def test_forkinthread(self):
+        running = True
+        status = "not set"
+
+        def thread1():
+            nonlocal running, status
+
+            # fork in a thread
+            pid = os.fork()
+            if pid == 0:
+                # child
+                try:
+                    os.close(self.read_fd)
+                    os.write(self.write_fd, b"OK")
+                finally:
+                    os._exit(0)
+            else:
+                # parent
+                os.close(self.write_fd)
+                pid, status = os.waitpid(pid, 0)
+                running = False
+
+        thread.start_new_thread(thread1, ())
+        self.assertEqual(os.read(self.read_fd, 2), b"OK",
+                         "Unable to fork() in thread")
+        while running:
+            time.sleep(POLL_SLEEP)
+        self.assertEqual(status, 0)
+
+    def tearDown(self):
+        try:
+            os.close(self.read_fd)
+        except OSError:
+            pass
+
+        try:
+            os.close(self.write_fd)
+        except OSError:
+            pass
+
+
+if __name__ == "__main__":
+    unittest.main()

graalpython/com.oracle.graal.python/src/com/oracle/graal/python/PythonLanguage.java

@@ -470,4 +470,9 @@ public PCode cacheCode(String filename, Supplier<PCode> createCode) {
     public static Shape freshShape() {
         return freshShape;
     }
+
+    @Override
+    protected boolean isThreadAccessAllowed(Thread thread, boolean singleThreaded) {
+        return true;
+    }
 }

graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/Python3Core.java

@@ -136,6 +136,8 @@
 import com.oracle.graal.python.builtins.objects.slice.SliceBuiltins;
 import com.oracle.graal.python.builtins.objects.str.StringBuiltins;
 import com.oracle.graal.python.builtins.objects.superobject.SuperBuiltins;
+import com.oracle.graal.python.builtins.objects.thread.LockBuiltins;
+import com.oracle.graal.python.builtins.objects.thread.ThreadBuiltins;
 import com.oracle.graal.python.builtins.objects.traceback.TracebackBuiltins;
 import com.oracle.graal.python.builtins.objects.tuple.TupleBuiltins;
 import com.oracle.graal.python.builtins.objects.type.PythonBuiltinClass;
@@ -287,6 +289,8 @@ private static final PythonBuiltins[] initializeBuiltins() {
                         new BufferBuiltins(),
                         new MemoryviewBuiltins(),
                         new ThreadModuleBuiltins(),
+                        new ThreadBuiltins(),
+                        new LockBuiltins(),
                         new SuperBuiltins(),
                         new BinasciiModuleBuiltins(),
                         new PosixSubprocessModuleBuiltins(),