This is an automated email from the ASF dual-hosted git repository. pabloem pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/beam.git
commit 9cdbbe03091ef0507203d977000cfc4e22e5dd6e Author: Pablo <[email protected]> AuthorDate: Fri May 18 10:53:20 2018 -0700 Adding flag to mask change --- sdks/python/apache_beam/runners/worker/operations.py | 2 +- sdks/python/apache_beam/runners/worker/sideinputs.py | 4 ++-- .../apache_beam/runners/worker/sideinputs_test.py | 2 +- .../apache_beam/runners/worker/statesampler_fast.pxd | 2 ++ .../apache_beam/runners/worker/statesampler_fast.pyx | 18 ------------------ 5 files changed, 6 insertions(+), 22 deletions(-) diff --git a/sdks/python/apache_beam/runners/worker/operations.py b/sdks/python/apache_beam/runners/worker/operations.py index f382837..9aa29b8 100644 --- a/sdks/python/apache_beam/runners/worker/operations.py +++ b/sdks/python/apache_beam/runners/worker/operations.py @@ -325,7 +325,7 @@ class DoOperation(Operation): sources.append(si.source) # The tracking of time spend reading and bytes read from side inputs is # behind an experiment flag to test its performance impact. - if 'sideinput_io_metrics' in RuntimeValueProvider.experiments: + if 'sideinput_io_metrics_v2' in RuntimeValueProvider.experiments: si_counter = opcounters.SideInputReadCounter( self.counter_factory, self.state_sampler, diff --git a/sdks/python/apache_beam/runners/worker/sideinputs.py b/sdks/python/apache_beam/runners/worker/sideinputs.py index a71557c..980a208 100644 --- a/sdks/python/apache_beam/runners/worker/sideinputs.py +++ b/sdks/python/apache_beam/runners/worker/sideinputs.py @@ -80,7 +80,7 @@ class PrefetchingSourceSetIterable(object): def add_byte_counter(self, reader): """Adds byte counter observer to a side input reader. - If the 'sideinput_io_metrics' experiment flag is not passed in, then + If the 'sideinput_io_metrics_v2' experiment flag is not passed in, then nothing is attached to the reader. Args: @@ -123,7 +123,7 @@ class PrefetchingSourceSetIterable(object): # The tracking of time spend reading and bytes read from side # inputs is kept behind an experiment flag to test performance # impact. - if 'sideinput_io_metrics' in RuntimeValueProvider.experiments: + if 'sideinput_io_metrics_v2' in RuntimeValueProvider.experiments: self.add_byte_counter(reader) returns_windowed_values = reader.returns_windowed_values for value in reader: diff --git a/sdks/python/apache_beam/runners/worker/sideinputs_test.py b/sdks/python/apache_beam/runners/worker/sideinputs_test.py index 050ecdc..c4240dd 100644 --- a/sdks/python/apache_beam/runners/worker/sideinputs_test.py +++ b/sdks/python/apache_beam/runners/worker/sideinputs_test.py @@ -92,7 +92,7 @@ class PrefetchingSourceIteratorTest(unittest.TestCase): def test_bytes_read_are_reported(self): RuntimeValueProvider.set_runtime_options( - {'experiments': ['sideinput_io_metrics', 'other']}) + {'experiments': ['sideinput_io_metrics_v2', 'other']}) mock_read_counter = mock.MagicMock() source_records = ['a', 'b', 'c', 'd'] sources = [ diff --git a/sdks/python/apache_beam/runners/worker/statesampler_fast.pxd b/sdks/python/apache_beam/runners/worker/statesampler_fast.pxd index a808a8e..76b379b 100644 --- a/sdks/python/apache_beam/runners/worker/statesampler_fast.pxd +++ b/sdks/python/apache_beam/runners/worker/statesampler_fast.pxd @@ -25,6 +25,8 @@ from libc.stdint cimport int32_t, int64_t cdef class StateSampler(object): """Tracks time spent in states during pipeline execution.""" cdef int _sampling_period_ms + cdef int _sampling_period_ms_start + cdef double _sampling_period_ratio cdef list scoped_states_by_index diff --git a/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx b/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx index f775c4f..fdf4969 100644 --- a/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx +++ b/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx @@ -71,24 +71,6 @@ cdef inline int64_t get_nsec_time() nogil: cdef class StateSampler(object): """Tracks time spent in states during pipeline execution.""" - cdef int _sampling_period_ms - cdef int _sampling_period_ms_start - cdef double _sampling_period_ratio - - cdef list scoped_states_by_index - - cdef public bint started - cdef public bint finished - cdef object sampling_thread - - # This lock guards members that are shared between threads, specificaly - # finished, scoped_states_by_index, and the nsecs field of each state therein. - cdef pythread.PyThread_type_lock lock - - cdef public int64_t state_transition_count - cdef public int64_t time_since_transition - - cdef int32_t current_state_index def __init__(self, sampling_period_ms, -- To stop receiving notification emails like this one, please contact [email protected].
