🤽🏽 🏷️ 🎯 System.Threading.Channels: productor-consumidor de alto rendimiento y asíncrono sin asignación y inmersión en pila 🚶 🚙 🙋🏻

Hola de nuevo. Hace algún tiempo, escribí sobre otra herramienta poco conocida para los amantes del alto rendimiento: System.IO.Pipelines . En esencia, el sistema considerado.Cubos.Canales (en adelante, "canales") se basa en principios similares a los de las tuberías, resolviendo el mismo problema: productor-consumidor. Sin embargo, a veces tiene una API más simple que se adapta con elegancia a cualquier tipo de código empresarial. Al mismo tiempo, usa la asincronía sin asignaciones y sin apilamiento, ¡incluso en el caso asincrónico! (No siempre, pero a menudo).

Tabla de contenido

Introducción

El problema Productor / Consumidor se encuentra en el camino de los programadores con bastante frecuencia y durante más de una docena de años. El propio Edsger Dijkstra participó en la resolución de este problema: se le ocurrió la idea de utilizar semáforos para sincronizar hilos al organizar el trabajo en función de productor / consumidor. Y aunque su solución en su forma más simple es conocida y bastante trivial, en el mundo real este patrón (Productor / Consumidor) puede ocurrir en una forma mucho más complicada. Además, los estándares de programación modernos dejan sus marcas, el código está escrito de una manera más simplificada y se desglosa para su posterior reutilización. Todo se hace para reducir el umbral para escribir código de calidad y simplificar este proceso. Y el espacio de nombres en cuestión, System.Threading.Channels, es el siguiente paso hacia este objetivo.

Estaba mirando System.IO.Pipelines hace un tiempo. Se requirió un trabajo más atento y una comprensión profunda del asunto, se utilizaron Span y Memory, y para un trabajo eficiente se requirió no llamar a métodos obvios (para evitar asignaciones de memoria innecesarias) y pensar constantemente en bytes. Debido a esto, la interfaz de programación de Pipeline no era trivial ni intuitiva.

System.Threading.Channels presenta al usuario una API mucho más simple para trabajar. Vale la pena mencionar que, a pesar de la simplicidad de la API, esta herramienta está altamente optimizada y lo más probable es que no asigne memoria durante su trabajo. Quizás esto se deba al hecho de que ValueTask se usa debajo del capó en todas partes , e incluso en el caso de una asincronía real, se usa IValueTaskSource, que se reutiliza para otras operaciones. Este es precisamente todo el interés de la implementación de los canales.

Los canales son genéricos, el tipo genérico es, como puede suponer, el tipo que se producirá y consumirá. Curiosamente, la implementación de la clase Channel, que cabe en 1 línea (fuente github ):

namespace System.Threading.Channels
{
    public abstract class Channel<T> : Channel<T, T> { }
}

Por lo tanto, la clase principal de canales está parametrizada por 2 tipos, por separado para el canal productor y el canal consumidor. Pero para canales realizados esto no se usa.

Para aquellos familiarizados con las tuberías, el enfoque general para comenzar les parecerá familiar. A saber. Creamos 1 clase central de la que extraemos por separado los productores ( ChannelWriter ) y los consumidores ( ChannelReader ). A pesar de los nombres, vale la pena recordar que este es exactamente el productor / consumidor, y no el lector / escritor de otra tarea clásica de subprocesamiento múltiple del mismo nombre. ChannelReader cambia el estado del canal general (extrae el valor), que ya no está disponible. Esto significa que prefiere no leer, sino que consume. Pero nos familiarizaremos con la implementación más adelante.

Comienzo del trabajo. Canal

Comenzar con los canales comienza con una clase abstracta Channel <T> y una clase de canal estático que crea la implementación más adecuada. Además, desde este canal común, puede obtener un ChannelWriter para escribir en el canal y un ChannelReader para el consumo del canal. Un canal es un depósito de información general para ChannelWriter y ChannelReader, de modo que es donde se almacenan todos los datos. Y la lógica de su grabación o consumo ya está dispersa en ChannelWriter y ChannelReader. Convencionalmente, los canales se pueden dividir en 2 grupos: ilimitados y limitados. Los primeros son más simples en la implementación, puede escribir en ellos sin límite (siempre que la memoria lo permita). Los segundos están limitados por un cierto valor máximo del número de registros.

