derrickaw commented on code in PR #38851:
URL: https://github.com/apache/beam/pull/38851#discussion_r3380326352
##########
sdks/python/apache_beam/runners/worker/statesampler_test.py:
##########
@@ -312,6 +313,46 @@ def test_do_operation_process_timer_with_exception(self,
mock_get_dofn_specs):
actual_value, state_duration_ms * (1.0 - margin_of_error))
_LOGGER.info("Exception test finished successfully.")
+ def test_concurrent_nsecs_reads(self):
+ """Verify that concurrent reads of nsecs behave correctly under thread
contention.
+
+ This test runs state transitions on the main thread and reads `nsecs`
properties
+ from a secondary Python thread, while the background sampler thread is
concurrently
+ updating counter states.
+ """
+ if not statesampler.FAST_SAMPLER:
+ self.skipTest('Requires FAST_SAMPLER')
+
+ counter_factory = CounterFactory()
+ sampler = statesampler.StateSampler(
+ 'concurrent', counter_factory, sampling_period_ms=1)
+
+ sampler.start()
+ state_a = sampler.scoped_state('step1', 'statea')
+ state_b = sampler.scoped_state('step1', 'stateb')
+
+ stop_signal = False
+
+ def read_nsecs_loop():
+ while not stop_signal:
+ _ = state_a.nsecs
+ _ = state_b.nsecs
+ time.sleep(0.001)
+
+ reader_thread = threading.Thread(target=read_nsecs_loop)
+ reader_thread.start()
+
+ try:
+ for _ in range(100):
+ with state_a:
+ time.sleep(0.001)
+ with state_b:
+ time.sleep(0.001)
+ finally:
+ stop_signal = True
+ reader_thread.join()
+ sampler.stop()
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
