Skip to content

Commit 41c3657

Browse files
committed
fix race condition with lock for reset
1 parent 8ccd1d7 commit 41c3657

1 file changed

Lines changed: 7 additions & 3 deletions

File tree

sdks/python/apache_beam/runners/worker/statesampler_fast.pyx

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -153,9 +153,13 @@ cdef class StateSampler(object):
153153
self.sampling_thread.join()
154154

155155
def reset(self):
156-
for state in self.scoped_states_by_index:
157-
(<ScopedState>state)._nsecs = 0
158-
self.started = self.finished = False
156+
pythread.PyThread_acquire_lock(self.lock, pythread.WAIT_LOCK)
157+
try:
158+
for state in self.scoped_states_by_index:
159+
(<ScopedState>state)._nsecs = 0
160+
self.started = self.finished = False
161+
finally:
162+
pythread.PyThread_release_lock(self.lock)
159163

160164
cpdef ScopedState current_state(self):
161165
return self.current_state_c()

0 commit comments

Comments
 (0)