Aquí es donde la naturaleza de la asincronía es ligeramente diferente. En canales no acotados, la operación de escritura siempre se completará sincrónicamente, no hay nada que deje de escribir en el canal. Para canales limitados, la situación es diferente. Con el comportamiento estándar (que se puede anular), la operación de escritura se completará sincrónicamente siempre que haya espacio en el canal para nuevas instancias. Tan pronto como el canal esté lleno, la operación de escritura no finalizará hasta que haya espacio libre (después de que el consumidor haya consumido el consumido). Por lo tanto, aquí la operación será realmente asíncrona con cambios de subprocesos y cambios asociados (o sin cambios, que se describirán un poco más adelante).

En su mayor parte, el comportamiento de los lectores es el mismo: si hay algo en el canal, entonces el lector simplemente lo lee y termina sincronizado. Si no hay nada, entonces espera a que alguien escriba algo.

La clase estática Channel contiene 4 métodos para crear los canales anteriores:

Channel<T> CreateUnbounded<T>();
Channel<T> CreateUnbounded<T>(UnboundedChannelOptions options);
Channel<T> CreateBounded<T>(int capacity);
Channel<T> CreateBounded<T>(BoundedChannelOptions options);

Si lo desea, puede especificar opciones más precisas para crear un canal, lo que ayudará a optimizarlo para las necesidades especificadas.

UnboundedChannelOptions contiene 3 propiedades, que se establecen en falso de forma predeterminada:

AllowSynchronousContinuations — , , . -. , . , , , . , , , . , - - , ;
SingleReader — , . , ;
SingleWriter — , ;

BoundedChannelOptions contiene las mismas 3 propiedades y 2 más en la parte superior

AllowSynchronousContinuations - lo mismo;
SingleReader es lo mismo;
SingleWriter es lo mismo;
Capacidad: el número de grabaciones para caber en el canal. Este parámetro también es un parámetro constructor;
FullMode: la enumeración BoundedChannelFullMode, que tiene 4 opciones, determina el comportamiento al intentar escribir en un canal completo:
- Esperar: espera espacio libre para completar la operación asincrónica
- DropNewest: el elemento que se está escribiendo sobrescribe el más nuevo existente, finaliza sincrónicamente
- DropOldest: el elemento que se está escribiendo sobrescribe el elemento existente más antiguo que finaliza sincrónicamente
- DropWrite: el elemento que se está escribiendo no se escribe, termina sincrónicamente

Dependiendo de los parámetros pasados y del método llamado, se creará una de las 3 implementaciones: SingleConsumerUnboundedChannel , UnboundedChannel , BoundedChannel . Pero esto no es tan importante, porque usaremos el canal a través de la clase base Channel <TWrite, TRead>.

Tiene 2 propiedades:

ChannelReader <TRead> Reader {get; conjunto protegido; }
ChannelWriter <TWrite> Writer {get; conjunto protegido; }

Y también, 2 operadores de conversión de tipo implícito a ChannelReader <TRead> y ChannelWriter <TWrite>.

Un ejemplo de cómo empezar con los canales:

Channel<int> channel = Channel.CreateUnbounded<int>();
//  
ChannelWriter<int> writer = channel.Writer;
ChannelReader<int> reader = channel.Reader; 
// 
ChannelWriter<int> writer = channel;
ChannelReader<int> reader = channel;

Los datos se almacenan en una cola. Para 3 tipos, se utilizan 3 colas diferentes: ConcurrentQueue <T>, Deque <T> y SingleProducerSingleConsumerQueue <T>. En este punto, me pareció que estaba desactualizado y me perdí un montón de nuevas colecciones simples. Pero me apresuro a decepcionar, no son para todos. Marcado interno, así que usarlos no funcionará. Pero si de repente los necesita en el producto, puede encontrarlos aquí (SingleProducerConsumerQueue) y aquí (Deque) . La implementación de este último es muy simple. Te aconsejo que lo leas, puedes estudiarlo muy rápido.

Entonces, comencemos a estudiar directamente ChannelReader y ChannelWriter, así como detalles de implementación interesantes. Todos se reducen a asincrónico, sin asignación de memoria usando IValueTaskSource.

ChannelReader - consumidor

Cuando se solicita un objeto de consumidor, se devuelve una de las implementaciones de la clase abstracta ChannelReader <T>. Nuevamente, a diferencia de API Pipelines, es simple y hay pocos métodos. Solo necesita conocer la lista de métodos para comprender cómo usarlo en la práctica.

