[BEAM-2732] Starting refactor of state tracking in Python by pabloem · Pull Request #4375 · apache/beam

pabloem · 2018-01-09T22:05:23Z

The goal for BEAM-2732 is to refactor the context trackers in the Python SDK so that they will all use the same mechanism.
Currently, Metrics, Logging and StateSampler keep their own contexts. BEAM-2732 aims to have all of them rely on the logic in StateSampler to keep their context (this is already the case in Java).

This PR does the following:

Clean up legacy code in the StateSampler.
Adds a global per-thread variable for state sampler, so each execution thread will be able to access the current state sampler from this global variable (this will then be used for LoggingContext, and MetricsEnvironment.current_container).
Gives the Python-only state sampler functionality to track context, so that non-Cythonized runners can rely on the StateSampler to track context.

…e Python-only state sampler full functionality.

pabloem · 2018-01-09T22:30:42Z

r: @robertwb

pabloem · 2018-01-10T19:50:37Z

Python tests passing.

pabloem · 2018-01-10T23:41:07Z

Run Python PostCommit

robertwb

Sorry for the delay, here's some initial comments.

robertwb · 2018-01-11T21:15:55Z

sdks/python/apache_beam/runners/worker/statesampler_slow_test.py

+from apache_beam.utils.counters import CounterName
+
+
+class StateSamplerSlowTest(unittest.TestCase):


