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/21 08:49:56 UTC
[PR] [Dataflow Streaming] Reduce contention on work submission [beam]
arunpandianp opened a new pull request, #33687:
URL: https://github.com/apache/beam/pull/33687
The monitor guarding BoundedQueueExecutor's counters was locked by both the work submitting threads and the harness threads (300 of them). The contention can slow down the work submission loop. This change reduces the contention by collapsing counter updates from the harness threads and letting one thread to lock the monitor and update the counters.
--
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] Reduce contention on work submission [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on PR #33687:
URL: https://github.com/apache/beam/pull/33687#issuecomment-2608778879
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] [Dataflow Streaming] Reduce contention on work submission [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on PR #33687:
URL: https://github.com/apache/beam/pull/33687#issuecomment-2604005814
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] Reduce contention on work submission [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33687:
URL: https://github.com/apache/beam/pull/33687#discussion_r1928385055
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -36,6 +38,9 @@ public class BoundedQueueExecutor {
// Used to guard elementsOutstanding and bytesOutstanding.
private final Monitor monitor = new Monitor();
+ private final ConcurrentLinkedQueue<Long> decrementQueue = new ConcurrentLinkedQueue<>();
+ private final Object decrementQueueDrainLock = new Object();
+ private final AtomicBoolean decrementBatchOpen = new AtomicBoolean(false);
Review Comment:
I read decrementBatchOpen as "decrement batchopen" at first and it was confusing.
Maybe "isDecrementBatchPending" or something to clarify?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -236,10 +241,37 @@ private void executeMonitorHeld(Runnable work, long workBytes) {
}
private void decrementCounters(long workBytes) {
- monitor.enter();
- --elementsOutstanding;
- bytesOutstanding -= workBytes;
- monitor.leave();
+ // All threads queue decrements and one thread grabs the monitor and updates
+ // counters. We do this to reduce contention on monitor which is locked by
+ // GetWork thread
+ decrementQueue.add(workBytes);
+ boolean submittedToExistingBatch = decrementBatchOpen.getAndSet(true);
+ if (submittedToExistingBatch) {
+ // There is already a thread about to drain the decrement queue
+ // Current thread does not need to drain.
+ return;
+ }
+ synchronized (decrementQueueDrainLock) {
+ decrementBatchOpen.set(false);
Review Comment:
Since it's a little subtle how about a comment.
// By setting false here, we may allow another decrement to claim submission of the next batch and start waiting on the decrementQueueDrainLock.
// However this prevents races that would leave decrements in the queue and unclaimed and we are ensured there is at most one additional thread blocked. This helps prevent the executor from creating threads over the limit if many were contending on the lock while their decrements were already applied.
--
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] Reduce contention on work submission [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #33687:
URL: https://github.com/apache/beam/pull/33687
--
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] Reduce contention on work submission [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33687:
URL: https://github.com/apache/beam/pull/33687#discussion_r1923371306
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -235,10 +241,32 @@ private void executeMonitorHeld(Runnable work, long workBytes) {
}
private void decrementCounters(long workBytes) {
- monitor.enter();
- --elementsOutstanding;
- bytesOutstanding -= workBytes;
- monitor.leave();
+ // All threads queue decrements and one thread grabs the monitor and updates
+ // counters. We do this to reduce contention on monitor which is locked by
+ // GetWork thread
+ decrementQueue.add(workBytes);
+ synchronized (decrementQueue) {
Review Comment:
might be worth just having a object dedicated to be the lock. There are warnings usually using synchronized on concurrent/atomic objects about misuse (which I think would be wrong in this case but are still annoying).
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -235,10 +241,32 @@ private void executeMonitorHeld(Runnable work, long workBytes) {
}
private void decrementCounters(long workBytes) {
- monitor.enter();
- --elementsOutstanding;
- bytesOutstanding -= workBytes;
- monitor.leave();
+ // All threads queue decrements and one thread grabs the monitor and updates
+ // counters. We do this to reduce contention on monitor which is locked by
+ // GetWork thread
+ decrementQueue.add(workBytes);
+ synchronized (decrementQueue) {
+ Long polledElement;
+ List<Long> bytesList = new ArrayList<>();
Review Comment:
if keeping concurrent queue, it seems like you could remove this list and just build up elements and bytes sums and then subtract in the monitor below.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -235,10 +241,32 @@ private void executeMonitorHeld(Runnable work, long workBytes) {
}
private void decrementCounters(long workBytes) {
- monitor.enter();
- --elementsOutstanding;
- bytesOutstanding -= workBytes;
- monitor.leave();
+ // All threads queue decrements and one thread grabs the monitor and updates
+ // counters. We do this to reduce contention on monitor which is locked by
+ // GetWork thread
+ decrementQueue.add(workBytes);
Review Comment:
think we get away with an atomic integer.
static long ITEM_COUNT = 1L << 46;
atomicCounters.getAndAdd(ITEM_COUNT & workBytes);
synchronized (...) {
long flushedCounter = atomicCounters.getAndSet(0);
long items = flushedCounter >> 50;
long bytes = flushedCounter & (ITEM_COUNT - 1);
// could check for overflow
monitor...
}
--
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] Reduce contention on work submission [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #33687:
URL: https://github.com/apache/beam/pull/33687#discussion_r1926266354
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -235,10 +241,32 @@ private void executeMonitorHeld(Runnable work, long workBytes) {
}
private void decrementCounters(long workBytes) {
- monitor.enter();
- --elementsOutstanding;
- bytesOutstanding -= workBytes;
- monitor.leave();
+ // All threads queue decrements and one thread grabs the monitor and updates
+ // counters. We do this to reduce contention on monitor which is locked by
+ // GetWork thread
+ decrementQueue.add(workBytes);
Review Comment:
I think keeping the queue is simpler than packing both values into an atomic + overflow checking. The queue should contain only one element per harness thread. Let's keep the queue.
--
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] Reduce contention on work submission [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #33687:
URL: https://github.com/apache/beam/pull/33687#discussion_r1928370851
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -235,10 +240,30 @@ private void executeMonitorHeld(Runnable work, long workBytes) {
}
private void decrementCounters(long workBytes) {
- monitor.enter();
- --elementsOutstanding;
- bytesOutstanding -= workBytes;
- monitor.leave();
+ // All threads queue decrements and one thread grabs the monitor and updates
+ // counters. We do this to reduce contention on monitor which is locked by
+ // GetWork thread
+ decrementQueue.add(workBytes);
+ synchronized (decrementQueueDrainLock) {
Review Comment:
good idea, 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] [Dataflow Streaming] Reduce contention on work submission [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33687:
URL: https://github.com/apache/beam/pull/33687#issuecomment-2604008397
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] Reduce contention on work submission [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33687:
URL: https://github.com/apache/beam/pull/33687#discussion_r1926715847
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -235,10 +240,30 @@ private void executeMonitorHeld(Runnable work, long workBytes) {
}
private void decrementCounters(long workBytes) {
- monitor.enter();
- --elementsOutstanding;
- bytesOutstanding -= workBytes;
- monitor.leave();
+ // All threads queue decrements and one thread grabs the monitor and updates
+ // counters. We do this to reduce contention on monitor which is locked by
+ // GetWork thread
+ decrementQueue.add(workBytes);
+ synchronized (decrementQueueDrainLock) {
Review Comment:
wondering if this will end up creating more threads in the pool if we have a bunch of threads blocking on this but the count is decremented by another thread and new work is admitted.
maybe we could improve to reduce the # of threads waiting on synchronized block using an atomic? Seems like something like the following would make it so there is at most one thread waiting for the synchronized block
addToQueue
bool coalesce = !atomic_coalescer.getandset(true);
if (!coalesce) return;
{
sync single
atomic_coalescer.set(false);
poll queue, return if empty
monitor {
}
}
--
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] Reduce contention on work submission [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #33687:
URL: https://github.com/apache/beam/pull/33687#discussion_r1926265185
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -235,10 +241,32 @@ private void executeMonitorHeld(Runnable work, long workBytes) {
}
private void decrementCounters(long workBytes) {
- monitor.enter();
- --elementsOutstanding;
- bytesOutstanding -= workBytes;
- monitor.leave();
+ // All threads queue decrements and one thread grabs the monitor and updates
+ // counters. We do this to reduce contention on monitor which is locked by
+ // GetWork thread
+ decrementQueue.add(workBytes);
+ synchronized (decrementQueue) {
Review Comment:
agree, done.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -235,10 +241,32 @@ private void executeMonitorHeld(Runnable work, long workBytes) {
}
private void decrementCounters(long workBytes) {
- monitor.enter();
- --elementsOutstanding;
- bytesOutstanding -= workBytes;
- monitor.leave();
+ // All threads queue decrements and one thread grabs the monitor and updates
+ // counters. We do this to reduce contention on monitor which is locked by
+ // GetWork thread
+ decrementQueue.add(workBytes);
+ synchronized (decrementQueue) {
+ Long polledElement;
+ List<Long> bytesList = new ArrayList<>();
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