Métodos:

Propiedad de solo obtención virtual Finalización de tarea {get; } Un

objeto de tipo Tarea que se completa cuando el canal está cerrado;
Propiedad virtual de solo obtención int Count {get; }

Aquí debe tenerse en cuenta que se devuelve el número actual de objetos disponibles para la lectura;
Propiedad virtual de solo obtención bool CanCount {get; }

Indica si la propiedad Count está disponible;
bool TryRead(out T item)

. bool, , . out ( null, );
ValueTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default)

ValueTask true, , . ValueTask false, ( );
ValueTask<T> ReadAsync(CancellationToken cancellationToken = default)

. , . .

, TryRead WaitToReadAsync. ( cancelation tokens), — TryRead. , while(true) WaitToReadAsync. true, , TryRead. TryRead , , . — , WaitToReadAsync, , , .

, , - .

ChannelWriter - productor

Todo es similar al consumidor, así que echemos un vistazo a los métodos de inmediato:

Método virtual bool TryComplete (¿Excepción? Error = nulo)

Intenta marcar el canal como completo, es decir demuestre que no se le escribirán más datos. La excepción que causó la terminación del canal se puede pasar como un parámetro opcional. Devuelve verdadero si se completó con éxito, de lo contrario falso (si el canal ya se ha completado o no admite la terminación);
Método abstracto bool TryWrite (elemento T)

Intenta escribir un valor en el canal. Devuelve verdadero si tiene éxito y falso si no.
Método abstracto ValueTask <bool> WaitToWriteAsync (CancellationToken cancellationToken = default)

Devuelve una ValueTask con un valor verdadero que se completará cuando haya espacio para escribir en el canal. El valor será falso si ya no se permiten las escrituras en el canal;
Método virtual ValueTask WriteAsync (elemento T, CancellationToken cancellationToken = default)

Escribe de forma asíncrona en el canal. Por ejemplo, si el canal está lleno, la operación será realmente asíncrona y completa solo después de liberar espacio para este registro;
Método void Complete (Exception? Error = null)

Solo intenta marcar el canal como completo con TryComplete, y si falla, lanza una excepción.

Un pequeño ejemplo de lo anterior (para comenzar fácilmente sus propios experimentos):

Channel<int> unboundedChannel = Channel.CreateUnbounded<int>();

//      ,        
ChannelWriter<int> writer = unboundedChannel;
ChannelReader<int> reader = unboundedChannel;

//     
int objectToWriteInChannel = 555;
await writer.WriteAsync(objectToWriteInChannel);
//  ,     ,   ,  
writer.Complete();

//         
int valueFromChannel = await reader.ReadAsync();

Ahora pasemos a la parte más interesante.

Asincronía sin asignaciones

En el proceso de escribir y estudiar el código, me di cuenta de que no hay casi nada interesante en la implementación de todas estas operaciones. En general, puede describirlo de esta manera: evite bloqueos innecesarios utilizando colecciones competitivas y el uso abundante de ValueTask, que es una estructura que ahorra memoria. Sin embargo, me apresuro a recordarle que no vale la pena reemplazarlo rápidamente para revisar todos los archivos de su PC y reemplazar todas las Tareas con ValueTask. Tiene sentido solo en casos donde la operación en la mayoría de los casos termina sincrónicamente. Después de todo, como recordamos, con la asincronía, es muy probable un cambio de flujo, lo que significa que la pila no será la misma que antes. De todos modos, un verdadero profesional en el campo de la productividad sabe: no optimice antes de que surjan problemas.

Una cosa es buena, no me escribiré como profesional y, por lo tanto, es hora de descubrir cuál es el secreto de escribir código asincrónico sin asignar memoria, lo que a primera vista parece demasiado bueno para la verdad. Pero también sucede.

Interfaz IValueTaskSource

Comencemos nuestro viaje desde el principio: la estructura ValueTask , que se agregó en .net core 2.0 y se modificó en 2.1. Dentro de esta estructura, hay un objeto complicado campo _obj. Es fácil adivinar, basándose en el nombre que habla, que una de las 3 cosas puede ocultarse en este campo: nulo, Tarea / Tarea <T> o IValueTaskSource. De hecho, se deduce de la forma en que se crea ValueTask.

