Index: trunk/src/org/openstreetmap/josm/data/cache/HostLimitQueue.java =================================================================== --- trunk/src/org/openstreetmap/josm/data/cache/HostLimitQueue.java (revision 11443) +++ trunk/src/org/openstreetmap/josm/data/cache/HostLimitQueue.java (revision 11444) @@ -84,5 +84,12 @@ job = pollFirst(timeout, unit); if (job != null) { - acquireSemaphore(job); + try { + boolean gotLock = tryAcquireSemaphore(job, timeout, unit); + return gotLock ? job : null; + } catch (InterruptedException e) { + // acquire my got interrupted, first offer back what was taken + offer(job); + throw e; + } } return job; @@ -96,5 +103,11 @@ } job = takeFirst(); - acquireSemaphore(job); + try { + acquireSemaphore(job); + } catch (InterruptedException e) { + // acquire my got interrupted, first offer back what was taken + offer(job); + throw e; + } return job; } @@ -176,4 +189,19 @@ } + private boolean tryAcquireSemaphore(Runnable job, long timeout, TimeUnit unit) throws InterruptedException { + boolean ret = true; + if (job instanceof JCSCachedTileLoaderJob) { + final JCSCachedTileLoaderJob jcsJob = (JCSCachedTileLoaderJob) job; + Semaphore limit = getSemaphore(jcsJob); + if (limit != null) { + ret = limit.tryAcquire(timeout, unit); + if (ret) { + jcsJob.setFinishedTask(() -> releaseSemaphore(jcsJob)); + } + } + } + return ret; + } + private void releaseSemaphore(JCSCachedTileLoaderJob job) { Semaphore limit = getSemaphore(job); Index: trunk/test/unit/org/openstreetmap/josm/data/cache/HostLimitQueueTest.java =================================================================== --- trunk/test/unit/org/openstreetmap/josm/data/cache/HostLimitQueueTest.java (revision 11444) +++ trunk/test/unit/org/openstreetmap/josm/data/cache/HostLimitQueueTest.java (revision 11444) @@ -0,0 +1,167 @@ +// License: GPL. For details, see LICENSE file. +package org.openstreetmap.josm.data.cache; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.net.URL; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.commons.jcs.access.behavior.ICacheAccess; +import org.junit.Rule; +import org.junit.Test; +import org.openstreetmap.josm.testutils.JOSMTestRules; +import org.openstreetmap.josm.tools.Utils; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; + +/** + * Simple tests for ThreadPoolExecutor / HostLimitQueue veryfing, that this pair works OK + * @author Wiktor Niesiobedzki + * + */ +public class HostLimitQueueTest { + /** + * Setup test. + */ + @Rule + @SuppressFBWarnings(value = "URF_UNREAD_PUBLIC_OR_PROTECTED_FIELD") + public JOSMTestRules test = new JOSMTestRules().preferences().timeout(20 * 1000); + + + private static ThreadPoolExecutor getNewThreadPoolExecutor(String nameFormat, int workers, int queueLimit) { + HostLimitQueue workQueue = new HostLimitQueue(queueLimit); + ThreadPoolExecutor executor = new ThreadPoolExecutor( + 0, // 0 so for unused thread pools threads will eventually die, freeing also the threadpool + workers, // do not this number of threads + 300, // keepalive for thread + TimeUnit.SECONDS, + workQueue, + Utils.newThreadFactory(nameFormat, Thread.NORM_PRIORITY) + ); + workQueue.setExecutor(executor); + return executor; + } + + /** + * Mock class for tests + */ + static class Task extends JCSCachedTileLoaderJob { + private URL url; + private AtomicInteger counter; + private int id; + + public Task(ICacheAccess cache, URL url, AtomicInteger counter, int id) { + super(cache, 1, 1, null); + this.url = url; + this.counter = counter; + this.id = id; + } + + @Override + public void run() { + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + } finally { + this.counter.incrementAndGet(); + executionFinished(); + } + } + + @Override + public String getCacheKey() { + return ""; + } + + @Override + public URL getUrl() throws IOException { + return this.url; + } + + @Override + protected CacheEntry createCacheEntry(byte[] content) { + return null; + } + } + + /** + * Check if single threaded execution works properly + * @throws InterruptedException + * @throws IOException + */ + @Test + public void test_single_thread_per_host() throws InterruptedException, IOException { + ThreadPoolExecutor tpe = getNewThreadPoolExecutor("test-%d", 3, 1); + ICacheAccess cache = JCSCacheManager.getCache("test", 3, 0, ""); + AtomicInteger counter = new AtomicInteger(0); + long start = System.currentTimeMillis(); + for (int i = 0; i < 10; i++) { + tpe.execute(new Task(cache, new URL("http://localhost/"+i), counter, i)); + } + tpe.shutdown(); + tpe.awaitTermination(15, TimeUnit.SECONDS); // at most it should take ~10 seconds, so after 15 it's already failed + long duration = System.currentTimeMillis() - start; + // check that all tasks were executed + assertEquals(10, counter.get()); + // although there are 3 threads, we can make only 1 parallel call to localhost + // so it should take ~10 seconds to finish + // if it's shorter, it means that host limit does not work + assertTrue("Expected duration between 9 and 11 seconds not met. Actual duration: " + (duration /1000), duration < 11*1000 & duration > 9*1000 ); + } + + /** + * Check if two threaded execution work properly + * @throws InterruptedException + * @throws IOException + */ + @Test + public void test_multiple_thread_per_host() throws InterruptedException, IOException { + ThreadPoolExecutor tpe = getNewThreadPoolExecutor("test-%d", 3, 2); + ICacheAccess cache = JCSCacheManager.getCache("test", 3, 0, ""); + AtomicInteger counter = new AtomicInteger(0); + long start = System.currentTimeMillis(); + for (int i = 0; i < 10; i++) { + tpe.execute(new Task(cache, new URL("http://hostlocal/"+i), counter, i)); + } + tpe.shutdown(); + tpe.awaitTermination(15, TimeUnit.SECONDS); + long duration = System.currentTimeMillis() - start; + // check that all tasks were executed + assertEquals(10, counter.get()); + // although there are 3 threads, we can make only 2 parallel call to localhost + // so it should take ~5 seconds to finish + // if it's shorter, it means that host limit does not work + assertTrue("Expected duration between 4 and 6 seconds not met. Actual duration: " + (duration /1000), duration < 6*1000 & duration > 4*1000 ); + } + + /** + * Check two hosts + * @throws InterruptedException + * @throws IOException + */ + @Test + public void test_two_hosts() throws InterruptedException, IOException { + ThreadPoolExecutor tpe = getNewThreadPoolExecutor("test-%d", 3, 1); + ICacheAccess cache = JCSCacheManager.getCache("test", 3, 0, ""); + AtomicInteger counter = new AtomicInteger(0); + long start = System.currentTimeMillis(); + for (int i = 0; i < 10; i++) { + String url = (i % 2 == 0) ? "http://localhost" : "http://hostlocal"; + tpe.execute(new Task(cache, new URL(url+i), counter, i)); + } + tpe.shutdown(); + tpe.awaitTermination(15, TimeUnit.SECONDS); + long duration = System.currentTimeMillis() - start; + // check that all tasks were executed + assertEquals(10, counter.get()); + // although there are 3 threads, we can make only 1 parallel per host, and we have 2 hosts + // so it should take ~5 seconds to finish + // if it's shorter, it means that host limit does not work + assertTrue("Expected duration between 4 and 6 seconds not met. Actual duration: " + (duration /1000), duration < 6*1000 & duration > 4*1000 ); + } + +}