You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "scwhittle (via GitHub)" <gi...@apache.org> on 2024/05/15 15:06:05 UTC
[PR] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
scwhittle opened a new pull request, #31304:
URL: https://github.com/apache/beam/pull/31304
Make the batcher Closeable which helps users remember to flush
Improve testing of multithreaded scenarios for commits, which would have failed with ConcurrentModificationException as observed in running pipeline.
Addresses #31303
------------------------
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #31304:
URL: https://github.com/apache/beam/pull/31304#issuecomment-2112811586
Sending this for pre-review, I think I have to fix a hotkeys test still. And I would like to increase the streampool size to match # of commit threads (or perhaps will do that separately).
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #31304:
URL: https://github.com/apache/beam/pull/31304#discussion_r1602807120
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitter.java:
##########
@@ -156,15 +159,18 @@ private void streamingCommitLoop() {
try (CloseableStream<CommitWorkStream> closeableCommitStream =
commitWorkStreamFactory.get()) {
CommitWorkStream commitStream = closeableCommitStream.stream();
- if (!tryAddToCommitStream(initialCommit, commitStream)) {
- throw new AssertionError("Initial commit on flushed stream should always be accepted.");
+ try (CommitWorkStream.RequestBatcher batcher = commitStream.batcher()) {
+ if (!tryAddToCommitBatch(initialCommit, batcher)) {
+ throw new AssertionError(
+ "Initial commit on flushed stream should always be accepted.");
+ }
+ // Batch additional commits to the stream and possibly make an un-batched commit the
+ // next
Review Comment:
done
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #31304:
URL: https://github.com/apache/beam/pull/31304#discussion_r1602163902
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -315,10 +307,23 @@ private Batcher() {
this.queue = new HashMap<>();
}
- boolean canAccept(PendingRequest request) {
- return queue.isEmpty()
- || (queue.size() < streamingRpcBatchLimit
- && (request.getBytes() + queuedBytes) < AbstractWindmillStream.RPC_STREAM_CHUNK_SIZE);
+ @Override
+ public boolean commitWorkItem(
+ String computation, WorkItemCommitRequest commitRequest, Consumer<CommitStatus> onDone) {
+ PendingRequest request = new PendingRequest(computation, commitRequest, onDone);
Review Comment:
if we can't accept the `PendingRequest` we just construct this object and throw it way. How about passing `long request.getSerializedSize() + computation.length()` which is what `PendingRequest.getBytes()` returns into `canAccept`, and only creating the `PendingRequest` when we call add.
and only
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #31304:
URL: https://github.com/apache/beam/pull/31304#discussion_r1602158486
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -51,11 +51,11 @@ public final class GrpcCommitWorkStream
private static final long HEARTBEAT_REQUEST_ID = Long.MAX_VALUE;
private final Map<Long, PendingRequest> pending;
- private final Batcher batcher;
private final AtomicLong idGenerator;
private final JobHeader jobHeader;
private final ThrottleTimer commitWorkThrottleTimer;
private final int streamingRpcBatchLimit;
+ private Batcher batcher = new Batcher();
Review Comment:
can this be final
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #31304:
URL: https://github.com/apache/beam/pull/31304#issuecomment-2114847430
The hotkeys test flakiness appeared to be due to the slow shutdown of the worker due to the committers. I changed them to interrupt without 10 second wait since the threads are going to be waiting on the queue most likely, and then improved the interrupt handling.
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #31304:
URL: https://github.com/apache/beam/pull/31304#discussion_r1602795043
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/WindmillStream.java:
##########
@@ -68,21 +70,34 @@ Windmill.KeyedGetDataResponse requestKeyedData(
/** Interface for streaming CommitWorkRequests to Windmill. */
@ThreadSafe
interface CommitWorkStream extends WindmillStream {
+ @NotThreadSafe
+ interface RequestBatcher extends Closeable {
Review Comment:
Closeable (in io package) extends AutoCloseable https://docs.oracle.com/javase/8/docs/api/java/io/Closeable.html
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #31304:
URL: https://github.com/apache/beam/pull/31304#discussion_r1601830357
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/WindmillStream.java:
##########
@@ -68,21 +70,34 @@ Windmill.KeyedGetDataResponse requestKeyedData(
/** Interface for streaming CommitWorkRequests to Windmill. */
@ThreadSafe
interface CommitWorkStream extends WindmillStream {
+ @NotThreadSafe
+ interface RequestBatcher extends Closeable {
Review Comment:
I think this may need to be AutoCloseable for close() to be automatically called after try with resources block. Or else may need to explicitly call close().
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitter.java:
##########
@@ -156,15 +159,18 @@ private void streamingCommitLoop() {
try (CloseableStream<CommitWorkStream> closeableCommitStream =
commitWorkStreamFactory.get()) {
CommitWorkStream commitStream = closeableCommitStream.stream();
- if (!tryAddToCommitStream(initialCommit, commitStream)) {
- throw new AssertionError("Initial commit on flushed stream should always be accepted.");
+ try (CommitWorkStream.RequestBatcher batcher = commitStream.batcher()) {
+ if (!tryAddToCommitBatch(initialCommit, batcher)) {
+ throw new AssertionError(
+ "Initial commit on flushed stream should always be accepted.");
+ }
+ // Batch additional commits to the stream and possibly make an un-batched commit the
+ // next
Review Comment:
Nit comment formatting
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #31304:
URL: https://github.com/apache/beam/pull/31304#discussion_r1602796983
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -51,11 +51,11 @@ public final class GrpcCommitWorkStream
private static final long HEARTBEAT_REQUEST_ID = Long.MAX_VALUE;
private final Map<Long, PendingRequest> pending;
- private final Batcher batcher;
private final AtomicLong idGenerator;
private final JobHeader jobHeader;
private final ThrottleTimer commitWorkThrottleTimer;
private final int streamingRpcBatchLimit;
+ private Batcher batcher = new Batcher();
Review Comment:
Oops, actually this was leftover from testing that my unit tests would catch the lack of thread-safety. We should be creating a new batcher for each call instead of sharing the non-thread-safe batcher (similar to existing code)
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #31304:
URL: https://github.com/apache/beam/pull/31304
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #31304:
URL: https://github.com/apache/beam/pull/31304#issuecomment-2112812440
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #31304:
URL: https://github.com/apache/beam/pull/31304#discussion_r1602806310
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -305,7 +297,7 @@ long getBytes() {
}
}
- private class Batcher {
+ private class Batcher implements CommitWorkStream.RequestBatcher {
private final Map<Long, PendingRequest> queue;
Review Comment:
See other comment, batcher is not thread-safe but we shouldn't have a batcher member just vend new batchers. Was leftover from testing.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -315,10 +307,23 @@ private Batcher() {
this.queue = new HashMap<>();
}
- boolean canAccept(PendingRequest request) {
- return queue.isEmpty()
- || (queue.size() < streamingRpcBatchLimit
- && (request.getBytes() + queuedBytes) < AbstractWindmillStream.RPC_STREAM_CHUNK_SIZE);
+ @Override
+ public boolean commitWorkItem(
+ String computation, WorkItemCommitRequest commitRequest, Consumer<CommitStatus> onDone) {
+ PendingRequest request = new PendingRequest(computation, commitRequest, onDone);
Review Comment:
Done
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #31304:
URL: https://github.com/apache/beam/pull/31304#issuecomment-2112809463
R: @m-trieu
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #31304:
URL: https://github.com/apache/beam/pull/31304#issuecomment-2114861528
Also changed to have the stream pool equal to the # of threads so that we have more streaming rpcs for the threads to use.
--
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] Make GrpcCommitWorkStream thread-safe as documented by moving batcher out of it. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #31304:
URL: https://github.com/apache/beam/pull/31304#discussion_r1602171640
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -305,7 +297,7 @@ long getBytes() {
}
}
- private class Batcher {
+ private class Batcher implements CommitWorkStream.RequestBatcher {
private final Map<Long, PendingRequest> queue;
Review Comment:
Wouldn't this still throw `ConcurrentModificationException` if `queue` is accessed by multiple threads? Should we make this `ConcurrentHashMap` vs `HashMap`?
--
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