Copilot commented on code in PR #4895:
URL: https://github.com/apache/arrow-adbc/pull/4895#discussion_r4278889780
##########
csharp/src/Apache.Arrow.Adbc/AdbcStatement.cs:
##########
@@ -83,9 +89,49 @@ public virtual void BindStream(IArrowArrayStream stream)
/// A <see cref="ValueTuple"/> where the first item is the number
/// of records and the second is the <see cref="IArrowArrayStream"/>.
/// </returns>
- public virtual async ValueTask<QueryResult> ExecuteQueryAsync()
+ /// <remarks>
+ /// <para>New code should call <see
cref="ExecuteQueryAsync(CancellationToken)"/>.</para>
+ /// <para>The default implementation runs <see cref="ExecuteQuery"/>
on the thread pool.
+ /// It doesn't call <see
cref="ExecuteQueryAsync(CancellationToken)"/>, because that
+ /// overload's default implementation calls this one. A driver that
overrides
+ /// <see cref="ExecuteQueryAsync(CancellationToken)"/> should also
override this
+ /// overload to call it with <see cref="CancellationToken.None"/>, so
that callers of
+ /// either overload get the driver's asynchronous
implementation.</para>
+ /// </remarks>
+ public virtual ValueTask<QueryResult> ExecuteQueryAsync()
+ {
+ return new ValueTask<QueryResult>(AdbcAsync.Run(ExecuteQuery,
Cancel, CancellationToken.None));
+ }
+
+ /// <summary>
+ /// Executes the statement and returns a tuple containing the number
+ /// of records and the <see cref="IArrowArrayStream"/>.
+ /// </summary>
+ /// <param name="cancellationToken">
+ /// A token used to cancel the operation. Cancellation calls <see
cref="Cancel"/>.
+ /// </param>
+ /// <returns>
+ /// A <see cref="ValueTuple"/> where the first item is the number
+ /// of records and the second is the <see cref="IArrowArrayStream"/>.
+ /// </returns>
+ /// <remarks>
+ /// The default implementation calls <see
cref="ExecuteQueryAsync()"/>, so that drivers
+ /// written before this overload existed keep their asynchronous
behavior, and calls
+ /// <see cref="Cancel"/> if the token is cancelled.
+ /// </remarks>
+ public virtual ValueTask<QueryResult>
ExecuteQueryAsync(CancellationToken cancellationToken)
{
- return await Task.Run(() => ExecuteQuery());
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return new
ValueTask<QueryResult>(Task.FromCanceled<QueryResult>(cancellationToken));
+ }
+
+ ValueTask<QueryResult> result = ExecuteQueryAsync();
Review Comment:
The legacy override is invoked before `WithCancellation` registers the
token. An override such as `new ValueTask<QueryResult>(ExecuteQuery())` can
therefore block here, and cancellation from another thread cannot reach
`Cancel` while it runs. The update overload at line 202 has the same problem
with overrides such as `Task.FromResult(ExecuteUpdate())`. Register
cancellation before invoking either legacy override and keep the registration
until its result completes, for example by passing an operation factory to the
helper. Add regression tests for both synchronously blocking overrides.
##########
csharp/src/Apache.Arrow.Adbc/AdbcStatement.cs:
##########
@@ -109,10 +166,45 @@ public virtual Schema ExecuteSchema()
/// Executes an update command and returns the number of
/// records effected.
/// </summary>
- /// <exception cref="NotImplementedException"></exception>
- public virtual async Task<UpdateResult> ExecuteUpdateAsync()
+ /// <remarks>
+ /// <para>New code should call <see
cref="ExecuteUpdateAsync(CancellationToken)"/>.</para>
+ /// <para>The default implementation runs <see cref="ExecuteUpdate"/>
on the thread pool.
+ /// It doesn't call <see
cref="ExecuteUpdateAsync(CancellationToken)"/>, because that
+ /// overload's default implementation calls this one. A driver that
overrides
+ /// <see cref="ExecuteUpdateAsync(CancellationToken)"/> should also
override this
+ /// overload to call it with <see cref="CancellationToken.None"/>, so
that callers of
+ /// either overload get the driver's asynchronous
implementation.</para>
+ /// </remarks>
+ public virtual Task<UpdateResult> ExecuteUpdateAsync()
{
- return await Task.Run(() => ExecuteUpdate());
+ return AdbcAsync.Run(ExecuteUpdate, Cancel,
CancellationToken.None);
+ }
+
+ /// <summary>
+ /// Executes an update command and returns the number of
+ /// records effected.
+ /// </summary>
+ /// <param name="cancellationToken">
+ /// A token used to cancel the operation. Cancellation calls <see
cref="Cancel"/>.
+ /// </param>
+ /// <remarks>
+ /// The default implementation calls <see
cref="ExecuteUpdateAsync()"/>, so that drivers
+ /// written before this overload existed keep their asynchronous
behavior, and calls
+ /// <see cref="Cancel"/> if the token is cancelled.
+ /// </remarks>
+ public virtual Task<UpdateResult> ExecuteUpdateAsync(CancellationToken
cancellationToken)
+ {
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return Task.FromCanceled<UpdateResult>(cancellationToken);
+ }
+
+ Task<UpdateResult> result = ExecuteUpdateAsync();
Review Comment:
If the token is cancelled while the default update task is queued,
`WithCancellation` calls `Cancel` before `ExecuteUpdate` starts. A driver that
only cancels an active operation cannot stop that queued work. The
parameterless overload schedules it with `CancellationToken.None`, so the
update still starts later, potentially mutating the database after cancellation
with no further cancellation attempt. The query overload has the same race.
Carry the supplied token into the synchronous fallback while preserving legacy
async overrides, and test cancellation before the queued operation starts.
##########
csharp/src/Apache.Arrow.Adbc/AdbcAsync.cs:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.
+ */
+
+using System;
+using System.Runtime.ExceptionServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Apache.Arrow.Adbc
+{
+ /// <summary>
+ /// Helpers for the default implementations of the asynchronous methods on
+ /// the ADBC base classes. Those defaults run the synchronous method on the
+ /// thread pool and translate a <see cref="CancellationToken"/> into a call
+ /// to the object's <c>Cancel</c> method.
+ /// </summary>
+ internal static class AdbcAsync
+ {
+ private static readonly Action<object?> s_cancel = TryCancel;
+
+ public static Task<T> Run<T>(Func<T> operation, Action? cancel,
CancellationToken cancellationToken)
+ {
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return Task.FromCanceled<T>(cancellationToken);
+ }
+
+ return Task.Run(() =>
+ {
+ CancelState state = new CancelState(cancel);
+ Exception failure;
+ using (Register(state, cancellationToken))
+ {
+ try
+ {
+ return operation();
+ }
+ catch (Exception e) when (!(e is
OperationCanceledException))
+ {
+ failure = e;
+ }
+ }
+
+ // Disposing the registration waits for an in-progress Cancel
to finish
+ throw Translate(failure, state, cancellationToken);
+ }, cancellationToken);
+ }
+
+ public static Task Run(Action operation, Action? cancel,
CancellationToken cancellationToken)
+ {
+ return Run<bool>(() => { operation(); return true; }, cancel,
cancellationToken);
+ }
+
+ /// <summary>
+ /// Attaches cancellation to a task produced by a driver override that
+ /// doesn't itself accept a <see cref="CancellationToken"/>.
+ /// </summary>
+ public static async Task<T> WithCancellation<T>(Task<T> task, Action?
cancel, CancellationToken cancellationToken)
+ {
+ CancelState state = new CancelState(cancel);
+ Exception failure;
+ using (Register(state, cancellationToken))
+ {
+ try
+ {
+ return await task.ConfigureAwait(false);
+ }
+ catch (Exception e) when (!(e is OperationCanceledException))
+ {
+ failure = e;
+ }
+ }
+
+ throw Translate(failure, state, cancellationToken);
+ }
+
+ private static CancellationTokenRegistration Register(CancelState
state, CancellationToken cancellationToken)
+ {
+ if (state.Cancel == null || !cancellationToken.CanBeCanceled)
+ {
+ return default;
+ }
+ return cancellationToken.Register(s_cancel, state);
+ }
+
+ /// <summary>
+ /// Reports a failure as cancellation only if cancellation plausibly
caused it: the
+ /// token was cancelled and either the driver accepted the call to
Cancel or the
+ /// failure itself says the operation was cancelled. Otherwise, a
driver that can't
+ /// cancel would have its own errors hidden behind <see
cref="OperationCanceledException"/>.
+ /// </summary>
+ private static Exception Translate(Exception failure, CancelState
state, CancellationToken cancellationToken)
+ {
+ bool cancelled = cancellationToken.IsCancellationRequested &&
+ (state.Delivered || (failure is AdbcException adbc &&
adbc.Status == AdbcStatusCode.Cancelled));
+ if (cancelled)
+ {
+ return new OperationCanceledException(failure.Message,
failure, cancellationToken);
+ }
+
+ ExceptionDispatchInfo.Capture(failure).Throw();
+ return failure; // not reached
+ }
+
+ private static void TryCancel(object? state)
+ {
+ CancelState cancelState = (CancelState)state!;
+ try
+ {
+ cancelState.Cancel!();
+ cancelState.Delivered = true;
Review Comment:
If `Cancel` faults a legacy override's plain `TaskCompletionSource`,
`WithCancellation` can resume inline inside this call. Disposing the
registration from its own callback does not wait for that callback to finish,
so `Translate` sees `Delivered == false`. A non-cancellation failure such as
`TimeoutException` is then propagated instead of `OperationCanceledException`,
even when `Cancel` subsequently returns successfully. Coordinate failure
translation with completion of the cancellation callback without blocking that
callback. Add a regression test with inline task completion and a
non-cancellation error; setting `Delivered` before calling `Cancel` would
incorrectly treat throwing cancellations as successful.
--
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]
