package ee.pri.rl.crawler.core.thread;import java.util.Collection;import ee.pri.rl.crawler.core.CrawlerURL;import ee.pri.rl.crawler.core.queue.CrawlerQueue;import ee.pri.rl.crawler.core.visited.CrawlerVisitedSet;import ee.pri.rl.crawler.util.Mutex;/** * A consept for correct locking * of queue and visitedset for crawler's main loop. * The type parameter T is for abstract URL representation * used by given crawler. * TODO add more comments */public class MainloopMutex<T extends CrawlerURL> { private static final int EMPTY_QUEUE_SLEEP = 500; private CrawlerQueue<T> queue; private CrawlerVisitedSet<T> visited; private Mutex readMutex; private ProcessingIndicator processingIndicator; public MainloopMutex(CrawlerQueue<T> queue, CrawlerVisitedSet<T> visited) { this.queue = queue; this.visited = visited; readMutex = new Mutex(); processingIndicator = new ProcessingIndicator(); } private class ProcessingIndicator { private int counter; public ProcessingIndicator() { counter = 0; } public synchronized boolean isProcessing() { return counter > 0; } public synchronized void increase() { counter++; } public synchronized void decrease() { counter--; } } public T read() throws MainloopTermination { // let only one thread to try reading the url // this quarantees the correct setting of 'processing' // variable T element = null; synchronized (readMutex) { // read cycle while (true) { // try to read and url from queue element = queue.pop(); if (null == element) { // if we got nothing then we can hope that other threads would // produce new urls if (processingIndicator.isProcessing()) { // sleep and let other threads to produce // new urls try { Thread.sleep(EMPTY_QUEUE_SLEEP); } catch (InterruptedException e) { // in case that thread got interrupted // we throw the end of this mainloop throw new MainloopTermination(); } } else { // if other threads were not processing any url // nothing will be put in queue // let other threads into read block and throw // termination of this crawling thread throw new MainloopTermination(); } } else { // we got an url // indicate that the current thread will process an url processingIndicator.increase(); return element; } } } } public void write(Collection<T> urls) { writeURLs(urls); processingIndicator.decrease(); } public void writeStarts(Collection<T> urls) { writeURLs(urls); } public void writeURLs(Collection<T> urls) { for (T url : urls) { if (!visited.isVisited(url)) { visited.visited(url); queue.push(url); } } }}