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