Support send(), throw(), and close() for generators.

Author: rhettinger

Doc/library/threading.rst

@@ -1461,7 +1461,9 @@ of the derived iterators.
 .. class:: serialize(iterable)
 
    Return an iterator wrapper that serializes concurrent calls to
-   :meth:`~iterator.__next__` using a lock.
+   :meth:`~iterator.__next__` using a lock.  For generators, will also
+   serialize calls to :meth:`~generator.send`, :meth:`~generator.throw`,
+   and :meth:`~generator.close`.
 
    This makes it possible to share a single iterator, including a generator
    iterator, between multiple threads. A lock assures that calls are handled

Lib/test/test_threading.py

@@ -2405,6 +2405,77 @@ def consumer(iterator):
 
         self.assertEqual(result, limit * (limit - 1) // 2)
 
+    def test_serialize_generator_methods(self):
+        # A generator that yields and receives
+        def echo():
+            try:
+                while True:
+                    val = yield "ready"
+                    yield f"received {val}"
+            except ValueError:
+                yield "caught"
+
+        it = threading.serialize(echo())
+
+        # Test __next__
+        self.assertEqual(next(it), "ready")
+
+        # Test send()
+        self.assertEqual(it.send("hello"), "received hello")
+        self.assertEqual(next(it), "ready")
+
+        # Test throw()
+        self.assertEqual(it.throw(ValueError), "caught")
+
+        # Test close()
+        it.close()
+        with self.assertRaises(StopIteration):
+            next(it)
+
+    def test_serialize_methods_attribute_error(self):
+        # A standard iterator that does not have send/throw/close
+        # should raise AttributeError when called.
+        standard_it = threading.serialize([1, 2, 3])
+
+        with self.assertRaises(AttributeError):
+            standard_it.send("foo")
+
+        with self.assertRaises(AttributeError):
+            standard_it.throw(ValueError)
+
+        with self.assertRaises(AttributeError):
+            standard_it.close()
+
+    def test_serialize_generator_methods_locking(self):
+        # Verifies that generator methods also acquire the lock.
+        # We can test this by checking if the lock is held during the call.
+
+        class LockCheckingGenerator:
+            def __init__(self, lock):
+                self.lock = lock
+            def __iter__(self):
+                return self
+            def send(self, value):
+                if not self.lock.locked():
+                    raise RuntimeError("Lock not held during send()")
+                return value
+            def throw(self, *args):
+                if not self.lock.locked():
+                    raise RuntimeError("Lock not held during throw()")
+            def close(self):
+                if not self.lock.locked():
+                    raise RuntimeError("Lock not held during close()")
+
+        # Manually create the serialize object to inspect the lock
+        it = threading.serialize([])
+        mock_gen = LockCheckingGenerator(it.lock)
+        it.iterator = mock_gen
+
+        # These should not raise RuntimeError
+        it.send(1)
+        it.throw(ValueError)
+        it.close()
+
     def test_synchronized_serializes_generator_instances(self):
         unique = 10
         repetitions = 5

Lib/threading.py

@@ -849,8 +849,8 @@ class BrokenBarrierError(RuntimeError):
 class serialize:
     """Wrap a non-concurrent iterator with a lock to enforce sequential access.
 
-    Applies a non-reentrant lock around calls to __next__, allowing
-    iterator and generator instances to be shared by multiple consumer
+    Applies a non-reentrant lock around calls to __next__, send, throw, and close.
+    Allows iterator and generator instances to be shared by multiple consumer
     threads.
     """
 
@@ -867,6 +867,30 @@ def __next__(self):
         with self.lock:
             return next(self.iterator)
 
+    def send(self, value, /):
+        """Send a value to a generator.
+
+        Raises AttributeError if not a generator.
+        """
+        with self.lock:
+            return self.iterator.send(value)
+
+    def throw(self, *args):
+        """Call throw() on a generator.
+
+        Raises AttributeError if not a generator.
+        """
+        with self.lock:
+            return self.iterator.throw(*args)
+
+    def close(self):
+        """Call close() on a generator.
+
+        Raises AttributeError if not a generator.
+        """
+        with self.lock:
+            return self.iterator.close()
+
 
 def synchronized(func):
     """Wrap an iterator-returning callable to make its iterators thread-safe.