Index: trunk/src/org/openstreetmap/josm/data/cache/JCSCachedTileLoaderJob.java =================================================================== --- trunk/src/org/openstreetmap/josm/data/cache/JCSCachedTileLoaderJob.java (revision 8313) +++ trunk/src/org/openstreetmap/josm/data/cache/JCSCachedTileLoaderJob.java (revision 8314) @@ -56,5 +56,35 @@ * maximum download threads that will be started */ - public static final IntegerProperty THREAD_LIMIT = new IntegerProperty("cache.jcs.max_threads", 10); + public final static IntegerProperty THREAD_LIMIT = new IntegerProperty("cache.jcs.max_threads", 10); + + public static class LIFOQueue extends LinkedBlockingDeque { + 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. + * + * 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. + * + * 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 + * the response, so later it could be used). We could actually cancel what is in LIFOQueue, but this is a tradeoff between simplicity + * and performance (we do want to have something to offer to worker threads before tasks will be resubmitted by class consumer) + */ private static Executor 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) @@ -63,18 +93,7 @@ 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 LinkedBlockingDeque(5) { - /* keep the queue size fairly small, we do not want to - download a lot of tiles, that user is not seeing anyway */ - @Override - public boolean offer(Runnable t) { - return super.offerFirst(t); - } - - @Override - public Runnable remove() { - return super.removeFirst(); - } - } + new LIFOQueue(5) ); + private static ConcurrentMap> inProgress = new ConcurrentHashMap<>(); private static ConcurrentMap useHead = new ConcurrentHashMap<>(); @@ -157,6 +176,11 @@ // object not in cache, so submit work to separate thread try { - // use getter method, so subclasses may override executors, to get separate thread pool - getDownloadExecutor().execute(JCSCachedTileLoaderJob.this); + 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 @@ -165,4 +189,21 @@ } } + } + + /** + * 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; + } + + /** + * This method is run when job has finished + */ + protected void executionFinished() { } @@ -219,4 +260,5 @@ } } finally { + executionFinished(); currentThread.setName(oldName); } Index: trunk/src/org/openstreetmap/josm/data/imagery/TMSCachedTileLoaderJob.java =================================================================== --- trunk/src/org/openstreetmap/josm/data/imagery/TMSCachedTileLoaderJob.java (revision 8313) +++ trunk/src/org/openstreetmap/josm/data/imagery/TMSCachedTileLoaderJob.java (revision 8314) @@ -8,5 +8,4 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executor; -import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadPoolExecutor; @@ -42,77 +41,57 @@ /** + * Limit definition for per host concurrent connections + */ + public final static 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<>(); + + /** * overrides the THREAD_LIMIT in superclass, as we want to have separate limit and pool for TMS */ - 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); - - - private static class LIFOQueue extends LinkedBlockingDeque { - public LIFOQueue(int capacity) { - super(capacity); - } - - private final static Semaphore getSemaphore(Runnable r) { - if (!(r instanceof TMSCachedTileLoaderJob)) - return null; - TMSCachedTileLoaderJob cachedJob = (TMSCachedTileLoaderJob) r; - Semaphore limit = HOST_LIMITS.get(cachedJob.getUrl().getHost()); - if (limit == null) { - synchronized(HOST_LIMITS) { - limit = HOST_LIMITS.get(cachedJob.getUrl().getHost()); - if (limit == null) { - limit = new Semaphore(HOST_LIMIT.get().intValue()); - HOST_LIMITS.put(cachedJob.getUrl().getHost(), limit); - } - } - } - return limit; - } - - private boolean acquireSemaphore(Runnable r) { - boolean ret = true; - Semaphore limit = getSemaphore(r); - if (limit != null) { - ret = limit.tryAcquire(); - if (!ret) { - Main.debug("rejecting job because of per host limit"); - } - } - return ret; - } - - @Override - public boolean offer(Runnable t) { - return acquireSemaphore(t) && super.offerFirst(t); - } - - private Runnable releaseSemaphore(Runnable r) { - Semaphore limit = getSemaphore(r); - if (limit != null) - limit.release(); - return r; - } - - @Override - public Runnable remove() { - return releaseSemaphore(super.removeFirst()); - } - - @Override - public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException { - return releaseSemaphore(super.poll(timeout, unit)); - } - - @Override - public Runnable take() throws InterruptedException { - return releaseSemaphore(super.take()); - } - } - - private static Map HOST_LIMITS = new ConcurrentHashMap<>(); + public final static IntegerProperty THREAD_LIMIT = new IntegerProperty("imagery.tms.tmsloader.maxjobs", 25); /** @@ -143,5 +122,4 @@ } - /** * Constructor for creating a job, to get a specific tile from cache @@ -175,5 +153,5 @@ * this doesn't needs to be synchronized, as it's not that costly to keep only one execution * in parallel, but URL creation and Tile.getUrl() are costly and are not needed when fetching - * data from cache + * data from cache, that's why URL creation is postponed until it's needed * * We need to have static url value for TileLoaderJob, as for some TileSources we might get different @@ -234,4 +212,14 @@ } + @Override + protected boolean executionGuard() { + return acquireSemaphore(); + } + + @Override + protected void executionFinished() { + releaseSemaphore(); + } + public void submit() { tile.initLoading(); @@ -246,4 +234,5 @@ case FAILURE: tile.setError("Problem loading tile"); + // no break intentional here case SUCCESS: handleNoTileAtZoom(); @@ -254,6 +243,7 @@ } } + // no break intentional here case REJECTED: - // do not set anything here, leave waiting sign + // do nothing } if (listener != null) {