You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@stormcrawler.apache.org by "GGraziadei (via GitHub)" <gi...@apache.org> on 2026/09/06 13:32:17 UTC
[PR] #2129 FetcherBolt: lock-free fetch queues ordered by next fetch time [stormcrawler]
GGraziadei opened a new pull request, #2132:
URL: https://github.com/apache/stormcrawler/pull/2132
Fixes #2129.
`FetchItemQueues` used a single monitor for adding, taking and finishing items, and `getFetchItem()` rotated a `LinkedHashMap` linearly over every queue whose crawl delay had not elapsed. With many hosts the fetcher threads held that lock most of the time and the executor thread calling `execute()` stalled on every incoming tuple.
### Change
- Queues live in a `ConcurrentHashMap`; the ones that may have an item ready are referenced from a `DelayQueue` of tickets ordered by their next fetch time. 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) uses atomics; a per-queue monitor is used only for the add/reap race. Empty queues are removed from the map as soon as they drain.
- `FetchItem` is built before any lock is taken, so a DNS lookup in `byIP` mode no longer blocks the fetcher threads.
### Behaviour preserved
Politeness per queue, `asap` release, `fetcher.threads.per.queue` and `fetcher.maxThreads.*`, `fetcher.max.queue.size`, crawl delay overrides from metadata and robots.txt, queue modes, metrics and the debug dump. One observable difference: among hosts that are ready at the same time, the order is "ready for the longest first" rather than round-robin by insertion order.
### Tests
New `FetchItemQueuesTest`: politeness, `asap`, several hosts back to back, queue size bound, multiple threads per queue, crawl delay from metadata, reaping of empty queues, and a concurrent producers/consumers test checking that nothing is lost or duplicated. Existing `FetcherBoltTest` and `SimpleFetcherBoltTest` pass unchanged.
### Benchmark
50 fetcher threads, 1 producer thread, 20 URLs per host, `fetcher.server.delay` 1s, 8s per run, best of the measured samples.
| hosts | metric | before | after |
|---|---|---|---|
| 20000 | `getFetchItem` avg | 1.58 ms | 24 us |
| 20000 | `getFetchItem` p99 | 5.0 ms | 24 us |
| 20000 | `addFetchItem` avg | 14.1 ms | 24 us |
| 20000 | `addFetchItem` p99 | 458 ms | 0.11 ms |
| 2000 | `addFetchItem` p99 | 2.5 ms | 0.13 ms |
| 20000 | fetched/s | 20001 | 20000 |
Throughput is unchanged, as expected: it is bounded by politeness. What goes away is the executor stall and the CPU burnt scanning queues that are not ready.
---
### For all changes
- [x] Is there a issue associated with this PR? Is it referenced in the commit message?
- [x] Does your PR title start with `#XXXX` where `XXXX` is the issue number you are trying to resolve?
- [x] Has your PR been rebased against the latest commit within the target branch (typically main)?
- [x] Is your initial contribution a single, squashed commit?
- [x] Is the code properly formatted with `mvn git-code-format:format-code -Dgcf.globPattern="**/*" -Dskip.format.code=false`?
### For code changes
- [x] Have you ensured that the full suite of tests is executed via `mvn clean verify`?
- [x] Have you written or updated unit tests to verify your changes?
- [x] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)? (no new dependencies)
- [x] If applicable, have you updated the LICENSE file, including the main LICENSE file? (not applicable)
- [x] If applicable, have you updated the NOTICE file, including the main NOTICE file? (not applicable)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@stormcrawler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] #2129 FetcherBolt: lock-free fetch queues ordered by next fetch time [stormcrawler]
Posted by "GGraziadei (via GitHub)" <gi...@apache.org>.
GGraziadei commented on code in PR #2132:
URL: https://github.com/apache/stormcrawler/pull/2132#discussion_r3944260878
##########
core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java:
##########
@@ -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<Entry<String, FetchItemQueue>> 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<String, FetchItemQueue> 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()) {
Review Comment:
Thanks, added the fix and tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@stormcrawler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] #2129 FetcherBolt: lock-free fetch queues ordered by next fetch time [stormcrawler]
Posted by "GGraziadei (via GitHub)" <gi...@apache.org>.
GGraziadei commented on code in PR #2132:
URL: https://github.com/apache/stormcrawler/pull/2132#discussion_r3952692297
##########
core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java:
##########
@@ -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<FetchItem> queue;
+ static class FetchItemQueue {
+ final Queue<FetchItem> 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();
Review Comment:
Confirmed and fixed in 2986b7c.
`poll()` now reserves the slot first with a CAS loop (`tryAcquireSlot()`, refusing when `inProgress >= maxThreads`), then dequeues, and releases the slot again if the queue was empty. While an item is in flight `inProgress` is never zero, so `reapIfEmpty` cannot remove the queue from under it. The CAS also makes the bound exact when two threads pass `hasFreeSlot()` together with one slot left, which the old `hasFreeSlot()` + `poll()` pair allowed to overshoot by one. In `getFetchItem` the null-poll branch re-issues the ticket only if a slot is free and the queue is not empty; otherwise the fetch that finishes next schedules it.
Tests, both failing on the previous revision:
- `fetchFinishingDuringPollDoesNotReapTheQueue`: a package-private no-op `afterDequeue()` hook in `poll()` lets the test pause right after the last item is taken. The other active fetch then finishes; the queue must still be in the map. After resuming, the item is returned, `inProgress == 1`, finishing it goes to the original queue, and only then is the queue reaped. Before the fix the queue was gone at the first assertion.
- `pollEnforcesMaxThreadsAtomically`: `maxThreads = 1`, two queued items, second `poll()` returns null without losing the item; after `finish()` it is handed out.
Full `core` verify: 454 tests, 0 failures.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@stormcrawler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] #2129 FetcherBolt: lock-free fetch queues ordered by next fetch time [stormcrawler]
Posted by "rzo1 (via GitHub)" <gi...@apache.org>.
rzo1 merged PR #2132:
URL: https://github.com/apache/stormcrawler/pull/2132
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@stormcrawler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] #2129 FetcherBolt: lock-free fetch queues ordered by next fetch time [stormcrawler]
Posted by "dpol1 (via GitHub)" <gi...@apache.org>.
dpol1 commented on code in PR #2132:
URL: https://github.com/apache/stormcrawler/pull/2132#discussion_r3951576598
##########
core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java:
##########
@@ -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<FetchItem> queue;
+ static class FetchItemQueue {
+ final Queue<FetchItem> 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();
Review Comment:
Reap can run between `queue.poll()` and `inProgress.incrementAndGet()`: with `fetcher.threads.per.queue` > 1 (or an equivalent per-queue override), another fetch on the same queue finishing in that window can observe an empty queue and `inProgress == 0`, and remove the queue while the just-polled item is being dispatched. When that item finishes, `finishFetchItem` either hits the unknown-queue branch or, if the queue has been recreated, decrements the replacement queue's counter. This can undercount active fetches, potentially make the counter negative, and allow `maxThreads` to be exceeded.
Could we increment `inProgress` before `queue.poll()`, undo it on `null`, and add a deterministic regression test for this interleaving? Pausing the poll after it removes the last queued item, then finishing the other active fetch, should leave the original queue in the map; after resuming, the remaining fetch should be counted and finish normally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@stormcrawler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] #2129 FetcherBolt: lock-free fetch queues ordered by next fetch time [stormcrawler]
Posted by "GGraziadei (via GitHub)" <gi...@apache.org>.
GGraziadei commented on PR #2132:
URL: https://github.com/apache/stormcrawler/pull/2132#issuecomment-5590014884
All three observations verified with tests before changing anything, fixed in 5f40940.
**Racy max on the crawl delays.** Confirmed. A spin-barrier test with two long-lived threads calling `getFetchItemQueue` for the same host with delays 1000/2000 lost the larger value in 2284 of 300000 rounds on the previous revision (a `CyclicBarrier` version never hit it: thread release skew is orders of magnitude above the window). `crawlDelay` and `minCrawlDelay` are now `AtomicLong`s raised with `accumulateAndGet(v, Math::max)`; the fetcher thread keeps setting them outright from robots.txt as it did before, which was already outside the old lock. Test `concurrentDelayUpdatesKeepTheMaximum`, 300000 rounds in about 0.2 s.
**Size decremented after the dequeue.** Confirmed, deterministically: a queue with `maxQueueSize = 1`, poll paused via the `afterDequeue()` hook right after the item is out, a concurrent `addFetchItem` returned false. The size is now reserved before the dequeue and given back if the queue was empty, mirroring `tryAcquireSlot`. The `offer()` javadoc now says why its own transient over-report is harmless (offers are serialised by the monitor) instead of the wrong claim about metrics; `getQueueSize()` clamps at zero for the metrics while a poll of an empty queue is in flight. Test `addDuringPollIsNotRefusedAtTheSizeBound`.
**Stale ticket returning null.** Confirmed, deterministically: `a.net` ticket at the head with its delay extended to 10 s by a `finish`, `b.net` due, `getFetchItem` returned null. Now `continue`; the re-issued ticket carries a future time so it cannot come straight back, and the loop is bounded anyway. Test `staleTicketDoesNotHideAnotherReadyQueue`. Same point as @dpol1's inline question.
Full `core` verify: 457 tests, 0 failures.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01C6YDC99NSJKU8JeWF2sGfY
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@stormcrawler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] #2129 FetcherBolt: lock-free fetch queues ordered by next fetch time [stormcrawler]
Posted by "GGraziadei (via GitHub)" <gi...@apache.org>.
GGraziadei commented on code in PR #2132:
URL: https://github.com/apache/stormcrawler/pull/2132#discussion_r3961086289
##########
core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java:
##########
@@ -430,55 +549,52 @@ public synchronized FetchItemQueue getFetchItemQueue(String id, Metadata metadat
return fiq;
}
- public synchronized FetchItem getFetchItem() {
- if (queues.isEmpty()) {
- return null;
- }
-
- FetchItemQueue start = null;
-
- do {
- Iterator<Entry<String, FetchItemQueue>> 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<String, FetchItemQueue> 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()));
Review Comment:
No reason that survives scrutiny: the ticket was due, only its queue is not, and other tickets may be. Changed to `continue` in 5f40940 (the re-issued ticket has a future time, so it cannot be polled straight back, and the loop is bounded). `staleTicketDoesNotHideAnotherReadyQueue` reproduces the stall: `a.net` at the head with its delay pushed to 10 s, `b.net` due, `getFetchItem` returned null on the previous revision.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@stormcrawler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] #2129 FetcherBolt: lock-free fetch queues ordered by next fetch time [stormcrawler]
Posted by "rzo1 (via GitHub)" <gi...@apache.org>.
rzo1 commented on code in PR #2132:
URL: https://github.com/apache/stormcrawler/pull/2132#discussion_r3944168127
##########
core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java:
##########
@@ -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<Entry<String, FetchItemQueue>> 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<String, FetchItemQueue> 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()) {
Review Comment:
Lost wakeup: this can leave a queue with items and a free slot but no ticket.
Interleaving: a fetch on this queue finishes between `hasFreeSlot()` returning false and `scheduled.set(false)` on line 540. `finishFetchItem` calls `schedule`, whose CAS from false to true fails because `scheduled` is still true, and then this thread sets it to false. Nothing re-issues a ticket until an unrelated URL for the same host arrives, so that host stalls.
The two branches below (lines 544 to 556) already clear the flag and then re-check, which is the pattern this one needs.
```suggestion
if (!fiq.hasFreeSlot()) {
// clear first, then re-check: a fetch finishing in between
// would otherwise find the flag still set, skip scheduling,
// and leave the queue with items but no ticket
fiq.scheduled.set(false);
if (fiq.hasFreeSlot() && !fiq.queue.isEmpty()) {
schedule(fiq, fiq.getNextFetchTime());
}
continue;
}
```
Please add a targeted test: `fetcher.threads.per.queue: 1`, take an item, assert the next one becomes available after `finishFetchItem`, driven from two threads so the race is reachable. Without it this failure shows up only as `concurrentProducersAndConsumersLoseNothing` occasionally hitting its 60 s timeout, which reads as flakiness.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@stormcrawler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] #2129 FetcherBolt: lock-free fetch queues ordered by next fetch time [stormcrawler]
Posted by "dpol1 (via GitHub)" <gi...@apache.org>.
dpol1 commented on code in PR #2132:
URL: https://github.com/apache/stormcrawler/pull/2132#discussion_r3956249037
##########
core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java:
##########
@@ -430,55 +549,52 @@ public synchronized FetchItemQueue getFetchItemQueue(String id, Metadata metadat
return fiq;
}
- public synchronized FetchItem getFetchItem() {
- if (queues.isEmpty()) {
- return null;
- }
-
- FetchItemQueue start = null;
-
- do {
- Iterator<Entry<String, FetchItemQueue>> 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<String, FetchItemQueue> 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()));
Review Comment:
Why `return null` here rather than `continue`? Returning here can make this fetcher wait even when another host is ready.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@stormcrawler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] #2129 FetcherBolt: lock-free fetch queues ordered by next fetch time [stormcrawler]
Posted by "GGraziadei (via GitHub)" <gi...@apache.org>.
GGraziadei commented on code in PR #2132:
URL: https://github.com/apache/stormcrawler/pull/2132#discussion_r3952697436
##########
core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java:
##########
@@ -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<FetchItem> queue;
+ static class FetchItemQueue {
+ final Queue<FetchItem> 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();
Review Comment:
Hi @dpol1 thanks for your comment.
The reported path is repeatable and unintentional. I agree with the proposed solution, fixing it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@stormcrawler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org