SynchronizationContext y aplicaciones de consola

Marcos de interfaz de usuario como Windows Forms, WPF y .NET MAUI instalan un SynchronizationContext en su subproceso de interfaz de usuario. Cuando se await una tarea en esos entornos, la continuación se publica automáticamente en el subproceso de la interfaz de usuario. Las aplicaciones de consola no instalan un SynchronizationContext, lo que significa await que las continuaciones se ejecutan en el grupo de subprocesos. En este artículo se explican las consecuencias y se muestra cómo crear un bucle de mensajes de un solo subproceso cuando se necesita uno.

Comportamiento predeterminado en una aplicación de consola

En una aplicación de consola, SynchronizationContext.Current devuelve null. Cuando un método devuelve un valor en await, la continuación se ejecuta en cualquier subproceso del grupo de subprocesos que esté disponible:

static void DefaultBehaviorDemo()
{
    DemoAsync().GetAwaiter().GetResult();
}

static async Task DemoAsync()
{
    var d = new Dictionary<int, int>();
    for (int i = 0; i < 10_000; i++)
    {
        int id = Thread.CurrentThread.ManagedThreadId;
        d[id] = d.TryGetValue(id, out int count) ? count + 1 : 1;

        await Task.Yield();
    }

    foreach (var pair in d)
        Console.WriteLine(pair);
}

Salida representativa de la ejecución de este programa:

[1, 1]
[3, 2687]
[4, 2399]
[5, 2397]
[6, 2516]

El subproceso 1 (el subproceso principal) solo aparece una vez, durante la primera iteración sincrónica antes await Task.Yield() de suspender el método. Todas las iteraciones posteriores se ejecutan en hilos del grupo de subprocesos.

Puntos de entrada asincrónicos modernos

A partir de C# 7.1, puede declarar Main como async Task o async Task<int>. En C# 9 y versiones posteriores, puede usar instrucciones de nivel superior con await directamente:

// Top-level statements (C# 9+)
await DemoAsync();

// async Task Main (C# 7.1+)
static async Task Main()
{
    await DemoAsync();
}

Estos puntos de entrada no instalan un SynchronizationContext. El tiempo de ejecución genera un bootstrap que llama a tu método asincrónico y se bloquea en el valor devuelto Task, similar a cuando se llama a .GetAwaiter().GetResult(). Las continuaciones se siguen ejecutando en el grupo de subprocesos.

Cuando necesitas afinidad de hilo

En muchas aplicaciones de consola, no hay ningún problema en ejecutar continuaciones en el grupo de subprocesos. Sin embargo, algunos escenarios requieren que todas las continuaciones se ejecuten en un único subproceso:

  • Ejecución serializada: varias operaciones asincrónicas simultáneas comparten el estado sin bloqueos ejecutando sus continuaciones en el mismo subproceso.
  • Requisitos de biblioteca: algunas bibliotecas o objetos COM requieren afinidad con un subproceso determinado.
  • Pruebas unitarias: es posible que los marcos de prueba necesiten una ejecución determinista y de un solo subproceso del código asincrónico.

Compilación de synchronizationContext de un solo subproceso

Para ejecutar todas las continuaciones en un hilo, necesita dos cosas:

  1. Un SynchronizationContext cuyo método Post pone en cola trabajos en una colección segura para subprocesos.
  2. Bucle de bomba de mensajes que procesa esa cola en el subproceso de destino.

Contexto personalizado

El contexto utiliza BlockingCollection<T> para coordinar a los productores (las continuaciones asincrónicas) y a un consumidor (el bucle de bombeo):

sealed class SingleThreadSynchronizationContext : SynchronizationContext
{
    private readonly
        BlockingCollection<KeyValuePair<SendOrPostCallback, object?>> _queue = new();

    public override void Post(SendOrPostCallback d, object? state)
    {
        _queue.Add(new KeyValuePair<SendOrPostCallback, object?>(d, state));
    }

    public void RunOnCurrentThread()
    {
        while (_queue.TryTake(out KeyValuePair<SendOrPostCallback, object?> workItem,
            Timeout.Infinite))
        {
            workItem.Key(workItem.Value);
        }
    }

    public void Complete() => _queue.CompleteAdding();
}

