Mediatr: Override PublishCore

[mpaul31]

Hello,

I am trying to override PublishCore in order to call all notification handlers and aggregate any exceptions that may occur. The problem I am running into is it seems the notification handler is executed during the enumeration.

protected override async Task PublishCore(IEnumerable<Task> allHandlers)
{
    foreach (var handler in allHandlers)  // <-- calls notification handler here
    {
        await handler.ConfigureAwait(false);
    }
}

could you please let me know how you suggest I handle this?

Thanks!

[lilasquared]

@mpaul31 notification handlers don't have a return type. Can you give me some idea about your approach so far? My first thought would be that catching and logging the exceptions would be the responsibility of the notification handler implementation and not the caller.

[mpaul31]

correct, the caller does not care about a return type but it does care if any of the side effects (handlers) caused any exceptions. since mediatr is in-process messaging so to speak i think this makes sense.

all i am attempting to do is ensure each handler gets its chance to execute and after all have finished, throw an AggregateException if any one or more exceptions occurred.

please feel free to let me know if you do not agree with this.

[FuncLun]

It appears that the tasks are started as the IEnumerable<Task> allHandlers is being enumerated. If a task throws, it is thrown during the foreach enumerator. You'll have to do the enumeration manually.

var exceptions = new List<Exception>();

using (var enumerator = allHandlers.GetEnumerator())
{
    while (true)
    {
        try
        {
            if (!enumerator.MoveNext())
                break;

            var handler = enumerator.Current;
            if (handler != null)
                await handler.ConfigureAwait(false);
        }
        catch (Exception exception)
        {
            exceptions.Add(exception);
        }
    }
}

if (exceptions.Any())
    throw new AggregateException(exceptions);

You could check for .Count == 1 and throw the .First() instead of putting just one in the AggregateException.

Considering these Handlers should be decoupled, an exception in one handler probably should not prevent others from executing,

Another consideration is these tasks are run serially. To run in parallel, you can create a List and add handler to it instead of awaiting it. Later, use the original foreach loop to await the tasks in the list.

[mpaul31]

Considering these Handlers should be decoupled, an exception in one handler probably should not prevent others from executing,

Well said, that is exactly what I am thinking. I'm surprised this behavior changed with the last major release.

Thanks for your help, I'll give your suggestions a go.

[mpaul31]

To run in parallel, you can create a List and add handler to it instead of awaiting it. Later, use the original foreach loop to await the tasks in the list.

unfortunately it looks like the call to MoveNext() executes the handlers so the only way I can figure out how to let them run in parallel is to go back to the original implementation and do:

return Task.WhenAll(allHandlers);

then it is up to the caller to perform an awaitor Wait() depending upon whether or not they want an unwrapped exception or aggregate.

[FuncLun]

I'm sorry, I don't think I was clear enough about creating the list.

Here's what I meant (coding in browser, not IDE, sorry for typos/errors):

var exceptions = new List<Exception>();
var tasks = new List<Task>();

using (var enumerator = allHandlers.GetEnumerator())
{
    while (true)
    {
        try
        {
            if (!enumerator.MoveNext())
                break;

            var handler = enumerator.Current;
            if (handler != null)
                tasks.Add(handler);
                //await handler.ConfigureAwait(false); //forgot to delete this line!
        }
        catch (Exception exception)
        {
            exceptions.Add(exception);
        }
    }
}

foreach (var task in tasks)
{
    try
    {
        await task.ConfigureAwait(false);
    }
    catch (Exception exception)
    {
        exceptions.Add(exception);
    }
}

if (exceptions.Any())
    throw new AggregateException(exceptions);

The first loop should start all the tasks, collecting exceptions. The second loop will await each one, collecting exceptions. Maybe the second loop might be usable with .WhenAll() or .WaitAll(), but I'm not sure if they wait for all tasks if one of them throws an exception.

[mpaul31]

this part confuses me:

if (!enumerator.MoveNext())  // <-- this lines causes the notification handler to execute
    break;

    var handler = enumerator.Current;
    if (handler != null)
        tasks.Add(handler);
        await handler.ConfigureAwait(false);  // <-- what is the point of this if it has already executed

let me know if my comments make sense.

[lilasquared]

@mpaul31 the task is started but the code continues to move forward. The await and configure await just tells the code it needs to wait for the task to finish and not worry about returning to the captured context

[lilasquared]

that being said I think the code will not function as @FuncLun is expecting since by the time it gets to the second loop there will be nothing to await anymore

[FuncLun]

I forgot to delete a line. That's what I get for coding in a browser instead of Visual Studio! Sorry!

@lilasquared After removing the first await, it should work as expected. All tasks will start, and be put in a list. We're not waiting/awaiting for for the task. Later, we await the tasks from the list. It should be okay if the tasks complete out of order.

@mpaul31 When not using foreach, I'd normally use an enumerator like this:

while (enumerator.MoveNext()) //stop loop if false
{

However, this enumerator will:

if (there are more records)
  1a) move internal pointer to the next record
  2a) start the task
  3a) return true
else
  1b) return false

Except, step 2a throws an exception if the task throws on start. We want to catch the exception, but not stop the while loop, so we need to break it apart:

