diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java index ac141a4ad..35367a7a2 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java @@ -26,22 +26,23 @@ import java.net.UnknownHostException; import java.time.Instant; import java.util.Arrays; -import java.util.Collections; import java.util.HashMap; -import java.util.Iterator; -import java.util.LinkedHashMap; import java.util.Locale; import java.util.Map; import java.util.Map.Entry; -import java.util.concurrent.BlockingDeque; +import java.util.Queue; import java.util.concurrent.CancellationException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.DelayQueue; +import java.util.concurrent.Delayed; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; -import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.regex.Pattern; @@ -138,7 +139,7 @@ public Map getComponentConfiguration() { } /** This class described the item to be fetched. */ - private static class FetchItem { + static class FetchItem { String queueId; String url; @@ -204,60 +205,92 @@ public static FetchItem create(URL u, String url, Tuple t, String queueMode) { * proto/IP pair). It also keeps track of requests in progress and elapsed time between * requests. */ - private static class FetchItemQueue { - final BlockingDeque queue; + static class FetchItemQueue { + final Queue queue = new ConcurrentLinkedQueue<>(); + + final String id; + + /** Number of items in {@link #queue}; bounded by maxQueueSize. */ + private final AtomicInteger size = new AtomicInteger(); private final AtomicInteger inProgress = new AtomicInteger(); private final AtomicLong nextFetchTime = new AtomicLong(); - private long minCrawlDelay; + /** Whether a ticket for this queue is currently present in the ready queue. */ + private final AtomicBoolean scheduled = new AtomicBoolean(false); + + /** Set when the queue has been removed from the map because it was empty. */ + private boolean removed = false; + + private final int maxQueueSize; private final int maxThreads; - long crawlDelay; + volatile long minCrawlDelay; + volatile long crawlDelay; public FetchItemQueue( - int maxThreads, long crawlDelay, long minCrawlDelay, int maxQueueSize) { + String id, int maxThreads, long crawlDelay, long minCrawlDelay, int maxQueueSize) { + this.id = id; this.maxThreads = maxThreads; this.crawlDelay = crawlDelay; this.minCrawlDelay = minCrawlDelay; - this.queue = new LinkedBlockingDeque<>(maxQueueSize); + this.maxQueueSize = maxQueueSize; // ready to start setNextFetchTime(System.currentTimeMillis(), true); } public int getQueueSize() { - return queue.size(); + return size.get(); } public int getInProgressSize() { return inProgress.get(); } - public void finishFetchItem(FetchItem it, boolean asap) { - if (it != null) { - inProgress.decrementAndGet(); - setNextFetchTime(System.currentTimeMillis(), asap); - } - } - - public boolean addFetchItem(FetchItem it) { - return queue.offer(it); + long getNextFetchTime() { + return nextFetchTime.get(); } - public FetchItem getFetchItem() { - if (inProgress.get() >= maxThreads) { - return null; + /** + * Must be called with the monitor of this queue held. The size is incremented before the + * bound is checked and decremented again on overflow, so {@link #getQueueSize()} can + * transiently over-report by one while an offer is being rejected. Harmless: the value is + * only used for metrics and the debug dump. + */ + boolean offer(FetchItem it) { + if (removed) { + return false; } - if (nextFetchTime.get() > System.currentTimeMillis()) { - return null; + if (size.incrementAndGet() > maxQueueSize) { + size.decrementAndGet(); + return false; } - FetchItem it = queue.pollFirst(); + queue.add(it); + return true; + } + + FetchItem poll() { + FetchItem it = queue.poll(); if (it != null) { + size.decrementAndGet(); inProgress.incrementAndGet(); } return it; } + boolean hasFreeSlot() { + return inProgress.get() < maxThreads; + } + + boolean isReady(long now) { + return nextFetchTime.get() <= now; + } + + void finish(boolean asap) { + inProgress.decrementAndGet(); + setNextFetchTime(System.currentTimeMillis(), asap); + } + private void setNextFetchTime(long endTime, boolean asap) { if (!asap) { nextFetchTime.set(endTime + (maxThreads > 1 ? minCrawlDelay : crawlDelay)); @@ -267,13 +300,36 @@ private void setNextFetchTime(long endTime, boolean asap) { } } + /** + * A ticket in the ready queue: a queue which may have an item to fetch at {@code time}. Kept + * separate from the queue itself so that the ordering key is immutable while in the heap. + */ + private record QueueTicket(FetchItemQueue fiq, long time) implements Delayed { + + @Override + public long getDelay(TimeUnit unit) { + return unit.convert(time - System.currentTimeMillis(), TimeUnit.MILLISECONDS); + } + + @Override + public int compareTo(Delayed o) { + return Long.compare(time, ((QueueTicket) o).time); + } + } + /** * Convenience class - a collection of queues that keeps track of the total number of items, and * provides items eligible for fetching from any queue. + * + *

Queues are kept in a {@link ConcurrentHashMap} and the ones which may have an item ready + * are referenced from a {@link DelayQueue} ordered by their next fetch time: taking an item is + * O(log n) and does not require a global lock, so the executor thread adding URLs is never + * blocked by the fetcher threads. */ - private static class FetchItemQueues { - final Map queues = - Collections.synchronizedMap(new LinkedHashMap<>()); + static class FetchItemQueues { + final Map queues = new ConcurrentHashMap<>(); + + private final DelayQueue ready = new DelayQueue<>(); AtomicInteger inQueues = new AtomicInteger(0); @@ -331,32 +387,61 @@ public FetchItemQueues(Config conf) { * * @return true if the URL has been added, false otherwise. */ - public synchronized boolean addFetchItem(URL u, String url, Tuple input) { - FetchItem it = FetchItem.create(u, url, input, queueMode); + public boolean addFetchItem(URL u, String url, Tuple input) { + // built outside any lock: in byIP mode this resolves the hostname + final FetchItem it = FetchItem.create(u, url, input, queueMode); final Metadata metadata = (Metadata) input.getValueByField("metadata"); - FetchItemQueue fiq = getFetchItemQueue(it.queueId, metadata); - boolean added = fiq.addFetchItem(it); - if (added) { + while (true) { + FetchItemQueue fiq = getFetchItemQueue(it.queueId, metadata); + synchronized (fiq) { + if (fiq.removed) { + // reaped concurrently: get a fresh one + continue; + } + if (!fiq.offer(it)) { + return false; + } + } inQueues.incrementAndGet(); + schedule(fiq, fiq.getNextFetchTime()); + LOG.debug("{} added to queue {}", url, it.queueId); + return true; } - - LOG.debug("{} added to queue {}", url, it.queueId); - - return added; } - public synchronized void finishFetchItem(FetchItem it, boolean asap) { + public void finishFetchItem(FetchItem it, boolean asap) { FetchItemQueue fiq = queues.get(it.queueId); if (fiq == null) { LOG.warn("Attempting to finish item from unknown queue: {}", it.queueId); return; } - fiq.finishFetchItem(it, asap); + fiq.finish(asap); + if (fiq.queue.isEmpty()) { + reapIfEmpty(fiq); + } else { + schedule(fiq, fiq.getNextFetchTime()); + } + } + + /** Puts a ticket for the queue in the ready queue, unless one is already there. */ + private void schedule(FetchItemQueue fiq, long time) { + if (fiq.scheduled.compareAndSet(false, true)) { + ready.add(new QueueTicket(fiq, time)); + } } - public synchronized FetchItemQueue getFetchItemQueue(String id, Metadata metadata) { - FetchItemQueue fiq = queues.get(id); + /** Removes the queue from the map if it holds nothing and nothing is in progress. */ + private void reapIfEmpty(FetchItemQueue fiq) { + synchronized (fiq) { + if (fiq.queue.isEmpty() && fiq.getInProgressSize() == 0 && !fiq.removed) { + if (queues.remove(fiq.id, fiq)) { + fiq.removed = true; + } + } + } + } + public FetchItemQueue getFetchItemQueue(String id, Metadata metadata) { long delay = crawlDelay; long minDelay = minCrawlDelay; @@ -387,36 +472,42 @@ public synchronized FetchItemQueue getFetchItemQueue(String id, Metadata metadat } } - if (fiq == null) { - int threadVal = defaultMaxThread; - // custom maxThread value? - for (Entry p : customMaxThreads.entrySet()) { - if (p.getKey().matcher(id).matches()) { - threadVal = p.getValue(); - break; - } - } - - // overridden at URL level - // custom thread number from metadata? - if (metadata != null) { - final String val = metadata.getFirstValue(CRAWL_MAX_THREAD_KEY_NAME); - if (val != null) { - try { - threadVal = Integer.parseInt(val); - } catch (NumberFormatException e) { - LOG.warn( - "Invalid max threads value '{}' in metadata for queue '{}', using default.", - val, - id); - } - } - } - - // initialize queue - fiq = new FetchItemQueue(threadVal, delay, minDelay, maxQueueSize); - queues.put(id, fiq); - } + final long queueDelay = delay; + final long queueMinDelay = minDelay; + + FetchItemQueue fiq = + queues.computeIfAbsent( + id, + k -> { + int threadVal = defaultMaxThread; + // custom maxThread value? + for (Entry p : customMaxThreads.entrySet()) { + if (p.getKey().matcher(k).matches()) { + threadVal = p.getValue(); + break; + } + } + + // overridden at URL level + // custom thread number from metadata? + if (metadata != null) { + final String val = + metadata.getFirstValue(CRAWL_MAX_THREAD_KEY_NAME); + if (val != null) { + try { + threadVal = Integer.parseInt(val); + } catch (NumberFormatException e) { + LOG.warn( + "Invalid max threads value '{}' in metadata for queue '{}', using default.", + val, + k); + } + } + } + + return new FetchItemQueue( + k, threadVal, queueDelay, queueMinDelay, maxQueueSize); + }); // in cases where we have different pages with the same key that will fall in the same // queue, each one with a custom min crawl delay, we take the less aggressive @@ -430,55 +521,51 @@ public synchronized FetchItemQueue getFetchItemQueue(String id, Metadata metadat return fiq; } - public synchronized FetchItem getFetchItem() { - if (queues.isEmpty()) { - return null; - } - - FetchItemQueue start = null; - - do { - Iterator> i = queues.entrySet().iterator(); - - if (!i.hasNext()) { + /** + * Returns an item from a queue whose crawl delay has elapsed and which has a free slot, or + * null if there is none right now. + */ + public FetchItem getFetchItem() { + // bounded so that a burst of stale tickets can not keep a thread busy for long + for (int attempt = 0; attempt < 1000; attempt++) { + final QueueTicket ticket = ready.poll(); + if (ticket == null) { + // nothing is due: the head of the heap is the earliest queue return null; } - - Map.Entry nextEntry = i.next(); - - if (nextEntry == null) { + final FetchItemQueue fiq = ticket.fiq(); + final long now = System.currentTimeMillis(); + if (!fiq.isReady(now)) { + // the delay was extended after the ticket was issued: re-issue it + ready.add(new QueueTicket(fiq, fiq.getNextFetchTime())); return null; } - - FetchItemQueue fiq = nextEntry.getValue(); - - // We remove the entry and put it at the end of the map - i.remove(); - - // reap empty queues - if (fiq.getQueueSize() == 0 && fiq.getInProgressSize() == 0) { + if (!fiq.hasFreeSlot()) { + fiq.scheduled.set(false); + // a fetch may have finished between the check and the clearing of the + // flag, in which case its schedule() found the flag still set: re-check + if (fiq.hasFreeSlot() && !fiq.queue.isEmpty()) { + schedule(fiq, fiq.getNextFetchTime()); + } continue; } - - // Put the entry at the end no matter the result - queues.put(nextEntry.getKey(), nextEntry.getValue()); - - // In case of we are looping - if (start == null) { - start = fiq; - } else if (fiq == start) { - return null; + final FetchItem it = fiq.poll(); + fiq.scheduled.set(false); + if (it == null) { + reapIfEmpty(fiq); + if (!fiq.queue.isEmpty()) { + // lost a race with a concurrent add + schedule(fiq, fiq.getNextFetchTime()); + } + continue; } - - FetchItem fit = fiq.getFetchItem(); - - if (fit != null) { - inQueues.decrementAndGet(); - return fit; + inQueues.decrementAndGet(); + if (fiq.hasFreeSlot() && !fiq.queue.isEmpty()) { + // multi-threaded queue: let another thread pick the next one + schedule(fiq, fiq.getNextFetchTime()); } - - } while (!queues.isEmpty()); - + return it; + } return null; } } @@ -1102,29 +1189,33 @@ public void execute(Tuple input) { } } + /** + * Logs the content of the queues and the URLs being fetched. Not synchronized: the queues and + * their items are iterated with weakly consistent iterators while the fetcher threads keep + * working, so the dump is a smear over the time it takes to produce it rather than a + * point-in-time snapshot. Sizes and item lists may not add up exactly. + */ private void logQueuesContent() { StringBuilder sb = new StringBuilder(); - synchronized (fetchQueues.queues) { - sb.append("\nNum queues : ").append(fetchQueues.queues.size()); - for (Entry entry : fetchQueues.queues.entrySet()) { - sb.append("\nQueue ID : ").append(entry.getKey()); - FetchItemQueue fiq = entry.getValue(); - sb.append("\t size : ").append(fiq.getQueueSize()); - sb.append("\t in progress : ").append(fiq.getInProgressSize()); - for (FetchItem fetchItem : fiq.queue) { - sb.append("\n\t").append(fetchItem.url); - } + sb.append("\nNum queues : ").append(fetchQueues.queues.size()); + for (Entry entry : fetchQueues.queues.entrySet()) { + sb.append("\nQueue ID : ").append(entry.getKey()); + FetchItemQueue fiq = entry.getValue(); + sb.append("\t size : ").append(fiq.getQueueSize()); + sb.append("\t in progress : ").append(fiq.getInProgressSize()); + for (FetchItem fetchItem : fiq.queue) { + sb.append("\n\t").append(fetchItem.url); } - LOG.info("Dumping queue content {}", sb.toString()); + } + LOG.info("Dumping queue content {}", sb.toString()); - StringBuilder sb2 = new StringBuilder("\n"); - // dump the list of URLs being fetched - for (int i = 0; i < beingFetched.length; i++) { - if (beingFetched[i].length() > 0) { - sb2.append("\n\tThread #").append(i).append(": ").append(beingFetched[i]); - } + StringBuilder sb2 = new StringBuilder("\n"); + // dump the list of URLs being fetched + for (int i = 0; i < beingFetched.length; i++) { + if (beingFetched[i].length() > 0) { + sb2.append("\n\tThread #").append(i).append(": ").append(beingFetched[i]); } - LOG.info("URLs being fetched {}", sb2.toString()); } + LOG.info("URLs being fetched {}", sb2.toString()); } } diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java new file mode 100644 index 000000000..0c28e9300 --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java @@ -0,0 +1,293 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.bolt; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.net.MalformedURLException; +import java.net.URL; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.storm.Config; +import org.apache.storm.tuple.Tuple; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.bolt.FetcherBolt.FetchItem; +import org.apache.stormcrawler.bolt.FetcherBolt.FetchItemQueue; +import org.apache.stormcrawler.bolt.FetcherBolt.FetchItemQueues; +import org.apache.stormcrawler.util.URLUtil; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +class FetchItemQueuesTest { + + private static FetchItemQueues queues(Object... kv) { + Config conf = new Config(); + for (int i = 0; i < kv.length; i += 2) { + conf.put((String) kv[i], kv[i + 1]); + } + return new FetchItemQueues(conf); + } + + private static Tuple tuple(Metadata md) { + Tuple t = mock(Tuple.class); + when(t.contains("key")).thenReturn(false); + when(t.getValueByField("metadata")).thenReturn(md); + return t; + } + + private static boolean add(FetchItemQueues q, String url) throws MalformedURLException { + return add(q, url, new Metadata()); + } + + private static boolean add(FetchItemQueues q, String url, Metadata md) + throws MalformedURLException { + URL u = URLUtil.toURL(url); + return q.addFetchItem(u, url, tuple(md)); + } + + private static FetchItem awaitItem(FetchItemQueues q, long maxMillis) + throws InterruptedException { + long deadline = System.currentTimeMillis() + maxMillis; + while (System.currentTimeMillis() < deadline) { + FetchItem it = q.getFetchItem(); + if (it != null) { + return it; + } + Thread.sleep(5); + } + return null; + } + + @Test + void itemIsReturnedOnceAndHostWaitsForCrawlDelay() throws Exception { + FetchItemQueues q = queues("fetcher.server.delay", 0.3f); + Assertions.assertTrue(add(q, "http://a.net/1")); + Assertions.assertTrue(add(q, "http://a.net/2")); + Assertions.assertEquals(2, q.inQueues.get()); + + FetchItem first = q.getFetchItem(); + Assertions.assertNotNull(first); + Assertions.assertEquals("http://a.net/1", first.url); + Assertions.assertEquals(1, q.inQueues.get()); + // one thread per host: nothing else from a.net while the fetch is in progress + Assertions.assertNull(q.getFetchItem()); + + q.finishFetchItem(first, false); + // still nothing: the crawl delay has not elapsed + Assertions.assertNull(q.getFetchItem()); + FetchItem second = awaitItem(q, 2000); + Assertions.assertNotNull(second); + Assertions.assertEquals("http://a.net/2", second.url); + Assertions.assertEquals(0, q.inQueues.get()); + } + + @Test + void finishingAsapMakesHostImmediatelyAvailable() throws Exception { + FetchItemQueues q = queues("fetcher.server.delay", 5.0f); + add(q, "http://a.net/1"); + add(q, "http://a.net/2"); + FetchItem first = q.getFetchItem(); + q.finishFetchItem(first, true); + FetchItem second = q.getFetchItem(); + Assertions.assertNotNull(second); + Assertions.assertEquals("http://a.net/2", second.url); + } + + @Test + void differentHostsAreServedBackToBack() throws Exception { + FetchItemQueues q = queues("fetcher.server.delay", 5.0f); + add(q, "http://a.net/1"); + add(q, "http://b.net/1"); + add(q, "http://c.net/1"); + Set got = ConcurrentHashMap.newKeySet(); + for (int i = 0; i < 3; i++) { + FetchItem it = q.getFetchItem(); + Assertions.assertNotNull(it); + got.add(it.queueId); + } + Assertions.assertEquals(Set.of("a.net", "b.net", "c.net"), got); + Assertions.assertNull(q.getFetchItem()); + } + + @Test + void maxQueueSizeRejectsExtraItems() throws Exception { + FetchItemQueues q = queues("fetcher.max.queue.size", 2); + Assertions.assertTrue(add(q, "http://a.net/1")); + Assertions.assertTrue(add(q, "http://a.net/2")); + Assertions.assertFalse(add(q, "http://a.net/3")); + Assertions.assertTrue(add(q, "http://b.net/1")); + Assertions.assertEquals(3, q.inQueues.get()); + } + + @Test + void multipleThreadsPerQueueAllowConcurrentFetchesFromSameHost() throws Exception { + FetchItemQueues q = queues("fetcher.threads.per.queue", 2, "fetcher.server.delay", 5.0f); + add(q, "http://a.net/1"); + add(q, "http://a.net/2"); + add(q, "http://a.net/3"); + FetchItem first = q.getFetchItem(); + FetchItem second = q.getFetchItem(); + Assertions.assertNotNull(first); + Assertions.assertNotNull(second); + // two in progress: the third has to wait + Assertions.assertNull(q.getFetchItem()); + q.finishFetchItem(first, true); + Assertions.assertNotNull(q.getFetchItem()); + } + + @Test + void crawlDelayFromMetadataOverridesDefault() throws Exception { + FetchItemQueues q = queues("fetcher.server.delay", 5.0f); + Metadata md = new Metadata(); + md.setValue("crawl.delay", "0"); + add(q, "http://a.net/1", md); + add(q, "http://a.net/2", md); + FetchItem first = q.getFetchItem(); + q.finishFetchItem(first, false); + Assertions.assertNotNull(awaitItem(q, 500)); + } + + @Test + void emptyQueuesAreRemoved() throws Exception { + FetchItemQueues q = queues("fetcher.server.delay", 0.0f); + add(q, "http://a.net/1"); + Assertions.assertEquals(1, q.queues.size()); + FetchItem it = q.getFetchItem(); + q.finishFetchItem(it, false); + // drained: the host must not stay in memory forever + Assertions.assertNull(awaitItem(q, 200)); + Assertions.assertEquals(0, q.queues.size()); + } + + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void concurrentProducersAndConsumersLoseNothing() throws Exception { + final int hosts = 200; + final int perHost = 50; + final int total = hosts * perHost; + FetchItemQueues q = queues("fetcher.server.delay", 0.0f); + AtomicInteger fetched = new AtomicInteger(); + Set seen = ConcurrentHashMap.newKeySet(); + AtomicBoolean failed = new AtomicBoolean(); + List threads = new ArrayList<>(); + for (int p = 0; p < 4; p++) { + final int producer = p; + threads.add( + new Thread( + () -> { + try { + for (int h = producer; h < hosts; h += 4) { + for (int u = 0; u < perHost; u++) { + if (!add(q, "http://h" + h + ".net/" + u)) { + failed.set(true); + } + } + } + } catch (Exception e) { + failed.set(true); + } + })); + } + for (int c = 0; c < 16; c++) { + threads.add( + new Thread( + () -> { + while (fetched.get() < total && !failed.get()) { + FetchItem it = q.getFetchItem(); + if (it == null) { + Thread.yield(); + continue; + } + if (!seen.add(it.url)) { + failed.set(true); + } + fetched.incrementAndGet(); + q.finishFetchItem(it, false); + } + })); + } + threads.forEach(Thread::start); + for (Thread t : threads) { + t.join(); + } + Assertions.assertFalse(failed.get(), "duplicate, lost or rejected item"); + Assertions.assertEquals(total, seen.size()); + Assertions.assertEquals(0, q.inQueues.get()); + } + + /** + * A fetch finishing between the "no free slot" check and the clearing of the scheduled flag + * must not leave the queue without a ticket: the URL waiting behind would never be fetched. + */ + @Test + @Timeout(value = 10, unit = TimeUnit.SECONDS) + void fetchFinishingDuringFreeSlotCheckDoesNotLoseTheWakeup() throws Exception { + FetchItemQueues q = queues("fetcher.server.delay", 0.0f); + CountDownLatch inCheck = new CountDownLatch(1); + CountDownLatch proceed = new CountDownLatch(1); + AtomicBoolean armed = new AtomicBoolean(false); + // queue whose "no free slot" answer pauses until the test lets it continue + FetchItemQueue hooked = + new FetchItemQueue("a.net", 1, 0, 0, Integer.MAX_VALUE) { + @Override + boolean hasFreeSlot() { + boolean free = super.hasFreeSlot(); + if (!free && armed.compareAndSet(true, false)) { + inCheck.countDown(); + try { + proceed.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + return free; + } + }; + q.queues.put("a.net", hooked); + + add(q, "http://a.net/1"); + FetchItem first = q.getFetchItem(); + Assertions.assertNotNull(first); + // arrives while the first is in progress: issues a ticket + add(q, "http://a.net/2"); + armed.set(true); + + FetchItem[] polled = new FetchItem[1]; + Thread poller = new Thread(() -> polled[0] = q.getFetchItem()); + poller.start(); + // the poller is now inside getFetchItem, having seen no free slot + inCheck.await(); + // the first fetch finishes: its schedule() finds the flag still set + q.finishFetchItem(first, true); + proceed.countDown(); + poller.join(); + + // whoever polls next must get the second URL + FetchItem second = polled[0] != null ? polled[0] : awaitItem(q, 2000); + Assertions.assertNotNull(second, "second URL lost: no ticket left for the queue"); + Assertions.assertEquals("http://a.net/2", second.url); + } +}