Como asegura el fabricante, esta estructura solo debe usarse obviamente, con la palabra clave wait. Es decir, no debe aplicar esperar muchas veces al mismo ValueTask, usar combinadores, agregar varias continuaciones, etc. Además, no debe obtener el resultado de ValueTask más de una vez. Y esto se debe al hecho de que estamos tratando de entender: la reutilización de todo esto sin asignar memoria.

Ya mencioné la interfaz IValueTaskSource . Es él quien ayuda a salvar la memoria. Esto se hace reutilizando el IValueTaskSource en varias ocasiones para muchas tareas. Pero precisamente por esta reutilización, no hay forma de disfrutar de ValueTask.

Entonces IValueTaskSource. Esta interfaz tiene 3 métodos, implementando los cuales ahorrará con éxito memoria y tiempo en la asignación de esos bytes atesorados.

GetResult : se llama una vez, cuando la máquina de estado , formada en tiempo de ejecución para métodos asincrónicos, necesita un resultado. ValueTask tiene un método GetResult, que llama al método de interfaz del mismo nombre, que, como recordamos, puede almacenarse en el campo _obj.
GetStatus : llamado por la máquina de estado para determinar el estado de la operación. También a través de ValueTask.
OnCompleted : nuevamente, la máquina de estado lo llama para agregar una continuación a la tarea que no se completó en ese momento.

Pero a pesar de la interfaz simple, la implementación requerirá algo de habilidad. Y aquí podemos recordar con qué comenzamos: canales . Esta implementación usa la clase AsyncOperationque es una implementación de IValueTaskSource. Esta clase está oculta detrás del modificador de acceso interno. Pero esto no interfiere con la comprensión de los mecanismos básicos. Esto plantea la pregunta, ¿por qué no dar la implementación de IValueTaskSource a las masas? La primera razón (por diversión) es cuando hay un martillo en la mano, las uñas están en todas partes, cuando una implementación IValueTaskSource está en las manos, hay un trabajo analfabeto con memoria en todas partes. La segunda razón (más plausible) es que, si bien la interfaz es simple y universal, la implementación real es óptima cuando se utilizan ciertos matices de aplicación. Y probablemente por esta razón, es posible encontrar implementaciones en varias partes de la gran y poderosa .net, como AsyncOperation bajo el capó de los canales, AsyncIOOperation dentro de la nueva API de socket, etc.

Sin embargo, para ser justos, todavía hay una implementación común:ManualResetValueTaskSourceCore . Pero esto ya está demasiado lejos del tema del artículo.

CompareExchange

Un método bastante popular de una clase popular que evita la sobrecarga de las primitivas de sincronización clásicas. Creo que la mayoría está familiarizada con él, pero aún vale la pena describirlo en 3 palabras, porque esta construcción se usa con bastante frecuencia en AsyncOperation.

En la literatura convencional, esta función se llama comparar e intercambiar (CAS). En .net está disponible en la clase Interlocked .

La firma es la siguiente:

public static T CompareExchange<T>(ref T location1, T value, T comparand) where T : class;

También hay sobrecargas con int, long, float, double, IntPtr, object.

El método en sí mismo es atómico, es decir, se ejecuta sin interrupción. Compara 2 valores y, si son iguales, asigna el nuevo valor a la variable. Resuelven el problema cuando necesita verificar el valor de una variable y cambiar la variable dependiendo de ella.

Suponga que desea incrementar una variable si su valor es menor que 10.

Luego vienen 2 hilos.

Stream 1	Stream 2
Comprueba el valor de una variable para alguna condición (es decir, si es menor que 10) que se activa	-
Entre verificar y cambiar un valor	Asigna a una variable un valor que no cumple una condición (por ejemplo, 15)
Cambia el valor, aunque no debería, porque la condición ya no se cumple	-

Al usar este método, puede cambiar exactamente el valor que desea o no cambiar, mientras recibe el valor real de la variable.

location1 es una variable cuyo valor queremos cambiar. Se compara con comparand, en caso de igualdad, el valor1 se escribe en location1. Si la operación tiene éxito, el método devolverá el valor pasado de la variable location1. De lo contrario, se devolverá el valor real de location1.

Más profundamente, hay una instrucción en lenguaje ensamblador, cmpxchg, que hace esto. Es ella quien se usa debajo del capó.

Pila de buceo