while (true) //always loop
{
    if (!enumerator.MoveNext()) break; //stop loop if false

Now, wrap the if in a try/catch. If MoveNext() doesn't throw, it gets the current handler, and put it's in the list. (I put in the null check because ReSharper told me to - it should never be null... That is, until it goes into production!)

If it does throw, we collect the exception and continue the loop. We could (possibly) get the failed task by looking at enumerator.Current, but I haven't tested that, and we really don't care about the failed task.

[lilasquared]

we've all been there! My confusion here is I thought something like Task.WhenAll(...) is supposed to be for this exact use case. Why does it not work for this?

[FuncLun]

await Task.WhenAll(tasks); may be perfect. I don't know how it handles task exceptions. It might collect the exceptions and throw an AggregateException, like we want.

But, I do not know the 'proper' way to do .ConfigureAwait(false). Maybe it should be await Task.WhenAll(tasks).ConfigureAwait(false). Or, does that need to be called for each task?

[lilasquared]

It looks like there are many things causing the behavior we are seeing. Let me start by saying I was able to get it to work correctly, without a lot of messy enumerators. There are two things preventing the clean solution though.

1. Notification handlers that are not async, yet throw exceptions, will cause the execution of the tasks to always stop during enumeration of the IEnumerable. I don't seem to have an explanation for this but let me show an example.

public class PongHandler : INotificationHandler<Ping>
{
    private readonly TextWriter _writer;

    public PongHandler(TextWriter writer)
    {
        _writer = writer;
    }

    public async Task Handle(Ping notification, CancellationToken cancellationToken)
    {
        throw new NotImplementedException();
        await _writer.WriteLineAsync(notification.Message + " Pong");
    }
}

public class PungHandler : INotificationHandler<Ping>
{
    private readonly TextWriter _writer;

    public PungHandler(TextWriter writer)
    {
        _writer = writer;
    }

    public Task Handle(Ping notification, CancellationToken cancellationToken)
    {
        return _writer.WriteLineAsync(notification.Message + " Pung");
    }
}

The PongHandler throws an exception, but is using the async/await keywords. PungHandler dows not throw exceptions and does not use async/await. If PongHandler did not use async / await it throws an exception immediately upon enumeration.

1. The PublishCore api is expecting and IEnumerable<Task>. This seems to contribute to the problem as well.

By changing the PublishCore api to accept Task[] instead, and ensuring to use async / await keywords, the following override of PublishCore works correctly and throws the AggregateException.

protected override Task PublishCore(Task[] handlers)
{
    Task.WhenAll(handlers).Wait();

    return Task.CompletedTask;
}

This require a change to the NotificationHandlerWrapper as well.

internal class NotificationHandlerWrapperImpl<TNotification> : NotificationHandlerWrapper
    where TNotification : INotification
{
    public override Task Handle(INotification notification, CancellationToken cancellationToken, ServiceFactory serviceFactory, Func<Task[], Task> publish)
    {
        var handlers = serviceFactory
            .GetInstances<INotificationHandler<TNotification>>()
            .Select(handler => handler.Handle((TNotification)notification, cancellationToken))
            .ToArray();

        return publish(handlers);
    }
}

[remcoros]

@lilasquared with that change to NotificationHandlerWrapper the handlers are executed immediately due to the call to ToArray().

This way, you don't have the ability anymore to enumerate over the handlers yourself (and wrap them with something, like a try/catch).

Enumerating over the tasks yourself like @FuncLun shows is the way to go if you want to wrap each call to "Handle(..)" with it's own try/catch.

[lilasquared]

They aren’t executed immediately if the signatures of the of the Handle methods are defined as async and use the await keyword. I have a working example I can publish if you’re interested.

With that being said I don’t think my solution is the right one because you can’t guarantee that, I just wanted to share some info I found out about how tasks behave.

[mpaul31]

i think the underlying issue is the fact that simply enumerating a list causes a side-effect (handler executing) which is strange behavior IMO.

I think I will stick with the following:

return Task.WhenAll(allHandlers);

thanks everyone!

[lilasquared]

@mpaul31 it is strange unless you consider the code.

var handlers = serviceFactory
            .GetInstances<INotificationHandler<TNotification>>()
            .Select(handler => handler.Handle((TNotification)notification, cancellationToken));

this returns the IEnumerable, which when iterated does call the handle method. If the handle method is not async / await then it executes immediately, if it is then it simply returns the task handle.

[mpaul31]

maybe the code should return an Action so it is up to the caller when execution should occur.

[FuncLun]

Enumerating the Select will start the tasks, which is what ToArray() will do. The select is calling Handle and returning the task.

Note: The foreach loop was changed from WhenAll due to issue #288 and #292. I haven't fully digested those issues, but it appears to be around the order that PipelineBehaviors are executed. Commit message by Jimmy: "Making all enumerable tasks foreach; fixes 228". Maybe raise a separate issue to get event handlers parallel again?

I realized there are 2 scenarios to handle, but I was only testing 1...

Throwing in an async handler
The exception is inside the task. It won't be realized until the task is observed (awaited, result, etc). foreach awaits each task serially, meaning the first task that throws will be thrown and the remaining tasks not started. WhenAll is parallel and will aggregate task exceptions. Note, when you await the WhenAll task, only 1 of the exceptions will be thrown (i.e. not the AggregateException). To get all the exceptions, you have to go back to the task like this:

var whenAllTask = Task.WhenAll(allHandlers).ConfigureAwait(false);
try { await whenAllTask; }
catch (Exception exception) //will only be 1 exception
{ throw whenAllTask.Exception; } //will be AggregateException with 1 or more exceptions

Throwing in a non-async handler.
The exception is not inside the task, but will throw during enumeration. Both foreach and WhenAll will fail to enumerate past the throwing task. Since foreach is serial and awaited, all previous tasks should be complete without exception. WhenAll is parallel and does not await until after enumeration, so previous tasks have been started, but we throw away those tasks - even if they are still running.

A complex fix is the GetEnumerator() code. But, this is available to override.

A simple fix is to make all handlers async, even if not awaiting anything. You could go change all your handlers and add async. await Task.Yield; if you have to.
Or, NotificationHandlerWrapperImpl<>.Handle can be changed to this: (I added async await and .ConfigureAwait(false))

        public override Task Handle(INotification notification, CancellationToken cancellationToken, ServiceFactory serviceFactory, Func<IEnumerable<Task>, Task> publish)
        {
            var handlers = serviceFactory
                .GetInstances<INotificationHandler<TNotification>>()
                .Select(async x => await x.Handle((TNotification)notification, cancellationToken).ConfigureAwait(false));

            return publish(handlers);
        }

Now, calling the Handle function is part of the task. Even if the handler function is non-async and throws, the task will catch it, allowing WhenAll to add it to the AggregateException.
There's really no change for foreach. Since it is serial, it would throw on the await line instead of the foreach line, but the outcome would be the same.

Going further...
When you await a task that throws AggregateException, it will only throw one of the exceptions. To access all the exceptions, you have to go back to the task object:

var task = Task.WhenAll(allHandlers).ConfigureAwait(false);
try { await task; }
catch (Exception exception) //will only be 1 exception
{ throw task.Exception; } //will be AggregateException with 1 or more exceptions

[mpaul31]

Note: The foreach loop was changed from WhenAll due to issue #288 and #292.

Behaviors do not even apply to notification handlers so i have a feeling it was changed just for consistency. makes sense for request handlers not notification handlers IMO

[lilasquared]

Do you need the ConfigureAwait(false) in the Select? Otherwise this is a really simple solution that makes sense to me, and probably would be a good change to MediatR without breaking anything.

Also as long as you don't await Task.WhenAll() it will throw the aggregate exception. Task.WhenAll(...).Wait() for example.

[remcoros]

@mpaul31 you mean like this: https://github.com/jbogard/MediatR/pull/330

[lilasquared]

@remcoros that is what I was thinking but maybe using the async await like @FuncLun suggested is better

[remcoros]

That solution still calls "Handle" as soon as the enumeration starts.

async/await or not, that's (mostly) just syntactic sugar for the ContinueWith(..) construct.

Any CPU-bound work before the first 'await' within 'Handle(..)' is still executed when MoveNext on the enumerator is called.

I understand that a lot of people can't really wrap their head around asynchronousity, await, cpu-bound io-bound. I too get so confused about the whole model.

But a fact is, anything before the first "await" of the called method, is still cpu-bound and gets executed immediately by the enumerator. Without an explicit call to something like Task.Run (which mediator does not do) at least the first part, is still synchronous and running on the callers thread.

Wrapping it in an Action/Func, actually stores a reference to the handler, to be called later by the implementor

[lilasquared]

I have tests that i wrote for this scenario and @FuncLun's proposed changes work actually. since the lambda that the select is using is what is async it won't execute the handler during enumeration.

public class ThrowingPangHandler : INotificationHandler<Ping>
{
    private readonly TextWriter _writer;

    public ThrowingPangHandler(TextWriter writer)
    {
        _writer = writer;
    }

    public Task Handle(Ping notification, CancellationToken cancellationToken)
    {
        throw new NotImplementedException();
        return _writer.WriteLineAsync(notification.Message + " Pang");
    }
}

public class ExceptionMediator : Mediator
{
    public ExceptionMediator(ServiceFactory serviceFactory)
        : base(serviceFactory)
    {
    }

    protected override Task PublishCore(IEnumerable<Task> allHandlers)
    {
        Task.WhenAll(allHandlers).Wait();

        return Task.CompletedTask;
    }
}

[Fact]
public async Task Should_override_and_throw_aggregate()
{
    var builder = new StringBuilder();
    var writer = new StringWriter(builder);

    var container = new Container(cfg =>
    {
        cfg.Scan(scanner =>
        {
            scanner.AssemblyContainingType(typeof(PublishTests));
            scanner.IncludeNamespaceContainingType<Ping>();
            scanner.WithDefaultConventions();
            scanner.AddAllTypesOf(typeof(INotificationHandler<>));
        });
        cfg.For<TextWriter>().Use(writer);
        cfg.For<IMediator>().Use<ExceptionMediator>();
        cfg.For<ServiceFactory>().Use<ServiceFactory>(ctx => t => ctx.GetInstance(t));
    });

    var mediator = container.GetInstance<IMediator>();

    try
    {
        await mediator.Publish(new Ping {Message = "Ping"});
    }
    catch (AggregateException ex)
    {
        Console.Write("caught it");
    }

    var result = builder.ToString().Split(new[] { Environment.NewLine }, StringSplitOptions.None);
    result.ShouldContain("Ping Pong");
    result.ShouldContain("Ping Pung");
}

[remcoros]

Task Handle()
{
    int i = 0;
    return _httpClient.GetSomeAsync(..)
        .ContinueWith(t=> {
            var result = t.Result;
            DoSomeWithResult(result);
        })
}

is the same as:

async Task Handle()
{
    int i = 0;
    var result = await _httpClient.GetSomeAsync(..);
    DoSomeWithResult(result);
}

In both situations "int i = 0;" is executed immediately by MoveNext, on the same thread as the caller, async/await or not.

[lilasquared]

that may be the case but using async / await does not throw the exeption, even if it executes the handle method right away. throw my tests into PublishTests and try it out. You'll see what I mean.

[remcoros]

that may be the case but using async / await does not throw the exeption, even if it executes the handle method right away. throw my tests into PublishTests and try it out. You'll see what I mean.

I know what you mean, and there are some nuances (which, I reckon, also can't wrap my head around 100% of the time). If you step into source, debug line by line (also the WhenAll), you see the order of execution.

Anyway, relying on the async execution model to influence how/when handlers get executed is not the way to go in my opinion.

Just passing on a reference to the Handle function, and leave it to the implementor to decide what to do with it, seems to be the most flexible (and least error prone) solution.

Using this, OP can do this:

foreach(handler in handlers)
{
  try { await handler().ConfigAwait(false); } catch(Exception ex) { ... }
}

(which, as already mentioned, can be done already, if you do the enumeration yourself)

[lilasquared]

what makes that better than

protected override async Task PublishCore(IEnumerable<Task> allHandlers)
{
    foreach(var handler in allHandlers)
    {
        try
        {
            await handler.ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            Console.Write("caught it");
        }
    }
}

since the async await keyword are what should be used for async work i would expect this to be the more appropriate usage.

Sorry for blowing you up really just curious more than anything. Like you said the async / await stuff isn't exactly trivial

[remcoros]

No problem, I love this stuff :) Let me try to cook something up, will follow up.

[mpaul31]

what makes this better than

protected override async Task PublishCore(IEnumerable<Task> allHandlers)
{
    foreach(var handler in allHandlers)
    {
        try
        {
            await handler.ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            Console.Write("caught it");
        }
    }
}

since the async await keyword are what should be used for async work i would expect this to be the more appropriate usage.

Sorry for blowing you up really just curious more than anything. Like you said the async / await stuff isn't exactly trivial

that's originally what i tried and the await will automagically "unwrap" the aggregate exception with an exception. there are other nuances with it but i can't remember at the moment.

[lilasquared]

yeah await does that. If you want the aggregate exception you have to use Task.WhenAll(...)

[mpaul31]

yeah await does that. If you want the aggregate exception you have to use Task.WhenAll(...)

yeah i get that, that's why i moved on and went back to the old way mediatr worked :)

[lilasquared]

@mpaul31 the current way mediator works will not call all of your handlers though. It is not actually using the functionality of Task.WhenAll(...). It will fail and throw an exception just iterating the list so the first one that fails will stop execution.

[mpaul31]

@mpaul31 the current way mediator works will not call all of your handlers though. It is not actually using the functionality of Task.WhenAll(...). It will fail and throw an exception just iterating the list so the first one that fails will stop execution.

Sorry should have been more specific. I created a derived class that overrides the PublishCore with the behavior it use to do.

[lilasquared]

the problem is in the NotificationHandlerWrapper - if any one of your handlers fails it will throw an exception and not continue with the rest of them. That's what we are trying to fix :smile:

[mpaul31]

no i have tested it, this works as desired and allows all the handlers to execute regardless if one fails

        protected override Task PublishCore(IEnumerable<Task> allHandlers)
        {
            // Override the default implementation so the handlers run in parallel.
            // Also, if an exception occurs while one handler is executing, it will
            // not prevent other handlers from running.
            return Task.WhenAll(allHandlers);
        }

[lilasquared]

I'm doing it right now with mediatr's unit tests and it doesn't. My first handler throws and the rest do not get executed.

[mpaul31]

hmm not sure my tests pass :)

