MessagePipe 의 병렬 발행은 앞이 던지면 뒤를 시작하지 않는다

subscriber.Subscribe(async (d, ct) =>
{
    if (d.Amount < 0) throw new ArgumentException();   // 인자 검사
    await ShowHitAsync(d, ct);
});
subscriber.Subscribe(async (d, ct) => await SaveAsync(d, ct));

await publisher.PublishAsync(new Damage(-1), ct);        // IAsyncPublisher<Damage>

비동기 핸들러 둘을 병렬로 발행했는데, 첫 핸들러가 인자 검사에서 던지자 저장 핸들러는 불리지도 않았습니다. async 로 썼으니 예외가 태스크에 담겨 다른 핸들러와 상관없을 것 같았습니다.

MessagePipe 의 병렬 발행은 핸들러를 차례로 시작하다가 이미 실패로 끝난 핸들러를 만나면 거기서 멈춥니다. 늦게 실패한 경우에도 나머지를 기다리지 않고 그 예외 하나로 끝납니다.

병렬 발행은 핸들러를 차례로 시작하다 즉시 실패에서 멈춘다

PublishAsync 를 전략 없이 부르면 옵션의 기본값 AsyncPublishStrategy.Parallel 로 돕니다. 병렬은 스레드를 여러 개 쓴다는 뜻이 아니라, 기다리지 않고 다음 핸들러를 시작한다는 뜻입니다. 그 시작을 전용 awaiter 의 생성자가 합니다.

// Runtime/Internal/AsyncHandlerWhenAll.cs (줄임)
foreach (var item in handlers)
{
    try
    {
        var awaiter = item.HandleAsync(message, ct).GetAwaiter();
        if (awaiter.IsCompleted) awaiter.GetResult();            // 이미 끝났으면 결과를 바로 꺼낸다
        else AwaiterNode.RegisterUnsafeOnCompleted(this, awaiter);
    }
    catch (Exception ex)
    {
        exception = ExceptionDispatchInfo.Capture(ex);
        TryInvokeContinuation();
        return;                                                  // 나머지 핸들러를 시작하지 않는다
    }
}

async 메서드가 첫 await 전에 던지면 돌려받은 태스크는 이미 실패로 끝나 있습니다. 생성자는 IsCompleted 를 보자마자 GetResult() 로 열어 보고, 그 자리에서 다시 던져져 catch 의 return 으로 빠집니다.

구독 순서 A, B, C 에서 A 는 50 ms 를 기다리고 B 만 던지게 한 결과입니다. .NET, Mono, IL2CPP 가 같았습니다.

늦게 실패하면 나머지를 기다리지 않는다

첫 await 뒤에 던지면 C 는 불리지만, PublishAsync 가 모든 핸들러를 기다렸다 예외를 모아 주지는 않습니다. B 가 던지는 순간 등록해 둔 노드가 예외를 기록하고 기다리던 쪽을 깨웁니다. A 가 아직 돌고 있어도 PublishAsync 는 B 의 예외로 끝나고, A 는 뒤에서 혼자 마저 돕니다.

두 핸들러가 둘 다 await 뒤에 거의 동시에 던지게 해 500번 돌렸습니다.

예외 필드 하나를 서로 덮어쓰는 구조라 AggregateException 으로 묶이지 않고 하나만, 어느 쪽인지 정해지지 않은 채 옵니다.

병렬 발행은 발행마다 할당한다

생성자에서 핸들러를 시작하는 그 awaiter(AsyncHandlerWhenAll<T>)를 발행마다 새로 만듭니다. 기다리는 핸들러마다 붙는 노드는 정적 풀에서 꺼내 씁니다.

핸들러가 실제로 기다리면 Parallel 은 .NET 342 B, Mono 336 B, IL2CPP 472 B 부터입니다(유니티는 플레이어 빌드). 대신 기다림이 겹칩니다. 20 ms 를 기다리는 핸들러 넷을 Parallel 은 2228 ms, Sequential 은 80117 ms 에 끝냈습니다.

주의할 점

1. 비동기 핸들러는 예외를 스스로 잡는다

안 지키면 생기는 문제. 인자 검사처럼 메서드 첫머리에서 던지는 핸들러 하나 때문에 저장이나 UI 갱신 같은 뒤 핸들러가 통째로 빠집니다. async 로 써도 첫 await 전이면 같고, 발행하는 쪽에서는 막을 방법이 없습니다.

이렇게 쓴다. 비동기 핸들러는 본문 전체를 try/catch 로 감싸 실패를 그 안에서 끝냅니다.

2. PublishAsync 가 끝났다고 모든 핸들러가 끝난 것은 아니다

안 지키면 생기는 문제. 하나가 실패하면 다른 핸들러가 도는 중에도 PublishAsync 가 끝나고, catch 에는 여러 실패 중 하나만 옵니다. “await 가 끝났으니 저장도 끝났다” 고 보고 다음 단계로 넘어가면 저장이 도는 도중에 씬이 바뀝니다.

이렇게 쓴다. 실패 기록은 각 핸들러가 잡아서 남기고, 발행 쪽 catch 는 마지막 안전망으로만 둡니다.

3. 매 프레임 쏘는 메시지는 Parallel 로 보내지 않는다

안 지키면 생기는 문제. Parallel 은 핸들러가 바로 끝나도 발행마다 40 B, 한 번이라도 기다리면 340~470 B 를 씁니다. 매 프레임 쏘면 프레임마다 쌓여 GC 를 부릅니다.

이렇게 쓴다. 자주 쏘는 메시지는 동기 IPublisher 나 할당이 0 인 Sequential 로 보냅니다. 기본값은 o.DefaultAsyncPublishStrategy = AsyncPublishStrategy.Sequential 로, 한 번만 다르게 할 때는 PublishAsync 의 전략 인자로 바꿉니다.

정리

MessagePipe 의 병렬 발행은 앞 핸들러가 곧바로 던지면 뒤 핸들러를 시작하지 않습니다.

• 핸들러는 awaiter 의 생성자에서 차례로 시작되고, 이미 실패한 태스크를 만나면 거기서 멈춥니다. async 여도 첫 await 전이면 같습니다.
• 늦게 실패하면 나머지를 기다리지 않고 예외 하나로 끝나며, 어느 예외가 올지는 정해지지 않습니다.
• Parallel 은 발행마다 40 B 부터 할당하고, Sequential 은 0 B 지만 기다림이 쌓입니다.