Class SingleThreadSynchronizationContext
    Inherits SynchronizationContext

    Private ReadOnly _queue As New _
        BlockingCollection(Of KeyValuePair(Of SendOrPostCallback, Object))()

    Public Overrides Sub Post(d As SendOrPostCallback, state As Object)
        _queue.Add(New KeyValuePair(Of SendOrPostCallback, Object)(d, state))
    End Sub

    Public Sub RunOnCurrentThread()
        Dim workItem As New KeyValuePair(Of SendOrPostCallback, Object)(Nothing, Nothing)
        While _queue.TryTake(workItem, Timeout.Infinite)
            workItem.Key.Invoke(workItem.Value)
        End While
    End Sub

    Public Sub Complete()
        _queue.CompleteAdding()
    End Sub
End Class

Método AsyncPump.Run

AsyncPump.Run instala el contexto personalizado, invoca el método asincrónico y bombea continuaciones en el subproceso que realiza la llamada hasta que el método finaliza:

static class AsyncPump
{
    public static void Run(Func<Task> func)
    {
        SynchronizationContext? prevCtx = SynchronizationContext.Current;
        try
        {
            var syncCtx = new SingleThreadSynchronizationContext();
            SynchronizationContext.SetSynchronizationContext(syncCtx);

            Task t;
            try
            {
                t = func();
            }
            catch
            {
                syncCtx.Complete();
                throw;
            }

            t.ContinueWith(
                _ => syncCtx.Complete(), TaskScheduler.Default);

            syncCtx.RunOnCurrentThread();

            t.GetAwaiter().GetResult();
        }
        finally
        {
            SynchronizationContext.SetSynchronizationContext(prevCtx);
        }
    }

Class AsyncPump
    Public Shared Sub Run(func As Func(Of Task))
        Dim prevCtx As SynchronizationContext = SynchronizationContext.Current
        Try
            Dim syncCtx As New SingleThreadSynchronizationContext()
            SynchronizationContext.SetSynchronizationContext(syncCtx)

            Dim t As Task
            Try
                t = func()
            Catch
                syncCtx.Complete()
                Throw
            End Try

            t.ContinueWith(
                Sub(unused) syncCtx.Complete(), TaskScheduler.Default)

            syncCtx.RunOnCurrentThread()

            t.GetAwaiter().GetResult()
        Finally
            SynchronizationContext.SetSynchronizationContext(prevCtx)
        End Try
    End Sub

Verlo en acción

Reemplace la llamada predeterminada por AsyncPump.Run:

static void AsyncPumpDemo()
{
    AsyncPump.Run(async () =>
    {
        var d = new Dictionary<int, int>();
        for (int i = 0; i < 10_000; i++)
        {
            int id = Thread.CurrentThread.ManagedThreadId;
            d[id] = d.TryGetValue(id, out int count) ? count + 1 : 1;

            await Task.Yield();
        }

        foreach (var pair in d)
            Console.WriteLine(pair);
    });
}

Sub AsyncPumpDemo()
    AsyncPump.Run(
        Async Function() As Task
            Dim d As New Dictionary(Of Integer, Integer)()
            For i As Integer = 0 To 9999
                Dim id As Integer = Thread.CurrentThread.ManagedThreadId
                Dim count As Integer
                If d.TryGetValue(id, count) Then
                    d(id) = count + 1
                Else
                    d(id) = 1
                End If

                Await Task.Yield()
            Next

            For Each pair In d
                Console.WriteLine(pair)
            Next
        End Function)
End Sub

Salida:

[1, 10000]

El identificador de subproceso específico puede diferir en función del entorno de ejecución y la plataforma, pero el resultado clave es que las 10 000 iteraciones se ejecutan en un único subproceso: el subproceso principal.

Control de métodos asincrónicos nulos

La sobrecarga Func<Task> realiza un seguimiento de la finalización a través del Task devuelto. Los métodos asincrónicos void no devuelven una tarea; en su lugar, notifican al SynchronizationContext actual a través de OperationStarted() y OperationCompleted(). Para admitir métodos asincrónicos void , extienda el contexto para realizar un seguimiento de las operaciones pendientes:

