Throw an exception when the local resources will never be available.

PiperOrigin-RevId: 755749640
Change-Id: I9216a472bc443c92c2a341044e96ff90296f8ada

Author: bigelephant29

src/main/java/com/google/devtools/build/lib/actions/ResourceManager.java

@@ -112,14 +112,15 @@ ResourceRequest getRequest() {
 
     /** Closing the ResourceHandle releases the resources associated with it. */
     @Override
-    public void close() throws IOException, InterruptedException {
+    public void close() throws IOException, InterruptedException, UserExecException {
       manager.releaseResources(request, worker);
       Profiler.instance()
           .completeTask(
               resourceAcquiredTime, ProfilerTask.LOCAL_ACTION_COUNTS, "Resources acquired");
     }
 
-    public void invalidateAndClose(@Nullable Exception e) throws IOException, InterruptedException {
+    public void invalidateAndClose(@Nullable Exception e)
+        throws IOException, InterruptedException, UserExecException {
       // If there is an exception, we need to set the kill cause before invalidating the object.
       // This ensures that the worker implementation updates their worker metrics accordingly
       // if/when it destroys itself.
@@ -279,14 +280,14 @@ public WindowUpdateRunner(String name) {
     public void run() {
       try {
         windowUpdate();
-      } catch (IOException | InterruptedException e) {
+      } catch (IOException | InterruptedException | UserExecException e) {
         logger.atWarning().withCause(e).log(
             "Exception while updating window of locally scheduled action: %s", e);
       }
     }
   }
 
-  synchronized void windowUpdate() throws IOException, InterruptedException {
+  synchronized void windowUpdate() throws IOException, InterruptedException, UserExecException {
     windowRequestIds.clear();
     windowEstimationCpu = 0.0;
     processAllWaitingRequests();
@@ -468,7 +469,7 @@ public boolean threadHasResources() {
    * @throws java.io.IOException if could not return worker to the workerPool
    */
   void releaseResources(ResourceRequest request, @Nullable Worker worker)
-      throws IOException, InterruptedException {
+      throws IOException, InterruptedException, UserExecException {
     Preconditions.checkNotNull(
         request.getResourceSet(),
         "releaseResources called with resources == NULL during %s",
@@ -500,7 +501,7 @@ public void acquireResourceOwnership() {
    * wait.
    */
   private synchronized ResourceLatch acquire(ResourceRequest request)
-      throws IOException, InterruptedException {
+      throws IOException, InterruptedException, UserExecException {
     if (areResourcesAvailable(request.getResourceSet())) {
       Worker worker = incrementResources(request);
       return new ResourceLatch(/* latch= */ null, worker);
@@ -523,7 +524,7 @@ private synchronized ResourceLatch acquire(ResourceRequest request)
 
   /** Release resources and process the queues of waiting threads. */
   private synchronized void release(ResourceRequest request, @Nullable Worker worker)
-      throws IOException, InterruptedException {
+      throws IOException, InterruptedException, UserExecException {
     if (worker != null) {
       this.workerPool.returnWorker(worker.getWorkerKey(), worker);
     }
@@ -555,14 +556,15 @@ private synchronized void release(ResourceRequest request, @Nullable Worker work
     processAllWaitingRequests();
   }
 
-  private synchronized void processAllWaitingRequests() throws IOException, InterruptedException {
+  private synchronized void processAllWaitingRequests()
+      throws IOException, InterruptedException, UserExecException {
     processWaitingRequests(localRequests);
     processWaitingRequests(dynamicWorkerRequests);
     processWaitingRequests(dynamicStandaloneRequests);
   }
 
   private synchronized void processWaitingRequests(Deque<WaitingRequest> requests)
-      throws IOException, InterruptedException {
+      throws IOException, InterruptedException, UserExecException {
     if (requests.isEmpty()) {
       return;
     }
@@ -607,10 +609,29 @@ private void assertResourcesTracked(ResourceSet resources) throws ExecException
     }
   }
 
-  private <T extends Number> boolean isAvailable(T available, T used, T requested) {
+  private <T extends Number> boolean isAvailable(
+      T available, T used, T requested, String resourceName) throws UserExecException {
+    if (!allowOneActionOnResourceUnavailable
+        && available.doubleValue() + used.doubleValue() < requested.doubleValue()) {
+      throw new UserExecException(
+          FailureDetails.FailureDetail.newBuilder()
+              .setMessage(
+                  String.format(
+                      "The `%s` resources are not enough to fulfill the request. To allow Bazel to"
+                          + " bypass this limitation, please adjust the --local_resources flag or"
+                          + " specify --allow_one_action_on_resource_unavailable in your Bazel"
+                          + " command.",
+                      resourceName))
+              .setLocalExecution(
+                  FailureDetails.LocalExecution.newBuilder()
+                      .setCode(FailureDetails.LocalExecution.Code.NOT_ENOUGH_LOCAL_RESOURCE)
+                      .build())
+              .build());
+    }
     // Resources are considered available if any one of the conditions below is true:
     // 1) If resource is not requested at all, it is available.
-    // 2) If resource is not used at the moment, it is considered to be
+    // 2) If resource is not used at the moment and the flag
+    // "allow_one_action_on_resource_unavailable" is enabled, it is considered to be
     // available regardless of how much is requested. This is necessary to
     // ensure that at any given time, at least one thread is able to acquire
     // resources even if it requests more than available.
@@ -622,7 +643,7 @@ private <T extends Number> boolean isAvailable(T available, T used, T requested)
 
   // Method will return true if all requested resources are considered to be available.
   @VisibleForTesting
-  synchronized boolean areResourcesAvailable(ResourceSet resources) {
+  synchronized boolean areResourcesAvailable(ResourceSet resources) throws UserExecException {
     Preconditions.checkNotNull(availableResources);
     // Comparison below is robust, since any calculation errors will be fixed
     // by the release() method.
@@ -640,7 +661,11 @@ synchronized boolean areResourcesAvailable(ResourceSet resources) {
     }
 
     int availableLocalTestCount = availableResources.getLocalTestCount();
-    if (!isAvailable(availableLocalTestCount, usedLocalTestCount, resources.getLocalTestCount())) {
+    if (!isAvailable(
+        availableLocalTestCount,
+        usedLocalTestCount,
+        resources.getLocalTestCount(),
+        "local_test_count")) {
       return false;
     }
 
@@ -662,14 +687,14 @@ synchronized boolean areResourcesAvailable(ResourceSet resources) {
           resource.getValue() * MIN_NECESSARY_RATIO.getOrDefault(key, DEFAULT_MIN_NECESSARY_RATIO);
       double used = usedResources.getOrDefault(key, 0.0);
       double available = availableResources.get(key);
-      if (!isAvailable(available, used, requested)) {
+      if (!isAvailable(available, used, requested, key)) {
         return false;
       }
     }
     return true;
   }
 
-  synchronized boolean isCpuAvailable(Map.Entry<String, Double> resource) {
+  synchronized boolean isCpuAvailable(Map.Entry<String, Double> resource) throws UserExecException {
     String key = resource.getKey();
 
     double requested =
@@ -684,10 +709,10 @@ synchronized boolean isCpuAvailable(Map.Entry<String, Double> resource) {
       if (runningActions >= MAX_ACTIONS_PER_CPU * availableResources.get(ResourceSet.CPU)) {
         return false;
       }
-      return isAvailable(available, windowEstimation + currentUsage, requested);
+      return isAvailable(available, windowEstimation + currentUsage, requested, key);
     }
 
-    return isAvailable(available, used, requested);
+    return isAvailable(available, used, requested, key);
   }
 
   @VisibleForTesting

src/main/java/com/google/devtools/build/lib/worker/WorkerSpawnRunner.java

@@ -645,15 +645,18 @@ private void finishWorkAsync(
 
                   w = null;
 
-                } catch (IOException | InterruptedException e2) {
+                } catch (IOException | InterruptedException | UserExecException e2) {
                   // The reaper thread can't do anything useful about this.
                 }
               } finally {
                 if (w != null) {
                   try {
                     resourceHandle.close();
-                  } catch (IOException | InterruptedException | IllegalStateException e) {
-                    // Error while returning worker to the pool. Could not do anythinng.
+                  } catch (IOException
+                      | InterruptedException
+                      | IllegalStateException
+                      | UserExecException e) {
+                    // Error while returning worker to the pool. Could not do anything.
                   }
                 }
               }

src/main/protobuf/failure_details.proto

@@ -778,6 +778,7 @@ message LocalExecution {
     LOCAL_EXECUTION_UNKNOWN = 0 [(metadata) = { exit_code: 37 }];
     LOCKFREE_OUTPUT_PREREQ_UNMET = 1 [(metadata) = { exit_code: 2 }];
     UNTRACKED_RESOURCE = 2 [(metadata) = { exit_code: 1 }];
+    NOT_ENOUGH_LOCAL_RESOURCE = 3 [(metadata) = { exit_code: 1 }];
   }
 
   Code code = 1;

src/test/java/com/google/devtools/build/lib/actions/ResourceManagerTest.java

@@ -137,7 +137,8 @@ private ResourceHandle acquire(
     return acquire(ram, cpu, extraResources, tests, ResourcePriority.LOCAL);
   }
 
-  private void release(ResourceHandle resourceHandle) throws IOException, InterruptedException {
+  private void release(ResourceHandle resourceHandle)
+      throws IOException, InterruptedException, UserExecException {
     manager.releaseResources(resourceHandle.getRequest(), /* worker= */ null);
   }
 
@@ -167,8 +168,8 @@ public void allowOneActionOnResourceUnavailable_success() throws Exception {
     // TODO: b/405364605 - Add a test for false case after we start throwing an exception.
     manager.setAllowOneActionOnResourceUnavailable(true);
 
-    // When nothing is consuming RAM,
-    // Then Resource Manager will successfully acquire an over-budget request for RAM:
+    // When nothing is consuming RAM, then ResourceManager will successfully acquire an over-budget
+    // request for RAM if allow_one_action_on_resource_unavailable is set to true.
     double bigRam = 10000.0;
     ResourceHandle bigRamHandle = acquire(bigRam, 0, 0);
     // When RAM is consumed,
@@ -202,6 +203,16 @@ public void allowOneActionOnResourceUnavailable_success() throws Exception {
     assertThat(manager.inUse()).isFalse();
   }
 
+  @Test
+  public void noAllowOneActionOnResourceUnavailable_exception() throws Exception {
+    assertThat(manager.inUse()).isFalse();
+
+    // When nothing is consuming RAM, then ResourceManager should fail to acquire an over-budget
+    // request for RAM, when allow_one_action_on_resource_unavailable is not explicitly set.
+    double bigRam = 10000.0;
+    assertThrows(UserExecException.class, () -> acquire(bigRam, 0, 0));
+  }
+
   @Test
   public void testThatCpuCanBeOverallocated() throws Exception {
     assertThat(manager.inUse()).isFalse();
@@ -395,14 +406,14 @@ public void testInterruptedAcquisitionClearsResources() throws Exception {
     assertThat(manager.inUse()).isFalse();
     // Acquire a small amount of resources so that future requests can block (the initial request
     // always succeeds even if it's for too much).
-    TestThread smallThread = new TestThread(() -> acquire(1, 0, 0));
+    TestThread smallThread = new TestThread(() -> acquire(400, 0, 0));
     smallThread.start();
     smallThread.joinAndAssertState(TestUtils.WAIT_TIMEOUT_MILLISECONDS);
     TestThread thread1 =
         new TestThread(
             () -> {
               Thread.currentThread().interrupt();
-              assertThrows(InterruptedException.class, () -> acquire(1999, 0, 0));
+              assertThrows(InterruptedException.class, () -> acquire(700, 0, 0));
             });
     thread1.start();
     thread1.joinAndAssertState(TestUtils.WAIT_TIMEOUT_MILLISECONDS);
@@ -413,7 +424,7 @@ public void testInterruptedAcquisitionClearsResources() throws Exception {
     TestThread thread2 =
         new TestThread(
             () -> {
-              ResourceHandle handle = acquire(1999, 0, 0);
+              ResourceHandle handle = acquire(700, 0, 0);
               release(handle);
             });
     thread2.start();
@@ -817,7 +828,8 @@ public void testCPULoadScheduling_cantAcquireX3Cpu() throws Exception {
     assertThat(e).hasCauseThat().hasMessageThat().contains("is still alive");
   }
 
-  synchronized boolean isAvailable(ResourceManager rm, double ram, double cpu, int localTestCount) {
+  synchronized boolean isAvailable(ResourceManager rm, double ram, double cpu, int localTestCount)
+      throws UserExecException {
     return rm.areResourcesAvailable(ResourceSet.create(ram, cpu, localTestCount));
   }