adressevedligehold/afstandandenrute/LookupWorker.java

package dk.daoas.adressevedligehold.afstandandenrute;

import java.sql.SQLException;
import java.util.ArrayList;
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 ags.utils.dataStructures.MaxHeap;
import ags.utils.dataStructures.trees.thirdGenKD.DistanceFunction;
import ags.utils.dataStructures.trees.thirdGenKD.KdTree;
import ags.utils.dataStructures.trees.thirdGenKD.SquareEuclideanDistanceFunction;
import dk.daoas.adressevedligehold.ServiceConfig;
import dk.daoas.adressevedligehold.beans.Address;
import dk.daoas.adressevedligehold.tasks.Task;

public class LookupWorker implements Runnable {

        static final int MAX_RESULTS = 99;


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

        OSRMHelper osrmHelper = new OSRMHelper();

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

        AtomicInteger antalFundne;

        int workerID;
        boolean verbose;

        Task task;

        DistanceFunction dist = new SquareEuclideanDistanceFunction();

        public LookupWorker(int workerID, 
                        Task task,
                        CyclicBarrier barrier, 
                        Queue<Address> ikkeDaekkede, 
                        Map<Short, KdTree<Address>> hoTrees, 
                        AtomicInteger antalFundne,
                        Database db,
                        boolean verbose,
                        Constants consts) {

                this.workerID = workerID;
                this.task = task;
                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");

                ServiceConfig conf = ServiceConfig.getInstance();

                try {
                        while ( ! ikkeDaekkede.isEmpty() ) {

                                if (task.isAborted()) {
                                        System.out.println( "Worker/" + workerID + " aborting");
                                        return;
                                }


                                Address qAddress = ikkeDaekkede.remove();

                                short ho = qAddress.ho;
                                KdTree<Address> addressTree = hoTrees.get(ho);

                                MaxHeap<Address> res = addressTree.findNearestNeighbors(qAddress.xyz, MAX_RESULTS, dist);
                                //Collection<Address> res = addressTree.nearestNeighbourSearch(MAX_RESULTS, qAddress);
                                int size = res.size();
                                if (size > MAX_RESULTS ) {
                                        System.out.println("Returnerede mere end MAX_RESULTS> " + res.size() );
                                        throw new Exception("Returnerede mere end MAX_RESULTS> " + res.size());
                                }

                                ArrayList<Address> adrList = new ArrayList<Address>();

                                for (int i=0; i<size; i++) {
                                        Address a = res.getMax();
                                        adrList.add(a);
                                        res.removeMax();
                                }
                                Address bedsteAdr = adrList.get(size-1); //da det er en max-heap er det elementet med størst distance der er først og dermed den sidste der er tættest
                                double bedsteAfstand = GeoPointHelper.beregnAfstand(qAddress, bedsteAdr);


                                if ( conf.useOsrm == true) {
                                        //Hvis at den er inden for afstand, så prøver vi at lave findNærmste på vej distancen
                                        if (bedsteAfstand <= consts.getMaxDistance()) {

                                                Address bedsteAdr2 = osrmHelper.getNearestTableHttp(qAddress, adrList); 

                                                if (bedsteAdr2 != null) {
                                                        bedsteAdr = bedsteAdr2;
                                                        bedsteAfstand = GeoPointHelper.beregnAfstand(qAddress, bedsteAdr);
                                                } else {
                                                        logger.info("OSRM returned null for " + qAddress);
                                                }
                                        }
                                }


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

                                        antalFundne.incrementAndGet();

                                        try {
                                                db.gemResultat(qAddress, bedsteAdr, bedsteAfstand);
                                        } catch (SQLException e) {
                                                logger.log(Level.WARNING, "Error saving result", e);
                                                logger.warning("Shutting down");
                                                throw e; // rethrow
                                        }
                                } else {
                                        if (verbose == true) {
                                                System.out.println("For stor afstand: " + qAddress + " > " + bedsteAfstand);
                                        }
                                }

                        }

                } catch (Exception e) {
                        logger.log( Level.SEVERE, "Worker/" + workerID, e);
                        task.doAbort();
                }