what you are describing is not what and how the documentation explains how Task.WhenAll behaves.

Now I'm going to have to go back into my source code and double check :)

[mpaul31]

what makes that better than

protected override async Task PublishCore(IEnumerable<Task> allHandlers)
{
    foreach(var handler in allHandlers)
    {
        try
        {
            await handler.ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            Console.Write("caught it");
        }
    }
}

since the async await keyword are what should be used for async work i would expect this to be the more appropriate usage.

Sorry for blowing you up really just curious more than anything. Like you said the async / await stuff isn't exactly trivial

i'm not at my desk at the moment but if i remember correctly the code you pasted will not allow all of the handlers to execute if one threw an exception. (99.9% sure)

[lilasquared]

yeah this code was in response to remcoros. I am using Task.WhenAll(...) in the tests as you are.

[remcoros]

@lilasquared I didn't see you wanted to run them in parallel, you can save the task in a list (not awaiting it), and use WhenAll on the list of saved tasks.

and about enumerating an IQueryable of: Tasks vs async/await Task vs Func:

https://github.com/remcoros/EnumeratingTasks/blob/master/Program.cs

The first two are the same (handler gets called on MoveNext)

[remcoros]

I'm lost at if you have a solution for now. With current mediatr you can use 'return Task.WhenAll' right?

