Fix calling watchdog from different thread Previously it was assumed that Response would only be accessed on the primary thread for a request. However, our own builtin Command Line Adaptor actually used Response from a different thread. With this change you can now use Response on a different thread than getDocContent(), although the object is still not thread-safe.

diff --git a/src/com/google/enterprise/adaptor/DocumentHandler.java b/src/com/google/enterprise/adaptor/DocumentHandler.java
index 813fa87..f72fb8e 100644
--- a/src/com/google/enterprise/adaptor/DocumentHandler.java
+++ b/src/com/google/enterprise/adaptor/DocumentHandler.java

@@ -205,7 +205,8 @@
       }
 
       DocumentRequest request = new DocumentRequest(ex, docId);
-      DocumentResponse response = new DocumentResponse(ex, docId);
+      DocumentResponse response
+          = new DocumentResponse(ex, docId, Thread.currentThread());
       journal.recordRequestProcessingStart();
       watchdog.processingStarting(headerTimeoutMillis);
       try {
@@ -537,6 +538,7 @@
    * need to be very aware of all the different possibilities.
    */
   private class DocumentResponse implements Response {
+    private Thread workingThread;
     private State state = State.SETUP;
     private HttpExchange ex;
     // Whether ex.getResponseBody().close() has been called while we are in the
@@ -562,9 +564,10 @@
     private boolean lock;
     private Map<String, Acl> fragments = new TreeMap<String, Acl>();
 
-    public DocumentResponse(HttpExchange ex, DocId docId) {
+    public DocumentResponse(HttpExchange ex, DocId docId, Thread thread) {
       this.ex = ex;
       this.docId = docId;
+      this.workingThread = thread;
     }
 
     @Override
@@ -841,8 +844,8 @@
       }
       // There are separate timeouts for sending headers and sending content.
       // Here we stop the headers timer and start the content timer.
-      watchdog.processingCompleted();
-      watchdog.processingStarting(contentTimeoutMillis);
+      watchdog.processingCompleted(workingThread);
+      watchdog.processingStarting(workingThread, contentTimeoutMillis);
       HttpExchanges.startResponse(
           ex, HttpURLConnection.HTTP_OK, contentType, hasContent);
       for (Map.Entry<String, Acl> fragment : fragments.entrySet()) {

diff --git a/src/com/google/enterprise/adaptor/Watchdog.java b/src/com/google/enterprise/adaptor/Watchdog.java
index 65f4da9..4fee053 100644
--- a/src/com/google/enterprise/adaptor/Watchdog.java
+++ b/src/com/google/enterprise/adaptor/Watchdog.java

@@ -33,8 +33,8 @@
  */
 class Watchdog {
   private final ScheduledExecutorService executor;
-  private final ThreadLocal<FutureInfo> inProcess
-      = new ThreadLocal<FutureInfo>();
+  private final ConcurrentMap<Thread, FutureInfo> inProcess
+      = new ConcurrentHashMap<Thread, FutureInfo>();
 
   /**
    * @param executor executor to schedule tasks
@@ -50,23 +50,36 @@
    * @param timeout maximum allowed duration in milliseconds
    */
   public void processingStarting(long timeout) {
-    if (inProcess.get() != null) {
+    processingStarting(Thread.currentThread(), timeout);
+  }
+
+  public void processingStarting(Thread thread, long timeout) {
+    if (inProcess.get(thread) != null) {
       throw new IllegalStateException("Processing is already occuring on the "
           + "thread");
     }
     AtomicBoolean interruptNeeded = new AtomicBoolean(true);
-    Runnable task = new Interrupter(Thread.currentThread(), interruptNeeded);
+    Runnable task = new Interrupter(thread, interruptNeeded);
     Future<?> future = executor.schedule(task, timeout, TimeUnit.MILLISECONDS);
-    inProcess.set(new FutureInfo(future, interruptNeeded));
+    FutureInfo info = new FutureInfo(future, interruptNeeded);
+    if (inProcess.putIfAbsent(thread, info) != null) {
+      // Reset state and try again.
+      future.cancel(false);
+      processingStarting(thread, timeout);
+      return;
+    }
   }
 
   public void processingCompleted() {
-    FutureInfo info = inProcess.get();
+    processingCompleted(Thread.currentThread());
+  }
+
+  public void processingCompleted(Thread thread) {
+    FutureInfo info = inProcess.remove(thread);
     if (info == null) {
       throw new IllegalStateException("No processing was started on the "
           + "thread");
     }
-    inProcess.remove();
     // Prevent Interrupter from running if it hasn't started already. It may
     // still be running after this call and Future doesn't tell us if it is
     // currently running.
@@ -80,7 +93,7 @@
         // Interrupter has interrupted this thread.
         // Clear the interrupt, if not already cleared, since we don't want to
         // interrupt this thread any further.
-        Thread.currentThread().interrupted();
+        thread.interrupted();
       }
     }
   }

diff --git a/test/com/google/enterprise/adaptor/MockWatchdog.java b/test/com/google/enterprise/adaptor/MockWatchdog.java
index 1b89c5b..a46153e 100644
--- a/test/com/google/enterprise/adaptor/MockWatchdog.java
+++ b/test/com/google/enterprise/adaptor/MockWatchdog.java

@@ -37,6 +37,14 @@
   }
 
   @Override
+  public void processingStarting(Thread thread, long timout) {
+  }
+
+  @Override
   public void processingCompleted() {
   }
+
+  @Override
+  public void processingCompleted(Thread thread) {
+  }
 }

diff --git a/test/com/google/enterprise/adaptor/WatchdogTest.java b/test/com/google/enterprise/adaptor/WatchdogTest.java
index ed6d078..bf2629c 100644
--- a/test/com/google/enterprise/adaptor/WatchdogTest.java
+++ b/test/com/google/enterprise/adaptor/WatchdogTest.java

@@ -20,6 +20,7 @@
 import org.junit.rules.ExpectedException;
 
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Tests for {@link Watchdog}.
@@ -56,6 +57,27 @@
   }
 
   @Test
+  public void testDifferentThread() throws InterruptedException {
+    watchdog = new Watchdog(executor);
+    final Thread thread = Thread.currentThread();
+    watchdog.processingStarting(1000);
+    final AtomicBoolean success = new AtomicBoolean();
+    Thread thread2 = new Thread() {
+      @Override
+      public void run() {
+        watchdog.processingCompleted(thread);
+        watchdog.processingStarting(thread, 1000);
+        success.set(true);
+      }
+    };
+    thread2.start();
+    thread2.join();
+    assertTrue(success.get());
+
+    watchdog.processingCompleted();
+  }
+
+  @Test
   public void testNoInterruption() throws InterruptedException {
     watchdog = new Watchdog(executor);
     watchdog.processingStarting(5);