[BEAM-8618] Tear down unused DoFns periodically in Python SDK harness. #10655

sunjincheng121 · 2020-01-22T05:38:55Z

Tear down unused DoFns periodically in Python SDK harness.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Gearpump	Samza	Spark
Go		---	---	---	---
Java
Python		---		---	---
XLang	---	---	---	---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

sunjincheng121 · 2020-01-22T05:39:38Z

R: @robertwb @lukecwik

sunjincheng121 · 2020-01-27T10:18:09Z

R: @mxm @lukecwik

mxm · 2020-01-28T14:00:02Z

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

@@ -280,6 +283,7 @@ def get(self, instruction_id, bundle_descriptor_id):
    try:
      # pop() is threadsafe
      processor = self.cached_bundle_processors[bundle_descriptor_id].pop()
+      self.last_access_time[bundle_descriptor_id] = time.time()
    except IndexError:


This won't update the access time when we first create the processor in the except block.

This is expected. The last_access_time represents the last time the cached_bundle_processors for some bundle_descriptor is accessed. If it exceeds the time limit, the remaining bundle processors cached in the cached_bundle_processors will be shutdown. What's your thought?

In my eyes, the first time it is accessed is when it is created.

The strategy here is that the bundle processors which are unused after an amount of time will be shutdown. When a bundle processor is created in the exception block, there are no cached(unused) bundle processors. The bundle processors become unused only when they are added to the cached bundle processors list. What about rename last_access_time to cached_bundle_processors_last_access_time to make it more clear?

I still do not understand, the method is named get, so we access the bundle processor independently of whether we create it or not. It is cached, regardless of whether it is created and added to the cache, or retrieved from the cache.

Logically, you might want to update the time when putting the processor into the cache. That would be in release.

What is the advantage of updating the time here? It should be sufficient to update it in release, directly before putting it back.

Hmm, If a bundle processor is retrieved from the cache, there is a high possibility that the remaining cached bundle processors will be needed in the future and so the last access time is updated.

If the bundle processor is newly created, it means that the cached bundle processor list is empty. This is the main reason that the last access time is only updated when bundle processor is retrieved from the cache. However, I think that it does no harm to update the last access time in both cases if it makes the code more readable.

What do you think?

If the bundle processor is newly created, it means that the cached bundle processor list is empty. This is the main reason that the last access time is only updated when bundle processor is retrieved from the cache.

Consider the case where we just have a single bundle processor. When we call get for the first time, we won't update the last-used time. However, every time we retrieve it afterwards, we will update the time, but the list of cached bundle processors will remain empty.

I think we should either (1) always update the last-used timestamp in get, regardless of creation or (2) update it only on release.

I'm leaning towards (2) because while a bundle processor is in-use, it can't be removed anyways. We update the timestamp when we put it back in release.

Regarding the single bundle processor case, it doesn't harm to update the time as the cached bundle processors is empty. However, in cases where there are multiple bundle processors, it will update the time for the remaining cached bundle processors and so improve the cache hit rate. I think this is main difference between solution (1) and (2). However, I'm fine with both solutions as I think both of them work. Will update the PR according to solution (2) if you are favor of it according to your experience.

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

sunjincheng121 · 2020-01-31T11:44:03Z

Thanks for the review and valuable comments. I have address the comments and I appreciate if you can have another look :) @mxm

mxm

LGTM. Thank you for addressing the comments.

mxm · 2020-01-31T13:41:12Z

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

+      for descriptor_id, last_access_time in \
+              self.cached_bundle_processors_last_access_time.items():
+        if time.time() - last_access_time > \
+                DEFAULT_BUNDLE_PROCESSOR_CACHE_THRESHOLD_S:


nit: you could surround the condition with paranthesis which would make it easier to read, e.g.

if (time.time() - last_access_time > DEFAULT_BUNDLE_PROCESSOR_CACHE_THRESHOLD_S):

mxm · 2020-01-31T13:42:16Z

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

@@ -69,6 +71,8 @@
 # 5 minutes * 60 seconds * 1020 millis * 1000 micros * 1000 nanoseconds
 DEFAULT_LOG_LULL_TIMEOUT_NS = 5 * 60 * 1000 * 1000 * 1000

+DEFAULT_BUNDLE_PROCESSOR_CACHE_THRESHOLD_S = 60


Suggested change

DEFAULT_BUNDLE_PROCESSOR_CACHE_THRESHOLD_S = 60

DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S = 60

mxm · 2020-01-31T13:42:31Z

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

+  def _schedule_periodic_shutdown(self):
+    def shutdown_inactive_bundle_processors():
+      for descriptor_id, last_access_time in \
+              self.cached_bundle_processors_last_access_time.items():