[lilasquared]

I don't think it works to run all tasks to completion, since the exception is thrown while it is enumerating. but @mpaul31 says it is working for him :man_shrugging:

[mpaul31]

I don't think it works to run all tasks to completion, since the exception is thrown while it is enumerating. but @mpaul31 says it is working for him 🤷‍♂️

@remcoros can you update your git sample to prove that?

[remcoros]

Well, only if no exception will be thrown before the first await call within 'Handle'. Anything after that is basically a callback and handled by .net TPL

See the implementation of WhenAll here: https://github.com/Microsoft/referencesource/blob/3b1eaf5203992df69de44c783a3eda37d3d4cd10/Microsoft.Bcl.Async/Microsoft.Threading.Tasks/Threading/Tasks/TaskEx.cs#L284

ToArray is called there on IEnumerable, which is actually the IQueryable of mediatr.

When ToArray is called by WhenAll, (or when enumerating yourself with foreach), this happens:

• 1 MoveNext()
  -> 2 x.Handle(..)
  -> 3 code within Handle (before await)
  -> 4 returns a Task, containing all the callbacks (ContinueWith, or anything after "await")
  -> 5 TPL stuff

If an exception happens at 1-3, it's thrown back to the caller. from 4 and onward, it's handled within TPL, and when every task is handled, you get back an AggregateException.

But ANY exception in code before the first await within Handle of the first and only first handler, will be thrown back at you.

[mpaul31]

Well, only if no exception will be thrown before the first await call within 'Handle'. Anything after that is basically a callback and handled by .net TPL

See the implementation of WhenAll here: https://github.com/Microsoft/referencesource/blob/3b1eaf5203992df69de44c783a3eda37d3d4cd10/Microsoft.Bcl.Async/Microsoft.Threading.Tasks/Threading/Tasks/TaskEx.cs#L284

ToArray is called there on IEnumerable, which is actually the IQueryable of mediatr.

When ToArray is called by WhenAll, (or when enumerating yourself with foreach), this happens:

• 1 MoveNext()
  -> 2 x.Handle(..)
  -> 3 code within Handle (before await)
  -> 4 returns a Task, containing all the callbacks (ContinueWith, or anything after "await")
  -> 5 TPL stuff

If an exception happens at 1-3, it's thrown back to the caller. from 4 and onward, it's handled within TPL, and when every task is handled, you get back an AggregateException.

But ANY exception in code before the first await within Handle of the first and only first handler, will be thrown back at you.

wow ok. so how best should i handle this to ensure all handlers get their own turn to execute regardless if an exception happens?

[lilasquared]

@remcoros @mpaul31 consider this.

