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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
