Experiment to work-around Frontier.managerThread bottleneck

* AbstractFrontier, WorkQueueFrontier
    synchronize around all InEvent/findEligibleURI actions, so that drainInbound and fillOutbound may be called outside managerThread
    whenever a ToeThread would block on enqueue() or next(), try the appropriate catch-up method before blocking
    remove assertions no longer true with activity happening outside managerThread

engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java

@@ -396,7 +396,9 @@ public abstract class AbstractFrontier
                                 reachedState(State.PAUSE);
                             }
                             // continue to process discovered and finished URIs
-                            inbound.take().process();
+                            synchronized(this) {
+                                inbound.take().process();
+                            }
                         }
                         break;
                     case FINISH:
@@ -405,7 +407,9 @@ public abstract class AbstractFrontier
                         // process all inbound
                         while (outbound.size() != getInProcessCount()) {
                             // continue to process discovered and finished URIs
-                            inbound.take().process();
+                            synchronized(this) {
+                                inbound.take().process();
+                            }
                         }
                         finalTasks(); 
                         // TODO: more cleanup?
@@ -469,14 +473,19 @@ public abstract class AbstractFrontier
     protected void drainInbound() throws InterruptedException {
         int batch = inbound.size();
         for(int i = 0; i < batch; i++) {
-            inbound.take().process();
+            InEvent ev = inbound.take();
+            synchronized(this) {
+                ev.process();
+            }
         }
         if(batch==0) {
             // always do at least one timed try
-            InEvent toProcess = inbound.poll(getMaxInWait(),
+            InEvent ev = inbound.poll(getMaxInWait(),
                     TimeUnit.MILLISECONDS);
-            if (toProcess != null) {
-                toProcess.process();
+            if (ev != null) {
+                synchronized(this) {
+                    ev.process();
+                }
             }
         }
     }
@@ -503,10 +512,12 @@ public abstract class AbstractFrontier
         outboundLock.readLock().unlock();
         
         
-        CrawlURI retval = outbound.take();
-        // TODO: consider optimizations avoiding this recalc of
-        // overrides when not necessary
-        sheetOverlaysManager.applyOverlaysTo(retval);
+        CrawlURI retval = outbound.poll();
+        if(retval==null) {
+            fillOutbound();
+            retval = outbound.take();
+        }
+        
 //      // TODO: consider if following necessary for maintaining throughput
 //        if(outbound.size()<=1) {
 //            doOrEnqueue(NOOP);
@@ -647,7 +658,7 @@ public abstract class AbstractFrontier
      * managerThread, as by an InEvent. 
      */
     protected void processSetTargetState(State target) {
-        assert Thread.currentThread() == managerThread;
+//        assert Thread.currentThread() == managerThread;
         targetState = target;
     }
     
@@ -750,8 +761,7 @@ public abstract class AbstractFrontier
     }
 
     /**
-     * Increment the running count of queued URIs. Synchronized because
-     * operations on longs are not atomic.
+     * Increment the running count of queued URIs.
      * 
      * @param increment
      *            amount to increment the queued count
@@ -1198,23 +1208,46 @@ public abstract class AbstractFrontier
      * and this is the managerThread. 
      * @param ev InEvent to be done
      */
-    protected void enqueueOrDo(InEvent ev) {
-        if(!inbound.offer(ev)) {
-            // if can't defer, 
-            if(Thread.currentThread()==managerThread) {
-                // if can't enqueue, ok to just do
-                ev.process();
-                return; 
-            } else {
-                try {
-                    inbound.put(ev);
-                } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                }
+    protected void enqueue(InEvent ev) {
+        if(!inbound.offer(ev)) {      
+            try {
+                drainInbound();
+                inbound.put(ev);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
             }
         }
     }
     
+    /**
+     * Arrange for the given InEvent to be done by the managerThread, via
+     * enqueueing with other events if possible, but directly if not possible
+     * and this is the managerThread. 
+     * @param ev InEvent to be done
+     */
+    protected void enqueueOrDo(InEvent ev) {
+        // for now, treat same as enqueue
+        // TODO: reevaluate
+        enqueue(ev); 
+//        if(!inbound.offer(ev)) {      
+//            // if can't defer, 
+//            if(Thread.currentThread()==managerThread) {
+//                // if can't enqueue, ok to just do
+//                synchronized(this) {
+//                    ev.process();
+//                }
+//                return; 
+//            } else {
+//                try {
+//                    drainInbound();
+//                    inbound.put(ev);
+//                } catch (InterruptedException e) {
+//                    throw new RuntimeException(e);
+//                }
+//            }
+//        }
+    }
+    
     /**
      * Arrange for the given InEvent to be done by the managerThread, 
      * immediately if this is the managerThread, of via enqueueing with
@@ -1224,14 +1257,12 @@ public abstract class AbstractFrontier
     protected void doOrEnqueue(InEvent ev) {
         if (Thread.currentThread() == managerThread) {
             // if can't enqueue, ok to just do
-            ev.process();
+            synchronized(this) {
+                ev.process();
+            }
             return;
         } else {
-            try {
-                inbound.put(ev);
-            } catch (InterruptedException e) {
-                throw new RuntimeException(e);
-            }
+            enqueue(ev);
         }
     }
     

engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java

@@ -341,7 +341,7 @@ implements Closeable,
      * @param caUri CrawlURI.
      */
     protected void processScheduleAlways(CrawlURI curi) {
-        assert Thread.currentThread() == managerThread;
+//        assert Thread.currentThread() == managerThread;
         assert KeyedProperties.overridesActiveFrom(curi); 
         
         prepForFrontier(curi);
@@ -400,7 +400,7 @@ implements Closeable,
      * @param curi
      */
     protected void sendToQueue(CrawlURI curi) {
-        assert Thread.currentThread() == managerThread;
+//        assert Thread.currentThread() == managerThread;
         
         WorkQueue wq = getQueueFor(curi);
         int originalPrecedence = wq.getPrecedence();
@@ -460,7 +460,7 @@ implements Closeable,
      * @param wq
      */
     private void readyQueue(WorkQueue wq) {
-        assert Thread.currentThread() == managerThread;
+//        assert Thread.currentThread() == managerThread;
 
         try {
             wq.setActive(this, true);
@@ -482,7 +482,7 @@ implements Closeable,
      * @param wq
      */
     protected void deactivateQueue(WorkQueue wq) {
-        assert Thread.currentThread() == managerThread;
+//        assert Thread.currentThread() == managerThread;
         
         int precedence = wq.getPrecedence();
         if(!wq.getOnInactiveQueues().contains(precedence)) {
@@ -543,7 +543,7 @@ implements Closeable,
      * @param wq
      */
     protected void retireQueue(WorkQueue wq) {
-        assert Thread.currentThread() == managerThread;
+//        assert Thread.currentThread() == managerThread;
 
         getRetiredQueues().add(wq.getClassKey());
         decrementQueuedCount(wq.getCount());
@@ -594,7 +594,7 @@ implements Closeable,
      * @param q
      */
     private void unretireQueue(WorkQueue q) {
-        assert Thread.currentThread() == managerThread;
+//        assert Thread.currentThread() == managerThread;
 
         deactivateQueue(q);
         q.setRetired(false); 
@@ -631,8 +631,8 @@ implements Closeable,
      *
      * @see org.archive.crawler.framework.Frontier#next()
      */
-    protected CrawlURI findEligibleURI() {
-            assert Thread.currentThread() == managerThread;
+    protected synchronized CrawlURI findEligibleURI() {
+//            assert Thread.currentThread() == managerThread;
             // wake any snoozed queues
             wakeQueues();
             // consider rescheduled URIS
@@ -692,6 +692,9 @@ implements Closeable,
                     
                     // from queues, override names persist but not map source
                     curi.setOverlayMapsSource(sheetOverlaysManager);
+                    // TODO: consider optimizations avoiding this recalc of
+                    // overrides when not necessary
+                    sheetOverlaysManager.applyOverlaysTo(curi);
                     // check if curi belongs in different queue
                     String currentQueueKey;
                     try {
@@ -745,7 +748,7 @@ implements Closeable,
      * Check for any future-scheduled URIs now eligible for reenqueuing
      */
     protected void checkFutures() {
-        assert Thread.currentThread() == managerThread;
+//        assert Thread.currentThread() == managerThread;
         // TODO: consider only checking this every set interval
         Iterator<CrawlURI> iter = 
             futureUris.headMap(System.currentTimeMillis())
@@ -763,7 +766,7 @@ implements Closeable,
      * Activate an inactive queue, if any are available. 
      */
     private void activateInactiveQueue() {
-        assert Thread.currentThread() == managerThread;
+//        assert Thread.currentThread() == managerThread;
 
         SortedMap<Integer,Queue<String>> inactiveQueuesByPrecedence = 
             getInactiveQueuesByPrecedence();
@@ -964,7 +967,7 @@ implements Closeable,
      * @see org.archive.crawler.framework.Frontier#finished(org.archive.modules.CrawlURI)
      */
     protected void processFinish(CrawlURI curi) {
-        assert Thread.currentThread() == managerThread;
+//        assert Thread.currentThread() == managerThread;
         
         long now = System.currentTimeMillis();
 
@@ -1451,6 +1454,11 @@ implements Closeable,
         w.print(exhaustedCount);
         w.print("\n");
         
+        int inCount = inbound.size();
+        int outCount = outbound.size();
+        State last = lastReachedState;
+        w.print("\n               Threadbound: "+last+ ": "+inCount+" in, "+outCount+" out");        
+        
         w.print("\n -----===== MANAGER THREAD =====-----\n");
         ToeThread.reportThread(managerThread, w);