public class Handler
{
    public Handler(string name)
    {
        Name = name;
    }

    public string Name { get; }

    public Task Handle()
    {
        if (Name == "2")
        {
            throw new InvalidOperationException();
        }
        Console.WriteLine($"Handler {Name}, before delay");
        Task.Delay(100);
        Console.WriteLine($"Handler {Name}, after delay");
        return Task.CompletedTask;
    }
}

```csharp
public async Task IQueryableAsyncTask()
{
Console.WriteLine("Enumerating an IQueryable");

var handlersOfAsyncTask = new[]
    {
        new Handler("1"),
        new Handler("2"),
        new Handler("3")
    }
    .Select(async x => await x.Handle());

try
{
    await Task.WhenAll(handlersOfAsyncTask);
}
catch (Exception ex)
{
    Console.WriteLine(ex.GetType());
}

}
// Handler 1, before delay
// Handler 1, after delay
// System.InvalidOperationException
// Handler 3, before delay
// Handler 3, after delay

```csharp
public async Task IQueryableTask()
{
    Console.WriteLine("Enumerating an IQueryable<Task>");

    var handlersOfTask = new[] {
            new Handler("1"), 
            new Handler("2"), 
            new Handler("3")
        }
        .Select(x => x.Handle());
    try
    {
        await Task.WhenAll(handlersOfTask);
    }
    catch (Exception ex)
    {
        Console.WriteLine(ex.GetType());
    }
}
// Handler 1, before delay
// Handler 1, after delay
// System.InvalidOperationException

The IQueryableTask method is how MediatR functions today

[mpaul31]

still confused. even with the first handler throwing an exception before the first await it seems they both execute (according to my breakpoints):

    public class PingMe : INotification { }

    public class PingOne : INotificationHandler<PingMe>
    {
        public async Task Handle(PingMe notification, CancellationToken cancellationToken)
        {
            throw new Exception("boom");

            await Task.Delay(1000, cancellationToken);
        }
    }

    public class PingTwo : INotificationHandler<PingMe>
    {
        public Task Handle(PingMe notification, CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }
    }

{
    await mediator.Publish(new PingMe());
}
catch (Exception e)
{
    Console.WriteLine(e);
    throw;
}

[lilasquared]

@mpaul31 if you don't use async await on PingOne then PingTwo will not execute.

[mpaul31]

@mpaul31 if you don't use async await on PingOne then PingTwo will not execute.

can i bribe you to try doing that in a simple console app? i swear using my debugger it works regardless if async await is there or not. the debugger may break but if you just continue execution it keeps on chugging.

[lilasquared]

the code i posted does just that, and the console output is already included

[mpaul31]

your code is not doing exactly the same.

your doing a Wait and looking for an aggregate exception. Here is mine:

        protected override Task PublishCore(IEnumerable<Task> allHandlers)
        {
            // Override the default implementation so the handlers run in parallel.
            // Also, if an exception occurs while one handler is executing, it will
            // not prevent other handlers from running.
            return Task.WhenAll(allHandlers);
        }

[lilasquared]

The code i am referring to is not using mediatr at all. I'm referring to this comment https://github.com/jbogard/MediatR/issues/328#issuecomment-439173541

[remcoros]

Updated my sample with yours:

PS C:\Projects\EnumeratingTasks> dotnet run
Enumerating an IQueryable<async Task>
Handler 1, before await
Handler 3, before await
Handler 3, after await
Handler 1, after await
System.InvalidOperationException
Enumerating an IQueryable<Task>
Handler 1, before await
Handler 3, before await
Handler 3, after await
Handler 1, after await
System.InvalidOperationException
PS C:\Projects\EnumeratingTasks>

~I'm lost..~

[remcoros]

public class Handler
{
    public Handler(string name)
    {
        Name = name;
    }

    public string Name { get; }

    public Task Handle()
    {
        if (Name == "2")
        {
            throw new InvalidOperationException();
        }
        Console.WriteLine($"Handler {Name}, before delay");
        Task.Delay(100);
        Console.WriteLine($"Handler {Name}, after delay");
        return Task.CompletedTask;
    }
}

@lilasquared you don't await the Task.Delay here, so that skews results.

[mpaul31]

@remcoros based on what you described would you not expect what I showed here https://github.com/jbogard/MediatR/issues/328#issuecomment-439174589 to have both handlers execute?

[lilasquared]

@remcoros that is the point I am trying to make. Without async / await explicitly used in the handle method the enumeration throws the exception and all handlers will not run. async / await is not required for a Task based method.

[remcoros]

Well yikes, yes. you're right. Not adding 'async' to a method, does not add the compiler magic around it, so if you do not wrap it in a task yourself and return that (but have code like those argument checks), it will run in line with MoveNext and all, because it's just a regular method call to 'Handle(..)'.

And then we're back to the question if it should be possible to be able to override how 'Handle' is executed.

Also with the pr I sent, it would be possible to do Task.Run(() => handler()) or something more sophisticated.

[lilasquared]

I personally like the idea of the function handles but that doesn't seem to jive with the typical c# paradigm.
(its more of a loosely typed language thing. Looking at you javascript). That't why I was thinking the async / await inside the select made sense.

[lilasquared]

maybe we can betelgeuse @jbogard into weighing in :100:

[mpaul31]

my head hurts :/

[remcoros]

Yes I see now why, also why it would work. From my understanding though, if you don't have to actually await anything and have a task, you can (should?) just return the task, as to avoid the extra compiler magic.

Must be one of those edge-cases :)

I'd say, open it up a bit more and leave it up to an implementor.

[remcoros]

Oh btw, if I wasn't entirely clear. If you have a "Task SomeMethod()", which is not async, it IS expected that if you call that, it's just a regular method and any exceptions get right back at you (because of the MoveNext). It wouldn't be any different than method which returns something else.

So in your Handler, it's just a regular method call (the delay doesn't have any effect, other than holding up another thread somewhere) and it returns a completed task, so any other task after it, can continue.

