Implement SemLock with eventfd where available (3.10+/Linux)

Otherwise SemLock is semaphore-based and blocks when RLock

eventfd-based solution doesn't work with `forkserver`

fixes #16

Author: arcivanov

.github/workflows/build.yml

@@ -25,6 +25,7 @@ jobs:
           - '3.6'
           - 'pypy-3.7'
           - 'pypy-3.8'
+          # - 'pypy-3.9'
         with-venv:
           - 'true'
           - 'false'

.idea/geventmp.iml

@@ -18,9 +18,10 @@
 
 from gevent import spawn, monkey
 from gevent.util import assert_switches
-from geventmp.monkey import GEVENT_SAVED_MODULE_SETTINGS
 from multiprocessing.util import get_logger
 
+from geventmp.monkey import GEVENT_SAVED_MODULE_SETTINGS
+
 logger = get_logger()
 
 
@@ -72,3 +73,34 @@ def count():
     task.kill()
     logger.info("exiting")
     sys.exit(10)
+
+
+def test_joinable_queues(r_q, w_q):
+    def count():
+        while True:
+            sleep(0.01)
+
+    task = spawn(count)
+    task.start()
+
+    with assert_switches():
+        sleep(1)
+
+    with assert_switches():
+        logger.info(r_q.get(timeout=5))
+
+    with assert_switches():
+        r_q.task_done()
+
+    with assert_switches():
+        sleep(1)
+
+    with assert_switches():
+        w_q.put(test_queues.__name__, timeout=5)
+
+    with assert_switches():
+        sleep(1)
+
+    task.kill()
+    logger.info("exiting")
+    sys.exit(10)

src/integrationtest/python/_mp_test_gevent.py

@@ -20,8 +20,8 @@
 if not getattr(current_process(), "_inheriting", False):
     monkey.patch_all()
 
-    # from multiprocessing.util import log_to_stderr
-    # log_to_stderr(1)
+# from multiprocessing.util import log_to_stderr
+# log_to_stderr(1)
 
 from unittest import TestCase, main
 import trace
@@ -59,8 +59,11 @@ def test_mp_queues_fork(self):
     def test_mp_queues_spawn(self):
         self.run_test_mp_queues("spawn", _mp_test_gevent.test_queues)
 
-    def test_mp_queues_forkserver(self):
-        self.run_test_mp_queues("forkserver", _mp_test.test_queues)
+    def test_mp_jqueues_fork(self):
+        self.run_test_mp_jqueues("fork", _mp_test_gevent.test_queues)
+
+    def test_mp_jqueues_spawn(self):
+        self.run_test_mp_jqueues("spawn", _mp_test_gevent.test_queues)
 
     def test_mp_no_args_fork(self):
         self.run_test_mp_no_args("fork", _mp_test_gevent.test_no_args)
@@ -118,6 +121,16 @@ def run_test_mp_queues(self, context, func, do_trace=False):
         else:
             self._test_mp_queues(p, r_q, w_q)
 
+    def run_test_mp_jqueues(self, context, func, do_trace=False):
+        ctx = mp.get_context(context)
+        r_q = ctx.JoinableQueue()
+        w_q = ctx.JoinableQueue()
+        p = ctx.Process(target=func, args=(w_q, r_q))
+        if do_trace:
+            trace.Trace(count=0).runfunc(self._test_mp_jqueues, p, r_q, w_q)
+        else:
+            self._test_mp_jqueues(p, r_q, w_q)
+
     def _test_mp_queues(self, p, r_q, w_q):
         async_counter = [0]
 
@@ -151,6 +164,42 @@ def count():
         logger.info("Async counter counted to %d" % async_counter[0])
         self.assertGreater(async_counter[0], 0)
 
+    def _test_mp_jqueues(self, p, r_q, w_q):
+        async_counter = [0]
+
+        def count():
+            while True:
+                idle()
+                async_counter[0] += 1
+                sleep(0.001)
+
+        task = spawn(count)
+
+        p.start()
+        self.assertTrue(p.pid > 0)
+
+        with assert_switches():
+            w_q.put("master", timeout=5)
+        with assert_switches():
+            self.assertEqual(r_q.get(timeout=5), "test_queues")
+        r_q.task_done()
+
+        with assert_switches():
+            start = clock()
+            p.join(15)
+            end = clock()
+            logger.info("Waited for child to die for %f" % (end - start))
+        task.kill()
+
+        # This is to ensure a greenlet flip
+        sleep(0.001)
+
+        logger.info(f"checking {p} is alive")
+        self.assertFalse(p.is_alive())
+        self.assertEqual(p.exitcode, 10)
+        logger.info("Async counter counted to %d" % async_counter[0])
+        self.assertGreater(async_counter[0], 0)
+
 
 if __name__ == '__main__':
     main()

