From 5e352afe2b35487ea2ced85ca79bd9f79858e648 Mon Sep 17 00:00:00 2001
From: larsrc <larsrc@google.com>
Date: Thu, 22 Jul 2021 02:25:30 -0700
Subject: [PATCH] Fix bug in WorkRequestHandler's handling of singleplex
 requests that would cause occasional hangs.

RELNOTES: None.
PiperOrigin-RevId: 386194200
---
 .../build/lib/worker/WorkRequestHandler.java  | 34 +++++++++++++------
 .../build/lib/worker/ExampleWorker.java       |  9 ++++-
 2 files changed, 32 insertions(+), 11 deletions(-)

diff --git a/src/main/java/com/google/devtools/build/lib/worker/WorkRequestHandler.java b/src/main/java/com/google/devtools/build/lib/worker/WorkRequestHandler.java
index 969f261fffa18f..6e897b241fa6cd 100644
--- a/src/main/java/com/google/devtools/build/lib/worker/WorkRequestHandler.java
+++ b/src/main/java/com/google/devtools/build/lib/worker/WorkRequestHandler.java
@@ -305,26 +305,40 @@ public WorkRequestHandler build() {
    * returns. If {@code in} reaches EOF, it also returns.
    */
   public void processRequests() throws IOException {
-    while (true) {
-      WorkRequest request = messageProcessor.readWorkRequest();
-      if (request == null) {
-        break;
-      }
-      if (request.getCancel()) {
-        respondToCancelRequest(request);
-      } else {
-        startResponseThread(request);
+    try {
+      while (true) {
+        WorkRequest request = messageProcessor.readWorkRequest();
+        if (request == null) {
+          break;
+        }
+        if (request.getCancel()) {
+          respondToCancelRequest(request);
+        } else {
+          startResponseThread(request);
+        }
       }
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      stderr.println("InterruptedException processing requests.");
     }
   }
 
   /** Starts a thread for the given request. */
-  void startResponseThread(WorkRequest request) {
+  void startResponseThread(WorkRequest request) throws InterruptedException {
     Thread currentThread = Thread.currentThread();
     String threadName =
         request.getRequestId() > 0
             ? "multiplex-request-" + request.getRequestId()
             : "singleplex-request";
+    // TODO(larsrc): See if this can be handled with a queue instead, without introducing more
+    // race conditions.
+    if (request.getRequestId() == 0) {
+      while (activeRequests.containsKey(request.getRequestId())) {
+        // b/194051480: Previous singleplex requests can still be in activeRequests for a bit after
+        // the response has been sent. We need to wait for them to vanish.
+        Thread.sleep(1);
+      }
+    }
     Thread t =
         new Thread(
             () -> {
diff --git a/src/test/java/com/google/devtools/build/lib/worker/ExampleWorker.java b/src/test/java/com/google/devtools/build/lib/worker/ExampleWorker.java
index a032c132aaf286..d0db58356dfecd 100644
--- a/src/test/java/com/google/devtools/build/lib/worker/ExampleWorker.java
+++ b/src/test/java/com/google/devtools/build/lib/worker/ExampleWorker.java
@@ -98,7 +98,14 @@ public void processRequests() throws IOException {
         if (request.getCancel()) {
           respondToCancelRequest(request);
         } else {
-          startResponseThread(request);
+          try {
+            startResponseThread(request);
+          } catch (InterruptedException e) {
+            // We don't expect interrupts at this level, only inside the individual request
+            // handling threads, so here we just abort on interrupt.
+            e.printStackTrace();
+            return;
+          }
         }
         if (workerOptions.exitAfter > 0 && workUnitCounter > workerOptions.exitAfter) {
           System.exit(0);