c# - net - ¿Cuál es el equivalente async/await de un servidor ThreadPool?

run method async c#

Question

Estoy trabajando en un servidor TCP que se parece a esto usando apis síncronas y el conjunto de hilos:

TcpListener listener;
void Serve(){
  while(true){
    var client = listener.AcceptTcpClient();
    ThreadPool.QueueUserWorkItem(this.HandleConnection, client);
    //Or alternatively new Thread(HandleConnection).Start(client)
  }
}

Suponiendo que mi objetivo es manejar tantas conexiones simultáneas como sea posible con el menor uso de recursos, esto parece que se limitará rápidamente por el número de subprocesos disponibles. Sospecho que al usar Apis de tareas no bloqueantes, podré manejar mucho más con menos recursos.

Mi impresión inicial es algo como:

async Task Serve(){
  while(true){
    var client = await listener.AcceptTcpClientAsync();
    HandleConnectionAsync(client); //fire and forget?
  }
}

Pero me parece que esto podría causar cuellos de botella. Tal vez HandleConnectionAsync tardará un tiempo inusualmente largo en alcanzar la primera espera, y detendrá el ciclo de aceptación principal. ¿Solo utilizará un hilo, o el tiempo de ejecución ejecutará mágicamente las cosas en varios hilos como crea conveniente?

¿Hay alguna manera de combinar estos dos enfoques para que mi servidor use exactamente el número de subprocesos que necesita para el número de tareas que se ejecutan activamente, pero para que no bloquee los subprocesos innecesariamente en las operaciones de IO?

¿Existe una forma idiomática de maximizar el rendimiento en una situación como esta?

Answer 1

¿Hay alguna razón por la que necesite aceptar conexiones asíncronas? Quiero decir, ¿la espera de alguna conexión de cliente te da algún valor? La única razón para hacerlo sería porque hay otros trabajos en curso en el servidor mientras se espera una conexión. Si hay, probablemente podrías hacer algo como esto:

    public async void Serve()
    {
        while (true)
        {
            var client = await _listener.AcceptTcpClientAsync();
            Task.Factory.StartNew(() => HandleClient(client), TaskCreationOptions.LongRunning);
        }
    }

De esta manera, la aceptación liberará la opción de salida del hilo actual para que se realicen otras cosas, y el manejo se ejecuta en un nuevo hilo. La única sobrecarga sería generar un nuevo hilo para el manejo del cliente antes de que vuelva a aceptar una nueva conexión.

Edit: Me di cuenta de que es casi el mismo código que escribiste. Creo que necesito leer tu pregunta otra vez para entender mejor lo que realmente estás preguntando: S

Edit2:

¿Hay alguna manera de combinar estos dos enfoques para que mi servidor use exactamente el número de subprocesos que necesita para el número de tareas que se ejecutan activamente, pero para que no bloquee los subprocesos innecesariamente en las operaciones de E / S?

Creo que mi solución realmente responde a esta pregunta. ¿Pero es realmente necesario?

Edit3: Made Task.Factory.StartNew () crea un nuevo hilo.

Answer 2

De acuerdo con el http://msdn.microsoft.com/en-AU/library/hh524395.aspx#BKMK_VoidReturnType Microsoft, el tipo void return no se debe usar porque no puede detectar excepciones. Como ha señalado, necesita tareas de "disparar y olvidar", por lo que mi conclusión es que siempre debe devolver la tarea (como ha dicho Microsoft), pero debería detectar el error utilizando:

TaskInstance.ContinueWith(i => { /* exception handler */ }, TaskContinuationOptions.OnlyOnFaulted);

Un ejemplo que usé como prueba está abajo:

public static void Main()
{
    Awaitable()
        .ContinueWith(
            i =>
                {
                    foreach (var exception in i.Exception.InnerExceptions)
                    {
                        Console.WriteLine(exception.Message);
                    }
                },
            TaskContinuationOptions.OnlyOnFaulted);
    Console.WriteLine("This needs to come out before my exception");
    Console.ReadLine();
}

public static async Task Awaitable()
{
    await Task.Delay(3000);
    throw new Exception("Hey I can catch these pesky things");
}

Answer 3

Dejaría que Framework administre los hilos y no crearía hilos adicionales, a menos que las pruebas de creación de perfiles sugieran que podría ser necesario. Especialmente, si las llamadas dentro de HandleConnectionAsync están principalmente vinculadas a IO.

De todos modos, si desea liberar el subproceso de llamada (el despachador) al comienzo de HandleConnectionAsync , hay una solución muy fácil. Puedes saltar a un nuevo hilo desde ThreadPool con await Yield() . Eso funciona si su servidor se ejecuta en el entorno de ejecución que no tiene ningún contexto de sincronización instalado en el subproceso inicial (una aplicación de consola, un servicio WCF), que normalmente es el caso de un servidor TCP.

[EDITADO] Lo siguiente ilustra esto (el código es originalmente de here ). Tenga en cuenta que el bucle while principal no crea ningún subproceso explícitamente:

using System;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;

class Program
{
    object _lock = new Object(); // sync lock 
    List<Task> _connections = new List<Task>(); // pending connections

    // The core server task
    private async Task StartListener()
    {
        var tcpListener = TcpListener.Create(8000);
        tcpListener.Start();
        while (true)
        {
            var tcpClient = await tcpListener.AcceptTcpClientAsync();
            Console.WriteLine("[Server] Client has connected");
            var task = StartHandleConnectionAsync(tcpClient);
            // if already faulted, re-throw any error on the calling context
            if (task.IsFaulted)
                task.Wait();
        }
    }

