From 47215be18a72786d7bdc487ad94ceb8ee1bdd44b Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Sun, 6 Sep 2026 15:05:47 +0200 Subject: [PATCH 1/3] #2129 FetcherBolt: lock-free fetch queues ordered by next fetch time FetchItemQueues used a single monitor for adding, taking and finishing items, and getFetchItem() rotated a LinkedHashMap linearly over all the queues that were not ready yet. With many hosts the fetcher threads held that lock most of the time and the executor thread calling execute() stalled on every incoming tuple. Queues are now kept in a ConcurrentHashMap and the ones that may have an item ready are referenced from a DelayQueue of tickets ordered by their next fetch time, so taking an item is O(log n) and adding never waits for the fetcher threads. Per-queue state (size bound, in-progress count, next fetch time, crawl delays) is handled with atomics and a per-queue monitor only for the add/reap race. Empty queues are removed from the map as soon as they drain. Fixes #2129. Behaviour preserved: politeness per queue, asap release, max threads per queue, max queue size, crawl delay overrides from metadata and robots.txt, queue modes, metrics and the debug dump. Covered by the new FetchItemQueuesTest, including a concurrent producers/consumers test. Benchmark (50 fetcher threads, 1 producer, 20 URLs per host, delay 1s): 20000 hosts getFetchItem avg 1.58 ms -> 24 us addFetchItem p99 458 ms -> 0.11 ms, max 571 ms -> 0.8 ms throughput unchanged (bounded by politeness) --- .../apache/stormcrawler/bolt/FetcherBolt.java | 342 +++++++++++------- .../bolt/FetchItemQueuesTest.java | 238 ++++++++++++ 2 files changed, 447 insertions(+), 133 deletions(-) create mode 100644 core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java 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..cf3ce95ff 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,87 @@ 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. */ + 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 +295,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 +382,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()); + } } - public synchronized FetchItemQueue getFetchItemQueue(String id, Metadata metadata) { - FetchItemQueue fiq = queues.get(id); + /** 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)); + } + } + /** 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 +467,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 +516,47 @@ 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()) { + // finishFetchItem will schedule it again + fiq.scheduled.set(false); 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; } } @@ -1104,27 +1182,25 @@ public void execute(Tuple input) { 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..99162a3f7 --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java @@ -0,0 +1,238 @@ +/* + * 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.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.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()); + } +} From 4f133ca409bbb3e56cad4940af1bb3e4f109c173 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Sun, 6 Sep 2026 16:32:58 +0200 Subject: [PATCH 2/3] #2129 FetchItemQueues: do not lose the wakeup when a fetch finishes during the free-slot check In getFetchItem(), a fetch on the same queue could finish between hasFreeSlot() returning false and the clearing of the scheduled flag. finishFetchItem() then found the flag still set and did not issue a ticket, after which the flag was cleared: the queue was left with items and a free slot but no ticket, and stalled until an unrelated URL for the same host arrived. Clear the flag first and re-check, as the other branches already do. Covered by a deterministic test that pauses the poller inside the free-slot check while the in-progress fetch finishes. --- .../apache/stormcrawler/bolt/FetcherBolt.java | 6 +- .../bolt/FetchItemQueuesTest.java | 55 +++++++++++++++++++ 2 files changed, 60 insertions(+), 1 deletion(-) 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 cf3ce95ff..7bf716ba7 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java @@ -536,8 +536,12 @@ public FetchItem getFetchItem() { return null; } if (!fiq.hasFreeSlot()) { - // finishFetchItem will schedule it again 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; } final FetchItem it = fiq.poll(); diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java index 99162a3f7..0c28e9300 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java @@ -26,6 +26,7 @@ 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; @@ -33,6 +34,7 @@ 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; @@ -235,4 +237,57 @@ void concurrentProducersAndConsumersLoseNothing() throws Exception { 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); + } } From 44421c87cf05bdc114c2ea7ebf0102f7563b20b2 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Sun, 6 Sep 2026 16:36:28 +0200 Subject: [PATCH 3/3] #2129 FetchItemQueues: document the transient over-report in offer() and the smeared debug dump offer() increments the size before checking the bound and decrements on overflow, so getQueueSize() can transiently over-report by one while an offer is rejected; the value only feeds metrics and the debug dump. logQueuesContent() is no longer synchronized: it iterates the queues and their items with weakly consistent iterators while the fetcher threads keep working, so the dump is a smear rather than a point-in-time snapshot. Both are now stated in the javadoc. --- .../org/apache/stormcrawler/bolt/FetcherBolt.java | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) 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 7bf716ba7..35367a7a2 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java @@ -251,7 +251,12 @@ long getNextFetchTime() { return nextFetchTime.get(); } - /** Must be called with the monitor of this queue held. */ + /** + * 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; @@ -1184,6 +1189,12 @@ 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(); sb.append("\nNum queues : ").append(fetchQueues.queues.size());