public void start() {if(isRunning()) {return; }uriUniqFilter.setDestination(this);super.start();try {initInternalQueues();} catch (Exception e) {throw new IllegalStateException(e);}}

 public void start() {if(isRunning()) {return; }if (getRecoveryLogEnabled()) try {initJournal(loggerModule.getPath().getFile().getAbsolutePath());} catch (IOException e) {throw new IllegalStateException(e);}pause();startManagerThread();}

/*** Start the dedicated thread with an independent view of the frontier's* state. */protected void startManagerThread() {managerThread = new Thread(this+".managerThread") {public void run() {AbstractFrontier.this.managementTasks();}};managerThread.setPriority(Thread.NORM_PRIORITY+1); managerThread.start();}

/*** Main loop of frontier's managerThread. Only exits when State.FINISH * is requested (perhaps automatically at URI exhaustion) and reached. * * General strategy is to try to fill outbound queue, then process an* item from inbound queue, and repeat. A HOLD (to be implemented) or * PAUSE puts frontier into a stable state that won't be changed* asynchronously by worker thread activity. */protected void managementTasks() {assert Thread.currentThread() == managerThread;try {loop: while (true) {try {State reachedState = null; switch (targetState) {case EMPTY:reachedState = State.EMPTY; case RUN:// enable outbound takes if previously lockedwhile(outboundLock.isWriteLockedByCurrentThread()) {outboundLock.writeLock().unlock();}if(reachedState==null) {reachedState = State.RUN; }reachedState(reachedState);Thread.sleep(1000);if(isEmpty()&&targetState==State.RUN) {requestState(State.EMPTY); } else if (!isEmpty()&&targetState==State.EMPTY) {requestState(State.RUN); }break;case HOLD:// TODO; for now treat same as PAUSEcase PAUSE:// pausing// prevent all outbound takes
                        outboundLock.writeLock().lock();// process all inboundwhile (targetState == State.PAUSE) {if (getInProcessCount()==0) {reachedState(State.PAUSE);}Thread.sleep(1000);}break;case FINISH:// prevent all outbound takes
                        outboundLock.writeLock().lock();// process all inboundwhile (getInProcessCount()>0) {Thread.sleep(1000);}finalTasks(); // TODO: more cleanup?
                        reachedState(State.FINISH);break loop;}} catch (RuntimeException e) {// log, try to pause, continuelogger.log(Level.SEVERE,"",e);if(targetState!=State.PAUSE && targetState!=State.FINISH) {requestState(State.PAUSE);}}}} catch (InterruptedException e) {throw new RuntimeException(e);} // try to leave in safely restartable state: targetState = State.PAUSE;while(outboundLock.isWriteLockedByCurrentThread()) {outboundLock.writeLock().unlock();}//TODO: ensure all other structures are cleanly reset on restart
        logger.log(Level.FINE,"ending frontier mgr thread");}

/*** Initializes internal queues.  May decide to keep all queues in memory based on* {@link QueueAssignmentPolicy#maximumNumberOfKeys}.  Otherwise invokes* {@link #initAllQueues()} to actually set up the queues.* * Subclasses should invoke this method with recycle set to "true" in * a private readObject method, to restore queues after a checkpoint.* * @param recycle* @throws IOException* @throws DatabaseException*/protected void initInternalQueues() throws IOException, DatabaseException {this.initOtherQueues();if (workQueueDataOnDisk()&& preparer.getQueueAssignmentPolicy().maximumNumberOfKeys() >= 0&& preparer.getQueueAssignmentPolicy().maximumNumberOfKeys() <= MAX_QUEUES_TO_HOLD_ALLQUEUES_IN_MEMORY) {this.allQueues = new ObjectIdentityMemCache<WorkQueue>(701, .9f, 100);} else {this.initAllQueues();}}

@Overrideprotected void initOtherQueues() throws DatabaseException {boolean recycle = (recoveryCheckpoint != null);// tiny risk of OutOfMemoryError: if giant number of snoozed// queues all wake-to-ready at oncereadyClassQueues = new LinkedBlockingQueue<String>();inactiveQueuesByPrecedence = new ConcurrentSkipListMap<Integer,Queue<String>>();retiredQueues = bdb.getStoredQueue("retiredQueues", String.class, recycle);// primary snoozed queuessnoozedClassQueues = new DelayQueue<DelayedWorkQueue>();// just in case: overflow for extreme situationssnoozedOverflow = bdb.getStoredMap("snoozedOverflow", Long.class, DelayedWorkQueue.class, true, false);this.futureUris = bdb.getStoredMap("futureUris", Long.class, CrawlURI.class, true, recoveryCheckpoint!=null);// initialize master map in which other queues livethis.pendingUris = createMultipleWorkQueues();}

@Overrideprotected void initAllQueues() throws DatabaseException {boolean isRecovery = (recoveryCheckpoint != null);this.allQueues = bdb.getObjectCache("allqueues", isRecovery, WorkQueue.class, BdbWorkQueue.class);if(isRecovery) {// restore simple instance fields JSONObject json = recoveryCheckpoint.loadJson(beanName);try {nextOrdinal.set(json.getLong("nextOrdinal"));queuedUriCount.set(json.getLong("queuedUriCount"));futureUriCount.set(json.getLong("futureUriCount"));succeededFetchCount.set(json.getLong("succeededFetchCount"));failedFetchCount.set(json.getLong("failedFetchCount"));disregardedUriCount.set(json.getLong("disregardedUriCount"));totalProcessedBytes.set(json.getLong("totalProcessedBytes"));JSONArray inactivePrecedences = json.getJSONArray("inactivePrecedences"); // restore all intended inactiveQueuesfor(int i = 0; i < inactivePrecedences.length(); i++) {int precedence = inactivePrecedences.getInt(i);inactiveQueuesByPrecedence.put(precedence,createInactiveQueueForPrecedence(precedence,true));}} catch (JSONException e) {throw new RuntimeException(e);}           // retired queues already restored with prior data in initOtherQueues// restore ready queues (those not already on inactive, retired)BufferedReader activeQueuesReader = null;try {activeQueuesReader = recoveryCheckpoint.loadReader(beanName,"active");String line; while((line = activeQueuesReader.readLine())!=null) {readyClassQueues.add(line); }} catch (IOException ioe) {throw new RuntimeException(ioe); } finally {IOUtils.closeQuietly(activeQueuesReader); }// TODO: restore largestQueues topNset?
        }}