daoas/fulddaekning/LookupWorker.java

package dk.daoas.fulddaekning;

import java.sql.SQLException;
import java.util.Collection;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.jwetherell.algorithms.data_structures.KdTree;

import dk.daoas.fulddaekning.osrm.OSRMHelper;

public class LookupWorker implements Runnable {

        static final int MAX_RESULTS = 100;


        final static Logger logger = Logger.getLogger( LookupWorker.class.toString() );

        OSRMHelper osrmHelper = new OSRMHelper();

        CyclicBarrier barrier;
        Queue<Adresse> ikkeDaekkede;
        Map<Short, KdTree<Adresse>> hoTrees;
        Database db;
        Constants consts;
        
        AtomicInteger antalFundne;

        int workerID;
        boolean verbose;
        
        public LookupWorker(int workerID, 
                                CyclicBarrier barrier, 
                                Queue<Adresse> ikkeDaekkede, 
                                Map<Short, KdTree<Adresse>> hoTrees, 
                                AtomicInteger antalFundne,
                                Database db,
                                boolean verbose,
                                Constants consts) {
                
                this.workerID = workerID;
                this.barrier = barrier;
                this.ikkeDaekkede = ikkeDaekkede;
                this.hoTrees = hoTrees;
                this.antalFundne = antalFundne;
                this.db = db;
                this.verbose = verbose;
                this.consts = consts;
                
        }


        @Override
        public void run() {
                System.out.println( "Worker/" + workerID + " starting");

                for (Adresse qAdresse : ikkeDaekkede) {

                        short ho = qAdresse.ho;
                        KdTree<Adresse> addressTree = hoTrees.get(ho);

                        Collection<Adresse> res = addressTree.nearestNeighbourSearch(MAX_RESULTS, qAdresse);
                        if (res.size() > MAX_RESULTS ) {
                                System.out.println("Returnerede mere end MAX_RESULTS> " + res.size() );
                                System.exit(0);
                        }


                        Adresse bedsteAdr = osrmHelper.getNearestViaTable(qAdresse, res);
                        //System.out.println("Bedste:" + bedsteAdr );

                        double bedsteAfstand = GeoPointHelper.beregnAfstand(qAdresse, bedsteAdr);


                        if (bedsteAfstand <= consts.getMaxDistance() ) { //Gem kun hvis distancen er under max_distance for distributøren
                                if (verbose == true) {
                                        logger.fine("(" +workerID + ") Bedste: " + qAdresse + "==>" + bedsteAdr + " ::: " + bedsteAfstand );
                                }

                                antalFundne.incrementAndGet();

                                try {
                                        db.gemResultat(qAdresse, bedsteAdr, bedsteAfstand);
                                } catch (SQLException e) {
                                        logger.log(Level.WARNING, "Error saving result", e);
                                        logger.warning("Shutting down");

                                        System.exit(0); //hvis at indsættelsen fejler skal vi afbryde med det samme
                                }
                        } else {
                                if (verbose == true) {
                                        System.out.println("For stor afstand: " + qAdresse + " > " + bedsteAfstand);
                                }
                        }

                }

                try {
                        barrier.await();
                } catch (BrokenBarrierException e) {
                        logger.log(Level.SEVERE, e.getMessage(), e);
                        System.exit(-1);
                } catch(InterruptedException e) {
                        //Do nothing
                }


                System.out.println( "Worker/" + workerID + " stopping");
        }

}
1	package dk.daoas.fulddaekning;
2
3	import java.sql.SQLException;
4	import java.util.Collection;
5	import java.util.Map;
6	import java.util.Queue;
7	import java.util.concurrent.BrokenBarrierException;
8	import java.util.concurrent.CyclicBarrier;
9	import java.util.concurrent.atomic.AtomicInteger;
10	import java.util.logging.Level;
11	import java.util.logging.Logger;
12
13	import com.jwetherell.algorithms.data_structures.KdTree;
14
15	import dk.daoas.fulddaekning.osrm.OSRMHelper;
16
17	public class LookupWorker implements Runnable {
18
19	static final int MAX_RESULTS = 100;
20
21
22
23	final static Logger logger = Logger.getLogger( LookupWorker.class.toString() );
24
25	OSRMHelper osrmHelper = new OSRMHelper();
26
27	CyclicBarrier barrier;
28	Queue<Adresse> ikkeDaekkede;
29	Map<Short, KdTree<Adresse>> hoTrees;
30	Database db;
31	Constants consts;
32
33	AtomicInteger antalFundne;
34
35	int workerID;
36	boolean verbose;
37
38	public LookupWorker(int workerID,
39	CyclicBarrier barrier,
40	Queue<Adresse> ikkeDaekkede,
41	Map<Short, KdTree<Adresse>> hoTrees,
42	AtomicInteger antalFundne,
43	Database db,
44	boolean verbose,
45	Constants consts) {
46
47	this.workerID = workerID;
48	this.barrier = barrier;
49	this.ikkeDaekkede = ikkeDaekkede;
50	this.hoTrees = hoTrees;
51	this.antalFundne = antalFundne;
52	this.db = db;
53	this.verbose = verbose;
54	this.consts = consts;
55
56	}
57
58
59	@Override
60	public void run() {
61	System.out.println( "Worker/" + workerID + " starting");
62
63	for (Adresse qAdresse : ikkeDaekkede) {
64
65	short ho = qAdresse.ho;
66	KdTree<Adresse> addressTree = hoTrees.get(ho);
67
68	Collection<Adresse> res = addressTree.nearestNeighbourSearch(MAX_RESULTS, qAdresse);
69	if (res.size() > MAX_RESULTS ) {
70	System.out.println("Returnerede mere end MAX_RESULTS> " + res.size() );
71	System.exit(0);
72	}
73
74
75	Adresse bedsteAdr = osrmHelper.getNearestViaTable(qAdresse, res);
76	//System.out.println("Bedste:" + bedsteAdr );
77
78	double bedsteAfstand = GeoPointHelper.beregnAfstand(qAdresse, bedsteAdr);
79
80
81	if (bedsteAfstand <= consts.getMaxDistance() ) { //Gem kun hvis distancen er under max_distance for distributøren
82	if (verbose == true) {
83	logger.fine("(" +workerID + ") Bedste: " + qAdresse + "==>" + bedsteAdr + " ::: " + bedsteAfstand );
84	}
85
86	antalFundne.incrementAndGet();
87
88	try {
89	db.gemResultat(qAdresse, bedsteAdr, bedsteAfstand);
90	} catch (SQLException e) {
91	logger.log(Level.WARNING, "Error saving result", e);
92	logger.warning("Shutting down");
93
94	System.exit(0); //hvis at indsættelsen fejler skal vi afbryde med det samme
95	}
96	} else {
97	if (verbose == true) {
98	System.out.println("For stor afstand: " + qAdresse + " > " + bedsteAfstand);
99	}
100	}
101
102	}
103
104	try {
105	barrier.await();
106	} catch (BrokenBarrierException e) {
107	logger.log(Level.SEVERE, e.getMessage(), e);
108	System.exit(-1);
109	} catch(InterruptedException e) {
110	//Do nothing
111	}
112
113
114	System.out.println( "Worker/" + workerID + " stopping");
115	}
116
117	}