Adding flag to mask change

Melodylican · Jun 15, 2018 · 9cdbbe0 · 9cdbbe0
1 parent 22f5a81
commit 9cdbbe0
Show file tree

Hide file tree

Showing 5 changed files with 6 additions and 22 deletions.
diff --git a/sdks/python/apache_beam/runners/worker/operations.py b/sdks/python/apache_beam/runners/worker/operations.py
@@ -325,7 +325,7 @@ def _read_side_inputs(self, tags_and_types):
         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
@@ -80,7 +80,7 @@ def __init__(self,
   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 @@ def _reader_thread(self):
               # 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
@@ -92,7 +92,7 @@ def test_bytes_read_behind_experiment(self):
 
   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
@@ -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
@@ -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,