                try {
                        logger.info( "Worker/" + workerID + " afventer barrier");
                        barrier.await();
                } catch (BrokenBarrierException e) {
                        logger.log(Level.SEVERE, e.getMessage(), e);
                        System.exit(-1);
                } catch(InterruptedException e) {
                        logger.log(Level.SEVERE, e.getMessage(), e);
                }


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

}
1	package dk.daoas.adressevedligehold.afstandandenrute;
2
3	import java.sql.SQLException;
4	import java.util.ArrayList;
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 ags.utils.dataStructures.MaxHeap;
14	import ags.utils.dataStructures.trees.thirdGenKD.DistanceFunction;
15	import ags.utils.dataStructures.trees.thirdGenKD.KdTree;
16	import ags.utils.dataStructures.trees.thirdGenKD.SquareEuclideanDistanceFunction;
17	import dk.daoas.adressevedligehold.ServiceConfig;
18	import dk.daoas.adressevedligehold.beans.Address;
19	import dk.daoas.adressevedligehold.tasks.Task;
20
21	public class LookupWorker implements Runnable {
22
23	static final int MAX_RESULTS = 99;
24
25
26	final static Logger logger = Logger.getLogger( LookupWorker.class.toString() );
27
28	OSRMHelper osrmHelper = new OSRMHelper();
29
30	CyclicBarrier barrier;
31	Queue<Address> ikkeDaekkede;
32	Map<Short, KdTree<Address>> hoTrees;
33	Database db;
34	Constants consts;
35
36	AtomicInteger antalFundne;
37
38	int workerID;
39	boolean verbose;
40
41	Task task;
42
43	DistanceFunction dist = new SquareEuclideanDistanceFunction();
44
45	public LookupWorker(int workerID,
46	Task task,
47	CyclicBarrier barrier,
48	Queue<Address> ikkeDaekkede,
49	Map<Short, KdTree<Address>> hoTrees,
50	AtomicInteger antalFundne,
51	Database db,
52	boolean verbose,
53	Constants consts) {
54
55	this.workerID = workerID;
56	this.task = task;
57	this.barrier = barrier;
58	this.ikkeDaekkede = ikkeDaekkede;
59	this.hoTrees = hoTrees;
60	this.antalFundne = antalFundne;
61	this.db = db;
62	this.verbose = verbose;
63	this.consts = consts;
64
65	}
66
67
68	@Override
69	public void run() {
70	System.out.println( "Worker/" + workerID + " starting");
71
72	ServiceConfig conf = ServiceConfig.getInstance();
73
74	try {
75	while ( ! ikkeDaekkede.isEmpty() ) {
76
77	if (task.isAborted()) {
78	System.out.println( "Worker/" + workerID + " aborting");
79	return;
80	}
81
82
83	Address qAddress = ikkeDaekkede.remove();
84
85	short ho = qAddress.ho;
86	KdTree<Address> addressTree = hoTrees.get(ho);
87
88	MaxHeap<Address> res = addressTree.findNearestNeighbors(qAddress.xyz, MAX_RESULTS, dist);
89	//Collection<Address> res = addressTree.nearestNeighbourSearch(MAX_RESULTS, qAddress);
90	int size = res.size();
91	if (size > MAX_RESULTS ) {
92	System.out.println("Returnerede mere end MAX_RESULTS> " + res.size() );
93	throw new Exception("Returnerede mere end MAX_RESULTS> " + res.size());
94	}
95
96	ArrayList<Address> adrList = new ArrayList<Address>();
97
98	for (int i=0; i<size; i++) {
99	Address a = res.getMax();
100	adrList.add(a);
101	res.removeMax();
102	}
103	Address bedsteAdr = adrList.get(size-1); //da det er en max-heap er det elementet med størst distance der er først og dermed den sidste der er tættest
104	double bedsteAfstand = GeoPointHelper.beregnAfstand(qAddress, bedsteAdr);
105
106
107	if ( conf.useOsrm == true) {
108	//Hvis at den er inden for afstand, så prøver vi at lave findNærmste på vej distancen
109	if (bedsteAfstand <= consts.getMaxDistance()) {
110
111	Address bedsteAdr2 = osrmHelper.getNearestTableHttp(qAddress, adrList);
112
113	if (bedsteAdr2 != null) {
114	bedsteAdr = bedsteAdr2;
115	bedsteAfstand = GeoPointHelper.beregnAfstand(qAddress, bedsteAdr);
116	} else {
117	logger.info("OSRM returned null for " + qAddress);
118	}
119	}
120	}
121
122
123
124
125
126	if (bedsteAfstand <= consts.getMaxDistance() ) { //Gem kun hvis distancen er under max_distance for distributøren
127	if (verbose == true) {
128	logger.fine("(" +workerID + ") Bedste: " + qAddress + "==>" + bedsteAdr + " ::: " + bedsteAfstand );
129	}
130
131	antalFundne.incrementAndGet();
132
133	try {
134	db.gemResultat(qAddress, bedsteAdr, bedsteAfstand);
135	} catch (SQLException e) {
136	logger.log(Level.WARNING, "Error saving result", e);
137	logger.warning("Shutting down");
138	throw e; // rethrow
139	}
140	} else {
141	if (verbose == true) {
142	System.out.println("For stor afstand: " + qAddress + " > " + bedsteAfstand);
143	}
144	}
145
146	}
147
148	} catch (Exception e) {
149	logger.log( Level.SEVERE, "Worker/" + workerID, e);
150	task.doAbort();
151	}
152
153
154
155	try {
156	logger.info( "Worker/" + workerID + " afventer barrier");
157	barrier.await();
158	} catch (BrokenBarrierException e) {
159	logger.log(Level.SEVERE, e.getMessage(), e);
160	System.exit(-1);
161	} catch(InterruptedException e) {
162	logger.log(Level.SEVERE, e.getMessage(), e);
163	}
164
165
166	System.out.println( "Worker/" + workerID + " stopping");
167	}
168
169	}