Skip to content

fix: contain throttled request setup failures #1007

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

Open
dkropachev wants to merge 1 commit into
base: scylla-4.x
Choose a base branch
from
Open

fix: contain throttled request setup failures #1007

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
80 changes: 56 additions & 24 deletions ...va/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,8 @@ public abstract class ContinuousRequestHandlerBase<StatementT extends Request, R
/** The callback selected to stream results back to the client. */
private final CompletableFuture<NodeResponseCallback> chosenCallback = new CompletableFuture<>();

private final AtomicBoolean terminal = new AtomicBoolean();

/**
* How many speculative executions are currently running (including the initial execution). We
* track this in order to know when to fail the request if all executions have reached the end of
Expand Down Expand Up @@ -254,22 +256,32 @@ protected abstract ResultSetT createResultSet(

@Override
public void onThrottleReady(boolean wasDelayed) {
DriverExecutionProfile executionProfile =
Conversions.resolveExecutionProfile(initialStatement, context);
if (wasDelayed
// avoid call to nanoTime() if metric is disabled:
&& sessionMetricUpdater.isEnabled(
DefaultSessionMetric.THROTTLING_DELAY, executionProfile.getName())) {
session
.getMetricUpdater()
.updateTimer(
DefaultSessionMetric.THROTTLING_DELAY,
executionProfile.getName(),
System.nanoTime() - startTimeNanos,
TimeUnit.NANOSECONDS);
try {
DriverExecutionProfile executionProfile =
Conversions.resolveExecutionProfile(initialStatement, context);
if (wasDelayed
// avoid call to nanoTime() if metric is disabled:
&& sessionMetricUpdater.isEnabled(
DefaultSessionMetric.THROTTLING_DELAY, executionProfile.getName())) {
session
.getMetricUpdater()
.updateTimer(
DefaultSessionMetric.THROTTLING_DELAY,
executionProfile.getName(),
System.nanoTime() - startTimeNanos,
TimeUnit.NANOSECONDS);
}
activeExecutionsCount.incrementAndGet();
sendRequest(initialStatement, null, 0, 0, specExecEnabled);
} catch (Throwable t) {
// The concurrency throttler contains ready-callback exceptions. Complete and release here so
// failures before sendRequest's guarded setup path cannot strand an admitted request.
if (abortGlobalRequestOrChosenCallback(t) && !(t instanceof CancellationException)) {
// Cancellation is propagated through the fetched result, whose cancellation listener owns
// the single signalCancel call.
throttler.signalError(this, t);
}
}
activeExecutionsCount.incrementAndGet();
sendRequest(initialStatement, null, 0, 0, specExecEnabled);
}

@Override
Expand All @@ -283,6 +295,8 @@ public void onThrottleFailure(@NonNull RequestThrottlingException error) {
}

private boolean abortGlobalRequestOrChosenCallback(@NonNull Throwable error) {
terminal.set(true);
cancelGlobalTimeout();
boolean completedChosenCallback = chosenCallback.completeExceptionally(error);
if (!completedChosenCallback) {
chosenCallback.thenAccept(callback -> callback.abort(error, false));
Expand All @@ -291,7 +305,16 @@ private boolean abortGlobalRequestOrChosenCallback(@NonNull Throwable error) {
}

public CompletionStage<ResultSetT> handle() {
globalTimeout = scheduleGlobalTimeout();
// Immediate admission happens in the continuous graph handler's constructor. If setup failed
// there, chosenCallback is already terminal and there is no live request to time out.
if (!terminal.get()) {
globalTimeout = scheduleGlobalTimeout();
Comment thread
dkropachev marked this conversation as resolved.
// Admission can race with handle() after the check above but before globalTimeout is
// assigned. Ensure a synchronous terminal setup failure cannot leave that timeout behind.
if (terminal.get()) {
cancelGlobalTimeout();
}
}
return fetchNextPage();
}

Expand Down Expand Up @@ -370,7 +393,7 @@ private void sendRequest(
}
} else if (!chosenCallback.isDone()) {
boolean writeSubmitted = false;
Throwable terminalPreWriteFailure = null;
Throwable terminalSetupFailure = null;
NodeResponseCallback nodeResponseCallback = null;
try {
nodeResponseCallback =
Expand All @@ -392,26 +415,35 @@ private void sendRequest(
writeSubmitted = true;
writeFuture.addListener(nodeResponseCallback);
} catch (Throwable t) {
if (!writeSubmitted && activeExecutionsCount.decrementAndGet() == 0) {
if (abortGlobalRequestOrChosenCallback(t)) {
terminalPreWriteFailure = t;
recordError(node, t);
if (activeExecutionsCount.decrementAndGet() == 0) {
if (abortGlobalRequestOrChosenCallback(t) && !(t instanceof CancellationException)) {
Comment thread
dkropachev marked this conversation as resolved.
terminalSetupFailure = t;
}
} else {
Loggers.warnWithException(
LOG, "[{}] Request setup failed, another execution is still active", logPrefix, t);
}
throw t;
} finally {
if (!writeSubmitted) {
if (nodeResponseCallback != null) {
inFlightCallbacks.remove(nodeResponseCallback);
}
try {
channel.cancelPreAcquireId();
} finally {
if (terminalPreWriteFailure != null) {
throttler.signalError(this, terminalPreWriteFailure);
} catch (Throwable cleanupFailure) {
if (terminalSetupFailure != null && terminalSetupFailure != cleanupFailure) {
terminalSetupFailure.addSuppressed(cleanupFailure);
} else {
Loggers.warnWithException(
LOG, "[{}] Failed to cancel stream ID reservation", logPrefix, cleanupFailure);
}
}
}
}
if (terminalSetupFailure != null) {
throttler.signalError(this, terminalSetupFailure);
}
} else {
channel.cancelPreAcquireId();
}
Expand Down
59 changes: 38 additions & 21 deletions core/src/main/java/com/datastax/dse/driver/internal/core/graph/GraphRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import static com.datastax.oss.driver.api.core.DriverTimeoutException.UNAVAILABLE;

import com.datastax.dse.driver.api.core.config.DseDriverOption;
import com.datastax.dse.driver.api.core.graph.AsyncGraphResultSet;
import com.datastax.dse.driver.api.core.graph.GraphNode;
import com.datastax.dse.driver.api.core.graph.GraphStatement;
Expand Down Expand Up @@ -131,8 +132,10 @@ public class GraphRequestHandler implements Throttled {
private final RequestThrottler throttler;
private final RequestTracker requestTracker;
private final SessionMetricUpdater sessionMetricUpdater;
private final DriverExecutionProfile initialExecutionProfile;
private final GraphBinaryModule graphBinaryModule;
private final GraphSupportChecker graphSupportChecker;
private volatile boolean admitted;

// The errors on the nodes that were already tried (lazily initialized on the first error).
// We don't use a map because nodes can appear multiple times.
Expand Down Expand Up @@ -176,7 +179,12 @@ public class GraphRequestHandler implements Throttled {
this.requestTracker = context.getRequestTracker();
this.sessionMetricUpdater = session.getMetricUpdater();

Duration timeout = GraphConversions.resolveGraphRequestTimeout(statement, context);
this.initialExecutionProfile = Conversions.resolveExecutionProfile(statement, context);
Duration statementTimeout = statement.getTimeout();
Duration timeout =
statementTimeout != null
? statementTimeout
: initialExecutionProfile.getDuration(DseDriverOption.GRAPH_TIMEOUT);
this.scheduledTimeout = scheduleTimeout(timeout);

this.throttler = context.getRequestThrottler();
Expand All @@ -185,25 +193,31 @@ public class GraphRequestHandler implements Throttled {

@Override
public void onThrottleReady(boolean wasDelayed) {
DriverExecutionProfile executionProfile =
Conversions.resolveExecutionProfile(initialStatement, context);
if (wasDelayed
// avoid call to nanoTime() if metric is disabled:
&& sessionMetricUpdater.isEnabled(
DefaultSessionMetric.THROTTLING_DELAY, executionProfile.getName())) {
sessionMetricUpdater.updateTimer(
DefaultSessionMetric.THROTTLING_DELAY,
executionProfile.getName(),
System.nanoTime() - startTimeNanos,
TimeUnit.NANOSECONDS);
admitted = true;
try {
if (wasDelayed
// avoid call to nanoTime() if metric is disabled:
&& sessionMetricUpdater.isEnabled(
DefaultSessionMetric.THROTTLING_DELAY, initialExecutionProfile.getName())) {
sessionMetricUpdater.updateTimer(
DefaultSessionMetric.THROTTLING_DELAY,
initialExecutionProfile.getName(),
System.nanoTime() - startTimeNanos,
TimeUnit.NANOSECONDS);
}
Queue<Node> queryPlan =
initialStatement.getNode() != null
? new SimpleQueryPlan(initialStatement.getNode())
: context
.getLoadBalancingPolicyWrapper()
.newQueryPlan(initialStatement, initialExecutionProfile.getName(), session);
sendRequest(initialStatement, null, queryPlan, 0, 0, true);
} catch (Throwable t) {
// Concurrency-limiting throttlers contain exceptions from ready callbacks so that a failed
// request can't interrupt queue draining. Complete through the normal terminal path to
// release this request's permit before returning to the throttler.
setFinalError(initialStatement, t, null, NO_SUCCESSFUL_EXECUTION);
}
Queue<Node> queryPlan =
initialStatement.getNode() != null
? new SimpleQueryPlan(initialStatement.getNode())
: context
.getLoadBalancingPolicyWrapper()
.newQueryPlan(initialStatement, executionProfile.getName(), session);
sendRequest(initialStatement, null, queryPlan, 0, 0, true);
}

public CompletionStage<AsyncGraphResultSet> handle() {
Expand Down Expand Up @@ -483,7 +497,9 @@ public void onThrottleFailure(@NonNull RequestThrottlingException error) {
private void setFinalError(
GraphStatement<?> statement, Throwable error, Node node, int execution) {
DriverExecutionProfile executionProfile =
Conversions.resolveExecutionProfile(statement, context);
statement == initialStatement
? initialExecutionProfile
: Conversions.resolveExecutionProfile(statement, context);
if (error instanceof DriverException) {
((DriverException) error)
.setExecutionInfo(
Expand All @@ -510,7 +526,8 @@ private void setFinalError(
throttler.signalTimeout(this);
sessionMetricUpdater.incrementCounter(
DseSessionMetric.GRAPH_CLIENT_TIMEOUTS, executionProfile.getName());
} else if (!(error instanceof RequestThrottlingException)) {
} else if (!(error instanceof CancellationException)
&& (admitted || !(error instanceof RequestThrottlingException))) {
throttler.signalError(this, error);
}
}
Expand Down
41 changes: 35 additions & 6 deletions ...src/main/java/com/datastax/oss/driver/internal/core/adminrequest/AdminRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
import com.datastax.oss.driver.internal.core.channel.DriverChannel;
import com.datastax.oss.driver.internal.core.channel.ResponseCallback;
import com.datastax.oss.driver.internal.core.util.Loggers;
import com.datastax.oss.driver.internal.core.util.concurrent.UncaughtExceptions;
import com.datastax.oss.driver.shaded.guava.common.collect.Maps;
import com.datastax.oss.protocol.internal.Frame;
Expand Down Expand Up @@ -101,8 +102,8 @@ public static AdminRequestHandler<AdminResult> query(
private final Class<? extends Result> expectedResponseType;
protected final CompletableFuture<ResultT> result = new CompletableFuture<>();

// This is only ever accessed on the channel's event loop, so it doesn't need to be volatile
private ScheduledFuture<?> timeoutFuture;
private volatile ScheduledFuture<?> timeoutFuture;
private volatile boolean writeSubmitted;

protected AdminRequestHandler(
DriverChannel channel,
Expand Down Expand Up @@ -134,7 +135,7 @@ public CompletionStage<ResultT> start() {
boolean writeSubmitted = false;
try {
Future<Void> writeFuture = channel.write(message, false, customPayload, this);
writeSubmitted = true;
this.writeSubmitted = writeSubmitted = true;
writeFuture.addListener(this::onWriteComplete);
} finally {
if (!writeSubmitted) {
Expand All @@ -161,13 +162,22 @@ protected final void cancelCallerOwnedPreAcquireId() {

private void onWriteComplete(Future<? super Void> future) {
if (future.isSuccess()) {
if (result.isDone()) {
cancelSubmittedRequest();
return;
}
LOG.debug("[{}] Successfully wrote {}, waiting for response", logPrefix, this);
if (timeout.toNanos() > 0) {
timeoutFuture =
ScheduledFuture<?> timeoutFuture =
channel
.eventLoop()
.schedule(this::fireTimeout, timeout.toNanos(), TimeUnit.NANOSECONDS);
this.timeoutFuture = timeoutFuture;
timeoutFuture.addListener(UncaughtExceptions::log);
// Terminal completion can race with timeout installation after the check above.
if (result.isDone()) {
cancelSubmittedRequest();
}
}
} else {
setFinalError(future.cause());
Expand All @@ -177,8 +187,27 @@ private void onWriteComplete(Future<? super Void> future) {
private void fireTimeout() {
setFinalError(
new DriverTimeoutException(String.format("%s timed out after %s", debugString, timeout)));
if (!channel.closeFuture().isDone()) {
channel.cancel(this);
cancelSubmittedRequest();
}

/** Cancels this callback if its write was submitted, along with any installed timeout. */
protected final void cancelSubmittedRequest() {
if (!writeSubmitted) {
return;
}
try {
if (timeoutFuture != null) {
timeoutFuture.cancel(true);
}
} catch (Throwable t) {
Loggers.warnWithException(LOG, "[{}] Error cancelling timeout for {}", logPrefix, this, t);
}
try {
if (!channel.closeFuture().isDone()) {
channel.cancel(this);
}
} catch (Throwable t) {
Loggers.warnWithException(LOG, "[{}] Error cancelling {}", logPrefix, this, t);
}
}