    public static void Run(Action asyncMethod)
    {
        SynchronizationContext? prevCtx = SynchronizationContext.Current;
        try
        {
            var syncCtx = new AsyncVoidSynchronizationContext();
            SynchronizationContext.SetSynchronizationContext(syncCtx);

            Exception? caughtException = null;

            syncCtx.OperationStarted();
            try
            {
                asyncMethod();
            }
            catch (Exception ex)
            {
                caughtException = ex;
                syncCtx.Complete();
            }
            finally
            {
                syncCtx.OperationCompleted();
            }

            syncCtx.RunOnCurrentThread();

            if (caughtException is not null)
            {
                System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(caughtException).Throw();
            }
        }
        finally
        {
            SynchronizationContext.SetSynchronizationContext(prevCtx);
        }
    }
}

sealed class AsyncVoidSynchronizationContext : SynchronizationContext
{
    private readonly
        BlockingCollection<KeyValuePair<SendOrPostCallback, object?>> _queue = new();
    private int _operationCount;

    public override void Post(SendOrPostCallback d, object? state)
    {
        _queue.Add(new KeyValuePair<SendOrPostCallback, object?>(d, state));
    }

    public override void OperationStarted() =>
        Interlocked.Increment(ref _operationCount);

    public override void OperationCompleted()
    {
        if (Interlocked.Decrement(ref _operationCount) == 0)
            Complete();
    }

    public void RunOnCurrentThread()
    {
        while (_queue.TryTake(out KeyValuePair<SendOrPostCallback, object?> workItem,
            Timeout.Infinite))
        {
            workItem.Key(workItem.Value);
        }
    }

    public void Complete() => _queue.CompleteAdding();
}

    Public Shared Sub Run(asyncMethod As Action)
        Dim prevCtx As SynchronizationContext = SynchronizationContext.Current
        Try
            Dim syncCtx As New AsyncVoidSynchronizationContext()
            SynchronizationContext.SetSynchronizationContext(syncCtx)

            syncCtx.OperationStarted()
            Try
                asyncMethod()
            Catch
                syncCtx.Complete()
                Throw
            Finally
                syncCtx.OperationCompleted()
            End Try

            syncCtx.RunOnCurrentThread()
        Finally
            SynchronizationContext.SetSynchronizationContext(prevCtx)
        End Try
    End Sub
End Class

Class AsyncVoidSynchronizationContext
    Inherits SynchronizationContext

    Private ReadOnly _queue As New _
        BlockingCollection(Of KeyValuePair(Of SendOrPostCallback, Object))()
    Private _operationCount As Integer

    Public Overrides Sub Post(d As SendOrPostCallback, state As Object)
        _queue.Add(New KeyValuePair(Of SendOrPostCallback, Object)(d, state))
    End Sub

    Public Overrides Sub OperationStarted()
        Interlocked.Increment(_operationCount)
    End Sub

    Public Overrides Sub OperationCompleted()
        If Interlocked.Decrement(_operationCount) = 0 Then
            Complete()
        End If
    End Sub

    Public Sub RunOnCurrentThread()
        Dim workItem As New KeyValuePair(Of SendOrPostCallback, Object)(Nothing, Nothing)
        While _queue.TryTake(workItem, Timeout.Infinite)
            workItem.Key.Invoke(workItem.Value)
        End While
    End Sub

    Public Sub Complete()
        _queue.CompleteAdding()
    End Sub
End Class

Con el seguimiento de operaciones habilitado, la bomba sale solo cuando se completan todos los métodos void asincrónicos pendientes, no solo la tarea de nivel superior.

Consideraciones prácticas

  • Riesgo de interbloqueo: si el código se ejecuta en bloques de AsyncPump.Run de forma sincrónica (por ejemplo, llamando a .Result o .Wait() en una tarea cuya continuación debe volver a publicarse en la bomba), el subproceso de la bomba no puede procesar esa continuación. El resultado es un interbloqueo. El mismo problema se describe en Contenedores sincrónicos para métodos asincrónicos.
  • Rendimiento: una bomba de un solo hilo limita el rendimiento a un hilo. Utilice este método solo cuando la afinidad de subprocesos sea importante.
  • Multiplataforma: la implementación de AsyncPump que se muestra aquí solo usa tipos de los espacios de nombres System.Collections.Concurrent y System.Threading. Funciona en todas las plataformas que .NET admite.

Consulte también

  • Last updated on