rzo1 commented on code in PR #2117:
URL: https://github.com/apache/stormcrawler/pull/2117#discussion_r3944168189
##########
external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/StatusUpdaterBolt.java:
##########
@@ -418,7 +751,142 @@ public void store(
KnownURLItem.newBuilder().setInfo(info).setRefetchableFromDate(date).build());
}
- requestObserver.onNext(itemBuilder.setID(url).build());
+ final URLItem item = itemBuilder.setID(url).build();
+
+ // discovered URLs travel in batches on the PutDiscovered endpoint,
known URLs keep
+ // using the streaming endpoint
+ if (status.equals(Status.DISCOVERED)) {
+ boolean shouldBatch;
+ boolean flushNow = false;
+ synchronized (batchLock) {
+ // re-read inside the lock: batching can be disabled
concurrently by the
+ // fallback for frontiers without the PutDiscovered endpoint
+ shouldBatch = batching;
+ if (shouldBatch) {
+ if (batchBuffer.isEmpty()) {
+ oldestBufferedAt = System.currentTimeMillis();
+ }
+ batchBuffer.add(item);
+ flushNow = batchBuffer.size() >= batchSize;
+ }
+ }
+ if (!shouldBatch) {
+ sendOnStreamingEndpoint(item);
+ return;
+ }
+ if (flushNow) {
+ flushBatch();
+ }
+ return;
+ }
+
+ if (batching) {
+ // the outlinks buffered so far belong to the page whose status is
now updated:
+ // a natural boundary for the batch
+ flushBatch();
+ }
+
+ sendOnStreamingEndpoint(item);
+ }
+
+ /** Sends the buffered discovered URLs as one batch, if any. */
+ private void flushBatch() {
+ flushBatch(false);
+ }
+
+ /**
+ * Sends the buffered discovered URLs as one batch, if any.
+ *
+ * @param awaitTransport whether to wait briefly for the transport to
become ready before
+ * sending; only the flusher thread may do so, the Storm executor
thread never stalls
+ */
+ private void flushBatch(boolean awaitTransport) {
+ final List<URLItem> items;
+ final StreamObserver<DiscoveredBatch> stream;
+ synchronized (batchLock) {
+ if (!batching || batchBuffer.isEmpty()) {
+ return;
+ }
+ if (batchRequestObserver == null) {
+ // the previous stream died: open a new one
+ batchRequestObserver = newPutDiscoveredStream();
+ }
+ stream = batchRequestObserver;
+ items = new ArrayList<>(batchBuffer);
+ batchBuffer.clear();
+ oldestBufferedAt = 0;
+ }
+
+ final String batchID = "batch-" + batchSequences.incrementAndGet();
+ final DiscoveredBatch.Builder batchBuilder =
DiscoveredBatch.newBuilder().setID(batchID);
+ for (URLItem buffered : items) {
+ batchBuilder.addItems(buffered.getDiscovered().getInfo());
+ }
+ final DiscoveredBatch batch = batchBuilder.build();
+
+ // registered before the send so that a fast ack can never miss it
+ synchronized (batchLock) {
+ pendingBatches.put(batchID, items);
+ }
+
+ try {
+ if (awaitTransport) {
+ // follow the transport's lead: wait briefly for it to take
the batch without
+ // buffering it, woken by the on-ready handler. The timeout is
a backstop, not a
+ // poll interval.
+ final ClientCallStreamObserver<DiscoveredBatch> transport =
batchTransport;
+ if (transport != null && !transport.isReady()) {
+ synchronized (flow) {
+ flow.wait(BATCH_FLUSH_DELAY_MS);
+ }
+ }
+ }
+ synchronized (sendLock) {
+ stream.onNext(batch);
+ }
+ eventCounter.scope("batched").incrBy(items.size());
+ eventCounter.scope("batches").incrBy(1);
+ LOG.debug("Sent batch {} with {} discovered URL(s).", batchID,
items.size());
+ } catch (InterruptedException e) {
Review Comment:
This strands a batch.
`pendingBatches.put(batchID, items)` on line 829 happens before the
`awaitTransport` wait. If that wait is interrupted, which
`disableBatchingAndResend` causes through `batchFlusher.shutdownNow()`, this
catch only re-sets the interrupt flag. The batch is never sent, stays in
`pendingBatches` forever, and its tuples sit in `waitAck` until cache expiry
(default one hour) fails them. Storm will have timed them out long before.
```suggestion
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
synchronized (batchLock) {
pendingBatches.remove(batchID);
}
for (URLItem failed : items) {
failTupleLocally(failed.getID());
}
```
##########
external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/StatusUpdaterBolt.java:
##########
@@ -294,43 +615,55 @@ public void store(
// First get processing permit. Otherwise, starvation possible.
var hasPermit = false;
var timeSpent = 0L;
+ boolean throttled = false;
while (!hasPermit) {
- try {
- hasPermit = inFlightSemaphore.tryAcquire(throttleTimeMS,
TimeUnit.MILLISECONDS);
- if (!hasPermit) {
- LOG.trace(
- "{} messages in flight, time spent throttling {}",
- inFlightSemaphore.getQueueLength(),
- timeSpent);
-
eventCounter.scope("timeSpentThrottling").incrBy(throttleTimeMS);
- timeSpent += throttleTimeMS;
- if (timeSpent >= 30000L) {
+ hasPermit = inFlightSemaphore.tryAcquire();
+ if (!hasPermit) {
+ throttled = true;
+ LOG.trace(
+ "{} messages in flight, time spent throttling {}",
+ inFlightSemaphore.getQueueLength(),
+ timeSpent);
+ // wait for room on the monitor: woken as soon as an ack
releases permits or the
+ // transport becomes ready again. The timeout is a backstop,
not a poll interval.
+ synchronized (flow) {
+ try {
+ flow.wait(throttleTimeMS);
+ } catch (InterruptedException e) {
LOG.warn(
- "Waiting more than {} ms for processing. There
are {} permits available for {} waiting threads.",
- timeSpent,
- inFlightSemaphore.availablePermits(),
+ "InterruptedException - (approx.) {} messages
in flight.",
inFlightSemaphore.getQueueLength());
- }
- // To prevent a deadlock, it is necessary to periodically
clean up the waitAck
- // cache. Otherwise, in case of a frontier-side or
connection-wise error, all
- // incoming URLs will after some time be all caught up in
this loop without
- // touching the cache, possibly leading to no eviction and
thus leading to no
- // release of inFlightSemaphore permits.
- waitAckLock.lock();
- try {
- waitAck.cleanUp();
- } finally {
- waitAckLock.unlock();
+ Thread.currentThread().interrupt();
}
}
- } catch (InterruptedException e) {
- LOG.warn(
- "InterruptedException - (approx.) {} messages in
flight.",
- inFlightSemaphore.getQueueLength());
- Thread.currentThread().interrupt();
+
eventCounter.scope("timeSpentThrottling").incrBy(throttleTimeMS);
Review Comment:
`timeSpentThrottling` is now incremented by the full `throttleTimeMS` even
when `flow.wait` returned immediately because an ack woke it, which is the
whole point of this change.
Measure the elapsed time around the `wait` instead, otherwise the metric
operators use to tune `urlfrontier.updater.max.messages` stops meaning anything.
##########
external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/StatusUpdaterBolt.java:
##########
@@ -253,10 +403,168 @@ public void onNext(final
crawlercommons.urlfrontier.Urlfrontier.AckMessage confi
return;
}
+ completeTuples(url, values, confirmation.getStatus());
+ }
+
+ /**
+ * Acknowledges a whole batch: one status per URL, in the order the batch
was sent.
+ *
+ * @param confirmation the BatchAck received from the PutDiscovered
endpoint
+ */
+ private void onNext(final BatchAck confirmation) {
+ if (closed) {
+ return;
+ }
+
+ final List<URLItem> items;
+ synchronized (batchLock) {
+ items = pendingBatches.remove(confirmation.getID());
+ }
+
+ if (items == null) {
+ LOG.debug("Could not find batch with ID `{}`.",
confirmation.getID());
+ return;
+ }
+
+ final List<AckMessage.Status> statuses =
confirmation.getStatusesList();
+ if (statuses.size() != items.size()) {
+ LOG.warn(
+ "BatchAck {} carries {} status(es) for {} URL(s).",
+ confirmation.getID(),
+ statuses.size(),
+ items.size());
+ }
+
+ // URLs without a status, e.g. on a protocol breach, are left to the
waitAck eviction
+ int numStatuses = Math.min(statuses.size(), items.size());
+ for (int i = 0; i < numStatuses; i++) {
+ final String url = items.get(i).getID();
+ final List<Tuple> values = detachWaitAck(url);
+ if (values == null) {
+ LOG.debug("Could not find unacked tuple for id `{}`.", url);
+ continue;
+ }
+ completeTuples(url, values, statuses.get(i));
+ }
+ }
+
+ private void onBatchError(final Throwable t) {
+ if (closed) {
+ return;
+ }
+
+ // a frontier older than 2.6 does not know the PutDiscovered endpoint;
instead of
+ // dropping the discovered URLs, send them individually on the
streaming endpoint
+ // and keep doing so for the rest of the bolt's life
+ if (t instanceof StatusRuntimeException
+ && ((StatusRuntimeException) t).getStatus().getCode()
+ == io.grpc.Status.Code.UNIMPLEMENTED) {
+ LOG.warn(
+ "The frontier does not implement PutDiscovered
(URLFrontier < 2.6) - sending discovered URLs individually on the streaming
endpoint.");
+ disableBatchingAndResend();
+ return;
+ }
+
+ LOG.error("Error received on the batch stream: {}", t.getMessage());
+ LOG.debug("Error received on the batch stream", t);
+
+ // the stream is dead: forget the batches it carried, their tuples are
failed by the
+ // waitAck cache eviction and replayed by Storm. A new stream is
opened on the next flush.
+ synchronized (batchLock) {
+ pendingBatches.clear();
+ batchRequestObserver = null;
+ batchTransport = null;
+ }
+ synchronized (flow) {
+ flow.notifyAll();
+ }
+ }
+
+ /** Stops batching and pushes everything buffered or in flight through the
streaming endpoint. */
+ private void disableBatchingAndResend() {
+ final List<URLItem> toResend = new ArrayList<>();
+ synchronized (batchLock) {
+ batching = false;
+ for (List<URLItem> items : pendingBatches.values()) {
+ toResend.addAll(items);
+ }
+ pendingBatches.clear();
Review Comment:
Same expiry-window problem as the interrupt path, and inconsistent with the
`RuntimeException` path in `flushBatch`, which calls `failTupleLocally`
straight away.
Relying on `waitAck` eviction means the tuples are only failed after
`urlfrontier.cache.expireafter.sec`, long after Storm has timed them out.
```suggestion
final List<URLItem> orphaned = new ArrayList<>();
synchronized (batchLock) {
for (List<URLItem> items : pendingBatches.values()) {
orphaned.addAll(items);
}
pendingBatches.clear();
batchRequestObserver = null;
batchTransport = null;
}
for (URLItem item : orphaned) {
failTupleLocally(item.getID());
}
```
--
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]
