Process tasks in the AbstractQueueVisitor in LIFO order as opposed to FIFO order. In general, there's no advantage in Blaze to FIFO, and it means that we effectively do breadth-first graph traversal. When we must hold state for incomplete nodes (as we do with action execution, or more generally, as we do in Skyframe), this increases our memory footprint.

LIFO is not exactly depth-first traversal, since we are multithreaded, but to a first approximation, it looks like a depth-first traversal with "width" the number of threads (at each level of the graph, #(threads) nodes are visited).

--
MOS_MIGRATED_REVID=105995014

2 files changed, 95 insertions, 13 deletions

diff --git a/src/main/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitor.java b/src/main/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitor.java
index 7a0ab061ac..51f054746c 100644
--- a/src/main/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitor.java
+++ b/src/main/java/com/google/devtools/build/lib/concurrent/AbstractQueueVisitor.java
@@ -11,7 +11,6 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
-
 package com.google.devtools.build.lib.concurrent;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -24,7 +23,6 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -240,11 +238,7 @@ public class AbstractQueueVisitor {
         concurrent
             ? executorFactory.apply(
                 new ExecutorParams(
-                    parallelism,
-                    keepAliveTime,
-                    units,
-                    poolName,
-                    new LinkedBlockingQueue<Runnable>()))
+                    parallelism, keepAliveTime, units, poolName, new BlockingStack<Runnable>()))
             : null;
   }
 
@@ -332,10 +326,6 @@ public class AbstractQueueVisitor {
     this.pool = executor;
   }
 
-  public AbstractQueueVisitor(ThreadPoolExecutor executor, boolean failFastOnException) {
-    this(/*concurrent=*/ true, executor, true, failFastOnException, true);
-  }
-
   /**
    * Create the AbstractQueueVisitor with concurrency enabled.
    *
@@ -360,8 +350,8 @@ public class AbstractQueueVisitor {
    * they may check the interrupted bit to see if the pool was interrupted.
    *
    * @param interruptWorkers if true, interrupt worker threads if main thread gets an interrupt or
-   *        if a worker throws a critical error (see {@link #isCriticalError(Throwable)}). If
-   *        false, just wait for them to terminate normally.
+   *        if a worker throws a critical error (see {@link #classifyError(Throwable)}. If false,
+   *        just wait for them to terminate normally.
    */
   protected final void work(boolean interruptWorkers) throws InterruptedException {
     if (concurrent) {
diff --git a/src/main/java/com/google/devtools/build/lib/concurrent/BlockingStack.java b/src/main/java/com/google/devtools/build/lib/concurrent/BlockingStack.java
new file mode 100644
index 0000000000..f409bc2d73
--- /dev/null
+++ b/src/main/java/com/google/devtools/build/lib/concurrent/BlockingStack.java
@@ -0,0 +1,92 @@
+// Copyright 2015 The Bazel Authors. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.google.devtools.build.lib.concurrent;
+
+import java.util.AbstractQueue;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingDeque;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+
+/** A {@link BlockingQueue} with LIFO (last-in-first-out) ordering. */
+class BlockingStack<E> extends AbstractQueue<E> implements BlockingQueue<E> {
+  // We just restrict to only using the *First methods on the deque, turning it into a stack.
+  private final BlockingDeque<E> deque;
+
+  BlockingStack() {
+    this.deque = new LinkedBlockingDeque<>();
+  }
+
+  @Override
+  public Iterator<E> iterator() {
+    return deque.iterator();
+  }
+
+  @Override
+  public int size() {
+    return deque.size();
+  }
+
+  @Override
+  public void put(E e) throws InterruptedException {
+    deque.putFirst(e);
+  }
+
+  @Override
+  public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
+    return deque.offerFirst(e, timeout, unit);
+  }
+
+  @Override
+  public E take() throws InterruptedException {
+    return deque.takeFirst();
+  }
+
+  @Override
+  public E poll(long timeout, TimeUnit unit) throws InterruptedException {
+    return deque.pollFirst(timeout, unit);
+  }
+
+  @Override
+  public int remainingCapacity() {
+    return deque.remainingCapacity();
+  }
+
+  @Override
+  public int drainTo(Collection<? super E> c) {
+    return deque.drainTo(c);
+  }
+
+  @Override
+  public int drainTo(Collection<? super E> c, int maxElements) {
+    return deque.drainTo(c, maxElements);
+  }
+
+  @Override
+  public boolean offer(E e) {
+    return deque.offerFirst(e);
+  }
+
+  @Override
+  public E poll() {
+    return deque.pollFirst();
+  }
+
+  @Override
+  public E peek() {
+    return deque.peekFirst();
+  }
+}