You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2016/04/08 23:26:51 UTC
[1/3] incubator-beam git commit: Use a weakValues LoadingCache for
serial TransformExecutorServices
Repository: incubator-beam
Updated Branches:
refs/heads/master 529bcdf56 -> 5d78420bf
Use a weakValues LoadingCache for serial TransformExecutorServices
This allows the garbage collector to clean up references to
TransformExecutorServices which are not currently in use.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/6f526374
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/6f526374
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/6f526374
Branch: refs/heads/master
Commit: 6f526374fbc743a0d22e37ad0f746f0a695785dd
Parents: ad58e26
Author: Thomas Groh <tg...@google.com>
Authored: Tue Mar 29 10:56:10 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Fri Apr 8 14:20:20 2016 -0700
----------------------------------------------------------------------
.../ExecutorServiceParallelExecutor.java | 36 ++++++++++++++------
1 file changed, 25 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6f526374/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/ExecutorServiceParallelExecutor.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/ExecutorServiceParallelExecutor.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/ExecutorServiceParallelExecutor.java
index 4d45e8f..c770735 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/ExecutorServiceParallelExecutor.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/ExecutorServiceParallelExecutor.java
@@ -31,6 +31,9 @@ import com.google.cloud.dataflow.sdk.values.PCollection;
import com.google.cloud.dataflow.sdk.values.PValue;
import com.google.common.base.MoreObjects;
import com.google.common.base.Optional;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableList;
import org.joda.time.Instant;
@@ -69,7 +72,7 @@ final class ExecutorServiceParallelExecutor implements InProcessExecutor {
private final InProcessEvaluationContext evaluationContext;
- private final ConcurrentMap<StepAndKey, TransformExecutorService> currentEvaluations;
+ private final LoadingCache<StepAndKey, TransformExecutorService> executorServices;
private final ConcurrentMap<TransformExecutor<?>, Boolean> scheduledExecutors;
private final Queue<ExecutorUpdate> allUpdates;
@@ -107,8 +110,12 @@ final class ExecutorServiceParallelExecutor implements InProcessExecutor {
this.transformEnforcements = transformEnforcements;
this.evaluationContext = context;
- currentEvaluations = new ConcurrentHashMap<>();
scheduledExecutors = new ConcurrentHashMap<>();
+ // Weak Values allows TransformExecutorServices that are no longer in use to be reclaimed.
+ // Executing TransformExecutorServices have a strong reference to their TransformExecutorService
+ // which stops the TransformExecutorServices from being prematurely garbage collected
+ executorServices =
+ CacheBuilder.newBuilder().weakValues().build(serialTransformExecutorServiceCacheLoader());
this.allUpdates = new ConcurrentLinkedQueue<>();
this.visibleUpdates = new ArrayBlockingQueue<>(20);
@@ -118,6 +125,16 @@ final class ExecutorServiceParallelExecutor implements InProcessExecutor {
defaultCompletionCallback = new DefaultCompletionCallback();
}
+ private CacheLoader<StepAndKey, TransformExecutorService>
+ serialTransformExecutorServiceCacheLoader() {
+ return new CacheLoader<StepAndKey, TransformExecutorService>() {
+ @Override
+ public TransformExecutorService load(StepAndKey stepAndKey) throws Exception {
+ return TransformExecutorServices.serial(executorService, scheduledExecutors);
+ }
+ };
+ }
+
@Override
public void start(Collection<AppliedPTransform<?, ?, ?>> roots) {
rootNodes = ImmutableList.copyOf(roots);
@@ -142,7 +159,12 @@ final class ExecutorServiceParallelExecutor implements InProcessExecutor {
if (bundle != null && isKeyed(bundle.getPCollection())) {
final StepAndKey stepAndKey =
StepAndKey.of(transform, bundle == null ? null : bundle.getKey());
- transformExecutor = getSerialExecutorService(stepAndKey);
+ // This executor will remain reachable until it has executed all scheduled transforms.
+ // The TransformExecutors keep a strong reference to the Executor, the ExecutorService keeps
+ // a reference to the scheduled TransformExecutor callable. Follow-up TransformExecutors
+ // (scheduled due to the completion of another TransformExecutor) are provided to the
+ // ExecutorService before the Earlier TransformExecutor callable completes.
+ transformExecutor = executorServices.getUnchecked(stepAndKey);
} else {
transformExecutor = parallelExecutorService;
}
@@ -174,14 +196,6 @@ final class ExecutorServiceParallelExecutor implements InProcessExecutor {
}
}
- private TransformExecutorService getSerialExecutorService(StepAndKey stepAndKey) {
- if (!currentEvaluations.containsKey(stepAndKey)) {
- currentEvaluations.putIfAbsent(
- stepAndKey, TransformExecutorServices.serial(executorService, scheduledExecutors));
- }
- return currentEvaluations.get(stepAndKey);
- }
-
@Override
public void awaitCompletion() throws Throwable {
VisibleExecutorUpdate update;
[2/3] incubator-beam git commit: Schedule all pending work before
firing timers
Posted by lc...@apache.org.
Schedule all pending work before firing timers
Pull all available work off of the ExecutorUpdate queue during each
execution of the MonitorRunnable.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/ad58e26d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/ad58e26d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/ad58e26d
Branch: refs/heads/master
Commit: ad58e26db132268c676ca9289b73e1ddab72627a
Parents: 529bcdf
Author: Thomas Groh <tg...@google.com>
Authored: Tue Mar 29 10:27:37 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Fri Apr 8 14:20:20 2016 -0700
----------------------------------------------------------------------
.../sdk/runners/inprocess/ExecutorServiceParallelExecutor.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ad58e26d/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/ExecutorServiceParallelExecutor.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/ExecutorServiceParallelExecutor.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/ExecutorServiceParallelExecutor.java
index 9af6f97..4d45e8f 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/ExecutorServiceParallelExecutor.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/inprocess/ExecutorServiceParallelExecutor.java
@@ -325,13 +325,15 @@ final class ExecutorServiceParallelExecutor implements InProcessExecutor {
Thread.currentThread().setName(runnableName);
try {
ExecutorUpdate update = allUpdates.poll();
- if (update != null) {
+ // pull all of the pending work off of the queue
+ while (update != null) {
LOG.debug("Executor Update: {}", update);
if (update.getBundle().isPresent()) {
scheduleConsumers(update.getBundle().get());
} else if (update.getException().isPresent()) {
visibleUpdates.offer(VisibleExecutorUpdate.fromThrowable(update.getException().get()));
}
+ update = allUpdates.poll();
}
boolean timersFired = fireTimers();
addWorkIfNecessary(timersFired);
[3/3] incubator-beam git commit: [BEAM-22] This closes #143
Posted by lc...@apache.org.
[BEAM-22] This closes #143
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/5d78420b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/5d78420b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/5d78420b
Branch: refs/heads/master
Commit: 5d78420bf31ca7e289fc956dc5365d920afd2439
Parents: 529bcdf 6f52637
Author: Luke Cwik <lc...@google.com>
Authored: Fri Apr 8 14:21:30 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Fri Apr 8 14:21:30 2016 -0700
----------------------------------------------------------------------
.../ExecutorServiceParallelExecutor.java | 40 ++++++++++++++------
1 file changed, 28 insertions(+), 12 deletions(-)
----------------------------------------------------------------------