--- dao/FuldDaekningWorker/src/dk/daoas/fulddaekning/LookupMain.java 2014/12/10 09:50:33 2241 +++ dao/FuldDaekningWorker/src/main/java/dk/daoas/fulddaekning/LookupMain.java 2015/10/15 15:34:27 2753 @@ -2,163 +2,236 @@ import java.io.File; import java.io.FileReader; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.atomic.AtomicInteger; import java.util.logging.FileHandler; import java.util.logging.Logger; import java.util.logging.SimpleFormatter; +import ags.utils.dataStructures.trees.thirdGenKD.KdTree; + + + public class LookupMain { static final String CONFIG_FILENAME = "fulddaekning.properties"; - - static boolean rename_tables; + + static boolean use_osrm; + static boolean test_mode; static SafeProperties conf; - + static int max_workers; static boolean verbose; - + static String distributor; final static Logger logger = Logger.getLogger( LookupMain.class.toString() ); - - - static Statistik flestDaekkede = new Statistik(); - static Statistik flestIkkeDaekkede = new Statistik(); - static Statistik mestBrugteTid = new Statistik(); - static Statistik stoersteDataset = new Statistik(); - - static Adresse[] alleDaekkedeAdresser; - - + + + static ThreadPoolExecutor threadPool; + + private static void setupThreadPool() { + threadPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(max_workers, new WorkerThreadFactory() ); + } + + + public static void main(String[] args) throws Exception { - + //Setup j.u.l Logger Logger root = Logger.getLogger(""); FileHandler fhandler = new FileHandler("fulddaekning.log"); // Ingen max størrelse, ingen rotation og ingen append fhandler.setFormatter( new SimpleFormatter() ); root.addHandler( fhandler ); - - - - + + + + File confFile = new File( CONFIG_FILENAME ); if (! confFile.exists() ) { logger.warning("Config file not found: " + CONFIG_FILENAME); System.exit(1); } - + + int antalIkkeDaekkede = -1; + conf = new SafeProperties(); conf.load( new FileReader(confFile) ); - - max_workers = Integer.parseInt( conf.getSafeProperty("MAX_WORKERS") ); - logger.info("Starting with MAX_WORKERS:" + max_workers); - + + + verbose = Boolean.parseBoolean( conf.getSafeProperty("VERBOSE") ); logger.info("Starting with VERBOSE:" + verbose); - - rename_tables = Boolean.parseBoolean( conf.getSafeProperty("RENAMETABLES") ); - logger.info("Starting with RENAMETABLES:" + rename_tables); - + + test_mode = Boolean.parseBoolean( conf.getSafeProperty("TEST") ); + logger.info("Starting with TEST:" + test_mode); + + use_osrm = Boolean.parseBoolean( conf.getSafeProperty("USE_OSRM") ); + logger.info("Starting with USE_OSRM:" + use_osrm); + distributor = conf.getSafeProperty("DISTRIBUTOR"); distributor = distributor.toUpperCase(); logger.info("Starting for DISTRIBUTOR:" + distributor); - + Constants.init(distributor); Constants consts = Constants.getInstance(); + + max_workers = Integer.parseInt( conf.getSafeProperty("MAX_WORKERS") ); + if (max_workers <= 0) { + logger.info("!!! AUTO-DETECT MAX_WORKERS !!!"); + int cores = Runtime.getRuntime().availableProcessors(); + cores -= 1;//Efterlad 1 core/cpu i reserve til systemet + + max_workers = Math.max(1, cores); //Dog skal der som minimum være 1 core til beregning + + } + if (test_mode) { + max_workers = 1; + } + logger.info("Starting with MAX_WORKERS:" + max_workers); + setupThreadPool(); + Database db = new Database(conf); - - + Queue ikkeDaekkede = db.hentAlleIkkedaekkedeAdresser(consts.getMinPostnr(), consts.getMaxPostnr() ); + antalIkkeDaekkede = ikkeDaekkede.size(); + logger.info("Antal ikke-daekkede: " + antalIkkeDaekkede ); + + - boolean testRun= false; - - + + AtomicInteger antalFundne = new AtomicInteger(0); + + long beregnStart =0; long start = System.currentTimeMillis(); - + int antalDaekkedeAdresser = -1; + if (testRun == false) { - + logger.info("Finder postnumre"); - List postnumre = db.hentPostnumre(); - + Set postnumre = db.hentPostnumreCache(); + // Først validerer vi BBox på alle postnummre, for at undgå fuldt stop midt i beregningen - for(String postnr : postnumre) { // + for(short postnr : postnumre) { // logger.info("Validerer BBox for " + postnr); BoundingBox bbox = db.getBoundingbox(postnr); bbox.validateBbox(); } - - if (consts.doExtendedLookup()) { - logger.info("Henter alle daekkede adresser"); - alleDaekkedeAdresser = db.hentAlleDaekkedeAdresser(); - logger.info( "AlleDaekkedeAdresser.length=" + alleDaekkedeAdresser.length); + + + logger.info("Henter alle daekkede adresser"); + antalDaekkedeAdresser = db.hentAlleDaekkedeAdresser(); + logger.info( "AlleDaekkedeAdresser.length=" + antalDaekkedeAdresser ); + + Map> addrHoList = db.getDaekkedeAdresserHO(); + + Map> hoTrees = new HashMap>(); + + for ( Map.Entry> entry : addrHoList.entrySet() ) { + short ho = entry.getKey(); + List geopoints = entry.getValue(); + + logger.info("Opbygger KDTree for " + ho + " - antal=" + geopoints.size() ); + + int bucketSize = 96*12; + + KdTree addressTree = new KdTree( 3, bucketSize ); + for(Adresse a: geopoints) { + addressTree.addPoint(a.xyz, a); + } + + hoTrees.put(ho, addressTree); } - - - //pre-check er ok - reset tmp tabel og start søgningen + db.resetResultTable(); + + beregnStart = System.currentTimeMillis(); + logger.info("Starter beregning"); + + //pre-check er ok - reset tmp tabel og start søgningen + + CyclicBarrier barrier = new CyclicBarrier(LookupMain.max_workers + 1); - for(String postnr : postnumre) { - Lookup lookup = new Lookup(postnr, db); - lookup.doLookup(); + for (int i=0; i flestDaekkede.antalDaekkede) { - flestDaekkede = stat; - } - if (stat.antalIkkeDaekkede > flestIkkeDaekkede.antalIkkeDaekkede) { - flestIkkeDaekkede = stat; - } - - if (stat.totalDataset > mestBrugteTid.totalDataset) { - stoersteDataset = stat; - } - - if (stat.forbrugtTid > mestBrugteTid.forbrugtTid) { - mestBrugteTid = stat; - } + logger.info( String.format("Fandt adresser til : %.2f %%", (antalFundne.get() *100.0)/antalIkkeDaekkede ) ); + //logger.info("Fandt adresser til : " + (antalFundne*100.0)/ikkeDaekkede.length + "%" ); } + + static String formatMilliSeconds(long milliseconds) { int mseconds = (int) milliseconds % 1000; int seconds = (int) (milliseconds / 1000) % 60 ; int minutes = (int) ((milliseconds / (1000*60)) % 60); int hours = (int) ((milliseconds / (1000*60*60)) % 24); - + return String.format("%02d:%02d:%02d.%03d", hours, minutes, seconds, mseconds); - } + } + + + static class WorkerThreadFactory implements ThreadFactory { + int count = 0; + + @Override + public Thread newThread(Runnable r) { + return new Thread(r, "lookupWorker/" + count++); + } + } }