Skip to content

Commit 64afb03

Browse files
Arran Cudbard-Bellarr2036
authored andcommitted
Fix SystemError in _DeferredCall.get() under GC pressure
CPython builds the argument tuple incrementally via _PyTuple_Resize when unpacking a generator: f(*(gen)). _PyTuple_Resize guards that Py_REFCNT(v) == 1 before resizing a non-empty tuple. Under sustained allocation pressure a GC cycle can run between generator yields and temporarily increment the refcount on the partial tuple, causing _PyTuple_Resize to call PyErr_BadInternalCall(). Change the generator expression to a list comprehension. CPython builds the list first and passes it to CALL_FUNCTION_EX, which takes the PySequence_Fast list path and never calls _PyTuple_Resize.
1 parent b0cc432 commit 64afb03

2 files changed

Lines changed: 39 additions & 1 deletion

File tree

sdks/python/apache_beam/runners/worker/sdk_worker.py

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1454,7 +1454,13 @@ def wait(self, timeout=None):
14541454

14551455
def get(self, timeout=None):
14561456
# type: (Optional[float]) -> T
1457-
return self._func(*(arg.get(timeout) for arg in self._args))
1457+
# List comprehension, not generator: *(gen) causes CPython to build the
1458+
# argument tuple incrementally via _PyTuple_Resize, which asserts
1459+
# Py_REFCNT(v)==1. A GC cycle between yields can increment that refcount,
1460+
# raising SystemError (Objects/tupleobject.c:927). See cpython/issues/127058
1461+
# (fixed in 3.14: cpython@5a23994). *[list] allocates the
1462+
# tuple once at its final size, avoiding the resize entirely.
1463+
return self._func(*[arg.get(timeout) for arg in self._args])
14581464

14591465
def set(self, value):
14601466
# type: (T) -> _Future[T]

sdks/python/apache_beam/runners/worker/sdk_worker_test.py

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -704,6 +704,38 @@ def testShortIdAssignment(self):
704704
% case.info)
705705

706706

707+
class DeferredCallTest(unittest.TestCase):
708+
"""Tests for _DeferredCall.get()."""
709+
def test_get_single_arg(self):
710+
f = sdk_worker._Future().set(42)
711+
call = sdk_worker._DeferredCall(lambda x: x, f)
712+
self.assertEqual(call.get(), 42)
713+
714+
def test_get_multiple_args(self):
715+
futures = [sdk_worker._Future().set(i) for i in range(5)]
716+
call = sdk_worker._DeferredCall(lambda *args: sum(args), *futures)
717+
self.assertEqual(call.get(), sum(range(5)))
718+
719+
def test_get_non_future_args_are_wrapped(self):
720+
# __init__ wraps non-Future values in _Future().set(v); get() must work.
721+
call = sdk_worker._DeferredCall(lambda x, y: x * y, 3, 7)
722+
self.assertEqual(call.get(), 21)
723+
724+
def test_get_mixed_future_and_value_args(self):
725+
a = sdk_worker._Future().set(10)
726+
call = sdk_worker._DeferredCall(lambda x, y: x + y, a, 5)
727+
self.assertEqual(call.get(), 15)
728+
729+
def test_get_zero_args(self):
730+
call = sdk_worker._DeferredCall(lambda: 99)
731+
self.assertEqual(call.get(), 99)
732+
733+
def test_get_preserves_return_value_type(self):
734+
f = sdk_worker._Future().set({'key': 'val'})
735+
call = sdk_worker._DeferredCall(lambda d: d, f)
736+
self.assertEqual(call.get(), {'key': 'val'})
737+
738+
707739
def monitoringInfoMetadata(info):
708740
return {
709741
descriptor.name: value

0 commit comments

Comments
 (0)