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.CyclicBarrier;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;

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;
import dk.daoas.adressevedligehold.tasks.TaskLogger;

public class LookupWorker implements Runnable {

        static final int MAX_RESULTS = 99;


        private TaskLogger logger = TaskLogger.getInstance();

        OSRMHelper osrmHelper = new OSRMHelper();

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

        AtomicInteger antalFundne;
        AtomicInteger antalBeregnet;

        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,
                        AtomicInteger antalBeregnet,
                        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.antalBeregnet = antalBeregnet;
                this.db = db;
                this.verbose = verbose;
                this.consts = consts;

        }


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

                ServiceConfig conf = ServiceConfig.getInstance();

                try (BarrierMarker marker = new BarrierMarker() ) {
                        while ( ! ikkeDaekkede.isEmpty() ) {

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


                                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.warning("Error saving result", e);
                                                logger.warning("Shutting down");
                                                throw e; // rethrow
                                        }
                                } else {
                                        if (verbose == true) {
                                                System.out.println("For stor afstand: " + qAddress + " > " + bedsteAfstand);
                                        }
                                }
                                
                                antalBeregnet.incrementAndGet();
                        }

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


                System.out.println( "Worker/" + workerID + " stopping");
        }
        
        
        /**
         * Makes sure the barrier is touched correctly no matter how the the current code block is exited.
         * @author thn
         *
         */
        public class BarrierMarker implements AutoCloseable {
                @Override
                public void close() {
                        try {
                                logger.info( "Worker/" + workerID + " afventer barrier");
                                barrier.await();
                        } catch (Exception e) {
                                logger.log(Level.SEVERE, e.getMessage(), e);
                        }
                }
        }

}
1	torben	2878	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.CyclicBarrier;
8			import java.util.concurrent.atomic.AtomicInteger;
9			import java.util.logging.Level;
10
11			import ags.utils.dataStructures.MaxHeap;
12			import ags.utils.dataStructures.trees.thirdGenKD.DistanceFunction;
13			import ags.utils.dataStructures.trees.thirdGenKD.KdTree;
14			import ags.utils.dataStructures.trees.thirdGenKD.SquareEuclideanDistanceFunction;
15			import dk.daoas.adressevedligehold.ServiceConfig;
16			import dk.daoas.adressevedligehold.beans.Address;
17			import dk.daoas.adressevedligehold.tasks.Task;
18	torben	2903	import dk.daoas.adressevedligehold.tasks.TaskLogger;
19	torben	2878
20			public class LookupWorker implements Runnable {
21
22			static final int MAX_RESULTS = 99;
23
24
25	torben	2903	private TaskLogger logger = TaskLogger.getInstance();
26	torben	2878
27			OSRMHelper osrmHelper = new OSRMHelper();
28
29			CyclicBarrier barrier;
30			Queue<Address> ikkeDaekkede;
31			Map<Short, KdTree<Address>> hoTrees;
32			Database db;
33			Constants consts;
34
35			AtomicInteger antalFundne;
36	torben	2891	AtomicInteger antalBeregnet;
37	torben	2878
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	torben	2891	AtomicInteger antalBeregnet,
52	torben	2878	Database db,
53			boolean verbose,
54			Constants consts) {
55
56			this.workerID = workerID;
57			this.task = task;
58			this.barrier = barrier;
59			this.ikkeDaekkede = ikkeDaekkede;
60			this.hoTrees = hoTrees;
61			this.antalFundne = antalFundne;
62	torben	2891	this.antalBeregnet = antalBeregnet;
63	torben	2878	this.db = db;
64			this.verbose = verbose;
65			this.consts = consts;
66
67			}
68
69
70			@Override
71			public void run() {
72			System.out.println( "Worker/" + workerID + " starting");
73
74			ServiceConfig conf = ServiceConfig.getInstance();
75
76	torben	2893	try (BarrierMarker marker = new BarrierMarker() ) {
77	torben	2878	while ( ! ikkeDaekkede.isEmpty() ) {
78
79			if (task.isAborted()) {
80			System.out.println( "Worker/" + workerID + " aborting");
81	torben	2893	break;
82	torben	2878	}
83
84
85			Address qAddress = ikkeDaekkede.remove();
86
87			short ho = qAddress.ho;
88			KdTree<Address> addressTree = hoTrees.get(ho);
89
90			MaxHeap<Address> res = addressTree.findNearestNeighbors(qAddress.xyz, MAX_RESULTS, dist);
91			//Collection<Address> res = addressTree.nearestNeighbourSearch(MAX_RESULTS, qAddress);
92			int size = res.size();
93			if (size > MAX_RESULTS ) {
94			System.out.println("Returnerede mere end MAX_RESULTS> " + res.size() );
95	torben	2887	throw new Exception("Returnerede mere end MAX_RESULTS> " + res.size());
96	torben	2878	}
97
98			ArrayList<Address> adrList = new ArrayList<Address>();
99
100			for (int i=0; i<size; i++) {
101			Address a = res.getMax();
102			adrList.add(a);
103			res.removeMax();
104			}
105			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
106			double bedsteAfstand = GeoPointHelper.beregnAfstand(qAddress, bedsteAdr);
107
108
109			if ( conf.useOsrm == true) {
110			//Hvis at den er inden for afstand, så prøver vi at lave findNærmste på vej distancen
111			if (bedsteAfstand <= consts.getMaxDistance()) {
112
113			Address bedsteAdr2 = osrmHelper.getNearestTableHttp(qAddress, adrList);
114
115			if (bedsteAdr2 != null) {
116			bedsteAdr = bedsteAdr2;
117			bedsteAfstand = GeoPointHelper.beregnAfstand(qAddress, bedsteAdr);
118			} else {
119			logger.info("OSRM returned null for " + qAddress);
120			}
121			}
122			}
123
124
125
126
127
128			if (bedsteAfstand <= consts.getMaxDistance() ) { //Gem kun hvis distancen er under max_distance for distributøren
129			if (verbose == true) {
130			logger.fine("(" +workerID + ") Bedste: " + qAddress + "==>" + bedsteAdr + " ::: " + bedsteAfstand );
131			}
132
133			antalFundne.incrementAndGet();
134
135			try {
136			db.gemResultat(qAddress, bedsteAdr, bedsteAfstand);
137			} catch (SQLException e) {
138	torben	2903	logger.warning("Error saving result", e);
139	torben	2878	logger.warning("Shutting down");
140	torben	2885	throw e; // rethrow
141	torben	2878	}
142			} else {
143			if (verbose == true) {
144			System.out.println("For stor afstand: " + qAddress + " > " + bedsteAfstand);
145			}
146			}
147	torben	2891
148			antalBeregnet.incrementAndGet();
149	torben	2878	}
150
151			} catch (Exception e) {
152			logger.log( Level.SEVERE, "Worker/" + workerID, e);
153	torben	2885	task.doAbort();
154	torben	2878	}
155
156
157	torben	2893
158	torben	2878
159			System.out.println( "Worker/" + workerID + " stopping");
160			}
161	torben	2893
162
163			/**
164			* Makes sure the barrier is touched correctly no matter how the the current code block is exited.
165			* @author thn
166			*
167			*/
168			public class BarrierMarker implements AutoCloseable {
169			@Override
170			public void close() {
171			try {
172			logger.info( "Worker/" + workerID + " afventer barrier");
173			barrier.await();
174			} catch (Exception e) {
175			logger.log(Level.SEVERE, e.getMessage(), e);
176			}
177			}
178			}
179	torben	2878
180			}