It's really obnoxious but kinda expected. And in this case, can easily break the pipeline.. maybe the framework should even help here..

[mpaul31]

so are you saying this is not considered an async method? (assuming GetVehiclesBySiteId returns a Task)

public Task<IEnumerable<Vehicle>> Handle(GetVehiclesBySiteId request, CancellationToken cancellationToken)
{
    return _repository.GetVehiclesBySiteId(request.SiteId);
}

[lilasquared]

That's true, however when you are thinking about from mediatr's perspective it doesn't know if you are async awaiting or not, since that isn't part of the interface (and can't be). From it's perspective all handle methods are just regular method calls. If the idea is that mediatr will allow you to call all notification handlers even if they fail, then it should support that. Currently it doesn't.

[lilasquared]

@mpaul31 it is an async method because it can be awaited. if you had something in there like this though:

public Task<IEnumerable<Vehicle>> Handle(GetVehiclesBySiteId request, CancellationToken cancellationToken)
{
    if (request.SiteId <= 0) {
        throw new ArgumentOutOfBoundsException();
    }

    return _repository.GetVehiclesBySiteId(request.SiteId);
}

then the exception fires synchronously. (unless the caller awaits it)

[remcoros]

        public Task<IEnumerable<Vehicle>> Handle(GetVehiclesBySiteId request, CancellationToken cancellationToken)
        {
            // everything here is executed in line with whoever called this method.
           // ...
            return _repository.GetVehiclesBySiteId(request.SiteId);
        }

See the comment above, and see this: https://sharplab.io/#v2:CYLg1APgAgDABFAjAbgLAChYMQVjejKAZgQCY4BhOAbwznoRKgDY4ApAVwGcAXFgCgCUNOgzFQA7AmYA6CgHsAtgAcANgFMe64C3xiAvhiPpxTABzS4AQS4BPAHYBjAcNomxDAPSeEUlnKU1TW1dUXpDdH0gA===

By adding 'async' the compiler wraps the method in a state machine, doing some of it's own exception handling.

It all comes down to the order of instructions, and who calls what.

[mpaul31]

That's true, however when you are thinking about from mediatr's perspective it doesn't know if you are async awaiting or not, since that isn't part of the interface (and can't be). From it's perspective all handle methods are just regular method calls. If the idea is that mediatr will allow you to call all notification handlers even if they fail, then it should support that. Currently it doesn't.

to me each handler is and should be de-coupled and i would think (hope) this will be supported.

[lilasquared]

@remcoros pushed the async / await alternative w/ tests - https://github.com/jbogard/MediatR/pull/332

[mpaul31]

thanks you both for diving into this. i hope this helps others in the future.

[FuncLun]

@mpaul31 it is an async method because it can be awaited. if you had something in there like this though:

Actually, it isn't an async method. Yes, it can be awaited, but only because it returns a task. async adds the compiler magic, and since it does not have that keyword, it doesn't have the magic. What it returns does have the magic.

then the exception fires synchronously. (unless the caller awaits it)

This is true - any code that is not in an async method, or is before any await statements in an async method is executed synchronously (including the throw).

The difference is async will wrap the function in a try/catch, holding onto the exception. The function will then return a Task that is Completed, but faulted. When observed (await, .Result, etc), the task will throw the stored exception.

Task task = NonAsyncThatThrowsImmediately(); //Throws here
await task;

Task task = AsyncThatThrowsImmediately();
await task; //Throws here

Summary: Root causes of not executing all Handlers:

1. The enumerator executes the handle function that's not guaranteed to be in the context of a Task (i.e. non-async). If the handle throws synchronously, the enumerator throws.
   Solution(s)
   A. Add async to the LINQ query like in PR #332

2. There's no try/catch around the await. This was issue introduced with v5.1.0
   Solutions
   A. Add try/catch, Add exceptions to list, throw AggregateException.
   B. Use Task.WhenAll() in Mediator.PublishCore().
   Side Affect: The Handlers are executed in parallel. They were in parallel in >= v5.0.1 and serial in v5.1.0.

[lilasquared]

So do you think part of the PR #332 should also add the exception handling to the default PublishCore implementation? I think parallel being opt in by overriding is more ideal. That way at least they all execute even if it is serial.

[FuncLun]

Yes, I think they both should be done.

This is one symptom (all handlers not executed) caused by two issues that look the same, but are actually slightly different:

1) exception thrown in non-async handler. Before the PR #332, the Exception will be thrown at enumeration. After the PR, it will be thrown in the task.

2) exception thrown in the task (an async handler is 'in the task'). If the Task throws an exception, it will be thrown at the await, not the enumeration.

So, if we fix issue 1, issue 2 will still be a problem... Actually, issue 1 exceptions become issue 2 exceptions.

As for parallel or serial... It went from parallel to serial 4 months ago (in July). That's a judgement call for the those in control.

[jbogard]

So....what's the final verdict here?

[lilasquared]

I think if I update the PR to include catch and aggregate exceptions in publish core that would be good? I saw you just merged #333, does that mean this fix can't make it into v5?

[FuncLun]

@jbogard Should events be handled in Parallel or Serial

Prior to v5.1.0, they were Parallel.
v5.1.0 made them Serial.

Either way is an easy update, and the behavior can be overridden.

[jbogard]

For most folks use cases, they would want serial, since people were doing things that resulted in unexpected behavior because it wasn't obvious that execution was parallel. So I made execution serial, but allowed you to override the behavior, since at that point, you knew what you were doing (hopefully).

[jbogard]

@lilasquared yeah I bumped the version because of adding strong naming. No breaking API changes from 5 to 6 though.

[remcoros]

