You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "arunpandianp (via GitHub)" <gi...@apache.org> on 2025/01/14 00:35:36 UTC
[PR] [Dataflow Streaming] Code micro optimizations [beam]
arunpandianp opened a new pull request, #33580:
URL: https://github.com/apache/beam/pull/33580
#33578
--
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] [Dataflow Streaming] Code micro optimizations [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33580:
URL: https://github.com/apache/beam/pull/33580#issuecomment-2588508100
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, 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] [Dataflow Streaming] Code micro optimizations (1/N) [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #33580:
URL: https://github.com/apache/beam/pull/33580
--
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] [Dataflow Streaming] Code micro optimizations [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on PR #33580:
URL: https://github.com/apache/beam/pull/33580#issuecomment-2588504885
R: @scwhittle
--
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] [Dataflow Streaming] Code micro optimizations (1/N) [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #33580:
URL: https://github.com/apache/beam/pull/33580#discussion_r1925183844
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -219,14 +220,12 @@ synchronized ImmutableList<RefreshableWork> getRefreshableWork(Instant refreshDe
.collect(toImmutableList());
}
- private void incrementActiveWorkBudget(Work work) {
- activeGetWorkBudget.updateAndGet(
- getWorkBudget -> getWorkBudget.apply(1, work.getSerializedWorkItemSize()));
+ private synchronized void incrementActiveWorkBudget(Work work) {
+ activeGetWorkBudget = activeGetWorkBudget.apply(1, work.getSerializedWorkItemSize());
}
- private void decrementActiveWorkBudget(Work work) {
- activeGetWorkBudget.updateAndGet(
- getWorkBudget -> getWorkBudget.subtract(1, work.getSerializedWorkItemSize()));
+ private synchronized void decrementActiveWorkBudget(Work work) {
Review Comment:
I suspected deadlock, but it was the n^2 loop inside failWorkForKey() holding the lock for too long that caused the delay.
This PR is not locking more mutexes than before, all calls to `decrementActiveWorkBudget`, `incrementActiveWorkBudget`, `currentActiveWorkBudget ` are done holding the ActiveWorkStae.this mutex.
Should be good to merge this change.
--
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] [Dataflow Streaming] Code micro optimizations (1/N) [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33580:
URL: https://github.com/apache/beam/pull/33580#discussion_r1925257235
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -219,14 +220,12 @@ synchronized ImmutableList<RefreshableWork> getRefreshableWork(Instant refreshDe
.collect(toImmutableList());
}
- private void incrementActiveWorkBudget(Work work) {
- activeGetWorkBudget.updateAndGet(
- getWorkBudget -> getWorkBudget.apply(1, work.getSerializedWorkItemSize()));
+ private synchronized void incrementActiveWorkBudget(Work work) {
+ activeGetWorkBudget = activeGetWorkBudget.apply(1, work.getSerializedWorkItemSize());
}
- private void decrementActiveWorkBudget(Work work) {
- activeGetWorkBudget.updateAndGet(
- getWorkBudget -> getWorkBudget.subtract(1, work.getSerializedWorkItemSize()));
+ private synchronized void decrementActiveWorkBudget(Work work) {
Review Comment:
currentActiveWorkBudget is public and called without synchronized currently, but it is also just used by test method.
--
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] [Dataflow Streaming] Code micro optimizations (1/N) [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33580:
URL: https://github.com/apache/beam/pull/33580#discussion_r1914521841
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ShardedKey.java:
##########
@@ -35,4 +35,9 @@ public static ShardedKey create(ByteString key, long shardingKey) {
public final String toString() {
return String.format("%016x", shardingKey());
}
+
+ @Override
+ public final int hashCode() {
+ return Long.hashCode(shardingKey());
Review Comment:
ditto add comment
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -191,7 +191,8 @@ public boolean throwExceptionsForLargeOutput() {
}
public boolean workIsFailed() {
- return Optional.ofNullable(work).map(Work::isFailed).orElse(false);
+ if (work != null) return work.isFailed();
Review Comment:
keep single line?
return work == null ? false : work.isFailed();
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillComputationKey.java:
##########
@@ -45,4 +46,9 @@ public final String toString() {
return String.format(
"%s: %s-%d", computationId(), TextFormat.escapeBytes(key()), shardingKey());
}
+
+ @Override
+ public final int hashCode() {
+ return Objects.hash(shardingKey(), computationId());
Review Comment:
add comment?
// Sharding key collisions are unexpected
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java:
##########
@@ -406,36 +404,27 @@ private static boolean useNewTimerTagEncoding(TimerData timerData) {
*/
public static ByteString timerTag(WindmillNamespacePrefix prefix, TimerData timerData) {
String tagString;
- ExposedByteArrayOutputStream out = new ExposedByteArrayOutputStream();
- try {
- if (useNewTimerTagEncoding(timerData)) {
- tagString =
- new StringBuilder()
- .append(prefix.byteString().toStringUtf8()) // this never ends with a slash
- .append(
- timerData.getNamespace().stringKey()) // this must begin and end with a slash
- .append('+')
- .append(timerData.getTimerId()) // this is arbitrary; currently unescaped
- .append('+')
- .append(timerData.getTimerFamilyId())
- .toString();
- out.write(tagString.getBytes(StandardCharsets.UTF_8));
- } else {
- // Timers without timerFamily would have timerFamily would be an empty string
- tagString =
- new StringBuilder()
- .append(prefix.byteString().toStringUtf8()) // this never ends with a slash
- .append(
- timerData.getNamespace().stringKey()) // this must begin and end with a slash
- .append('+')
- .append(timerData.getTimerId()) // this is arbitrary; currently unescaped
- .toString();
- out.write(tagString.getBytes(StandardCharsets.UTF_8));
- }
- return ByteString.readFrom(new ExposedByteArrayInputStream(out.toByteArray()));
- } catch (IOException e) {
- throw new RuntimeException(e);
+ if (useNewTimerTagEncoding(timerData)) {
+ tagString =
+ new StringBuilder()
+ .append(prefix.byteString().toStringUtf8()) // this never ends with a slash
+ .append(timerData.getNamespace().stringKey()) // this must begin and end with a slash
+ .append('+')
+ .append(timerData.getTimerId()) // this is arbitrary; currently unescaped
+ .append('+')
+ .append(timerData.getTimerFamilyId())
+ .toString();
+ } else {
+ // Timers without timerFamily would have timerFamily would be an empty string
+ tagString =
+ new StringBuilder()
+ .append(prefix.byteString().toStringUtf8()) // this never ends with a slash
+ .append(timerData.getNamespace().stringKey()) // this must begin and end with a slash
+ .append('+')
+ .append(timerData.getTimerId()) // this is arbitrary; currently unescaped
+ .toString();
}
+ return ByteString.copyFromUtf8(tagString);
Review Comment:
There is also beam's ByteStringOutputStream but given that these are short I'm guessing this could be faster.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -332,7 +330,9 @@ private synchronized ImmutableMap<ShardedKey, WorkId> getStuckCommitsAt(
* {@link ActiveWorkState}, and not committed back to Windmill.
*/
GetWorkBudget currentActiveWorkBudget() {
- return activeGetWorkBudget.get();
+ synchronized (this) {
+ return activeGetWorkBudget;
Review Comment:
just make method synchronized
--
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] [Dataflow Streaming] Code micro optimizations (1/N) [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on PR #33580:
URL: https://github.com/apache/beam/pull/33580#issuecomment-2598867023
Addressed comments and fixed merge conflicts
--
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] [Dataflow Streaming] Code micro optimizations (1/N) [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33580:
URL: https://github.com/apache/beam/pull/33580#discussion_r1925127733
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ActiveWorkState.java:
##########
@@ -219,14 +220,12 @@ synchronized ImmutableList<RefreshableWork> getRefreshableWork(Instant refreshDe
.collect(toImmutableList());
}
- private void incrementActiveWorkBudget(Work work) {
- activeGetWorkBudget.updateAndGet(
- getWorkBudget -> getWorkBudget.apply(1, work.getSerializedWorkItemSize()));
+ private synchronized void incrementActiveWorkBudget(Work work) {
+ activeGetWorkBudget = activeGetWorkBudget.apply(1, work.getSerializedWorkItemSize());
}
- private void decrementActiveWorkBudget(Work work) {
- activeGetWorkBudget.updateAndGet(
- getWorkBudget -> getWorkBudget.subtract(1, work.getSerializedWorkItemSize()));
+ private synchronized void decrementActiveWorkBudget(Work work) {
Review Comment:
how about removing this file from the PR and we can merge the rest?
in invalidateStuckCommits we call a function under the sychronized block which could be contributing to deadlock. Or could be printActiveWork holding synchronized as it calls into Work
--
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] [Dataflow Streaming] Code micro optimizations (1/N) [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #33580:
URL: https://github.com/apache/beam/pull/33580#discussion_r1915840487
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java:
##########
@@ -406,36 +404,27 @@ private static boolean useNewTimerTagEncoding(TimerData timerData) {
*/
public static ByteString timerTag(WindmillNamespacePrefix prefix, TimerData timerData) {
String tagString;
- ExposedByteArrayOutputStream out = new ExposedByteArrayOutputStream();
- try {
- if (useNewTimerTagEncoding(timerData)) {
- tagString =
- new StringBuilder()
- .append(prefix.byteString().toStringUtf8()) // this never ends with a slash
- .append(
- timerData.getNamespace().stringKey()) // this must begin and end with a slash
- .append('+')
- .append(timerData.getTimerId()) // this is arbitrary; currently unescaped
- .append('+')
- .append(timerData.getTimerFamilyId())
- .toString();
- out.write(tagString.getBytes(StandardCharsets.UTF_8));
- } else {
- // Timers without timerFamily would have timerFamily would be an empty string
- tagString =
- new StringBuilder()
- .append(prefix.byteString().toStringUtf8()) // this never ends with a slash
- .append(
- timerData.getNamespace().stringKey()) // this must begin and end with a slash
- .append('+')
- .append(timerData.getTimerId()) // this is arbitrary; currently unescaped
- .toString();
- out.write(tagString.getBytes(StandardCharsets.UTF_8));
- }
- return ByteString.readFrom(new ExposedByteArrayInputStream(out.toByteArray()));
- } catch (IOException e) {
- throw new RuntimeException(e);
+ if (useNewTimerTagEncoding(timerData)) {
+ tagString =
+ new StringBuilder()
+ .append(prefix.byteString().toStringUtf8()) // this never ends with a slash
+ .append(timerData.getNamespace().stringKey()) // this must begin and end with a slash
+ .append('+')
+ .append(timerData.getTimerId()) // this is arbitrary; currently unescaped
+ .append('+')
+ .append(timerData.getTimerFamilyId())
+ .toString();
+ } else {
+ // Timers without timerFamily would have timerFamily would be an empty string
+ tagString =
+ new StringBuilder()
+ .append(prefix.byteString().toStringUtf8()) // this never ends with a slash
+ .append(timerData.getNamespace().stringKey()) // this must begin and end with a slash
+ .append('+')
+ .append(timerData.getTimerId()) // this is arbitrary; currently unescaped
+ .toString();
}
+ return ByteString.copyFromUtf8(tagString);
Review Comment:
I think so, `ByteString.copyFromUtf8` uses `LiteralByteString` internally and doesn't create a copy of `byte[]` from `tagString.getBytes`.
--
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