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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
