[BEAM-8944] Change to use single thread in py sdk bundle progress report

[y1chi]

Use UnboundedThreadWorkerExecutor in bundle progress report introduces performance complications (by using too many threads. Experiment shows that a typical streaming wordcount pipeline which can deal with 360 messages/(second, worker) is only capable of process around 200 messages/(second, worker) with same resource. This PR aim to mitigate the performance and cost change of existing pipeline by change the use of UnboundedThreadWorkerExecutor in bundle progress report to just single thread executor.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Flink	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.

[y1chi]

R: @angoenka @lukecwik

[lukecwik]

Would it be better to have the runner request progress less frequently?

[y1chi]

Would it be better to have the runner request progress less frequently?

I think that helps too. I believe right now JRH requests every 0.1 sec. Not exactly sure how the frequency is picked.

[y1chi]

Run Python PreCommit

[lukecwik]

Would it be better to have the runner request progress less frequently?

I think that helps too. I believe right now JRH requests every 0.1 sec. Not exactly sure how the frequency is picked.

0.1 secs is a lot and doesn't seem right.

[y1chi]

Would it be better to have the runner request progress less frequently?

I think that helps too. I believe right now JRH requests every 0.1 sec. Not exactly sure how the frequency is picked.

0.1 secs is a lot and doesn't seem right.

I think this is where it is set, we can try to tune that down.
https://github.com/apache/beam/blob/master/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/fn/control/BeamFnMapTaskExecutor.java#L296

Single thread should be good enough for progress report in python sdk harness and shouldn't incur stuckness issues, and it could also help limiting its impact on the bundle processing critical path.

[lukecwik]

Would it be better to have the runner request progress less frequently?

I think that helps too. I believe right now JRH requests every 0.1 sec. Not exactly sure how the frequency is picked.

0.1 secs is a lot and doesn't seem right.

I think this is where it is set, we can try to tune that down.
https://github.com/apache/beam/blob/master/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/fn/control/BeamFnMapTaskExecutor.java#L296

Single thread should be good enough for progress report in python sdk harness and shouldn't incur stuckness issues, and it could also help limiting its impact on the bundle processing critical path.

Yeah, thats the wrong constant to use since that is meant to be used in a tight read loop using a lock which is inappropriate for API calls. I believe the Dataflow service has used 30 seconds for progress updates between the server and the worker so we could do anywhere between 5 and 30 seconds and still be fine.

[y1chi]

Would it be better to have the runner request progress less frequently?

I think that helps too. I believe right now JRH requests every 0.1 sec. Not exactly sure how the frequency is picked.

0.1 secs is a lot and doesn't seem right.

I think this is where it is set, we can try to tune that down.
https://github.com/apache/beam/blob/master/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/fn/control/BeamFnMapTaskExecutor.java#L296
Single thread should be good enough for progress report in python sdk harness and shouldn't incur stuckness issues, and it could also help limiting its impact on the bundle processing critical path.

Yeah, thats the wrong constant to use since that is meant to be used in a tight read loop using a lock which is inappropriate for API calls. I believe the Dataflow service has used 30 seconds for progress updates between the server and the worker so we could do anywhere between 5 and 30 seconds and still be fine.

it seems that it will slow down the bundle processing scheduling if I increase the period and the pipeline can become super slow. I'm guessing it will affect the time when a bundle process instruction can be considered finished.

[lukecwik]

Would it be better to have the runner request progress less frequently?

I think that helps too. I believe right now JRH requests every 0.1 sec. Not exactly sure how the frequency is picked.

0.1 secs is a lot and doesn't seem right.

I think this is where it is set, we can try to tune that down.
https://github.com/apache/beam/blob/master/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/fn/control/BeamFnMapTaskExecutor.java#L296
Single thread should be good enough for progress report in python sdk harness and shouldn't incur stuckness issues, and it could also help limiting its impact on the bundle processing critical path.

Yeah, thats the wrong constant to use since that is meant to be used in a tight read loop using a lock which is inappropriate for API calls. I believe the Dataflow service has used 30 seconds for progress updates between the server and the worker so we could do anywhere between 5 and 30 seconds and still be fine.

it seems that it will slow down the bundle processing scheduling if I increase the period and the pipeline can become super slow. I'm guessing it will affect the time when a bundle process instruction can be considered finished.

Progress updates are independent of the bundle processing and it finishing since those are separate messages over the control channel so updating the progress update interval will only impact how "fresh" the progress data. See this call flow for processing a bundle and this call flow for progress updates.

[y1chi]

Then I guess delayed update on

beam/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/fn/control/BeamFnMapTaskExecutor.java

Line 368 in 6a63c06

grpcWriteOperationElementsProcessed.accept((int) elementsConsumed);

might have a throttling effect?

I've tried 2, 3, 5 secs, and the performance looks pretty poor for any of these periods.

[lukecwik]

Then I guess delayed update on

beam/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/fn/control/BeamFnMapTaskExecutor.java

Line 368 in 6a63c06

grpcWriteOperationElementsProcessed.accept((int) elementsConsumed);

might have a throttling effect?
I've tried 2, 3, 5 secs, and the performance looks pretty poor for any of these periods.

I see that there is throttling going on there but it is unnecessary now. The SDK is responsible for performing push back. The server needs to be configured to not use too much buffer space. At this point in time we may just want to leave this as is and wait for the UW to be released since it doesn't rely on this 0.1 second polling interval or go with your original plan of using a single thread with a TODO to remove it.

WDYT?

[y1chi]

Then I guess delayed update on

beam/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/fn/control/BeamFnMapTaskExecutor.java

Line 368 in 6a63c06

grpcWriteOperationElementsProcessed.accept((int) elementsConsumed);

might have a throttling effect?
I've tried 2, 3, 5 secs, and the performance looks pretty poor for any of these periods.

I see that there is throttling going on there but it is unnecessary now. The SDK is responsible for performing push back. The server needs to be configured to not use too much buffer space. At this point in time we may just want to leave this as is and wait for the UW to be released since it doesn't rely on this 0.1 second polling interval or go with your original plan of using a single thread with a TODO to remove it.

WDYT?

Then I guess delayed update on

beam/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/fn/control/BeamFnMapTaskExecutor.java

Line 368 in 6a63c06

grpcWriteOperationElementsProcessed.accept((int) elementsConsumed);

might have a throttling effect?
I've tried 2, 3, 5 secs, and the performance looks pretty poor for any of these periods.

I see that there is throttling going on there but it is unnecessary now. The SDK is responsible for performing push back. The server needs to be configured to not use too much buffer space. At this point in time we may just want to leave this as is and wait for the UW to be released since it doesn't rely on this 0.1 second polling interval or go with your original plan of using a single thread with a TODO to remove it.

WDYT?

I fully agree that we should try to remove the throttling and lower the frequency in JRH, I opened https://issues.apache.org/jira/browse/BEAM-8998, meanwhile I think it's still better we use a single thread as a intermediate solution since user may take sometime to adopt UW. Also removing the throttling and lowering the frequency in JRH seems a bigger change, so needs sometime to test.
Also added TODO.

[angoenka]

Thanks!
LGTM too.