You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "lrakla (via GitHub)" <gi...@apache.org> on 2023/10/24 18:23:35 UTC
[PR] Fixes dictionary size change in direct runners [beam]
lrakla opened a new pull request, #29122:
URL: https://github.com/apache/beam/pull/29122
Addresses #27501 by adding a `self._lock` in `shutdown_inactive_bundle_processors` similar to other functions in sdk_worker.py as suggested by @tvalentyn
Before
``` python
def shutdown_inactive_bundle_processors():
# type: () -> None
for descriptor_id, last_access_time in self.last_access_times.items():
if (time.time() - last_access_time >
DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
BundleProcessorCache._shutdown_cached_bundle_processors(
self.cached_bundle_processors[descriptor_id])
```
After
``` python
def shutdown_inactive_bundle_processors():
# type: () -> None
with self._lock:
for descriptor_id, last_access_time in self.last_access_times.items():
if (time.time() - last_access_time >
DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
BundleProcessorCache._shutdown_cached_bundle_processors(
self.cached_bundle_processors[descriptor_id])
```
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn merged PR #29122:
URL: https://github.com/apache/beam/pull/29122
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #29122:
URL: https://github.com/apache/beam/pull/29122#issuecomment-1777823391
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/29122?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#29122](https://app.codecov.io/gh/apache/beam/pull/29122?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (ebecc73) into [master](https://app.codecov.io/gh/apache/beam/commit/0d95b4b400e94b2d47303c336852bb3192cf53b1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (0d95b4b) will **decrease** coverage by `0.01%`.
> Report is 2 commits behind head on master.
> The diff coverage is `0.00%`.
```diff
@@ Coverage Diff @@
## master #29122 +/- ##
==========================================
- Coverage 38.36% 38.36% -0.01%
==========================================
Files 686 686
Lines 101689 101690 +1
==========================================
Hits 39018 39018
- Misses 61091 61092 +1
Partials 1580 1580
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/29122/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [python](https://app.codecov.io/gh/apache/beam/pull/29122/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `29.98% <0.00%> (-0.01%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files](https://app.codecov.io/gh/apache/beam/pull/29122?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...ks/python/apache\_beam/runners/worker/sdk\_worker.py](https://app.codecov.io/gh/apache/beam/pull/29122?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvc2RrX3dvcmtlci5weQ==) | `78.56% <0.00%> (-0.13%)` | :arrow_down: |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "lrakla (via GitHub)" <gi...@apache.org>.
lrakla commented on code in PR #29122:
URL: https://github.com/apache/beam/pull/29122#discussion_r1373544136
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -583,11 +583,12 @@ def _schedule_periodic_shutdown(self):
# type: () -> None
def shutdown_inactive_bundle_processors():
# type: () -> None
- for descriptor_id, last_access_time in self.last_access_times.items():
- if (time.time() - last_access_time >
- DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
- BundleProcessorCache._shutdown_cached_bundle_processors(
- self.cached_bundle_processors[descriptor_id])
+ with self._lock:
Review Comment:
@AnandInguva I will try it out too, and make the change @tvalentyn suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #29122:
URL: https://github.com/apache/beam/pull/29122#discussion_r1382034441
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -583,11 +583,12 @@ def _schedule_periodic_shutdown(self):
# type: () -> None
def shutdown_inactive_bundle_processors():
# type: () -> None
- for descriptor_id, last_access_time in self.last_access_times.items():
- if (time.time() - last_access_time >
- DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
- BundleProcessorCache._shutdown_cached_bundle_processors(
- self.cached_bundle_processors[descriptor_id])
+ with self._lock:
Review Comment:
@AnandInguva did you also try lowering DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S to some small value (0.1) ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #29122:
URL: https://github.com/apache/beam/pull/29122#discussion_r1370750624
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -583,11 +583,12 @@ def _schedule_periodic_shutdown(self):
# type: () -> None
def shutdown_inactive_bundle_processors():
# type: () -> None
- for descriptor_id, last_access_time in self.last_access_times.items():
- if (time.time() - last_access_time >
- DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
- BundleProcessorCache._shutdown_cached_bundle_processors(
- self.cached_bundle_processors[descriptor_id])
+ with self._lock:
Review Comment:
It is a good idea to test the changes and yes it may be tricky to test this one.
According to code coverage report on this PR, this codepath might not be covered in unit tests, but would be triggered in long-running pipelines. To test locally, it should be possible to adjust a https://github.com/apache/beam/blob/0f7698143261271cf0fdb527b833e73997f10748/sdks/python/apache_beam/runners/worker/sdk_worker.py#L82 to a lower value, that might help reproducing the original issue.
As for test, we could decrease the value via `unittest.mock.patch` inside a unit test that exercises an SDK worker. @AnandInguva feel free to TAL and suggest a test that would be helpful. We could modify this param in an existing FnAPI runner test scenario and verify that this codepath is triggered.
Also, I would move up the lock definition in. https://github.com/apache/beam/blob/0f7698143261271cf0fdb527b833e73997f10748/sdks/python/apache_beam/runners/worker/sdk_worker.py#L442 a few lines above before we schedule this thread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "AnandInguva (via GitHub)" <gi...@apache.org>.
AnandInguva commented on code in PR #29122:
URL: https://github.com/apache/beam/pull/29122#discussion_r1376409808
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -583,11 +583,12 @@ def _schedule_periodic_shutdown(self):
# type: () -> None
def shutdown_inactive_bundle_processors():
# type: () -> None
- for descriptor_id, last_access_time in self.last_access_times.items():
- if (time.time() - last_access_time >
- DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
- BundleProcessorCache._shutdown_cached_bundle_processors(
- self.cached_bundle_processors[descriptor_id])
+ with self._lock:
Review Comment:
I tried few simple examples with 1000, 10000, 100000 elements by setting `direct_num_worker=4` and `direct_running_mode=multi_threading`. I haven't been able to reproduced the error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "AnandInguva (via GitHub)" <gi...@apache.org>.
AnandInguva commented on code in PR #29122:
URL: https://github.com/apache/beam/pull/29122#discussion_r1370957683
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -583,11 +583,12 @@ def _schedule_periodic_shutdown(self):
# type: () -> None
def shutdown_inactive_bundle_processors():
# type: () -> None
- for descriptor_id, last_access_time in self.last_access_times.items():
- if (time.time() - last_access_time >
- DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
- BundleProcessorCache._shutdown_cached_bundle_processors(
- self.cached_bundle_processors[descriptor_id])
+ with self._lock:
Review Comment:
@lrakla ah sorry. I wasn't aware this is your first contribution. I am not sure on how to reproduce the error yet but I can try it on my local and let you know of any results and the right place to add test(if we could reproduce it).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "AnandInguva (via GitHub)" <gi...@apache.org>.
AnandInguva commented on code in PR #29122:
URL: https://github.com/apache/beam/pull/29122#discussion_r1370669270
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -583,11 +583,12 @@ def _schedule_periodic_shutdown(self):
# type: () -> None
def shutdown_inactive_bundle_processors():
# type: () -> None
- for descriptor_id, last_access_time in self.last_access_times.items():
- if (time.time() - last_access_time >
- DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
- BundleProcessorCache._shutdown_cached_bundle_processors(
- self.cached_bundle_processors[descriptor_id])
+ with self._lock:
Review Comment:
from the issue, I see this is hard to reproduce but asking anyway, did you reproduce the error by any chance? If yes, can we add a test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #29122:
URL: https://github.com/apache/beam/pull/29122#issuecomment-1777795713
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #29122:
URL: https://github.com/apache/beam/pull/29122#discussion_r1371976122
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -583,11 +583,12 @@ def _schedule_periodic_shutdown(self):
# type: () -> None
def shutdown_inactive_bundle_processors():
# type: () -> None
- for descriptor_id, last_access_time in self.last_access_times.items():
- if (time.time() - last_access_time >
- DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
- BundleProcessorCache._shutdown_cached_bundle_processors(
- self.cached_bundle_processors[descriptor_id])
+ with self._lock:
Review Comment:
Thanks, @AnandInguva . @lrakla yes, that what I meant.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on PR #29122:
URL: https://github.com/apache/beam/pull/29122#issuecomment-1852068695
@AnandInguva @tvalentyn @lrakla what are next steps here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "AnandInguva (via GitHub)" <gi...@apache.org>.
AnandInguva commented on code in PR #29122:
URL: https://github.com/apache/beam/pull/29122#discussion_r1382458038
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -583,11 +583,12 @@ def _schedule_periodic_shutdown(self):
# type: () -> None
def shutdown_inactive_bundle_processors():
# type: () -> None
- for descriptor_id, last_access_time in self.last_access_times.items():
- if (time.time() - last_access_time >
- DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
- BundleProcessorCache._shutdown_cached_bundle_processors(
- self.cached_bundle_processors[descriptor_id])
+ with self._lock:
Review Comment:
tried it with 1. I will try with 0.1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "lrakla (via GitHub)" <gi...@apache.org>.
lrakla commented on PR #29122:
URL: https://github.com/apache/beam/pull/29122#issuecomment-1777793704
R: @tvalentyn
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "lrakla (via GitHub)" <gi...@apache.org>.
lrakla commented on code in PR #29122:
URL: https://github.com/apache/beam/pull/29122#discussion_r1370940999
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -583,11 +583,12 @@ def _schedule_periodic_shutdown(self):
# type: () -> None
def shutdown_inactive_bundle_processors():
# type: () -> None
- for descriptor_id, last_access_time in self.last_access_times.items():
- if (time.time() - last_access_time >
- DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
- BundleProcessorCache._shutdown_cached_bundle_processors(
- self.cached_bundle_processors[descriptor_id])
+ with self._lock:
Review Comment:
@AnandInguva I was not sure how to reproduce it. I can take a look based on @tvalentyn comments and try reproducing it locally. This is my first contribution to the project so any guidance is appreciated.
@tvalentyn Do you mean moving the lock definition before `self._schedule_periodic_shutdown()` ? i.e. swap lines 441 and 442?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes dictionary size change in direct runners [beam]
Posted by "AnandInguva (via GitHub)" <gi...@apache.org>.
AnandInguva commented on code in PR #29122:
URL: https://github.com/apache/beam/pull/29122#discussion_r1376409808
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -583,11 +583,12 @@ def _schedule_periodic_shutdown(self):
# type: () -> None
def shutdown_inactive_bundle_processors():
# type: () -> None
- for descriptor_id, last_access_time in self.last_access_times.items():
- if (time.time() - last_access_time >
- DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
- BundleProcessorCache._shutdown_cached_bundle_processors(
- self.cached_bundle_processors[descriptor_id])
+ with self._lock:
Review Comment:
I tried few simple examples with 1000, 10000, 100000 elements by setting `direct_num_worker=4` and `direct_running_mode=multi_threading`. I haven't been able to reproduced the error.
Example: List of elements, groupByKey and print
##########
sdks/python/apache_beam/runners/worker/sdk_worker.py:
##########
@@ -583,11 +583,12 @@ def _schedule_periodic_shutdown(self):
# type: () -> None
def shutdown_inactive_bundle_processors():
# type: () -> None
- for descriptor_id, last_access_time in self.last_access_times.items():
- if (time.time() - last_access_time >
- DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S):
- BundleProcessorCache._shutdown_cached_bundle_processors(
- self.cached_bundle_processors[descriptor_id])
+ with self._lock:
Review Comment:
I tried few simple examples with 1000, 10000, 100000 elements by setting `direct_num_worker=4` and `direct_running_mode=multi_threading`. I haven't been able to reproduced the error.
Example: Create list of elements, groupByKey and print
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org