Same here, use parenthesis.

It seems that it doesn't support parenthesis in the for loop？

It should, there are many examples of for loops using parentheses in the code base already.

Ah, I have added a local variable and it should have no this problem now. Is that make sense to you?

kamilwu · 2020-02-06T12:30:44Z

Just to let you know that we've just introduced Python autoformatter. Your merge conflict might be a result of this.
Here you can find an instruction on how to run autoformatter: https://cwiki.apache.org/confluence/display/BEAM/Python+Tips, section Formatting.
Sorry for inconvenience.

sunjincheng121 · 2020-02-07T08:08:04Z

squash the commits and rebase the code.

sunjincheng121 · 2020-02-10T08:10:25Z

R: @lukecwik @mxm

mxm · 2020-02-24T10:50:06Z

retest this please

mxm

Linting needs to be fixed. See comment from @kamilwu.

sunjincheng121 · 2020-02-25T06:44:54Z

Run PythonLint PreCommit

mxm · 2020-02-25T09:01:57Z

08:17:21 > Task :sdks:python:test-suites:tox:py2:lintPy27_3
08:17:21 ************* Module apache_beam.runners.worker.sdk_worker
08:17:21 W:372,14: dict.items referenced when not iterating (dict-items-not-iterating)

https://builds.apache.org/job/beam_PreCommit_PythonLint_Phrase/70/

…harness.

sunjincheng121 · 2020-02-25T09:25:17Z

Thanks for the info @mxm , I have fix the style issue, and update the PR :)

sunjincheng121 force-pushed the BEAM-8618-PR branch from 99d025b to 94d67c7 Compare January 22, 2020 09:11

mxm reviewed Jan 28, 2020

View reviewed changes

mxm approved these changes Jan 31, 2020

View reviewed changes

[BEAM-8618] Tear down unused DoFns periodically in Python SDK harness.

8d97572

sunjincheng121 force-pushed the BEAM-8618-PR branch from d3e1b4d to 8d97572 Compare February 7, 2020 08:06

mxm reviewed Feb 24, 2020

View reviewed changes

fixup! [BEAM-8618] Tear down unused DoFns periodically in Python SDK …

4c288d1

…harness.

probot-autolabeler bot added the python label Feb 25, 2020

robertwb merged commit 4a25aa0 into apache:master Feb 27, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-8618] Tear down unused DoFns periodically in Python SDK harness. #10655

[BEAM-8618] Tear down unused DoFns periodically in Python SDK harness. #10655

sunjincheng121 commented Jan 22, 2020

sunjincheng121 commented Jan 22, 2020

sunjincheng121 commented Jan 27, 2020

mxm Jan 28, 2020

sunjincheng121 Jan 29, 2020

mxm Jan 29, 2020

sunjincheng121 Jan 29, 2020

mxm Jan 30, 2020

sunjincheng121 Jan 30, 2020

mxm Jan 31, 2020

sunjincheng121 Jan 31, 2020 •

edited

Loading

sunjincheng121 commented Jan 31, 2020

mxm left a comment

mxm Jan 31, 2020

mxm Jan 31, 2020

mxm Jan 31, 2020

sunjincheng121 Feb 1, 2020

lukecwik Feb 5, 2020

sunjincheng121 Feb 7, 2020

kamilwu commented Feb 6, 2020

sunjincheng121 commented Feb 7, 2020

sunjincheng121 commented Feb 10, 2020

mxm commented Feb 24, 2020

mxm left a comment

sunjincheng121 commented Feb 25, 2020

mxm commented Feb 25, 2020

sunjincheng121 commented Feb 25, 2020

	DEFAULT_BUNDLE_PROCESSOR_CACHE_THRESHOLD_S = 60
	DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S = 60

[BEAM-8618] Tear down unused DoFns periodically in Python SDK harness. #10655

[BEAM-8618] Tear down unused DoFns periodically in Python SDK harness. #10655

Conversation

sunjincheng121 commented Jan 22, 2020

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

sunjincheng121 commented Jan 22, 2020

sunjincheng121 commented Jan 27, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sunjincheng121 Jan 31, 2020 • edited Loading

Choose a reason for hiding this comment

sunjincheng121 commented Jan 31, 2020

mxm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kamilwu commented Feb 6, 2020

sunjincheng121 commented Feb 7, 2020

sunjincheng121 commented Feb 10, 2020

mxm commented Feb 24, 2020

mxm left a comment

Choose a reason for hiding this comment

sunjincheng121 commented Feb 25, 2020

mxm commented Feb 25, 2020

sunjincheng121 commented Feb 25, 2020

sunjincheng121 Jan 31, 2020 •

edited

Loading