Skip to content

#2134 fetcher.thread.timeout: cancel the okhttp call, bounded helper pool for other protocols #2135

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
GGraziadei wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

#2134 fetcher.thread.timeout: cancel the okhttp call, bounded helper pool for other protocols #2135

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions core/src/main/java/org/apache/stormcrawler/Constants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,4 +60,13 @@ public class Constants {
public static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

private Constants() {}

/** Hard deadline in seconds for a single fetch, -1 to disable. */
public static final String FETCH_TIMEOUT_PARAM_KEY = "fetcher.thread.timeout";

/**
* Maximum number of helper threads on which fetches with a timeout are run for protocols that
* can not enforce the timeout themselves. Defaults to twice fetcher.threads.number.
*/
public static final String FETCH_TIMEOUT_HELPERS_PARAM_KEY = "fetcher.thread.timeout.helpers";
}
157 changes: 157 additions & 0 deletions core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,157 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to you under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.stormcrawler.bolt;

import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.storm.task.TopologyContext;
import org.apache.stormcrawler.Constants;
import org.apache.stormcrawler.metrics.CrawlerMetrics;
import org.apache.stormcrawler.protocol.Protocol;
import org.apache.stormcrawler.util.ConfUtils;

/**
* Enforces {@code fetcher.thread.timeout} on protocol calls (robots.txt lookup and fetch) for the
* fetcher bolts.
*
* <p>When the timeout is off, or the protocol enforces it itself (see {@link
* Protocol#supportsFetchTimeout()}, true for the default okhttp protocol), the call runs on the
* calling thread. Otherwise it runs on a helper thread and is abandoned there when the deadline
* passes: the helper stays busy until the protocol gives up on its own, so the pool is bounded
* ({@code fetcher.thread.timeout.helpers}) and a call that finds every helper busy fails at once
* with {@link SaturatedException}. Helper threads are created on demand and released after a minute
* of inactivity: with the default protocol none is ever created.
*/
final class FetchTimeoutHelpers {

/** Thrown when a call run on a helper thread did not complete within the timeout. */
static final class TimeoutException extends Exception {
TimeoutException(String url, long timeoutSecs) {
super("Fetch timed out after " + timeoutSecs + "s fetching " + url);
}
}

/** Thrown when no helper thread is available to run a call with a timeout. */
static final class SaturatedException extends Exception {
SaturatedException(String url) {
super("No fetch helper available for " + url);
}
}

private final long timeoutSecs;
private final ThreadPoolExecutor helpers;

/**
* @param conf the bolt configuration
* @param defaultMaxHelpers pool bound used unless {@code fetcher.thread.timeout.helpers} is set
* @param threadNamePrefix prefix of the helper thread names
*/
FetchTimeoutHelpers(Map<String, Object> conf, int defaultMaxHelpers, String threadNamePrefix) {
this.timeoutSecs = ConfUtils.getLong(conf, Constants.FETCH_TIMEOUT_PARAM_KEY, -1);
final int maxHelpers =
ConfUtils.getInt(
conf, Constants.FETCH_TIMEOUT_HELPERS_PARAM_KEY, defaultMaxHelpers);
final AtomicInteger helperNum = new AtomicInteger();
this.helpers =
new ThreadPoolExecutor(
0,
Math.max(1, maxHelpers),
60L,
TimeUnit.SECONDS,
new SynchronousQueue<>(),
r -> {
Thread t =
new Thread(r, threadNamePrefix + helperNum.incrementAndGet());
t.setDaemon(true);
return t;
});
}

/** Registers the {@code fetchhelpers} gauge: number of helper threads busy with a call. */
void registerMetrics(TopologyContext context, Map<String, Object> conf, int bucketSecs) {
CrawlerMetrics.registerGauge(
context, conf, "fetchhelpers", helpers::getActiveCount, bucketSecs);
}

/** Whether a timeout is configured at all. */
boolean enabled() {
return timeoutSecs > 0;
}

/** Timeout in seconds, -1 when disabled. */
long timeoutSecs() {
return timeoutSecs;
}

/** Pool bound. */
int maxHelpers() {
return helpers.getMaximumPoolSize();
}

/** Largest number of helper threads ever alive. */
int largestPoolSize() {
return helpers.getLargestPoolSize();
}

/**
* Runs a protocol call under the timeout.
*
* @throws SaturatedException when every helper is busy
* @throws TimeoutException when the deadline passed
* @throws Exception the protocol's own exception
*/
<T> T call(Callable<T> call, Protocol protocol, String url) throws Exception {
if (timeoutSecs <= 0 || protocol.supportsFetchTimeout()) {
return call.call();
}
final Future<T> future;
try {
future = helpers.submit(call);
} catch (RejectedExecutionException e) {
throw new SaturatedException(url);
}
try {
return future.get(timeoutSecs, TimeUnit.SECONDS);
} catch (java.util.concurrent.TimeoutException e) {
future.cancel(true);
throw new TimeoutException(url, timeoutSecs);
} catch (CancellationException e) {
throw new Exception("Fetch cancelled for " + url);
} catch (ExecutionException e) {
// unwrap the real cause so the bolts' classification sees it
Throwable cause = e.getCause();
if (cause instanceof Exception) {
throw (Exception) cause;
}
throw new Exception(cause);
}
}

void shutdown() {
helpers.shutdownNow();
}
}
100 changes: 34 additions & 66 deletions core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,14 +34,7 @@
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Pattern;
Expand Down Expand Up @@ -86,14 +79,6 @@ public class FetcherBolt extends StatusEmitterBolt {
*/
public static final String QUEUED_TIMEOUT_PARAM_KEY = "fetcher.timeout.queue";

/**
* Hard timeout in seconds for a single call to {@link Protocol#getProtocolOutput}. If a fetch
* exceeds this duration the thread is interrupted, the URL is marked as FETCH_ERROR, and the
* thread moves on to the next item. A value of {@code -1} (the default) disables the bolt-level
* timeout, relying solely on the protocol-level socket timeouts.
*/
public static final String FETCH_TIMEOUT_PARAM_KEY = "fetcher.thread.timeout";

/** Key name of the custom crawl delay for a queue that may be present in the metadata. */
private static final String CRAWL_DELAY_KEY_NAME = "crawl.delay";

Expand Down Expand Up @@ -129,6 +114,14 @@ public class FetcherBolt extends StatusEmitterBolt {

private String[] beingFetched;

/** Runs protocol calls under fetcher.thread.timeout, see {@link FetchTimeoutHelpers}. */
private FetchTimeoutHelpers fetchHelpers;

/** Largest number of helper threads ever alive; for tests. */
int helperPoolSize() {
return fetchHelpers == null ? 0 : fetchHelpers.largestPoolSize();
}

@Override
public Map<String, Object> getComponentConfiguration() {
Config conf = new Config();
Expand Down Expand Up @@ -498,15 +491,6 @@ private class FetcherThread extends Thread {

private long timeoutInQueues = -1;

/** Hard timeout in seconds for a single protocol fetch. -1 means disabled. */
private long fetchTimeout = -1;

/**
* Single-thread executor used to run the protocol call so that it can be interrupted via
* {@link Future#cancel(boolean)} when the bolt-level timeout fires.
*/
private final ExecutorService fetchExecutor;

// by default remains as is-pre 1.17
private String protocolMetadataPrefix = "";

Expand All @@ -520,24 +504,11 @@ public FetcherThread(Config conf, int num) {
this.crawlDelayForce = ConfUtils.getBoolean(conf, "fetcher.server.delay.force", false);
this.threadNum = num;
timeoutInQueues = ConfUtils.getLong(conf, QUEUED_TIMEOUT_PARAM_KEY, timeoutInQueues);
fetchTimeout = ConfUtils.getLong(conf, FETCH_TIMEOUT_PARAM_KEY, fetchTimeout);
protocolMetadataPrefix =
ConfUtils.getString(
conf,
ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM,
protocolMetadataPrefix);

if (fetchTimeout > 0) {
fetchExecutor =
Executors.newSingleThreadExecutor(
r -> {
Thread t = new Thread(r, "FetcherTimeout #" + num);
t.setDaemon(true);
return t;
});
} else {
fetchExecutor = null;
}
}

@Override
Expand Down Expand Up @@ -598,7 +569,9 @@ public void run() {
"No protocol implementation found for " + fit.url);
}

BaseRobotRules rules = protocol.getRobotRules(fit.url);
BaseRobotRules rules =
fetchHelpers.call(
() -> protocol.getRobotRules(fit.url), protocol, fit.url);
boolean fromCache = false;
if (rules instanceof RobotRules
&& ((RobotRules) rules).getContentLengthFetched().length == 0) {
Expand Down Expand Up @@ -733,33 +706,11 @@ public void run() {

final Metadata fetchMetadata = metadata;
ProtocolResponse response;
if (fetchExecutor != null) {
Future<ProtocolResponse> future =
fetchExecutor.submit(
() -> protocol.getProtocolOutput(fit.url, fetchMetadata));
try {
response = future.get(fetchTimeout, TimeUnit.SECONDS);
} catch (TimeoutException e) {
future.cancel(true);
throw new Exception(
"Fetch timed out after "
+ fetchTimeout
+ "s fetching "
+ fit.url,
e);
} catch (CancellationException e) {
throw new Exception("Fetch cancelled for " + fit.url);
} catch (ExecutionException e) {
// unwrap the real cause so existing catch logic handles it
Throwable cause = e.getCause();
if (cause instanceof Exception) {
throw (Exception) cause;
}
throw new Exception(cause);
}
} else {
response = protocol.getProtocolOutput(fit.url, metadata);
}
response =
fetchHelpers.call(
() -> protocol.getProtocolOutput(fit.url, fetchMetadata),
protocol,
fit.url);

long timeFetching = System.currentTimeMillis() - start;

Expand Down Expand Up @@ -871,10 +822,19 @@ public void run() {
}

// common exceptions for which we log only a short message
if (exece.getCause() instanceof java.util.concurrent.TimeoutException
if (exece instanceof FetchTimeoutHelpers.TimeoutException
|| exece instanceof java.io.InterruptedIOException
|| message.contains(" timed out")) {
LOG.info("Socket timeout fetching {}", fit.url);
message = "Socket timeout fetching";
eventCounter.scope("fetch.timeout").incrBy(1);
} else if (exece instanceof FetchTimeoutHelpers.SaturatedException) {
eventCounter.scope("fetch.helper.rejected").incrBy(1);
LOG.warn(
"{}: all {} fetch helpers are busy",
message,
fetchHelpers.maxHelpers());
message = "No fetch helper available";
} else if (exece.getCause() instanceof java.net.UnknownHostException
|| exece instanceof java.net.UnknownHostException) {
LOG.info("Unknown host {}", fit.url);
Expand Down Expand Up @@ -984,6 +944,11 @@ public void prepare(
int threadCount = ConfUtils.getInt(conf, "fetcher.threads.number", 10);
int startDelay = ConfUtils.getInt(conf, "fetcher.threads.start.delay", 10);

// helpers for protocols which can not cancel a fetch themselves; no thread until needed
fetchHelpers =
new FetchTimeoutHelpers(conf, Math.max(1, threadCount * 2), "FetcherTimeout-");
fetchHelpers.registerMetrics(context, stormConf, metricsTimeBucketSecs);

for (int i = 0; i < threadCount; i++) {
if (startDelay > 0 && i > 0) {
// short delay to avoid that DNS or other resources are temporarily
Expand Down Expand Up @@ -1030,6 +995,9 @@ public void declareOutputFields(OutputFieldsDeclarer declarer) {
@Override
public void cleanup() {
super.cleanup();
if (fetchHelpers != null) {
fetchHelpers.shutdown();
}
protocolFactory.cleanup();
}

Expand Down
Loading