jenkins-bot has submitted this change. (
https://gerrit.wikimedia.org/r/c/pywikibot/core/+/1336057?usp=email )
Change subject: threading: Propagate producer errors to generator consumers
......................................................................
threading: Propagate producer errors to generator consumers
A failed ThreadedGenerator producer exits without signalling completion,
leaving consumers waiting indefinitely for another result.
Capture producer exceptions, drain queued results using the existing
completion path, and re-raise the original exception in the consumer.
Change-Id: I7e5b747bd3f09dcf3a23c07cd86f8940c3522304
---
M pywikibot/tools/threading.py
M tests/tools_threading_tests.py
2 files changed, 73 insertions(+), 16 deletions(-)
Approvals:
jenkins-bot: Verified
Xqt: Looks good to me, approved
diff --git a/pywikibot/tools/threading.py b/pywikibot/tools/threading.py
index e112249..c344960 100644
--- a/pywikibot/tools/threading.py
+++ b/pywikibot/tools/threading.py
@@ -48,6 +48,9 @@
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
.. version-added:: 3.0
+ .. version-changed:: 11.8
+ Exceptions from the producer are re-raised during iteration after
+ yielding any queued results.
"""
def __init__(self, group=None, target=None, name: str = 'GeneratorThread',
@@ -71,6 +74,7 @@
super().__init__(group=group, name=name)
self.queue: queue.Queue[Any] = queue.Queue(qsize)
self.finished = threading.Event()
+ self._exception: BaseException | None = None
def __iter__(self):
"""Iterate results from the queue."""
@@ -85,28 +89,35 @@
except KeyboardInterrupt:
self.stop()
+ if self._exception is not None:
+ raise self._exception
+
def stop(self) -> None:
"""Stop the background thread."""
self.finished.set()
def run(self) -> None:
"""Run the generator and store the results on the queue."""
- iterable = any(hasattr(self.generator, key)
- for key in ('__iter__', '__getitem__'))
- if iterable and not self.args and not self.kwargs:
- self.__gen = self.generator
- else:
- self.__gen = self.generator(*self.args, **self.kwargs)
- for result in self.__gen:
- while True:
- if self.finished.is_set():
- return
- try:
- self.queue.put_nowait(result)
- except queue.Full:
- time.sleep(0.25)
- continue
- break
+ try:
+ iterable = any(hasattr(self.generator, key)
+ for key in ('__iter__', '__getitem__'))
+ if iterable and not self.args and not self.kwargs:
+ self.__gen = self.generator
+ else:
+ self.__gen = self.generator(*self.args, **self.kwargs)
+ for result in self.__gen:
+ while True:
+ if self.finished.is_set():
+ return
+ try:
+ self.queue.put_nowait(result)
+ except queue.Full:
+ time.sleep(0.25)
+ continue
+ break
+ except BaseException as e: # noqa: B036
+ # Re-raised in the consumer thread by __iter__.
+ self._exception = e
# wait for queue to be emptied, then kill the thread
while not self.finished.is_set() and not self.queue.empty():
time.sleep(0.25)
diff --git a/tests/tools_threading_tests.py b/tests/tools_threading_tests.py
index 20f12bb..8ff2c70 100755
--- a/tests/tools_threading_tests.py
+++ b/tests/tools_threading_tests.py
@@ -50,6 +50,52 @@
thd_gen.start()
self.assertEqual(list(thd_gen), list(iterable))
+ def test_producer_failure(self) -> None:
+ """Test producer failures reach consumers after queued results."""
+ values = (1, 2, 3)
+ failure = ValueError('producer failed')
+
+ def generate():
+ yield from values
+ raise failure
+
+ self._check_failure(generate, values, failure)
+
+ def test_target_failure(self) -> None:
+ """Test failure while creating the iterable reaches the consumer."""
+ failure = RuntimeError('target failed')
+
+ def target():
+ raise failure
+
+ self._check_failure(target, (), failure)
+
+ def _check_failure(self, target, values, failure) -> None:
+ """Consume with a timeout so a broken producer cannot hang tests."""
+ generator = ThreadedGenerator(target=target, qsize=1)
+ received = []
+ errors = []
+
+ def consume() -> None:
+ try:
+ received.extend(generator)
+ except type(failure) as e:
+ errors.append(e)
+
+ consumer = Thread(target=consume, daemon=True)
+ consumer.start()
+ try:
+ consumer.join(3)
+ self.assertFalse(consumer.is_alive())
+ self.assertEqual(received, list(values))
+ self.assertEqual(errors, [failure])
+ self.assertIs(errors[0], failure)
+ finally:
+ generator.stop()
+ consumer.join(3)
+ generator.join(3)
+ self.assertFalse(generator.is_alive())
+
class BoundedThreadPoolTests(TestCase):
--
To view, visit
https://gerrit.wikimedia.org/r/c/pywikibot/core/+/1336057?usp=email
To unsubscribe, or for help writing mail filters, visit
https://gerrit.wikimedia.org/r/settings?usp=email
Gerrit-MessageType: merged
Gerrit-Project: pywikibot/core
Gerrit-Branch: master
Gerrit-Change-Id: I7e5b747bd3f09dcf3a23c07cd86f8940c3522304
Gerrit-Change-Number: 1336057
Gerrit-PatchSet: 5
Gerrit-Owner: Mahveotm <[email protected]>
Gerrit-Reviewer: Xqt <[email protected]>
Gerrit-Reviewer: jenkins-bot
_______________________________________________
Pywikibot-commits mailing list -- [email protected]
To unsubscribe send an email to [email protected]
