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 860d73eb66a71260e0288aeb7f2117864f813eaf Author: Pablo <[email protected]> AuthorDate: Mon May 7 13:51:10 2018 -0700 Improving state sampler Cython annotations. --- .../apache_beam/runners/worker/opcounters.pxd | 3 +- .../runners/worker/statesampler_fast.pxd | 59 ++++++++++++++++++++++ .../runners/worker/statesampler_fast.pyx | 10 +--- 3 files changed, 62 insertions(+), 10 deletions(-) diff --git a/sdks/python/apache_beam/runners/worker/opcounters.pxd b/sdks/python/apache_beam/runners/worker/opcounters.pxd index 0bcd428..1d7f296 100644 --- a/sdks/python/apache_beam/runners/worker/opcounters.pxd +++ b/sdks/python/apache_beam/runners/worker/opcounters.pxd @@ -19,13 +19,14 @@ cimport cython cimport libc.stdint from apache_beam.utils.counters cimport Counter +from apache_beam.runners.worker cimport statesampler_fast cdef class TransformIOCounter(object): cdef readonly object _counter_factory cdef readonly object _state_sampler cdef Counter bytes_read_counter - cdef object scoped_state + cdef statesampler_fast.ScopedState scoped_state cdef object _latest_step cpdef update_current_step(self) diff --git a/sdks/python/apache_beam/runners/worker/statesampler_fast.pxd b/sdks/python/apache_beam/runners/worker/statesampler_fast.pxd new file mode 100644 index 0000000..a808a8e --- /dev/null +++ b/sdks/python/apache_beam/runners/worker/statesampler_fast.pxd @@ -0,0 +1,59 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +cimport cython + +from apache_beam.metrics.execution cimport MetricsContainer + +from cpython cimport pythread +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 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 + + cpdef _scoped_state(self, counter_name, output_counter, metrics_container) + +cdef class ScopedState(object): + """Context manager class managing transitions for a given sampler state.""" + + cdef readonly StateSampler sampler + cdef readonly int32_t state_index + cdef readonly object counter + cdef readonly object name + cdef readonly int64_t _nsecs + cdef int32_t old_state_index + cdef readonly MetricsContainer _metrics_container + + cpdef __enter__(self) + + cpdef __exit__(self, unused_exc_type, unused_exc_value, unused_traceback) diff --git a/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx b/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx index 67d9cc8..f775c4f 100644 --- a/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx +++ b/sdks/python/apache_beam/runners/worker/statesampler_fast.pyx @@ -172,7 +172,7 @@ cdef class StateSampler(object): return self.scoped_states_by_index[self.current_state_index] cpdef _scoped_state(self, counter_name, output_counter, - metrics_container=None): + metrics_container): """Returns a context manager managing transitions for a given state. Args: counter_name: A CounterName object with information about the execution @@ -201,14 +201,6 @@ cdef class StateSampler(object): cdef class ScopedState(object): """Context manager class managing transitions for a given sampler state.""" - cdef readonly StateSampler sampler - cdef readonly int32_t state_index - cdef readonly object counter - cdef readonly object name - cdef readonly int64_t _nsecs - cdef int32_t old_state_index - cdef readonly MetricsContainer _metrics_container - def __init__( self, sampler, name, state_index, counter=None, metrics_container=None): self.sampler = sampler -- To stop receiving notification emails like this one, please contact [email protected].
