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