src/integrationtest/python/monkey_mp_tests.py

@@ -0,0 +1,143 @@
+#   -*- coding: utf-8 -*-
+#   Copyright 2022 Karellen, Inc. and contributors
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+
+from multiprocessing.process import current_process
+
+from gevent import monkey
+
+if not getattr(current_process(), "_inheriting", False):
+    monkey.patch_all()
+
+    # from multiprocessing.util import log_to_stderr
+    # log_to_stderr(1)
+
+from unittest import TestCase, main
+
+from geventmp.monkey import GEVENT_SAVED_MODULE_SETTINGS
+
+import multiprocessing as mp
+from multiprocessing.util import get_logger
+import sys
+from time import monotonic
+
+logger = get_logger()
+
+
+class TestSynchronizers(TestCase):
+    def setUp(self):
+        self.assertTrue(monkey.saved[GEVENT_SAVED_MODULE_SETTINGS].get("geventmp"),
+                        "GeventMP patch has not run!")
+        self.tearDown()
+
+    def tearDown(self):
+        sys.stdout.flush()
+        sys.stderr.flush()
+        logger.info("=====================")
+        sys.stdout.flush()
+
+    def test_semaphore_fork(self):
+        self._test_semaphore("fork")
+
+    def test_semaphore_spawn(self):
+        self._test_semaphore("spawn")
+
+    def test_semaphore_forkserver(self):
+        self._test_semaphore("forkserver")
+
+    def test_bounded_semaphore_fork(self):
+        self._test_bounded_semaphore("fork")
+
+    def test_bounded_semaphore_spawn(self):
+        self._test_bounded_semaphore("spawn")
+
+    def test_bounded_semaphore_forkserver(self):
+        self._test_bounded_semaphore("forkserver")
+
+    def _test_semaphore(self, ctx_name):
+        ctx = mp.get_context(ctx_name)
+        s = ctx.Semaphore()
+        self.assertEqual(s.get_value(), 1)
+        self.assertEqual(s._semlock._count(), 0)
+        repr(s)
+
+        s.release()
+        self.assertEqual(s.get_value(), 2)
+        self.assertEqual(s._semlock._count(), -1)
+        repr(s)
+
+        self.assertTrue(s.acquire())
+        self.assertEqual(s.get_value(), 1)
+        self.assertEqual(s._semlock._count(), 0)
+        repr(s)
+
+        self.assertTrue(s.acquire())
+        self.assertEqual(s.get_value(), 0)
+        self.assertEqual(s._semlock._count(), 1)
+        repr(s)
+
+        self.assertFalse(s.acquire(block=False))
+        self.assertEqual(s.get_value(), 0)
+        self.assertEqual(s._semlock._count(), 1)
+        repr(s)
+
+        start = monotonic()
+        self.assertFalse(s.acquire(timeout=1))
+        duration = monotonic() - start
+        self.assertGreater(duration, 1)
+        self.assertLess(duration, 3)
+        s.release()
+        self.assertEqual(s.get_value(), 1)
+        self.assertEqual(s._semlock._count(), 0)
+
+    def _test_bounded_semaphore(self, ctx_name):
+        ctx = mp.get_context(ctx_name)
+        s = ctx.BoundedSemaphore(2)
+        repr(s)
+
+        self.assertEqual(s.get_value(), 2)
+        self.assertEqual(s._semlock._count(), 0)
+
+        self.assertTrue(s.acquire())
+        self.assertEqual(s.get_value(), 1)
+        self.assertEqual(s._semlock._count(), 1)
+        repr(s)
+
+        self.assertTrue(s.acquire())
+        self.assertEqual(s.get_value(), 0)
+        self.assertEqual(s._semlock._count(), 2)
+        repr(s)
+
+        self.assertFalse(s.acquire(block=False))
+        self.assertEqual(s.get_value(), 0)
+        self.assertEqual(s._semlock._count(), 2)
+        repr(s)
+
+        start = monotonic()
+        self.assertFalse(s.acquire(timeout=1))
+        duration = monotonic() - start
+        self.assertGreater(duration, 1)
+        self.assertLess(duration, 3)
+        s.release()
+        self.assertEqual(s.get_value(), 1)
+        self.assertEqual(s._semlock._count(), 1)
+        s.release()
+        self.assertEqual(s.get_value(), 2)
+        self.assertEqual(s._semlock._count(), 0)
+        with self.assertRaises(ValueError):
+            s.release()
+
+
+if __name__ == '__main__':
+    main()