Index: trunk/src/org/openstreetmap/josm/data/cache/HostLimitQueue.java =================================================================== --- trunk/src/org/openstreetmap/josm/data/cache/HostLimitQueue.java (revision 8403) +++ trunk/src/org/openstreetmap/josm/data/cache/HostLimitQueue.java (revision 8403) @@ -0,0 +1,148 @@ +// License: GPL. For details, see LICENSE file. +package org.openstreetmap.josm.data.cache; + +import java.util.Iterator; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; + +import org.openstreetmap.josm.Main; + +/** + * @author Wiktor Niesiobędzki + * + * Queue for ThreadPoolExecutor that implements per-host limit. It will acquire a semaphore for each task + * and it will set a runnable task with semaphore release, when job has finished. + * + * This implementation doesn't guarantee to have at most hostLimit connections per host[1], and it doesn't + * guarantee that all threads will be busy, when there is work for them[2]. + * + * [1] More connection per host may happen, when ThreadPoolExecutor is growing its pool, and thus + * tasks do not go through the Queue + * [2] If we have a queue, and for all hosts in queue we will fail to acquire semaphore, the thread + * take the first available job and wait for semaphore. It might be the case, that semaphore was released + * for some task further in queue, but this implementation doesn't try to detect such situation + * + * + */ + +public class HostLimitQueue extends LinkedBlockingDeque { + + private Map hostSemaphores = new ConcurrentHashMap<>(); + private int hostLimit; + + /** + * Creates an unbounded queue + * @param hostLimit how many parallel calls to host to allow + */ + public HostLimitQueue(int hostLimit) { + super(); // create unbounded queue + this.hostLimit = hostLimit; + } + + private JCSCachedTileLoaderJob findJob() { + for (Iterator it = iterator(); it.hasNext();) { + Runnable r = it.next(); + if (r instanceof JCSCachedTileLoaderJob) { + JCSCachedTileLoaderJob job = (JCSCachedTileLoaderJob) r; + if (tryAcquireSemaphore(job)) { + if (remove(job)) { + return job; + } else { + // we have acquired the semaphore, but we didn't manage to remove it, as someone else did + // release the semaphore and look for another candidate + releaseSemaphore(job); + } + } else { + Main.info("TMS - Skipping job {0} because host limit reached", job.getUrl()); + } + } + } + return null; + } + + @Override + public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException { + Runnable job = findJob(); + if (job != null) { + return job; + } + job = pollFirst(timeout, unit); + if (job != null) { + acquireSemaphore(job); + } + return job; + } + + @Override + public Runnable take() throws InterruptedException { + Runnable job = findJob(); + if (job != null) { + return job; + } + job = takeFirst(); + if (job != null) { + acquireSemaphore(job); + } + return job; + } + + private Semaphore getSemaphore(JCSCachedTileLoaderJob job) { + String host = ((JCSCachedTileLoaderJob)job).getUrl().getHost(); + Semaphore limit = hostSemaphores.get(host); + if (limit == null) { + synchronized(hostSemaphores) { + limit = hostSemaphores.get(host); + if (limit == null) { + limit = new Semaphore(hostLimit); + hostSemaphores.put(host, limit); + } + } + } + return limit; + } + + private void acquireSemaphore(Runnable job) throws InterruptedException { + if (job instanceof JCSCachedTileLoaderJob) { + final JCSCachedTileLoaderJob jcsJob = (JCSCachedTileLoaderJob) job; + Semaphore limit = getSemaphore(jcsJob); + if (limit != null) { + limit.acquire(); + jcsJob.setFinishedTask(new Runnable() { + public void run() { + releaseSemaphore(jcsJob); + } + }); + } + } + } + + + private boolean tryAcquireSemaphore(final JCSCachedTileLoaderJob job) { + boolean ret = true; + Semaphore limit = getSemaphore(job); + if (limit != null) { + ret = limit.tryAcquire(); + if (ret) { + job.setFinishedTask(new Runnable() { + public void run() { + releaseSemaphore(job); + } + }); + } + } + return ret; + } + + private void releaseSemaphore(JCSCachedTileLoaderJob job) { + Semaphore limit = getSemaphore(job); + if (limit != null) { + limit.release(); + if (limit.availablePermits() > hostLimit) { + Main.warn("More permits than it should be"); + } + } + } +} Index: trunk/src/org/openstreetmap/josm/data/cache/ICachedLoaderListener.java =================================================================== --- trunk/src/org/openstreetmap/josm/data/cache/ICachedLoaderListener.java (revision 8401) +++ trunk/src/org/openstreetmap/josm/data/cache/ICachedLoaderListener.java (revision 8403) @@ -11,5 +11,5 @@ SUCCESS, FAILURE, - REJECTED + CANCELED } /** Index: trunk/src/org/openstreetmap/josm/data/cache/JCSCachedTileLoaderJob.java =================================================================== --- trunk/src/org/openstreetmap/josm/data/cache/JCSCachedTileLoaderJob.java (revision 8401) +++ trunk/src/org/openstreetmap/josm/data/cache/JCSCachedTileLoaderJob.java (revision 8403) @@ -16,7 +16,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Executor; +import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingDeque; -import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -35,4 +35,5 @@ * * @param cache entry key type + * @param cache value type * * Generic loader for HTTP based tiles. Uses custom attribute, to check, if entry has expired @@ -62,41 +63,8 @@ public static final IntegerProperty THREAD_LIMIT = new IntegerProperty("cache.jcs.max_threads", 10); - public static class LIFOQueue extends LinkedBlockingDeque { - - /** - * Constructs a new {@code LIFOQueue} with a capacity of {@link Integer#MAX_VALUE}. - */ - public LIFOQueue() { - super(); - } - - /** - * Constructs a new {@code LIFOQueue} with the given (fixed) capacity. - * @param capacity the capacity of this deque - * @throws IllegalArgumentException if {@code capacity} is less than 1 - */ - public LIFOQueue(int capacity) { - super(capacity); - } - - @Override - public boolean offer(Runnable t) { - return super.offerFirst(t); - } - - @Override - public Runnable remove() { - return super.removeFirst(); - } - } - - - /** - * ThreadPoolExecutor starts new threads, until THREAD_LIMIT is reached. Then it puts tasks into LIFOQueue, which is fairly - * small, but we do not want a lot of outstanding tasks queued, but rather prefer the class consumer to resubmit the task, which are - * important right now. + /* + * ThreadPoolExecutor starts new threads, until THREAD_LIMIT is reached. Then it puts tasks into LinkedBlockingDeque. * - * This way, if some task gets outdated (for example - user paned the map, and we do not want to download this tile any more), - * the task will not be resubmitted, and thus - never queued. + * The queue works FIFO, so one needs to take care about ordering of the entries submitted * * There is no point in canceling tasks, that are already taken by worker threads (if we made so much effort, we can at least cache @@ -104,5 +72,6 @@ * and performance (we do want to have something to offer to worker threads before tasks will be resubmitted by class consumer) */ - private static Executor DEFAULT_DOWNLOAD_JOB_DISPATCHER = new ThreadPoolExecutor( + + private static ThreadPoolExecutor DEFAULT_DOWNLOAD_JOB_DISPATCHER = new ThreadPoolExecutor( 2, // we have a small queue, so threads will be quickly started (threads are started only, when queue is full) THREAD_LIMIT.get().intValue(), // do not this number of threads @@ -110,6 +79,17 @@ TimeUnit.SECONDS, // make queue of LIFO type - so recently requested tiles will be loaded first (assuming that these are which user is waiting to see) - new LIFOQueue(5) + new LinkedBlockingDeque(), + getNamedThreadFactory("JCS downloader") ); + + public static ThreadFactory getNamedThreadFactory(final String name) { + return new ThreadFactory(){ + public Thread newThread(Runnable r) { + Thread t = Executors.defaultThreadFactory().newThread(r); + t.setName(name); + return t; + } + }; + } private static ConcurrentMap> inProgress = new ConcurrentHashMap<>(); @@ -127,5 +107,6 @@ private int readTimeout; private Map headers; - private Executor downloadJobExecutor; + private ThreadPoolExecutor downloadJobExecutor; + private Runnable finishTask; /** @@ -139,5 +120,5 @@ int connectTimeout, int readTimeout, Map headers, - Executor downloadJobExecutor) { + ThreadPoolExecutor downloadJobExecutor) { this.cache = cache; @@ -209,29 +190,6 @@ } // object not in cache, so submit work to separate thread - try { - if (executionGuard()) { - // use getter method, so subclasses may override executors, to get separate thread pool - getDownloadExecutor().execute(this); - } else { - log.log(Level.FINE, "JCS - guard rejected job for: {0}", getCacheKey()); - finishLoading(LoadResult.REJECTED); - } - } catch (RejectedExecutionException e) { - // queue was full, try again later - log.log(Level.FINE, "JCS - rejected job for: {0}", getCacheKey()); - finishLoading(LoadResult.REJECTED); - } - } - } - - /** - * Guard method for execution. If guard returns true, the execution of download task will commence - * otherwise, execution will finish with result LoadResult.REJECTED - * - * It is responsibility of the overriding class, to handle properly situation in finishLoading class - * @return - */ - protected boolean executionGuard() { - return true; + getDownloadExecutor().execute(this); + } } @@ -240,4 +198,7 @@ */ protected void executionFinished() { + if (finishTask != null) { + finishTask.run(); + } } @@ -269,5 +230,5 @@ * this needs to be non-static, so it can be overridden by subclasses */ - protected Executor getDownloadExecutor() { + protected ThreadPoolExecutor getDownloadExecutor() { return downloadJobExecutor; } @@ -498,3 +459,34 @@ } } + + /** + * TODO: move to JobFactory + * cancels all outstanding tasks in the queue. + */ + public void cancelOutstandingTasks() { + ThreadPoolExecutor downloadExecutor = getDownloadExecutor(); + for(Runnable r: downloadExecutor.getQueue()) { + if (downloadExecutor.remove(r)) { + if (r instanceof JCSCachedTileLoaderJob) { + ((JCSCachedTileLoaderJob) r).handleJobCancellation(); + } + } + } + } + + /** + * Sets a job, that will be run, when job will finish execution + * @param runnable that will be executed + */ + public void setFinishedTask(Runnable runnable) { + this.finishTask = runnable; + + } + + /** + * Marks this job as canceled + */ + public void handleJobCancellation() { + finishLoading(LoadResult.CANCELED); + } } Index: trunk/src/org/openstreetmap/josm/data/imagery/TMSCachedTileLoader.java =================================================================== --- trunk/src/org/openstreetmap/josm/data/imagery/TMSCachedTileLoader.java (revision 8401) +++ trunk/src/org/openstreetmap/josm/data/imagery/TMSCachedTileLoader.java (revision 8403) @@ -16,6 +16,7 @@ import org.openstreetmap.gui.jmapviewer.interfaces.TileSource; import org.openstreetmap.josm.data.cache.BufferedImageCacheEntry; +import org.openstreetmap.josm.data.cache.HostLimitQueue; import org.openstreetmap.josm.data.cache.JCSCacheManager; -import org.openstreetmap.josm.data.cache.JCSCachedTileLoaderJob.LIFOQueue; +import org.openstreetmap.josm.data.cache.JCSCachedTileLoaderJob; import org.openstreetmap.josm.data.preferences.IntegerProperty; @@ -34,5 +35,7 @@ private TileLoaderListener listener; private static final String PREFERENCE_PREFIX = "imagery.tms.cache."; - // average tile size is about 20kb + /** + * how many object on disk should be stored for TMS region. Average tile size is about 20kb + */ public static final IntegerProperty MAX_OBJECTS_ON_DISK = new IntegerProperty(PREFERENCE_PREFIX + "max_objects_disk", 25000); // 25000 is around 500MB under this assumptions @@ -41,4 +44,10 @@ */ public static final IntegerProperty THREAD_LIMIT = new IntegerProperty("imagery.tms.tmsloader.maxjobs", 25); + + /** + * Limit definition for per host concurrent connections + */ + public static final IntegerProperty HOST_LIMIT = new IntegerProperty("imagery.tms.tmsloader.maxjobsperhost", 6); + /** @@ -54,8 +63,6 @@ 30, // keepalive for thread TimeUnit.SECONDS, - // make queue of LIFO type - so recently requested tiles will be loaded first (assuming that these are which user is waiting to see) - new LIFOQueue() - /* keep the queue size fairly small, we do not want to - download a lot of tiles, that user is not seeing anyway */ + new HostLimitQueue(HOST_LIMIT.get().intValue()), + JCSCachedTileLoaderJob.getNamedThreadFactory("TMS downloader") ); } @@ -123,35 +130,13 @@ /** - * Sets the download executor for this tile loader factory. Enables to use different queuing method - * for this factory. - * @param downloadExecutor - */ - public void setDownloadExecutor(ThreadPoolExecutor downloadExecutor) { - this.downloadExecutor = downloadExecutor; - } - - /** - * @return Executor that handles the jobs for this tile loader - */ - public ThreadPoolExecutor getDownloadExecutor() { - return downloadExecutor; - } - - /** * cancels all outstanding tasks in the queue. This rollbacks the state of the tiles in the queue * to loading = false / loaded = false */ public void cancelOutstandingTasks() { - for(Runnable elem: downloadExecutor.getQueue()) { - if (elem instanceof TMSCachedTileLoaderJob) { - TMSCachedTileLoaderJob loaderJob = (TMSCachedTileLoaderJob) elem; - if (downloadExecutor.remove(loaderJob)) { - Tile t = loaderJob.getTile(); - t.finishLoading(); - t.setLoaded(false); - } + for(Runnable r: downloadExecutor.getQueue()) { + if (downloadExecutor.remove(r) && r instanceof TMSCachedTileLoaderJob) { + ((TMSCachedTileLoaderJob)r).handleJobCancellation(); } } } - } Index: trunk/src/org/openstreetmap/josm/data/imagery/TMSCachedTileLoaderJob.java =================================================================== --- trunk/src/org/openstreetmap/josm/data/imagery/TMSCachedTileLoaderJob.java (revision 8401) +++ trunk/src/org/openstreetmap/josm/data/imagery/TMSCachedTileLoaderJob.java (revision 8403) @@ -13,6 +13,5 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Executor; -import java.util.concurrent.Semaphore; +import java.util.concurrent.ThreadPoolExecutor; import java.util.logging.Level; import java.util.logging.Logger; @@ -25,5 +24,4 @@ import org.openstreetmap.gui.jmapviewer.interfaces.TileSource; import org.openstreetmap.gui.jmapviewer.tilesources.AbstractTMSTileSource; -import org.openstreetmap.josm.Main; import org.openstreetmap.josm.data.cache.BufferedImageCacheEntry; import org.openstreetmap.josm.data.cache.CacheEntry; @@ -31,5 +29,4 @@ import org.openstreetmap.josm.data.cache.ICachedLoaderListener; import org.openstreetmap.josm.data.cache.JCSCachedTileLoaderJob; -import org.openstreetmap.josm.data.preferences.IntegerProperty; /** @@ -49,59 +46,4 @@ /** - * Limit definition for per host concurrent connections - */ - public static final IntegerProperty HOST_LIMIT = new IntegerProperty("imagery.tms.tmsloader.maxjobsperhost", 6); - - /* - * Host limit guards the area - between submission to the queue up to loading is finished. It uses executionGuard method - * from JCSCachedTileLoaderJob to acquire the semaphore, and releases it - when loadingFinished is called (but not when - * LoadResult.GUARD_REJECTED is set) - * - */ - - private Semaphore getSemaphore() { - String host = getUrl().getHost(); - Semaphore limit = HOST_LIMITS.get(host); - if (limit == null) { - synchronized(HOST_LIMITS) { - limit = HOST_LIMITS.get(host); - if (limit == null) { - limit = new Semaphore(HOST_LIMIT.get().intValue()); - HOST_LIMITS.put(host, limit); - } - } - } - return limit; - } - - private boolean acquireSemaphore() { - boolean ret = true; - Semaphore limit = getSemaphore(); - if (limit != null) { - ret = limit.tryAcquire(); - if (!ret) { - Main.debug("rejecting job because of per host limit"); - } - } - return ret; - } - - private void releaseSemaphore() { - Semaphore limit = getSemaphore(); - if (limit != null) { - limit.release(); - } - } - - private static Map HOST_LIMITS = new ConcurrentHashMap<>(); - - /** - * Reconfigures download dispatcher using current values of THREAD_LIMIT and HOST_LIMIT - */ - public static final void reconfigureDownloadDispatcher() { - HOST_LIMITS = new ConcurrentHashMap<>(); - } - - /** * Constructor for creating a job, to get a specific tile from cache * @param listener @@ -111,9 +53,10 @@ * @param readTimeout when connecting to remote resource * @param headers to be sent together with request + * @param downloadExecutor that will be executing the jobs */ public TMSCachedTileLoaderJob(TileLoaderListener listener, Tile tile, ICacheAccess cache, int connectTimeout, int readTimeout, Map headers, - Executor downloadExecutor) { + ThreadPoolExecutor downloadExecutor) { super(cache, connectTimeout, readTimeout, headers, downloadExecutor); this.tile = tile; @@ -207,14 +150,4 @@ } - @Override - protected boolean executionGuard() { - return acquireSemaphore(); - } - - @Override - protected void executionFinished() { - releaseSemaphore(); - } - public void submit() { tile.initLoading(); @@ -234,7 +167,4 @@ tile.finishLoading(); // whatever happened set that loading has finished switch(result){ - case FAILURE: - tile.setError("Problem loading tile"); - // no break intentional here case SUCCESS: handleNoTileAtZoom(); @@ -249,6 +179,9 @@ tile.setError(tr("HTTP error {0} when loading tiles", httpStatusCode)); } + break; + case FAILURE: + tile.setError("Problem loading tile"); // no break intentional here - case REJECTED: + case CANCELED: // do nothing } Index: trunk/src/org/openstreetmap/josm/gui/layer/TMSLayer.java =================================================================== --- trunk/src/org/openstreetmap/josm/gui/layer/TMSLayer.java (revision 8401) +++ trunk/src/org/openstreetmap/josm/gui/layer/TMSLayer.java (revision 8403) @@ -22,4 +22,5 @@ import java.util.ArrayList; import java.util.Collections; +import java.util.Comparator; import java.util.HashMap; import java.util.LinkedList; @@ -1194,8 +1195,26 @@ } + private Comparator getTileDistanceComparator() { + final int centerX = (int) Math.ceil((x0 + x1) / 2); + final int centerY = (int) Math.ceil((y0 + y1) / 2); + return new Comparator() { + private int getDistance(Tile t) { + return Math.abs(t.getXtile() - centerX) + Math.abs(t.getYtile() - centerY); + } + @Override + public int compare(Tile o1, Tile o2) { + int distance1 = getDistance(o1); + int distance2 = getDistance(o2); + return Integer.compare(distance1, distance2); + } + }; + } + private void loadAllTiles(boolean force) { if (!autoLoad && !force) return; - for (Tile t : this.allTilesCreate()) { + List allTiles = allTilesCreate(); + Collections.sort(allTiles, getTileDistanceComparator()); + for (Tile t : allTiles) { //, getTileDistanceComparator())) { loadTile(t, false); } Index: trunk/src/org/openstreetmap/josm/gui/preferences/imagery/TMSSettingsPanel.java =================================================================== --- trunk/src/org/openstreetmap/josm/gui/preferences/imagery/TMSSettingsPanel.java (revision 8401) +++ trunk/src/org/openstreetmap/josm/gui/preferences/imagery/TMSSettingsPanel.java (revision 8403) @@ -45,5 +45,5 @@ maxElementsOnDisk = new JSpinner(new SpinnerNumberModel(TMSCachedTileLoader.MAX_OBJECTS_ON_DISK.get().intValue(), 0, Integer.MAX_VALUE, 1)); maxConcurrentDownloads = new JSpinner(new SpinnerNumberModel(TMSCachedTileLoaderJob.THREAD_LIMIT.get().intValue(), 0, Integer.MAX_VALUE, 1)); - maxDownloadsPerHost = new JSpinner(new SpinnerNumberModel(TMSCachedTileLoaderJob.HOST_LIMIT.get().intValue(), 0, Integer.MAX_VALUE, 1)); + maxDownloadsPerHost = new JSpinner(new SpinnerNumberModel(TMSCachedTileLoader.HOST_LIMIT.get().intValue(), 0, Integer.MAX_VALUE, 1)); add(new JLabel(tr("Auto zoom by default: ")), GBC.std()); @@ -98,5 +98,5 @@ this.maxElementsOnDisk.setValue(TMSCachedTileLoader.MAX_OBJECTS_ON_DISK.get()); this.maxConcurrentDownloads.setValue(TMSCachedTileLoaderJob.THREAD_LIMIT.get()); - this.maxDownloadsPerHost.setValue(TMSCachedTileLoaderJob.HOST_LIMIT.get()); + this.maxDownloadsPerHost.setValue(TMSCachedTileLoader.HOST_LIMIT.get()); } @@ -117,9 +117,18 @@ TMSLayer.setMinZoomLvl((Integer)this.minZoomLvl.getValue()); - TMSCachedTileLoader.MAX_OBJECTS_ON_DISK.put((Integer) this.maxElementsOnDisk.getValue()); + if (!TMSCachedTileLoader.MAX_OBJECTS_ON_DISK.get().equals(this.maxElementsOnDisk.getValue())) { + TMSCachedTileLoader.MAX_OBJECTS_ON_DISK.put((Integer) this.maxElementsOnDisk.getValue()); + restartRequired = true; + } - TMSCachedTileLoaderJob.THREAD_LIMIT.put((Integer) this.maxConcurrentDownloads.getValue()); - TMSCachedTileLoaderJob.HOST_LIMIT.put((Integer) this.maxDownloadsPerHost.getValue()); - TMSCachedTileLoaderJob.reconfigureDownloadDispatcher(); + if(!TMSCachedTileLoader.THREAD_LIMIT.get().equals(this.maxConcurrentDownloads.getValue())) { + TMSCachedTileLoader.THREAD_LIMIT.put((Integer) this.maxConcurrentDownloads.getValue()); + restartRequired = true; + } + + if(!TMSCachedTileLoader.HOST_LIMIT.get().equals(this.maxDownloadsPerHost.getValue())) { + TMSCachedTileLoader.HOST_LIMIT.put((Integer) this.maxDownloadsPerHost.getValue()); + restartRequired = true; + } if (!TMSLayer.PROP_TILECACHE_DIR.get().equals(this.tilecacheDir.getText())) {