Почему UnobservedTaskException?

Question

Почему этот код вызывает необработанное исключение.Я ожидал, что исключение будет проглочено или сообщено в обработчике OnError в зависимости от того, удалена ли подписка Rx или нет.

Есть ли хороший способ избежать этого, кроме написания продолжения в каждой задаче, чтобы проверить, не является ли оно ошибочным, а затем посмотреть на его свойство исключения?

Я бы думал Rx сделал это для нас, но явно нет.

public static void Main(string[] args)
    {
        TaskScheduler.UnobservedTaskException += (s, a) =>
        {
            Console.WriteLine("***************\nUnhandled '{0}' detected.", a.Exception.InnerException.GetType());
        };

        var d = Observable.FromAsync(() => Task.WhenAny(Do1(), Do2()))
            .Subscribe(x => { }, ex => Console.WriteLine($"Error {ex}"), () => Console.WriteLine("Completed Inner"));

        Console.WriteLine("Press any key to dispose");
        Console.ReadKey();
        d.Dispose();
        GC.Collect();
        GC.WaitForPendingFinalizers();
        Console.ReadKey();
    }

    private static Task Do1()
    {
        return Task.Run(async () =>
        {
            await Task.Delay(500);
            throw new AccessViolationException("Oops 1");
        });
    }

    private static Task Do2()
    {
        return Task.Run(async () =>
        {
            await Task.Delay(600);
            throw new AccessViolationException("Oops 2");
        });
    }

Answer 1

Rx делает замечательную работу по набору ошибок и их устранению, когда происходит одно из следующих действий:

• Наблюдаемое отменяется (посредством удаления подписки)
• Наблюдаемое завершение.
• Наблюдаемое совершает ошибку.

Но для этого вся вычислительная цепочка должна находиться под контролем наблюдаемой подписки.

В вашем коде этого не произошло.

Observable.FromAsync принимает Func<Task<T>> и возвращает и IObservable<T>.Ваш Observable.FromAsync возвращает IObservable<Task>, что означает, что ваш параметр возвращает Task<Task>.И это то, что делает Task.WhenAny(Do1(), Do2()).Это означает, что внутренняя задача Task<Task> не находится под контролем наблюдаемой подписки.

Чтобы исправить код, вы можете сделать две вещи.

Первая - исправитьTPL, так что он находится под контролем наблюдаемого.Это значит сделать его отменяемым, а также распаковать Task.WhenAny.

Вот новые Do методы:

private static Task Do1(CancellationToken ct)
{
    return Task.Run(async () =>
    {
        await Task.Delay(500, ct);
        throw new AccessViolationException("Oops 1");
    }, ct);
}

private static Task Do2(CancellationToken ct)
{
    return Task.Run(async () =>
    {
        await Task.Delay(600, ct);
        throw new AccessViolationException("Oops 2");
    }, ct);
}

Теперь вы можете написать свой код следующим образом:

using (
    Observable
        .FromAsync(ct => Task.WhenAny(Do1(ct), Do2(ct)).Unwrap())
        .Subscribe(
            x => { },
            ex => Console.WriteLine($"Error {ex}"),
            () => Console.WriteLine("Completed Inner")))
{

    Console.WriteLine("Press any key to dispose");
    Console.ReadLine();
}

Теперь это работает как положено.

Второе, что вы можете сделать, это не писать это с помощью задач.Используйте чистый код Rx.

Вот альтернатива Do s:

private static IObservable<Unit> Do1() =>
    from x in Observable.Timer(TimeSpan.FromMilliseconds(500))
    from y in Observable.Throw<AccessViolationException>(new AccessViolationException("Oops 1"))
    select Unit.Default;

private static IObservable<Unit> Do2() =>
    from x in Observable.Timer(TimeSpan.FromMilliseconds(600))
    from y in Observable.Throw<AccessViolationException>(new AccessViolationException("Oops 2"))
    select Unit.Default;    

Вот основной код:

using (
    Observable
        .Amb(Do1(), Do2())
        .Subscribe(
            x => { },
            ex => Console.WriteLine($"Error {ex}"), 
            () => Console.WriteLine("Completed Inner")))
{
    Console.WriteLine("Press any key to dispose");
    Console.ReadLine();
}

Это тоже работает.

Я предпочитаю Rx способ сделать это.

Answer 2

Это не обработано;это ненаблюдаемое.

Незаметное исключение происходит из-за этого кода: Task.WhenAny(Do1(), Do2()) и не имеет ничего общего с Rx.Тип возврата Task.WhenAny - Task<Task>, то есть будущее, которое разрешается для задачи, которая завершилась.Чтобы наблюдать результат этой внутренней задачи (включая исключения), вам нужно await задача, которая завершилась, а не задача, возвращенная из Task.WhenAny.

Итак, в вашем коде прямо сейчас вы завершаетес наблюдаемой последовательностью задач , что странно, но законно.FromAsync возвращает IObservable<Task>, и x в вашей Subscribe лямбда имеет тип Task.

Я полагаю, вы хотите, чтобы результат Do1 или Do2 был фактическими даннымив наблюдаемом, и чтобы сделать это, вам нужно развернуть эту задачу.«Двойное ожидание» не редкость при вызове Task.WhenAny:

var d = Observable.FromAsync(async () => await await Task.WhenAny(Do1(), Do2()))
    .Subscribe(x => { }, ex => Console.WriteLine($"Error {ex}"), () => Console.WriteLine("Completed Inner"));

В качестве альтернативы, вы можете использовать метод Unwrap:

var d = Observable.FromAsync(() => Task.WhenAny(Do1(), Do2()).Unwrap())
    .Subscribe(x => { }, ex => Console.WriteLine($"Error {ex}"), () => Console.WriteLine("Completed Inner"));

В зависимости от того, что лучше работает для вашего реальногокод мира.

Обновление: Пример метода расширения для предотвращения ненаблюдаемых исключений только для определенных задач:

public static Task IgnoreUnobservedExceptions(this Task task)
{
  Ignore(task);
  return task;
}

public static Task<T> IgnoreUnobservedExceptions<T>(this Task<T> task)
{
  Ignore(task);
  return task;
}

private static async void Ignore(this Task task)
{
  try { await task.ConfigureAwait(false); }
  catch { }
}

Использование: Task.WhenAny(Do1().IgnoreUnobservedExceptions(), Do2().IgnoreUnobservedExceptions())