diff --git a/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java b/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java index bf8d3a19..a5f6a2ff 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java +++ b/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); } } diff --git a/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java b/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java index b7038cfe..aa412d83 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java +++ b/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 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> 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);