Mientras miraba todo este código, encontré referencias a "Stack Dive" más de una vez. Esta es una cosa muy interesante e interesante que en realidad es muy indeseable. La conclusión es que con la ejecución sincrónica de continuaciones, podemos agotar los recursos de la pila.

Digamos que tenemos 10,000 tareas, con estilo.

//code1
await ...
//code2

Supongamos que la primera tarea completa la ejecución y, por lo tanto, libera la continuación de la segunda, que inmediatamente comenzamos a ejecutar sincrónicamente en este hilo, es decir, tomar un pedazo de la pila con el marco de la continuación dada. A su vez, esta continuación desbloquea la continuación de la tercera tarea, que también comenzamos a realizar de inmediato. Etc. Si no hay más esperas en la secuela o algo que de alguna manera deje caer la pila, simplemente consumiremos el espacio de la pila hasta el final. Lo que podría causar StackOverflow y el bloqueo de la aplicación. En la revisión del código, mencionaré cómo AsyncOperation combate esto.

AsyncOperation como una implementación IValueTaskSource

El código fuente .

Dentro de AsyncOperation, hay un campo _continuación de tipo Acción <objeto>. El campo se usa para, créalo o no, las continuaciones. Pero, como suele ser el caso en el código que es demasiado moderno, los campos tienen responsabilidades adicionales (como un recolector de basura y el último bit en un enlace de tabla de método). El campo _continuación es de la misma serie. Hay 2 valores especiales que se pueden almacenar en este campo, excepto la continuación en sí y nulo. s_availableSentinel y s_completedSentinel . Estos campos indican que la operación está disponible y se completa en consecuencia. Se puede acceder solo para su reutilización para una operación completamente asíncrona.

AsyncOperation también implementa IThreadPoolWorkItemcon un solo método: void Execute () => SetCompletionAndInvokeContinuation (). El método SetCompletionAndInvokeContinuation realiza la continuación. Y este método se llama directamente en el código AsyncOperation, o mediante el mencionado Ejecutar. Después de todo, los tipos que implementan IThreadPoolWorkItem pueden arrojarse al grupo de subprocesos de alguna manera como este ThreadPool.UnsafeQueueUserWorkItem (this, preferLocal: false).

El grupo de subprocesos ejecutará el método Execute.

La ejecución de la continuación en sí es bastante trivial.

La continuación de _continuación se copia en una variable local, s_completedSentinel se escribe en su lugar- un objeto títere artificial (o un centinela, no sé cómo hablarme en nuestro discurso), lo que indica que la tarea se ha completado. Bueno, entonces la copia local de la continuación real simplemente se ejecuta. Con un ExecutionContext, estas acciones se publican en el contexto. No hay secreto aquí. La clase puede invocar este código directamente, simplemente llamando al método que encapsula estas acciones o a través de la interfaz IThreadPoolWorkItem en el grupo de subprocesos. Ahora puede adivinar cómo funciona la función con ejecución de continuación sincrónicamente.

El primer método de la interfaz IValueTaskSource es GetResult ( github ).

Es simple, él:

_currentId.

_currentId — , . . ;
_continuation - s_availableSentinel. , , AsyncOperation . , (pooled = true);
_result.

_result TrySetResult .

Método TrySetResult ( github ).

El método es trivial. - almacena el parámetro recibido en _result y completa la señal, es decir, llama al método SignalCompleteion , que es bastante interesante.

Método SignalCompletion ( github ).

Este método usa todo lo que hablamos al principio.

Al principio, si _continuación == nulo, escribimos la marioneta s_completedSentinel.

Además, el método se puede dividir en 4 bloques. Te lo diré de inmediato para facilitar la comprensión del circuito, el bloque 4 es solo una ejecución sincrónica de la continuación. Es decir, la ejecución trivial de la continuación a través del método, como describí en el párrafo sobre IThreadPoolWorkItem.

_schedulingContext == null, .. ( if).

_runContinuationsAsynchronously == true, , — ( if).

IThreadPoolWorkItem . AsyncOperation . .

, if ( , ), , 2 3 , — .. 4 ;
_schedulingContext is SynchronizationContext, ( if).

_runContinuationsAsynchronously = true. . , , . , . 2 , :
```
sc.Post(s => ((AsyncOperation<TResult>)s).SetCompletionAndInvokeContinuation(), this);
```
. , , ( , ), 4 — ;
, 2 . .

, _schedulingContext TaskScheduler, . , 2, .. _runContinuationsAsynchronously = true TaskScheduler . , Task.Factory.StartNew . .
— . , .

