Завершение ManualResetEvent как ожидаемая задача
Я хотел бы дождаться события ручного сброса с тайм-аутом и наблюдением отмены. Я придумал что-то вроде ниже. Объект события ручного сброса предоставляется API вне моего контроля. Есть ли способ сделать это, не принимая и не блокируя поток из ThreadPool?
static Task<bool> TaskFromWaitHandle(WaitHandle mre, int timeout, CancellationToken ct)
{
return Task.Run(() =>
{
bool s = WaitHandle.WaitAny(new WaitHandle[] { mre, ct.WaitHandle }, timeout) == 0;
ct.ThrowIfCancellationRequested();
return s;
}, ct);
}
// ...
if (await TaskFromWaitHandle(manualResetEvent, 1000, cts.Token))
{
// true if event was set
}
else
{
// false if timed out, exception if cancelled
}
[Отредактировано] Видимо, имеет смысл использовать RegisterWaitForSingleObject, Я попробую.
8 ответов
RegisterWaitForSingleObject будет объединять ожидания на выделенные потоки официантов, каждый из которых может ожидать несколько дескрипторов (в частности, 63 из них, что MAXIMUM_WAIT_OBJECTS минус один для "контрольной" ручки).
Таким образом, вы должны иметь возможность использовать что-то вроде этого (предупреждение: не проверено):
public static class WaitHandleExtensions
{
public static Task AsTask(this WaitHandle handle)
{
return AsTask(handle, Timeout.InfiniteTimeSpan);
}
public static Task AsTask(this WaitHandle handle, TimeSpan timeout)
{
var tcs = new TaskCompletionSource<object>();
var registration = ThreadPool.RegisterWaitForSingleObject(handle, (state, timedOut) =>
{
var localTcs = (TaskCompletionSource<object>)state;
if (timedOut)
localTcs.TrySetCanceled();
else
localTcs.TrySetResult(null);
}, tcs, timeout, executeOnlyOnce: true);
tcs.Task.ContinueWith((_, state) => ((RegisteredWaitHandle)state).Unregister(null), registration, TaskScheduler.Default);
return tcs.Task;
}
}
Вы также можете использовать SemaphoreSlim.WaitAsync(), который похож на ManualResetEvent
Решение Стивена Клири выглядит идеально. Microsoft предоставляет аналогичный .
Поскольку я не видел примера с логикой отмены.
Вот:
public static class WaitHandleExtensions
{
public static Task WaitOneAsync(this WaitHandle waitHandle, CancellationToken cancellationToken, int timeoutMilliseconds = Timeout.Infinite)
{
if (waitHandle == null)
throw new ArgumentNullException(nameof(waitHandle));
TaskCompletionSource<bool> tcs = new TaskCompletionSource<bool>();
CancellationTokenRegistration ctr = cancellationToken.Register(() => tcs.SetCanceled());
TimeSpan timeout = timeoutMilliseconds > Timeout.Infinite ? TimeSpan.FromMilliseconds(timeoutMilliseconds) : Timeout.InfiniteTimeSpan;
RegisteredWaitHandle rwh = ThreadPool.RegisterWaitForSingleObject(waitHandle,
(_, timedOut) =>
{
if (timedOut)
{
tcs.TrySetCanceled();
}
else
{
tcs.TrySetResult(true);
}
},
null, timeout, true);
Task<bool> task = tcs.Task;
_ = task.ContinueWith(_ =>
{
rwh.Unregister(null);
return ctr.Unregister();
}, CancellationToken.None);
return task;
}
}
Вы можете дать этому шанс, https://www.badflyer.com/asyncmanualresetevent, пытаясь использовать пример на https://blogs.msdn.com/b/pfxteam/archive/2012/02/11/10266920.aspx для поддержки тайм-аутов и отмены.
using System;
using System.Threading;
using System.Threading.Tasks;
/// <summary>
/// An async manual reset event.
/// </summary>
public sealed class ManualResetEventAsync
{
// Inspiration from https://devblogs.microsoft.com/pfxteam/building-async-coordination-primitives-part-1-asyncmanualresetevent/
// and the .net implementation of SemaphoreSlim
/// <summary>
/// The timeout in milliseconds to wait indefinitly.
/// </summary>
private const int WaitIndefinitly = -1;
/// <summary>
/// True to run synchronous continuations on the thread which invoked Set. False to run them in the threadpool.
/// </summary>
private readonly bool runSynchronousContinuationsOnSetThread = true;
/// <summary>
/// The current task completion source.
/// </summary>
private volatile TaskCompletionSource<bool> completionSource = new TaskCompletionSource<bool>();
/// <summary>
/// Initializes a new instance of the <see cref="ManualResetEventAsync"/> class.
/// </summary>
/// <param name="isSet">True to set the task completion source on creation.</param>
public ManualResetEventAsync(bool isSet)
: this(isSet: isSet, runSynchronousContinuationsOnSetThread: true)
{
}
/// <summary>
/// Initializes a new instance of the <see cref="ManualResetEventAsync"/> class.
/// </summary>
/// <param name="isSet">True to set the task completion source on creation.</param>
/// <param name="runSynchronousContinuationsOnSetThread">If you have synchronous continuations, they will run on the thread which invokes Set, unless you set this to false.</param>
public ManualResetEventAsync(bool isSet, bool runSynchronousContinuationsOnSetThread)
{
this.runSynchronousContinuationsOnSetThread = runSynchronousContinuationsOnSetThread;
if (isSet)
{
this.completionSource.TrySetResult(true);
}
}
/// <summary>
/// Wait for the manual reset event.
/// </summary>
/// <returns>A task which completes when the event is set.</returns>
public Task WaitAsync()
{
return this.AwaitCompletion(ManualResetEventAsync.WaitIndefinitly, default(CancellationToken));
}
/// <summary>
/// Wait for the manual reset event.
/// </summary>
/// <param name="token">A cancellation token.</param>
/// <returns>A task which waits for the manual reset event.</returns>
public Task WaitAsync(CancellationToken token)
{
return this.AwaitCompletion(ManualResetEventAsync.WaitIndefinitly, token);
}
/// <summary>
/// Wait for the manual reset event.
/// </summary>
/// <param name="timeout">A timeout.</param>
/// <param name="token">A cancellation token.</param>
/// <returns>A task which waits for the manual reset event. Returns true if the timeout has not expired. Returns false if the timeout expired.</returns>
public Task<bool> WaitAsync(TimeSpan timeout, CancellationToken token)
{
return this.AwaitCompletion((int)timeout.TotalMilliseconds, token);
}
/// <summary>
/// Wait for the manual reset event.
/// </summary>
/// <param name="timeout">A timeout.</param>
/// <returns>A task which waits for the manual reset event. Returns true if the timeout has not expired. Returns false if the timeout expired.</returns>
public Task<bool> WaitAsync(TimeSpan timeout)
{
return this.AwaitCompletion((int)timeout.TotalMilliseconds, default(CancellationToken));
}
/// <summary>
/// Set the completion source.
/// </summary>
public void Set()
{
if (this.runSynchronousContinuationsOnSetThread)
{
this.completionSource.TrySetResult(true);
}
else
{
// Run synchronous completions in the thread pool.
Task.Run(() => this.completionSource.TrySetResult(true));
}
}
/// <summary>
/// Reset the manual reset event.
/// </summary>
public void Reset()
{
// Grab a reference to the current completion source.
var currentCompletionSource = this.completionSource;
// Check if there is nothing to be done, return.
if (!currentCompletionSource.Task.IsCompleted)
{
return;
}
// Otherwise, try to replace it with a new completion source (if it is the same as the reference we took before).
Interlocked.CompareExchange(ref this.completionSource, new TaskCompletionSource<bool>(), currentCompletionSource);
}
/// <summary>
/// Await completion based on a timeout and a cancellation token.
/// </summary>
/// <param name="timeoutMS">The timeout in milliseconds.</param>
/// <param name="token">The cancellation token.</param>
/// <returns>A task (true if wait succeeded). (False on timeout).</returns>
private async Task<bool> AwaitCompletion(int timeoutMS, CancellationToken token)
{
// Validate arguments.
if (timeoutMS < -1 || timeoutMS > int.MaxValue)
{
throw new ArgumentException("The timeout must be either -1ms (indefinitely) or a positive ms value <= int.MaxValue");
}
CancellationTokenSource timeoutToken = null;
// If the token cannot be cancelled, then we dont need to create any sort of linked token source.
if (false == token.CanBeCanceled)
{
// If the wait is indefinite, then we don't need to create a second task at all to wait on, just wait for set.
if (timeoutMS == -1)
{
return await this.completionSource.Task;
}
timeoutToken = new CancellationTokenSource();
}
else
{
// A token source which will get canceled either when we cancel it, or when the linked token source is canceled.
timeoutToken = CancellationTokenSource.CreateLinkedTokenSource(token);
}
using (timeoutToken)
{
// Create a task to account for our timeout. The continuation just eats the task cancelled exception, but makes sure to observe it.
Task delayTask = Task.Delay(timeoutMS, timeoutToken.Token).ContinueWith((result) => { var e = result.Exception; }, TaskContinuationOptions.ExecuteSynchronously);
var resultingTask = await Task.WhenAny(this.completionSource.Task, delayTask).ConfigureAwait(false);
// The actual task finished, not the timeout, so we can cancel our cancellation token and return true.
if (resultingTask != delayTask)
{
// Cancel the timeout token to cancel the delay if it is still going.
timeoutToken.Cancel();
return true;
}
// Otherwise, the delay task finished. So throw if it finished because it was canceled.
token.ThrowIfCancellationRequested();
return false;
}
}
}
Альтернативное решение: дождитесь дескрипторов задачи и события ручного сброса
У меня были утечки памяти при использовании
Task.WaitAny() с
Task (возвращается SqlConnection.OpenAsync()') и событие ручного сброса, полученное как параметр и заключенное в
Task с участием
AsTask(). Эти объекты не удалялись:
TaskCompletionSource<Object>, Task<Object>, StandardTaskContinuation, RegisteredWaitHandle, RegisteredWaithandleSafe, ContinuationResultTaskFromresultTask<Object,bool>, _ThreadPoolWaitOrTimerCallback).
Это реальный производственный код, используемый в службе Windows, функции, которая пытается открыть соединение с базой данных в цикле до тех пор, пока соединение не будет открыто, или операция не завершится ошибкой, или
ManualResetEvent _finishRequest, полученный как параметр в функции, содержащей этот код, сигнализируется кодом в любом другом потоке.
Чтобы избежать утечки, я решил сделать наоборот: дождаться ручки
_finishRequest и
Task вернулся
OpenAsync():
Task asyncOpening = sqlConnection.OpenAsync();
// Wait for the async open to finish, or until _finishRequest is signaled
var waitHandles = new WaitHandle[]
{
// index 0 in array: extract the AsyncWaitHandle from the Task
((IAsyncResult)asyncOpening).AsyncWaitHandle,
// index 1:
_finishRequest
};
// Check if finish was requested (index of signaled handle in the array = 1)
int whichFinished = WaitHandle.WaitAny(waitHandles);
finishRequested = whichFinished == 1;
// If so, break the loop to exit immediately
if (finishRequested)
break;
// If not, check if OpenAsync finished with error (it's a Task)
if (asyncOpening.IsFaulted)
{
// Extract the exception from the task, and throw it
// NOTE: adapt it to your case. In mine, I'm interested in the inner exception,
// but you can check the exception itself, for example to see if it was a timeout,
// if you specified it in the call to the async function that returns the Task
var ex = asyncOpening?.Exception?.InnerExceptions?[0];
if (ex != null) throw ex;
}
else
{
Log.Verbose("Connection to database {Database} on server {Server}", database, server);
break;
}
Если вам также нужен тайм-аут, вы можете включить его в вызов
OpenAsync, или вы выполняете асинхронную функцию, а затем проверьте, был ли результат асинхронной операции отменен из-за тайм-аута: проверьте состояние задачи по завершении, как вы можете видеть в ПРИМЕЧАНИЕ в комментарии к коду.
В 2021 году стоит обратить внимание на TaskCompletionSource .
/questions/31555624/kogda-sleduet-ispolzovat-taskcompletionsourcet/31555643#31555643
УстановитеMicrosoft.VisualStudio.Threadingпакет, то вы сможете использоватьAsyncManualResetEventкласс, который в своей документации гласит:
Разновидность ManualResetEvent, которую можно ожидать асинхронно.
ManualResetEvent SomePublicSignal = new ManualResetEvent();
...
...
// Only thing this task does is wait for the signal.
await Task.Run(() => { someSignal.WaitOne(); });
...
...
// Then can use in a Task.WaitAny(....)
Task.WaitAny(
new Task[] {
Task.Run(() => { someSignal.WaitOne(); }),
Task.Delay(200, stoppingToken) } );