scwhittle commented on a change in pull request #16901:
URL: https://github.com/apache/beam/pull/16901#discussion_r814013955
##########
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java
##########
@@ -18,62 +18,135 @@
package org.apache.beam.runners.dataflow.worker.util;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.Monitor;
+import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.Monitor.Guard;
-/** Executor that blocks on execute() if its queue is full. */
+/** An executor for executing work on windmill items. */
@SuppressWarnings({
"nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
})
-public class BoundedQueueExecutor extends ThreadPoolExecutor {
- private static class ReducableSemaphore extends Semaphore {
- ReducableSemaphore(int permits) {
- super(permits);
- }
-
- @Override
- public void reducePermits(int permits) {
- super.reducePermits(permits);
- }
- }
+public class BoundedQueueExecutor {
+ private final ThreadPoolExecutor executor;
+ private final int maximumElementsOutstanding;
+ private final long maximumBytesOutstanding;
- private ReducableSemaphore semaphore;
+ private final Monitor monitor = new Monitor();
+ private int elementsOutstanding = 0;
+ private long bytesOutstanding = 0;
public BoundedQueueExecutor(
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
- int maximumQueueSize,
+ int maximumElementsOutstanding,
+ long maximumBytesOutstanding,
ThreadFactory threadFactory) {
- super(
- maximumPoolSize,
- maximumPoolSize,
- keepAliveTime,
- unit,
- new LinkedBlockingQueue<Runnable>(),
- threadFactory);
- this.semaphore = new ReducableSemaphore(maximumQueueSize);
- allowCoreThreadTimeOut(true);
+ executor =
+ new ThreadPoolExecutor(
+ maximumPoolSize,
+ maximumPoolSize,
+ keepAliveTime,
+ unit,
+ new LinkedBlockingQueue<>(),
+ threadFactory);
+ executor.allowCoreThreadTimeOut(true);
+ this.maximumElementsOutstanding = maximumElementsOutstanding;
+ this.maximumBytesOutstanding = maximumBytesOutstanding;
+ }
+
+ // Before adding a Work to the queue, check that there are enough bytes of
space or no other
+ // outstanding elements of work.
+ public void execute(Runnable work, long workBytes) {
+ monitor.enterWhenUninterruptibly(
+ new Guard(monitor) {
+ @Override
+ public boolean isSatisfied() {
+ return elementsOutstanding == 0
+ || (bytesAvailable() >= workBytes
+ && elementsOutstanding < maximumElementsOutstanding);
+ }
+ });
+ executeLockHeld(work, workBytes);
}
- // Before adding a Runnable to the queue, acquire the semaphore.
- @Override
- public void execute(Runnable r) {
- semaphore.acquireUninterruptibly();
- super.execute(r);
+ public void forceExecute(Runnable work, long workBytes) {
+ monitor.enter();
+ executeLockHeld(work, workBytes);
}
// Forcibly add something to the queue, ignoring the length limit.
public void forceExecute(Runnable r) {
- semaphore.reducePermits(1);
- super.execute(r);
+ forceExecute(r, 0);
+ }
+
+ public void shutdown() throws InterruptedException {
+ executor.shutdown();
+ if (!executor.awaitTermination(5, TimeUnit.MINUTES)) {
+ throw new RuntimeException("Work executor did not terminate within 5
minutes");
+ }
+ }
+
+ public boolean executorQueueIsEmpty() {
+ return executor.getQueue().isEmpty();
+ }
+
+ public String summaryHtml() {
+ monitor.enter();
+ try {
+ StringBuilder builder = new StringBuilder();
+ builder.append("Worker Threads: ");
+ builder.append(executor.getPoolSize());
+ builder.append("/");
+ builder.append(executor.getMaximumPoolSize());
+ builder.append("<br>/n");
+
+ builder.append("Active Threads: ");
+ builder.append(executor.getActiveCount());
+ builder.append("<br>/n");
+
+ builder.append("Work Queue Size: ");
+ builder.append(elementsOutstanding);
+ builder.append("/");
+ builder.append(maximumElementsOutstanding);
+ builder.append("<br>/n");
+
+ builder.append("Work Queue Bytes: ");
+ builder.append(bytesOutstanding);
+ builder.append("/");
+ builder.append(maximumBytesOutstanding);
+ builder.append("<br>/n");
+
+ return builder.toString();
+ } finally {
+ monitor.leave();
+ }
+ }
+
+ private void executeLockHeld(Runnable work, long workBytes) {
+ Runnable workRunnable =
+ () -> {
+ try {
+ work.run();
+ } finally {
+ monitor.enter();
+ --elementsOutstanding;
+ bytesOutstanding -= workBytes;
+ monitor.leave();
+ }
+ };
+ try {
+ bytesOutstanding += workBytes;
+ ++elementsOutstanding;
+ executor.execute(workRunnable);
Review comment:
seems like you should reduce monitor scope to just increment counters.
Then make runnable and call execute outside
Might deadlock otherwise if forceexecutes have pushed the executor queue
over maxelements and execute blocks queueing with monitor held. Then work
won't be able to enter and decrement when done and threads in executor will be
tied up.
##########
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java
##########
@@ -18,62 +18,135 @@
package org.apache.beam.runners.dataflow.worker.util;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.Monitor;
+import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.Monitor.Guard;
-/** Executor that blocks on execute() if its queue is full. */
+/** An executor for executing work on windmill items. */
@SuppressWarnings({
"nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
})
-public class BoundedQueueExecutor extends ThreadPoolExecutor {
- private static class ReducableSemaphore extends Semaphore {
- ReducableSemaphore(int permits) {
- super(permits);
- }
-
- @Override
- public void reducePermits(int permits) {
- super.reducePermits(permits);
- }
- }
+public class BoundedQueueExecutor {
+ private final ThreadPoolExecutor executor;
+ private final int maximumElementsOutstanding;
+ private final long maximumBytesOutstanding;
- private ReducableSemaphore semaphore;
+ private final Monitor monitor = new Monitor();
+ private int elementsOutstanding = 0;
+ private long bytesOutstanding = 0;
public BoundedQueueExecutor(
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
- int maximumQueueSize,
+ int maximumElementsOutstanding,
+ long maximumBytesOutstanding,
ThreadFactory threadFactory) {
- super(
- maximumPoolSize,
- maximumPoolSize,
- keepAliveTime,
- unit,
- new LinkedBlockingQueue<Runnable>(),
- threadFactory);
- this.semaphore = new ReducableSemaphore(maximumQueueSize);
- allowCoreThreadTimeOut(true);
+ executor =
+ new ThreadPoolExecutor(
+ maximumPoolSize,
+ maximumPoolSize,
+ keepAliveTime,
+ unit,
+ new LinkedBlockingQueue<>(),
+ threadFactory);
+ executor.allowCoreThreadTimeOut(true);
+ this.maximumElementsOutstanding = maximumElementsOutstanding;
+ this.maximumBytesOutstanding = maximumBytesOutstanding;
+ }
+
+ // Before adding a Work to the queue, check that there are enough bytes of
space or no other
+ // outstanding elements of work.
+ public void execute(Runnable work, long workBytes) {
+ monitor.enterWhenUninterruptibly(
+ new Guard(monitor) {
+ @Override
+ public boolean isSatisfied() {
+ return elementsOutstanding == 0
+ || (bytesAvailable() >= workBytes
+ && elementsOutstanding < maximumElementsOutstanding);
+ }
+ });
+ executeLockHeld(work, workBytes);
}
- // Before adding a Runnable to the queue, acquire the semaphore.
- @Override
- public void execute(Runnable r) {
- semaphore.acquireUninterruptibly();
- super.execute(r);
+ public void forceExecute(Runnable work, long workBytes) {
Review comment:
comment
##########
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java
##########
@@ -18,62 +18,135 @@
package org.apache.beam.runners.dataflow.worker.util;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.Monitor;
+import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.Monitor.Guard;
-/** Executor that blocks on execute() if its queue is full. */
+/** An executor for executing work on windmill items. */
@SuppressWarnings({
"nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
})
-public class BoundedQueueExecutor extends ThreadPoolExecutor {
- private static class ReducableSemaphore extends Semaphore {
- ReducableSemaphore(int permits) {
- super(permits);
- }
-
- @Override
- public void reducePermits(int permits) {
- super.reducePermits(permits);
- }
- }
+public class BoundedQueueExecutor {
+ private final ThreadPoolExecutor executor;
+ private final int maximumElementsOutstanding;
+ private final long maximumBytesOutstanding;
- private ReducableSemaphore semaphore;
+ private final Monitor monitor = new Monitor();
+ private int elementsOutstanding = 0;
+ private long bytesOutstanding = 0;
public BoundedQueueExecutor(
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
- int maximumQueueSize,
+ int maximumElementsOutstanding,
+ long maximumBytesOutstanding,
ThreadFactory threadFactory) {
- super(
- maximumPoolSize,
- maximumPoolSize,
- keepAliveTime,
- unit,
- new LinkedBlockingQueue<Runnable>(),
- threadFactory);
- this.semaphore = new ReducableSemaphore(maximumQueueSize);
- allowCoreThreadTimeOut(true);
+ executor =
+ new ThreadPoolExecutor(
+ maximumPoolSize,
+ maximumPoolSize,
+ keepAliveTime,
+ unit,
+ new LinkedBlockingQueue<>(),
+ threadFactory);
+ executor.allowCoreThreadTimeOut(true);
+ this.maximumElementsOutstanding = maximumElementsOutstanding;
+ this.maximumBytesOutstanding = maximumBytesOutstanding;
+ }
+
+ // Before adding a Work to the queue, check that there are enough bytes of
space or no other
+ // outstanding elements of work.
+ public void execute(Runnable work, long workBytes) {
+ monitor.enterWhenUninterruptibly(
+ new Guard(monitor) {
+ @Override
+ public boolean isSatisfied() {
+ return elementsOutstanding == 0
+ || (bytesAvailable() >= workBytes
+ && elementsOutstanding < maximumElementsOutstanding);
+ }
+ });
+ executeLockHeld(work, workBytes);
}
- // Before adding a Runnable to the queue, acquire the semaphore.
- @Override
- public void execute(Runnable r) {
- semaphore.acquireUninterruptibly();
- super.execute(r);
+ public void forceExecute(Runnable work, long workBytes) {
+ monitor.enter();
+ executeLockHeld(work, workBytes);
}
// Forcibly add something to the queue, ignoring the length limit.
public void forceExecute(Runnable r) {
- semaphore.reducePermits(1);
- super.execute(r);
+ forceExecute(r, 0);
+ }
+
+ public void shutdown() throws InterruptedException {
+ executor.shutdown();
+ if (!executor.awaitTermination(5, TimeUnit.MINUTES)) {
+ throw new RuntimeException("Work executor did not terminate within 5
minutes");
+ }
+ }
+
+ public boolean executorQueueIsEmpty() {
+ return executor.getQueue().isEmpty();
+ }
+
+ public String summaryHtml() {
+ monitor.enter();
+ try {
+ StringBuilder builder = new StringBuilder();
+ builder.append("Worker Threads: ");
+ builder.append(executor.getPoolSize());
+ builder.append("/");
+ builder.append(executor.getMaximumPoolSize());
+ builder.append("<br>/n");
+
+ builder.append("Active Threads: ");
+ builder.append(executor.getActiveCount());
+ builder.append("<br>/n");
+
+ builder.append("Work Queue Size: ");
+ builder.append(elementsOutstanding);
+ builder.append("/");
+ builder.append(maximumElementsOutstanding);
+ builder.append("<br>/n");
+
+ builder.append("Work Queue Bytes: ");
+ builder.append(bytesOutstanding);
+ builder.append("/");
+ builder.append(maximumBytesOutstanding);
+ builder.append("<br>/n");
+
+ return builder.toString();
+ } finally {
+ monitor.leave();
+ }
+ }
+
+ private void executeLockHeld(Runnable work, long workBytes) {
+ Runnable workRunnable =
+ () -> {
+ try {
+ work.run();
+ } finally {
+ monitor.enter();
+ --elementsOutstanding;
+ bytesOutstanding -= workBytes;
+ monitor.leave();
+ }
+ };
+ try {
+ bytesOutstanding += workBytes;
+ ++elementsOutstanding;
+ executor.execute(workRunnable);
+ } finally {
+ monitor.leave();
Review comment:
can adding to executor throw exeception? If so you should decrement
counters before throwing exception in case it gets caught and handled above.
##########
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java
##########
@@ -18,62 +18,135 @@
package org.apache.beam.runners.dataflow.worker.util;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.Monitor;
+import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.Monitor.Guard;
-/** Executor that blocks on execute() if its queue is full. */
+/** An executor for executing work on windmill items. */
@SuppressWarnings({
"nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
})
-public class BoundedQueueExecutor extends ThreadPoolExecutor {
- private static class ReducableSemaphore extends Semaphore {
- ReducableSemaphore(int permits) {
- super(permits);
- }
-
- @Override
- public void reducePermits(int permits) {
- super.reducePermits(permits);
- }
- }
+public class BoundedQueueExecutor {
+ private final ThreadPoolExecutor executor;
+ private final int maximumElementsOutstanding;
+ private final long maximumBytesOutstanding;
- private ReducableSemaphore semaphore;
+ private final Monitor monitor = new Monitor();
+ private int elementsOutstanding = 0;
+ private long bytesOutstanding = 0;
public BoundedQueueExecutor(
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
- int maximumQueueSize,
+ int maximumElementsOutstanding,
+ long maximumBytesOutstanding,
ThreadFactory threadFactory) {
- super(
- maximumPoolSize,
- maximumPoolSize,
- keepAliveTime,
- unit,
- new LinkedBlockingQueue<Runnable>(),
- threadFactory);
- this.semaphore = new ReducableSemaphore(maximumQueueSize);
- allowCoreThreadTimeOut(true);
+ executor =
+ new ThreadPoolExecutor(
+ maximumPoolSize,
+ maximumPoolSize,
+ keepAliveTime,
+ unit,
+ new LinkedBlockingQueue<>(),
+ threadFactory);
+ executor.allowCoreThreadTimeOut(true);
+ this.maximumElementsOutstanding = maximumElementsOutstanding;
+ this.maximumBytesOutstanding = maximumBytesOutstanding;
+ }
+
+ // Before adding a Work to the queue, check that there are enough bytes of
space or no other
+ // outstanding elements of work.
+ public void execute(Runnable work, long workBytes) {
+ monitor.enterWhenUninterruptibly(
+ new Guard(monitor) {
+ @Override
+ public boolean isSatisfied() {
+ return elementsOutstanding == 0
+ || (bytesAvailable() >= workBytes
+ && elementsOutstanding < maximumElementsOutstanding);
+ }
+ });
+ executeLockHeld(work, workBytes);
}
- // Before adding a Runnable to the queue, acquire the semaphore.
- @Override
- public void execute(Runnable r) {
- semaphore.acquireUninterruptibly();
- super.execute(r);
+ public void forceExecute(Runnable work, long workBytes) {
+ monitor.enter();
+ executeLockHeld(work, workBytes);
}
// Forcibly add something to the queue, ignoring the length limit.
public void forceExecute(Runnable r) {
- semaphore.reducePermits(1);
- super.execute(r);
+ forceExecute(r, 0);
Review comment:
nit: since this is kind of gross, I'd remove it from this class and make
the caller pass zero. Then callers at realize they should pass real bytes or
use another executor.
--
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]