El segundo método de la interfaz IValueTaskSource es GetStatus ( github )

Al igual que una dona de Petersburgo.

Si _continuation! = _CompletedSentinel, a continuación, volver ValueTaskSourceStatus.Pending

Si el error == null, a continuación, volver ValueTaskSourceStatus.Succeeded

Si _error.SourceException es OperationCanceledException, a continuación, volver ValueTaskSourceStatus.Canceled

Bueno, ya que gran parte llegó hasta aquí, volver ValueTaskSourceStatus.Faulted

tercera y última , pero el método más complejo de la interfaz IValueTaskSource es OnCompleted ( github ).

El método agrega una continuación que se ejecuta al finalizar.

Captura ExecutionContext y SynchronizationContext según sea necesario.

A continuación, Interlocked.CompareExchange , descrito anteriormente, se usa para guardar la continuación en el campo, comparándola con nulo. Les recuerdo que CompareExchange devuelve el valor actual de la variable.

Si se ha guardado la continuación, el valor que estaba en la variable antes de que se devuelva la actualización, es decir, nulo. Esto significa que la operación aún no se ha completado cuando se escribe la continuación. Y el que lo complete él mismo lo descubrirá (como vimos anteriormente). Y no tiene sentido que realicemos acciones adicionales. Y esto completa el trabajo del método.

Si el valor no se guardó, es decir, se devolvió algo distinto de nulo de CompareExchange. En este caso, alguien logró poner valor más rápido que nosotros. Es decir, ocurrió una de 2 situaciones: o la tarea se completó más rápido de lo que llegamos aquí, o hubo un intento de escribir más de 1 continuación, lo que no se puede hacer.

Por lo tanto, verificamos el valor devuelto, si es igual a s_completedSentinel: sería exactamente lo que se escribiría en caso de finalización.

Si esto no es s_completedSentinel , entonces no nos utilizaron de acuerdo con el plan; intentaron agregar más de una continuación. Es decir, el que ya se ha escrito y el que estamos escribiendo. Y esta es una situación excepcional;
s_completedSentinel, , , . , _runContinuationsAsynchronously = false.

, , OnCompleted, awaiter'. . , AsyncOperation — System.Threading.Channels. , . , . , , ( ) . , awaiter' , , . awaiter'.

Para evitar esta situación, debe ejecutar la continuación de forma asincrónica pase lo que pase. Se ejecuta de acuerdo con los mismos esquemas que los primeros 3 bloques en el método SignalCompleteion, solo en un grupo, en un contexto o a través de una fábrica y un planificador

Y aquí hay un ejemplo de continuaciones sincrónicas:

class Program
    {
        static async Task Main(string[] args)
        {
            Channel<int> unboundedChannel = Channel.CreateUnbounded<int>(new UnboundedChannelOptions
            {
                AllowSynchronousContinuations = true
            });

            ChannelWriter<int> writer = unboundedChannel;
            ChannelReader<int> reader = unboundedChannel;

            Console.WriteLine($"Main, before await. Thread id: {Thread.CurrentThread.ManagedThreadId}");

            var writerTask = Task.Run(async () =>
            {
                Thread.Sleep(500);
                int objectToWriteInChannel = 555;
                Console.WriteLine($"Created thread for writing with delay, before await write. Thread id: {Thread.CurrentThread.ManagedThreadId}");
                await writer.WriteAsync(objectToWriteInChannel);
                Console.WriteLine($"Created thread for writing with delay, after await write. Thread id: {Thread.CurrentThread.ManagedThreadId}");
            });

            //Blocked here because there are no items in channel
            int valueFromChannel = await reader.ReadAsync();
            Console.WriteLine($"Main, after await (will be processed by created thread for writing). Thread id: {Thread.CurrentThread.ManagedThreadId}");

            await writerTask;

            Console.Read();
        }
    }

Salida:

Principal, antes de esperar. Identificación del hilo: 1

hilo creado para escribir con retraso, antes de esperar a escribir. Identificación del hilo: 4

Principal, después de esperar (será procesado por el hilo creado para escritura). Identificación del hilo: 4

Hilo creado para escribir con retraso, después de esperar la escritura. Identificación del hilo: 4

System.Threading.Channels: productor-consumidor de alto rendimiento y asíncrono sin asignación y inmersión en pila