I personally would go with my PR (give a Func to PublishCore, so the implementor has all the freedom to execute the handlers.

This gives most flexibility, and you can override Mediatr and control how handlers are called exactly how you see fit.

With the async/await in the IQueryable.Select(), any cpu-bound code in the notification 'Handle' method is still executed during the first MoveNext.

Though, in case of exceptions, the async statemachine makes sure that it is handled within the task. This means that what has been discussed here, about exceptions, doesn't manifest itself anymore.

BUT that code is still executed in line with MoveNext.

If I want to truly implement a parallel notification publisher with Task.Run(() => handler.Handle(), this is NOT possible at the moment.

Only if we get a reference to the Handle method, we can truly control how it is executed.

[jbogard]

Fixed by #330

[mpaul31]

so with the new changes to execute is parallel we should enumerate the handlers and add them to a list then perform a WhenAll?

[remcoros]

so with the new changes to execute is parallel we should enumerate the handlers and add them to a list then perform a WhenAll?

• Truly parallel, means you need to execute the handler with Task.Run(). This means every handler is executed on its own thread, making cpu AND io bound code, run truly parallel

• If you want to run them synchronously, but do NOT want to stop on the first exception: use WhenAll like you said

• If you want to run them synchronously, and want to stop on the first exception: it's the default.

Note that with option 2, you have to really think hard about where you need to do the try/catch:

Calling "handler()" (now that it's a Func), still actually executes this method on the callers thread. So if that Handle method does not have the 'async' keyword, any exceptions before the first await within that Handle method, are still thrown to the caller and not captured within a Task.
That's just how async/await works,

I'll work on some samples.

[jbogard]

Oh wait, this makes the handlers execute in parallel? That’s not what we
want for the default behavior. I only want to enable parallel execution. It
should be serial by default.

On Fri, Nov 16, 2018 at 7:53 PM Remco Ros [e-mail] wrote:

so with the new changes to execute is parallel we should enumerate the
handlers and add them to a list then perform a WhenAll?

-

Truly parallel, means you need to execute the handler with
Task.Run(). This means every handler is executed on its own thread, making
cpu AND io bound code, run truly parallel
-

If you want to run them synchronously, but do NOT want to stop on the
first exception: use WhenAll like you said
-

If you want to run them synchronously, and want to stop on the first
exception: it's the default.

Note that with option 2, you have to really think hard about where you
need to do the try/catch:

Calling "handler()" (now that it's a Func), still actually executes this
method on the callers thread. So if that Handle method does not have the
'async' keyword, any exceptions before the first await within that Handle
method, are still thrown to the caller and not captured within a Task.
That's just how async/await works,

I'll work on some samples.

—
You are receiving this because you modified the open/close state.

Reply to this email directly, view it on GitHub
https://github.com/jbogard/MediatR/issues/328#issuecomment-439473996,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AAGYMmdD5KqKBQQBRy0gsUt5t0dDoIadks5uvvu0gaJpZM4YaQd6
.

[remcoros]

Oh wait, this makes the handlers execute in parallel? That’s not what we want for the default behavior. I only want to enable parallel execution. It should be serial by default.
…

3 is the default, so nothing has changed, all good. We just moved the actual execution of Handle(), from the IQueryable.Select (in NotificationHandlerWrapperImpl) to PublishCore.

Working on some samples atm.

[remcoros]

@lilasquared @mpaul31 I worked out a few examples: https://github.com/jbogard/MediatR/pull/334

[mpaul31]

thanks examples look great! im torn between async or ParallelWhenAll. any suggestions on why i would choose one over the other?

[remcoros]

It depends on your use-case. The reason why the default implementation switched to synchronously waiting for each handler to finish, is because people ran into issues with concurrently accessing DbContext (I think).

If you don't have that, I'd go with Async, this runs your (cpu) code on the calling thread and also handles the case when a 'Handle' method doesn't use the 'async' keyword.

If your handlers do cpu intensive work (so you actually need the parallel threads), you could try something like I showed with Task.Run.

But, in that case, I wonder if it's not better to use a more sophisticated messaging framework for that, as the example implementation is of-course pretty naive.

[mpaul31]

makes sense thx!

[mpaul31]

@lilasquared @remcoros @FuncLun @jbogard Before I update my MediatR package and change my source code can you please help me understand why this works (I verified by running in debug and non-debug mode)? Based on everything said in this thread I would not expect all the handlers to run. I must be missing something with my understanding

public class PingMe : INotification
{
}

public class PingMeOne : INotificationHandler<PingMe>
{
    public Task Handle(PingMe notification, CancellationToken cancellationToken)
    {
        Console.WriteLine("PingMeOne");
        throw new System.NotImplementedException();
    }
}

public class PingMeTwo : INotificationHandler<PingMe>
{
    public async Task Handle(PingMe notification, CancellationToken cancellationToken)
    {

        await Task.Delay(1000, cancellationToken);

        Console.WriteLine("PingMeTwo");
    }
}

public class PingMeThree : INotificationHandler<PingMe>
{
    public Task Handle(PingMe notification, CancellationToken cancellationToken)
    {
        Console.WriteLine("PingMeThree");

        Task.Delay(100);

        Console.WriteLine("PingMeThree After");

        return Task.CompletedTask;
    }
}

public class CustomMediator : Mediator
{
    public CustomMediator(ServiceFactory serviceFactory) : base(serviceFactory)
    {
    }

    protected override Task PublishCore(IEnumerable<Task> allHandlers)
    {
        // Override the default implementation so the handlers run in parallel.
        // Also, if an exception occurs while one handler is executing, it will
        // not prevent other handlers from running.
        return Task.WhenAll(allHandlers);
    }
}



md5-b5877490d12b3dc96774cad7969647a9



try
{
    await mediator.Publish(new PingMe());
}
catch (Exception e)
{
    Console.WriteLine(e);
    throw;
}

[remcoros]

correct, If "PingMeOneHandler" is the first to be executed, I would expect:

PingMeOne
NotImplementedException+stacktrace

I'll have a look when back home.

[remcoros]

@mpaul31 https://github.com/remcoros/MediatrPublishExample

I copied/pasted exactly what you posted. And it DOES throw (and stop) on the exception thrown by PingMeOne:

PS C:\Projects\MediatRPublishExample> dotnet run
PingMeOne
System.NotImplementedException: The method or operation is not implemented.
   at MediatRPublishExample.PingMeOne.Handle(PingMe notification, CancellationToken cancellationToken) in C:\Projects\MediatRPublishExample\Program.cs:line 62
   at System.Linq.Enumerable.SelectArrayIterator`2.MoveNext()
   at System.Threading.Tasks.Task.WhenAll(IEnumerable`1 tasks)
   at MediatRPublishExample.CustomMediator.PublishCore(IEnumerable`1 allHandlers) in C:\Projects\MediatRPublishExample\Program.cs:line 49
   at MediatR.Internal.NotificationHandlerWrapperImpl`1.Handle(INotification notification, CancellationToken cancellationToken, ServiceFactory serviceFactory, Func`2 publish)
   at MediatR.Mediator.Publish[TNotification](TNotification notification, CancellationToken cancellationToken)
   at MediatRPublishExample.Program.Main(String[] args) in C:\Projects\MediatRPublishExample\Program.cs:line 28

Unhandled Exception: System.NotImplementedException: The method or operation is not implemented.
   at MediatRPublishExample.PingMeOne.Handle(PingMe notification, CancellationToken cancellationToken) in C:\Projects\MediatRPublishExample\Program.cs:line 62
   at System.Linq.Enumerable.SelectArrayIterator`2.MoveNext()
   at System.Threading.Tasks.Task.WhenAll(IEnumerable`1 tasks)
   at MediatRPublishExample.CustomMediator.PublishCore(IEnumerable`1 allHandlers) in C:\Projects\MediatRPublishExample\Program.cs:line 49
   at MediatR.Internal.NotificationHandlerWrapperImpl`1.Handle(INotification notification, CancellationToken cancellationToken, ServiceFactory serviceFactory, Func`2 publish)
   at MediatR.Mediator.Publish[TNotification](TNotification notification, CancellationToken cancellationToken)
   at MediatRPublishExample.Program.Main(String[] args) in C:\Projects\MediatRPublishExample\Program.cs:line 28
   at MediatRPublishExample.Program.<Main>(String[] args)

The reason why that is, can be seen from the stacktrace.

• Publish calls NotificationHandlerWrapperImpl.Handle
  -> which calls PublishCore
  -> which calls WhenAll
  -> which calls SelectArrayIterator.MoveNext (because it converts the IEnumerable (IQueryable) into an Task[]
  -> which then calls Handle on PingMeOne, and throws

All this, are just regular method calls, no async/await magic, because there is none (PingMeOne method does not have the "async" keyword, so the compiler does not wrap it into a Task state machine).

---

Now, when you add "async" to PingMeOne.Handle, everything changes. The compiler wraps your method in a try/catch and adds some TPL magic.

See this: https://sharplab.io/#v2:CYLg1APgAgDABFAjAbgLAChYMQVjejKAZgQCY4BhOAbwznoRKgDY4ApAVwGcAXFgCgCUNOgzFQA7AmYA6CgHsAtgAcANgFMe64C3xiAvhiPpxTABzS4AQS4BPAHYBjAcNomxDAPSeEUlnKU1TW1dUXpDdH0gA===

The "JustTask" method, is just a regular method, which happens to return a Task.
"AsyncTask" has the "async" keyword, and by that, a lot of extra code is generated (you can see the try/catch there).

int Foo() { }

Foo();

Task Bar() { }

Bar();

There is litterally not a single difference between Foo and Bar. So if you throw an exception in Bar(), it will bubble up the stack like normal, until someone catches it.

Now, if you add the 'async' keyword, the compiler adds some code for you which does this "catching", wrapping it in a Task and using SetResult/SetException on that task to make it play nice with other tasks.

The thing to really understand here, is that concurrency is not parallelism.

Concurrency is about the order of execution, and breaking stuff up in smaller (to be executed later?) parts.

Parallelism is about scheduling work at the same time (with multiple threads).

The hard thing is, they are kind of intertwined. Since, when you can break your code up in smaller pieces (concurrency), you can run those smaller pieces on multiple threads (parallelism).

Here is a really good video about it: https://blog.golang.org/concurrency-is-not-parallelism

---

So if you want to "Run all your handlers at the same time", you WILL have to use Task.Run. Because that will actually start a new thread and the 'CPU code' AND 'IO waits' of all your handlers are executed at the same time, using multiple threads.

If instead, you want to run them concurrently, which means: CPU code is still executed in the order of execution, BUT in case something needs to wait for IO completion (network/disk access/etc which blocks the thread from running CPU code), the thread will be made available for other CPU instructions (your other handlers).

I hope you get this. It's all about running CPU-bound code vs waiting for IO completion.

[remcoros]

Stephen Cleary has a lot of good posts about the subject. Start with https://blog.stephencleary.com/2013/11/there-is-no-thread.html :)

[mpaul31]

@remcoros wow thank for the detailed response! everything you said made sense and i even copied the code into a separate console application and witnessed the behavior you are describing. However, I am now convinced the reason it works in my other project without the changes because it has something to do with the Azure Functions runtime. I have validated it numerous times and can't explain it otherwise. :)

I'll put together a screen video to show anyone out there to prove I am not crazy ;)

I will however update my MediatR package and update my code.

Thanks again!

[remcoros]

I believe you, hard to argue with what you observe :-). Haven't worked with Azure functions myself, so that would be interesting to see indeed. ,,It's always you, it's never the compiler'' is what I've always been told :)

[mpaul31]

ok i finally figured out what's going on! i was digging through my source code and forgot we are using Polly .NET to handle automatic retries. :)

Even though the exception that is thrown is not something that is kicking in the retry policy, the call is wrapped in the following:

public Task Handle(T notification, CancellationToken cancellationToken)
{
    return _concurrencyRetryPolicy.Execute(() => _innerHandler.Handle(notification, cancellationToken));
}

remove Polly from the equation it behaves as expected.

makes sense now! NOW i can move forward :)

thanks all!