Expand Down
18 changes: 11 additions & 7 deletions ...java/com/datastax/oss/driver/internal/core/adminrequest/ThrottledAdminRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,7 @@ public static ThrottledAdminRequestHandler<ByteBuffer> prepare(
private final long startTimeNanos;
private final RequestThrottler throttler;
private final SessionMetricUpdater metricUpdater;
private volatile boolean admitted;
private final AtomicBoolean holdsExternalReservation;

protected ThrottledAdminRequestHandler(
Expand Down Expand Up @@ -140,16 +141,21 @@ public CompletionStage<ResultT> start() {
throttler.register(this);
} catch (Throwable t) {
cancelExternalReservation();
// Registration can fail before the throttler admits this request, so complete the result
// without calling this class's override, which would signal a permit that was never acquired.
super.setFinalError(t);
if (admitted) {
setFinalError(t);
cancelSubmittedRequest();
} else {
// Registration failed before admission, so there is no throttler permit to release.
super.setFinalError(t);
}
throw t;
}
return result;
}

@Override
public void onThrottleReady(boolean wasDelayed) {
admitted = true;
try {
if (wasDelayed) {
metricUpdater.updateTimer(
Expand All @@ -163,7 +169,7 @@ public void onThrottleReady(boolean wasDelayed) {
} catch (Throwable t) {
cancelExternalReservation();
setFinalError(t);
throw t;
cancelSubmittedRequest();
}
}

Expand All @@ -175,8 +181,6 @@ public void onThrottleFailure(@NonNull RequestThrottlingException error) {
}

private void cancelExternalReservation() {
// register() can invoke onThrottleReady() synchronously. If that callback throws, both
// onThrottleReady() and start() catch the same failure, so cancellation must be idempotent.
if (holdsExternalReservation.compareAndSet(true, false)) {
cancelCallerOwnedPreAcquireId();
}
Expand All @@ -197,7 +201,7 @@ protected boolean setFinalError(Throwable error) {
if (wasSet) {
if (error instanceof DriverTimeoutException) {
throttler.signalTimeout(this);
} else if (!(error instanceof RequestThrottlingException)) {
} else if (admitted) {
throttler.signalError(this, error);
}
}
Expand Down
Loading
Loading