    // Register and handle the connection
    private async Task StartHandleConnectionAsync(TcpClient tcpClient)
    {
        // start the new connection task
        var connectionTask = HandleConnectionAsync(tcpClient);

        // add it to the list of pending task 
        lock (_lock)
            _connections.Add(connectionTask);

        // catch all errors of HandleConnectionAsync
        try
        {
            await connectionTask;
            // we may be on another thread after "await"
        }
        catch (Exception ex)
        {
            // log the error
            Console.WriteLine(ex.ToString());
        }
        finally
        {
            // remove pending task
            lock (_lock)
                _connections.Remove(connectionTask);
        }
    }

    // Handle new connection
    private async Task HandleConnectionAsync(TcpClient tcpClient)
    {
        await Task.Yield();
        // continue asynchronously on another threads

        using (var networkStream = tcpClient.GetStream())
        {
            var buffer = new byte[4096];
            Console.WriteLine("[Server] Reading from client");
            var byteCount = await networkStream.ReadAsync(buffer, 0, buffer.Length);
            var request = Encoding.UTF8.GetString(buffer, 0, byteCount);
            Console.WriteLine("[Server] Client wrote {0}", request);
            var serverResponseBytes = Encoding.UTF8.GetBytes("Hello from server");
            await networkStream.WriteAsync(serverResponseBytes, 0, serverResponseBytes.Length);
            Console.WriteLine("[Server] Response has been written");
        }
    }

    // The entry point of the console app
    static void Main(string[] args)
    {
        Console.WriteLine("Hit Ctrl-C to exit.");
        new Program().StartListener().Wait();
    }
}

Alternativamente, el código puede verse a continuación, sin await Task.Yield() . Tenga en cuenta que paso un lambda async a Task.Run , porque todavía quiero beneficiarme de las API asíncronas dentro de HandleConnectionAsync y uso await allí:

// Handle new connection
private static Task HandleConnectionAsync(TcpClient tcpClient)
{
    return Task.Run(async () =>
    {
        using (var networkStream = tcpClient.GetStream())
        {
            var buffer = new byte[4096];
            Console.WriteLine("[Server] Reading from client");
            var byteCount = await networkStream.ReadAsync(buffer, 0, buffer.Length);
            var request = Encoding.UTF8.GetString(buffer, 0, byteCount);
            Console.WriteLine("[Server] Client wrote {0}", request);
            var serverResponseBytes = Encoding.UTF8.GetBytes("Hello from server");
            await networkStream.WriteAsync(serverResponseBytes, 0, serverResponseBytes.Length);
            Console.WriteLine("[Server] Response has been written");
        }
    });
}

[ACTUALIZACIÓN] Basado en el comentario: si este va a ser un código de biblioteca, el entorno de ejecución es ciertamente desconocido y puede tener un contexto de sincronización no predeterminado. En este caso, prefiero ejecutar el bucle del servidor principal en un subproceso de grupo (que está libre de cualquier contexto de sincronización):

private static Task StartListener()
{
    return Task.Run(async () => 
    {
        var tcpListener = TcpListener.Create(8000);
        tcpListener.Start();
        while (true)
        {
            var tcpClient = await tcpListener.AcceptTcpClientAsync();
            Console.WriteLine("[Server] Client has connected");
            var task = StartHandleConnectionAsync(tcpClient);
            // if already faulted, re-throw any error on the calling context
            if (task.IsFaulted)
                task.Wait();
        }
    });
}

De esta manera, todas las tareas secundarias creadas dentro de StartListener no se verían afectadas por el contexto de sincronización del código del cliente. Por lo tanto, no tendría que llamar a Task.ConfigureAwait(false) ningún lugar explícitamente.

Answer 4

Las respuestas existentes se han propuesto correctamente para usar Task.Run(() => HandleConnection(client)); , pero no explica por qué.

Aquí le HandleConnectionAsync por qué: le preocupa que HandleConnectionAsync pueda tardar un tiempo en alcanzar la primera espera. Si sigue usando IO asíncrono (como debería hacerlo en este caso), esto significa que HandleConnectionAsync está haciendo trabajo vinculado a la CPU sin ningún bloqueo. Este es un caso perfecto para el grupo de subprocesos. Está hecho para ejecutar un trabajo de CPU corto y sin bloqueo.

Y tiene razón, que HandleConnectionAsync aceleraría el HandleConnectionAsync de HandleConnectionAsync tardando mucho tiempo antes de regresar (tal vez porque hay un importante trabajo relacionado con la CPU). Esto debe evitarse si necesita una alta frecuencia de nuevas conexiones.

Si está seguro de que no hay un trabajo importante que limite el ciclo, puede guardar la Task adicional del grupo de subprocesos y no hacerlo.

Alternativamente, puede tener múltiples aceptaciones ejecutándose al mismo tiempo. Reemplace await Serve(); por (por ejemplo):

var serverTasks =
    Enumerable.Range(0, Environment.ProcessorCount)
    .Select(_ => Serve());
await Task.WhenAll(serverTasks);

Esto elimina los problemas de escalabilidad. Tenga en cuenta que await a tragar todos menos un error aquí.

Answer 5

Tratar

TcpListener listener;
void Serve(){
  while(true){
    var client = listener.AcceptTcpClient();
    Task.Run(() => this.HandleConnection(client));
    //Or alternatively new Thread(HandleConnection).Start(client)
  }
}