It would be better to run the same test on both implementations rather than have a copy. (Possibly conditionally test those parts that aren't implemented of course.)

robertwb · 2018-01-11T21:15:55Z

sdks/python/apache_beam/runners/worker/statesampler_slow.py

+DEFAULT_SAMPLING_PERIOD_MS = 200
+
+
+class StateSampler(object):


If we're going to provide a full implementation here, let's make a common base class and put anything that can be shared there. (Possibly the default sampling period as well.)

robertwb · 2018-01-11T21:15:56Z

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

-    # the worker.
-    # We stop using prefixes with included dash.
-    self.prefix = prefix[:-1] if prefix[-1] == '-' else prefix
+    EXECUTION_STATE_SAMPLERS.set_sampler(self)


That the act of creating a StateSampler should not modify global state.

robertwb · 2018-01-11T21:15:56Z

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

+    self._current_sampler = sampler
+
+
+EXECUTION_STATE_SAMPLERS = ExecutionStateSamplers()


I'd really like to avoid more proliferation of global variables. This is somewhat needed for counters (as the user invokes counter operations without access to the underlying state) but we should be able to pass this state around explicitly.

pabloem · 2018-01-12T06:01:45Z

Run Python PostCommit

pabloem

Refactored the change. Let me know what you think.

pabloem · 2018-01-12T06:02:59Z

sdks/python/apache_beam/runners/worker/statesampler_slow_test.py

+from apache_beam.utils.counters import CounterName
+
+
+class StateSamplerSlowTest(unittest.TestCase):


robertwb wrote:
It would be better to run the same test on both implementations rather than have a copy. (Possibly conditionally test those parts that aren't implemented of course.)

Acknowledged.

pabloem · 2018-01-12T06:02:59Z

sdks/python/apache_beam/runners/worker/statesampler_slow.py

+DEFAULT_SAMPLING_PERIOD_MS = 200
+
+
+class StateSampler(object):


robertwb wrote:
If we're going to provide a full implementation here, let's make a common base class and put anything that can be shared there. (Possibly the default sampling period as well.)

I've restructured it into a class hierarchy. Let me know what you think.

pabloem · 2018-01-12T06:03:00Z

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

-    # the worker.
-    # We stop using prefixes with included dash.
-    self.prefix = prefix[:-1] if prefix[-1] == '-' else prefix
+    EXECUTION_STATE_SAMPLERS.set_sampler(self)


robertwb wrote:
That the act of creating a StateSampler should not modify global state.

Added a register function.

pabloem · 2018-01-12T06:03:00Z

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

+    self._current_sampler = sampler
+
+
+EXECUTION_STATE_SAMPLERS = ExecutionStateSamplers()


robertwb wrote:
I'd really like to avoid more proliferation of global variables. This is somewhat needed for counters (as the user invokes counter operations without access to the underlying state) but we should be able to pass this state around explicitly.

I'll document that we want to pass a tracker around as much as possible - but the aim of this global variable is to be able remove global state within MetricsEnvironment, and PerThreadLoggingContext, and have this be the canonical global state.

How does that sound?

pabloem · 2018-01-12T06:31:49Z

Run Python PostCommit

pabloem · 2018-01-12T18:21:07Z

@robertwb - I've created a class hierarchy. Passes Python PostCommit, and PreCommit.

robertwb · 2018-01-16T21:50:58Z

sdks/python/apache_beam/runners/portability/fn_api_runner_test.py

-  from apache_beam.runners.worker.statesampler import DEFAULT_SAMPLING_PERIOD_MS
-except ImportError:
-  DEFAULT_SAMPLING_PERIOD_MS = 0
+DEFAULT_SAMPLING_PERIOD_MS = 0


Nit: single assignment is easier to reason about. Put this into an else clause.

robertwb · 2018-01-16T21:57:02Z

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

-    self.counter_factory = counter_factory
-    self.sampling_period_ms = sampling_period_ms
+  def __init__(self, *args):
+    #TODO(pabloem): Figure out how to pass arguments without errors.


What errors were you getting?

robertwb · 2018-01-17T21:51:21Z

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


  cdef public int64_t state_transition_count
-  cdef int64_t time_since_transition
+  cdef int64_t _time_since_transition


Why this change? Or should we be changing all the others as well for consistency?

robertwb · 2018-01-17T21:52:48Z

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

@@ -190,60 +158,28 @@ cdef class StateSampler(object):
    if self.started and not self.finished:


Is this method also common to both?

robertwb · 2018-01-17T21:55:16Z

sdks/python/apache_beam/utils/counters.py


  def __str__(self):
-    return '%s' % self._str_internal()
+    return '<CounterName<%s> at %s>' % (self._str_internal(), hex(id(self)))


__str__ defaults to __repr__, if you want them to be the same remove this one.

robertwb · 2018-01-17T21:57:27Z

sdks/python/apache_beam/runners/worker/statesampler_test.py

-      # pylint: disable=global-variable-not-assigned
-      global statesampler
-      from apache_beam.runners.worker import statesampler
+      # pylint: disable=unused-variable


You could also put this method in a

@classmethod def setUpClass(cls): ...

robertwb · 2018-01-17T22:00:44Z

sdks/python/apache_beam/runners/worker/statesampler.py

+    self._current_sampler = sampler
+
+
+EXECUTION_STATE_SAMPLERS = ExecutionStateSamplers()


Please get rid of this unused global state.

robertwb · 2018-01-18T01:42:01Z

Let's not add the global variable until it's actually used, in which case we can weigh the pros and cons.

pabloem

robertwb wrote:
Let's not add the global variable until it's actually used, in which case we can weigh the pros and cons.

Removed the local variable. * I mean global.

pabloem · 2018-01-18T02:01:39Z

sdks/python/apache_beam/runners/portability/fn_api_runner_test.py

-  from apache_beam.runners.worker.statesampler import DEFAULT_SAMPLING_PERIOD_MS
-except ImportError:
-  DEFAULT_SAMPLING_PERIOD_MS = 0
+DEFAULT_SAMPLING_PERIOD_MS = 0


robertwb wrote:
Nit: single assignment is easier to reason about. Put this into an else clause.

Done.

pabloem · 2018-01-18T02:01:39Z

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

@@ -190,60 +158,28 @@ cdef class StateSampler(object):
    if self.started and not self.finished:


robertwb wrote:
Is this method also common to both?

You're right. Done.

pabloem · 2018-01-18T02:01:39Z

sdks/python/apache_beam/utils/counters.py


  def __str__(self):
-    return '%s' % self._str_internal()
+    return '<CounterName<%s> at %s>' % (self._str_internal(), hex(id(self)))


robertwb wrote:
__str__ defaults to __repr__, if you want them to be the same remove this one.

Done.

pabloem · 2018-01-18T02:01:40Z

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


  cdef public int64_t state_transition_count
-  cdef int64_t time_since_transition
+  cdef int64_t _time_since_transition


robertwb wrote:
Why this change? Or should we be changing all the others as well for consistency?

I had thought this was only c-accessible. I've kept it consistent.

pabloem · 2018-01-18T02:01:40Z

sdks/python/apache_beam/runners/worker/statesampler.py

+    self._current_sampler = sampler
+
+
+EXECUTION_STATE_SAMPLERS = ExecutionStateSamplers()


robertwb wrote:
Please get rid of this unused global state.

I plan to use the global state to track metrics and logging (and eventually remove the old style metrics context). I can add this now, or on the next PR. What do you think?
Next PR: https://github.com/apache/beam/pull/4387/files#diff-89072ff532dd15a0af899957de4c26f3R152

pabloem · 2018-01-18T02:01:40Z

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

-    self.counter_factory = counter_factory
-    self.sampling_period_ms = sampling_period_ms
+  def __init__(self, *args):
+    #TODO(pabloem): Figure out how to pass arguments without errors.


robertwb wrote:
What errors were you getting?

hmmm this is fixed now.

pabloem · 2018-01-18T18:00:30Z

Run Python PostCommit

robertwb

Thanks. LGTM.

robertwb · 2018-01-18T19:34:48Z

sdks/python/apache_beam/runners/worker/statesampler_test.py

-      from apache_beam.runners.worker import statesampler
+      # pylint: disable=unused-variable
+      from apache_beam.runners.worker import statesampler_fast
+      cls.slow_sampler = False


Actually, instead of duplicating the logic, perhaps just query state_sampler.FAST_SAMPLER directly below.

pabloem · 2018-01-18T21:24:46Z

Fixed, and done.

robertwb · 2018-01-18T21:38:45Z

Thanks. Non-Java runs are green. Merging.

Also giving the Python-only state sampler full functionality.

Adding a global per-thread variable for state sampler. Also giving th…

bb69e90

…e Python-only state sampler full functionality.

Fixing lint issues

4f9a58f

pabloem mentioned this pull request Jan 10, 2018

[BEAM-2732] Metrics rely on statesampler state #4387

Merged

pabloem mentioned this pull request Jan 11, 2018

[BEAM-2732] Logging relies on statesampler state #4395

Closed

robertwb reviewed Jan 11, 2018

View reviewed changes

Redesigning StateSampler class structure.

ef8ce93

pabloem commented Jan 12, 2018

View reviewed changes

Fixing imports.

995d67b

robertwb reviewed Jan 17, 2018

View reviewed changes

pabloem commented Jan 18, 2018

View reviewed changes

pabloem force-pushed the global-sampler-var branch from 7323851 to 5c30712 Compare January 18, 2018 02:03

robertwb approved these changes Jan 18, 2018

View reviewed changes

Addressing comments

d3d0cf6

pabloem force-pushed the global-sampler-var branch from 17b329e to d3d0cf6 Compare January 18, 2018 21:19

robertwb merged commit 24e6bf8 into apache:master Jan 18, 2018

pabloem deleted the global-sampler-var branch January 18, 2018 21:40

boyuanzz pushed a commit to boyuanzz/beam that referenced this pull request Feb 6, 2018

[BEAM-2732] Starting refactor of state tracking in Python (apache#4375)

00e86fb

Also giving the Python-only state sampler full functionality.

		from apache_beam.utils.counters import CounterName


		class StateSamplerSlowTest(unittest.TestCase):

		self._current_sampler = sampler


		EXECUTION_STATE_SAMPLERS = ExecutionStateSamplers()

		@@ -190,60 +158,28 @@ cdef class StateSampler(object):
		if self.started and not self.finished:

Conversation

pabloem commented Jan 9, 2018

Uh oh!

pabloem commented Jan 9, 2018

Uh oh!

pabloem commented Jan 10, 2018

Uh oh!

pabloem commented Jan 10, 2018

Uh oh!

robertwb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pabloem commented Jan 12, 2018

Uh oh!

pabloem left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pabloem commented Jan 12, 2018

Uh oh!

pabloem commented Jan 12, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

robertwb commented Jan 18, 2018

Uh oh!

pabloem left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pabloem commented Jan 18, 2018

Uh oh!

robertwb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pabloem commented Jan 18, 2018

Uh oh!

robertwb commented Jan 18, 2018

pabloem commented Jan 12, 2018 •

edited

Loading

pabloem left a comment •

edited

Loading