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