You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "m-trieu (via GitHub)" <gi...@apache.org> on 2024/01/19 04:25:05 UTC
[PR] Add support for current active work budget. [beam]
m-trieu opened a new pull request, #30048:
URL: https://github.com/apache/beam/pull/30048
Add support for current active work budget.
This will be used by direct path to pass to the `GetWorkBudgetDistributor`.
r: @scwhittle
------------------------
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] Add support for current active work budget. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on PR #30048:
URL: https://github.com/apache/beam/pull/30048#issuecomment-1899908534
assign set of reviewers
--
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] Add support for current active work budget. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #30048:
URL: https://github.com/apache/beam/pull/30048#issuecomment-1929396138
Reminder, please take a look at this pr: @damondouglas
--
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] Add support for current active work budget. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on PR #30048:
URL: https://github.com/apache/beam/pull/30048#issuecomment-1899938844
run java precommit
--
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] Add support for current active work budget. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #30048:
URL: https://github.com/apache/beam/pull/30048#discussion_r1470339493
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -168,6 +198,18 @@ synchronized void failWorkForKey(Map<Long, List<FailedTokens>> failedWork) {
}
}
+ private synchronized void incrementActiveWorkBudget(Work work) {
+ GetWorkBudget currentActiveWorkBudget = activeGetWorkBudget.get();
+ activeGetWorkBudget.set(
+ currentActiveWorkBudget.apply(1, work.getWorkItem().getSerializedSize()));
+ }
+
+ private synchronized void decrementActiveWorkBudget(Work work) {
Review Comment:
will never go negative since guards are in `GetWorkBudget`
```
public final GetWorkBudget build() {
setItems(Math.max(0, items()));
setBytes(Math.max(0, bytes()));
return autoBuild();
}
```
--
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] Add support for current active work budget. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #30048:
URL: https://github.com/apache/beam/pull/30048#discussion_r1458915472
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -238,6 +239,23 @@ private static Stream<KeyedGetDataRequest> toKeyedGetDataRequestStream(
.build());
}
+ /**
+ * Returns the current aggregate {@link GetWorkBudget} that is active on the user worker. Active
+ * means that the work is received from Windmill, being processed or queued to be processed in
+ * {@link ActiveWorkState}, and not committed back to Windmill.
+ */
+ synchronized GetWorkBudget currentActiveWorkBudget() {
Review Comment:
depending on how often this is called, it might be better to keep a running total as we add/remove things instead of calculating by iterating over everything each time this is called.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ComputationState.java:
##########
@@ -126,6 +127,13 @@ public ImmutableList<KeyedGetDataRequest> getKeysToRefresh(
return activeWorkState.getKeysToRefresh(refreshDeadline, sampler);
}
+ @SuppressWarnings(
Review Comment:
add test to avoid this? or just remove this part for now and add when it actually is used.
--
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] Add support for current active work budget. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #30048:
URL: https://github.com/apache/beam/pull/30048#discussion_r1470343075
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -288,6 +331,15 @@ private static Stream<HeartbeatRequest> toHeartbeatRequestStream(
.build());
}
+ /**
+ * Returns the current aggregate {@link GetWorkBudget} that is active on the user worker. Active
+ * means that the work is received from Windmill, being processed or queued to be processed in
+ * {@link ActiveWorkState}, and not committed back to Windmill.
+ */
+ GetWorkBudget currentActiveWorkBudget() {
+ return activeGetWorkBudget.get();
+ }
+
synchronized void printActiveWork(PrintWriter writer, Instant now) {
Review Comment:
added
--
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] Add support for current active work budget. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #30048:
URL: https://github.com/apache/beam/pull/30048
--
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] Add support for current active work budget. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on PR #30048:
URL: https://github.com/apache/beam/pull/30048#issuecomment-1931489370
any more comments here? @scwhittle
thanks!
--
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] Add support for current active work budget. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #30048:
URL: https://github.com/apache/beam/pull/30048#discussion_r1459562668
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -238,6 +239,23 @@ private static Stream<KeyedGetDataRequest> toKeyedGetDataRequestStream(
.build());
}
+ /**
+ * Returns the current aggregate {@link GetWorkBudget} that is active on the user worker. Active
+ * means that the work is received from Windmill, being processed or queued to be processed in
+ * {@link ActiveWorkState}, and not committed back to Windmill.
+ */
+ synchronized GetWorkBudget currentActiveWorkBudget() {
Review Comment:
good idea
added so that we increment when work is activated (ready to execute or queued)
and decrement when work is completed.
--
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] Add support for current active work budget. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #30048:
URL: https://github.com/apache/beam/pull/30048#discussion_r1469562154
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -70,11 +73,21 @@ public final class ActiveWorkState {
@GuardedBy("this")
private final WindmillStateCache.ForComputation computationStateCache;
+ /**
+ * Current budget that is being processed or queued on the user worker. Incremented when work is
+ * activated in {@link #activateWorkForKey(ShardedKey, Work)}, and decremented when work is
+ * completed in {@link #completeWorkAndGetNextWorkForKey(ShardedKey, long)}.
+ *
+ * @implNote Reads are lock free using {@link AtomicReference#get()}, writes are synchronized.
Review Comment:
an alternative is to use AtomicReference.updateAndGet if you aren't going to be synchronizing with more on writes
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -168,6 +198,18 @@ synchronized void failWorkForKey(Map<Long, List<FailedTokens>> failedWork) {
}
}
+ private synchronized void incrementActiveWorkBudget(Work work) {
+ GetWorkBudget currentActiveWorkBudget = activeGetWorkBudget.get();
+ activeGetWorkBudget.set(
+ currentActiveWorkBudget.apply(1, work.getWorkItem().getSerializedSize()));
+ }
+
+ private synchronized void decrementActiveWorkBudget(Work work) {
Review Comment:
should we have some guards against going negative?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -288,6 +331,15 @@ private static Stream<HeartbeatRequest> toHeartbeatRequestStream(
.build());
}
+ /**
+ * Returns the current aggregate {@link GetWorkBudget} that is active on the user worker. Active
+ * means that the work is received from Windmill, being processed or queued to be processed in
+ * {@link ActiveWorkState}, and not committed back to Windmill.
+ */
+ GetWorkBudget currentActiveWorkBudget() {
+ return activeGetWorkBudget.get();
+ }
+
synchronized void printActiveWork(PrintWriter writer, Instant now) {
Review Comment:
can you add to the status page?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -121,16 +140,27 @@ synchronized ActivateWorkResult activateWorkForKey(ShardedKey shardedKey, Work w
// Queue the work for later processing.
workQueue.addLast(work);
+ incrementActiveWorkBudget(work);
return ActivateWorkResult.QUEUED;
}
- public static final class FailedTokens {
- public long workToken;
- public long cacheToken;
+ @AutoValue
+ public abstract static class FailedTokens {
+ public abstract long workToken();
+
+ public abstract long cacheToken();
+
+ public static Builder newBuilder() {
+ return new AutoValue_ActiveWorkState_FailedTokens.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setWorkToken(long value);
- public FailedTokens(long workToken, long cacheToken) {
- this.workToken = workToken;
- this.cacheToken = cacheToken;
+ public abstract Builder setCacheToken(long value);
+
+ public abstract FailedTokens build();
Review Comment:
indent looks off
--
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] Add support for current active work budget. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #30048:
URL: https://github.com/apache/beam/pull/30048#issuecomment-1899909690
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @damondouglas added as fallback since no labels match configuration
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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] Add support for current active work budget. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #30048:
URL: https://github.com/apache/beam/pull/30048#issuecomment-1899776595
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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] Add support for current active work budget. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #30048:
URL: https://github.com/apache/beam/pull/30048#discussion_r1470338660
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -121,16 +140,27 @@ synchronized ActivateWorkResult activateWorkForKey(ShardedKey shardedKey, Work w
// Queue the work for later processing.
workQueue.addLast(work);
+ incrementActiveWorkBudget(work);
return ActivateWorkResult.QUEUED;
}
- public static final class FailedTokens {
- public long workToken;
- public long cacheToken;
+ @AutoValue
+ public abstract static class FailedTokens {
+ public abstract long workToken();
+
+ public abstract long cacheToken();
+
+ public static Builder newBuilder() {
+ return new AutoValue_ActiveWorkState_FailedTokens.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setWorkToken(long value);
- public FailedTokens(long workToken, long cacheToken) {
- this.workToken = workToken;
- this.cacheToken = cacheToken;
+ public abstract Builder setCacheToken(long value);
+
+ public abstract FailedTokens build();
Review Comment:
fixed
--
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] Add support for current active work budget. [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #30048:
URL: https://github.com/apache/beam/pull/30048#discussion_r1470337934
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -70,11 +73,21 @@ public final class ActiveWorkState {
@GuardedBy("this")
private final WindmillStateCache.ForComputation computationStateCache;
+ /**
+ * Current budget that is being processed or queued on the user worker. Incremented when work is
+ * activated in {@link #activateWorkForKey(ShardedKey, Work)}, and decremented when work is
+ * completed in {@link #completeWorkAndGetNextWorkForKey(ShardedKey, long)}.
+ *
+ * @implNote Reads are lock free using {@link AtomicReference#get()}, writes are synchronized.
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