Определение завершения длительных асинхронных задач

[Обновлено 18 апреля 2018 г. с примером LinqPad — см. конец]

Мое приложение получает список заданий:

var jobs = await myDB.GetWorkItems();

(Примечание: мы используем .ConfigureAwait(false) везде, я просто не показываю его в этих фрагментах псевдокода.)

Для каждого задания мы создаем долгоиграющую задачу. Однако мы не хотим ждать завершения этой длительной задачи.

jobs.ForEach(job =>
{
    var module = Factory.GetModule(job.Type);
    var task = Task.Run(() => module.ExecuteAsync(job.Data));
    this.NonAwaitedTasks.Add(task, module);
};

Задача и связанный с ней экземпляр модуля добавляются в ConcurrentDictionary, чтобы они не выходили за рамки.

В другом месте у меня есть другой метод, который иногда вызывается и содержит следующее:

foreach (var entry in this.NonAwaitedTasks.Where(e => e.Key.IsCompleted))
{
    var module = entry.Value as IDisposable;
    module?.Dispose();
    this.NonAwaitedTasks.Remove(entry.Key);
}

(Обратите внимание, что NonAwaitedTasks дополнительно блокируется с помощью SemaphoreSlim...)

Итак, идея состоит в том, что этот метод найдет все те задачи, которые были завершены, а затем избавится от связанного с ними модуля и удалит их из этого словаря.

Однако....

Во время отладки в Visual Studio 2017 я извлекаю одно задание из БД, и пока я трачу время на отладку в одном экземпляре модуля, в этом модуле вызывается Dispose. Глядя в стек вызовов, я вижу, что Dispose был вызван в методе выше, и это потому, что задача имеет IsCompleted == true. Но, очевидно, это не может быть завершено, потому что я все еще отлаживаю его.

• Является ли свойство .IsCompleted неправильным свойством для проверки?
• Это просто артефакт отладки в Visual Studio?
• Я иду об этом неправильно?

Дополнительная информация

В комментариях ниже меня попросили предоставить некоторую дополнительную информацию о потоке, потому что то, что я описал, казалось невозможным (и я действительно надеялся, что этого не может быть). Ниже приведена урезанная версия моего кода (я удалил проверки токена отмены и защитного кодирования, но ничего, что влияет на поток).

Точка входа приложения

Это служба Windows. В OnStart() есть следующая строка:

this.RunApplicationTask = 
Task.Run(() => myApp.DoWorkAsync().ConfigureAwait(false), myService.CancelSource.Token);

«RunApplicationTask» — это просто свойство, позволяющее сохранять выполняемую задачу в области действия в течение всего срока службы службы.

ДоворкАсинк()

public async Task DoWorkAsync()
{
    do
    {
        await this.ExecuteSingleIterationAsync().ConfigureAwait(false);
        await Task.Delay(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
    }
    while (myApp.ServiceCancellationToken.IsCancellationRequested == false);

    await Task.WhenAll(this.NonAwaitedTasks.Keys).ConfigureAwait(false);
    await this.ClearCompletedTasksAsync().ConfigureAwait(false);
    this.WorkItemsTaskCompletionSource.SetResult(true);

    return;
}

Итак, пока я отлаживаю, это итерация DO-LOOP, она не попадает в Task.WhenAll(....).

Также обратите внимание, что после вызова запроса на отмену и завершения всех задач я вызываю ClearCompletedTasksAsync(). Об этом позже....

ExecuteSingleIterationAsync

private async Task ExecuteSingleIterationAsync()
{
    var getJobsResponse = await DB.GetJobsAsync().ConfigureAwait(false);
    await this.ProcessWorkLoadAsync(getJobsResponse.Jobs).ConfigureAwait(false);
    await this.ClearCompletedTasksAsync().ConfigureAwait(false);
}

Процессворковлоадасинк

private async Task ProcessWorkLoadAsync(IList<Job> jobs)
{
    if (jobs.NoItems())
    {
        return ;
    }

    jobs.ForEach(job =>
    {
        // The processor instance is disposed of when removed from the NonAwaitedTasks collection.
        IJobProcessor processor = ProcessorFactory.GetProcessor(workItem, myApp.ServiceCancellationToken);
        try
        {
            var task = Task.Run(() => processor.ExecuteAsync(job).ConfigureAwait(false), myApp.ServiceCancellationToken);
            this.NonAwaitedTasks.Add(task, processor);
        }
        catch (Exception e)
        {
            ...
        }
    });

    return;
}

Каждый процессор реализует следующий метод интерфейса: Задача ExecuteAsync (рабочее задание);

Когда я нахожусь в ExecuteAsync, .Dispose() вызывается для экземпляра процессора, который я использую.

ProcessorFactory.GetProcessor()

public static IJobProcessor GetProcessor(Job job, CancellationToken token)
{
    .....
    switch (someParamCalculatedAbove)
    {
        case X:
            {
                return new XProcessor(...);
            }

        case Y:
            {
                return new YProcessor(...);
            }

        default:
            {
                return null;
            }
    }
}

Итак, здесь мы получаем экземпляр новый.

ClearCompletedTasksAsync()

private async Task ClearCompletedTasksAsync()
{
    await myStatic.NonAwaitedTasksPadlock.WaitAsync().ConfigureAwait(false);
    try
    {
        foreach (var taskEntry in this.NonAwaitedTasks.Where(entry => entry.Key.IsCompleted).ToArray())
        {
            var processorInstance = taskEntry.Value as IDisposable;
            processorInstance?.Dispose();
            this.NonAwaitedTasks.Remove(taskEntry.Key);
        }
    }
    finally
    {
        myStatic.NonAwaitedTasksPadlock.Release();
    }
}

Это называется каждой итерацией Do-Loop. Его цель состоит в том, чтобы гарантировать, что список неожидаемых задач остается небольшим.

Вот и все... Кажется, что Dispose вызывается только при отладке.

Пример LinqPad

async Task Main()
{
    SetProcessorRunning();

    await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false);

    do
    {
        foreach (var entry in NonAwaitedTasks.Where(e => e.Key.IsCompleted).ToArray())
        {
            "Task is completed, so will dispose of the Task's processor...".Dump();
            var p = entry.Value as IDisposable;
            p?.Dispose();
            NonAwaitedTasks.Remove(entry.Key);
        }
    }
    while (NonAwaitedTasks.Count > 0);
}

// Define other methods and classes here

public void SetProcessorRunning()
{
    var p = new Processor();

    var task = Task.Run(() => p.DoWorkAsync().ConfigureAwait(false));
    NonAwaitedTasks.Add(task, p);
}

public interface IProcessor
{
    Task DoWorkAsync();
}

public static Dictionary<Task, IProcessor> NonAwaitedTasks = new Dictionary<Task, IProcessor>();

public class Processor : IProcessor, IDisposable
{
    bool isDisposed = false;
    public void Dispose()
    {
        this.isDisposed = true;
        "I have been disposed of".Dump();
    }

    public async Task DoWorkAsync()
    {
        await Task.Delay(TimeSpan.FromSeconds(5)).ConfigureAwait(false);

        if (this.isDisposed)
        {
            $"I have been disposed of (isDispose = {this.isDisposed}) but I've not finished work yet...".Dump();
        }

        await Task.Delay(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
    }
}

Выход:

Task is completed, so will dispose of the Task's processor...

I have been disposed of

I have been disposed of (isDispose = True) but I've not finished work yet...