(beam) branch master updated: fix race condition in statesampler with lock for reset (#39129)

[derrick]

This is an automated email from the ASF dual-hosted git repository.

derrickaw pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 31175f9ab82 fix race condition in statesampler with lock for reset 
(#39129)
31175f9ab82 is described below

commit 31175f9ab82974644a6add4cddf761699ba9214e
Author: Derrick Williams <derric...@google.com>
AuthorDate: Mon Jun 29 10:13:03 2026 -0400

    fix race condition in statesampler with lock for reset (#39129)
    
    * fix race condition with lock for reset
    
    * add nogil
---
 sdks/python/apache_beam/runners/worker/statesampler_fast.pyx | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx 
b/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx
index 7075ef47017..5a67d6764ea 100644
--- a/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx
+++ b/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx
@@ -153,9 +153,14 @@ cdef class StateSampler(object):
     self.sampling_thread.join()
 
   def reset(self):
-    for state in self.scoped_states_by_index:
-      (<ScopedState>state)._nsecs = 0
-    self.started = self.finished = False
+    with nogil:
+      pythread.PyThread_acquire_lock(self.lock, pythread.WAIT_LOCK)
+    try:
+      for state in self.scoped_states_by_index:
+        (<ScopedState>state)._nsecs = 0
+      self.started = self.finished = False
+    finally:
+      pythread.PyThread_release_lock(self.lock)
 
   cpdef ScopedState current_state(self):
     return self.current_state_c()