lunes, 10 de abril de 2017

Uso de Sockets con Alto Desempeño (High Performance)

Recientemente me tope con un problema que desafío mi manera de atender multiples conexiones por  Sockets (Mi Paradigma),  en este caso C#; y es que ese fue el problema principal: Usar un paradigma sin considerar la carga de trabajo que recibiría el servicio.

El Entorno de Trabajo


  • La aplicación es un Windows Service, hecho en C#, el cual levanta un Socket Server para atender multiples peticiones (Requests) de manera simultánea, por medio de TCP.
  • Cada Request implica leer el comando de entrada, el cual tiene un tamaño fijo de 612 bytes.
  • Una vez que se haya terminado de atender el Request se procede a enviar la respuesta (Response), con un tamaño fijo de 612 bytes.
  • Una vez que se termino de enviar el response se debe cerrar la conexión de socket.
  • En hora pico se tiene una carga promedio de 15 mil Requests por Hora.

Mi Paradigma

El paradigma que utilizaba para atender cada una de las peticiones era:


  1. Crear un Thread para levantar un Socket Server.
  2. Por cada petición de conexión que recibía el Socket Server se crea un nuevo Thread para atender la petición del Socket Client.
  3. Dentro del nuevo Thread se procede a leer los datos que envío el Socket Cliente (Request).
  4. Una vez que se tiene el Request se procede a trabajarlo dentro del mismo thread.
  5. Una vez que se haya procesado el Request se envía la respuesta (Response).
  6. Cuando se haya terminado de enviar el Response se cierra el Socket.
  7. Una vez terminado el envío del Response se cierra el Socket.


El problema que tiene este paradigma es que bajo una carga de trabajo, como la que se describe en la sección "El Entorno de Trabajo" es que se presento una degradación en la atención de las peticiones y el servicio, presentando los siguientes sintomas:
  • Retraso de hasta 40 segundos para recibir y responder una petición.
  • Consumo de procesador de hasta 50%, antes de colapsar la atención de los Request.
  • Las aplicaciones que realizaban la operación de Socket Client, no recibían la respuesta a tiempo y generaban una excepción de TimeOut en la lectura del Response.

Considere que la atención de la petición puede llevar al Thread a esperar, por medio de los metodos Sleep y Wait.

La Solución


Después de batallar por algunos días, la respuesta al dilema fue atender las peticiones de manera Asíncrona y no por medio de un nuevo Thread dedicado.

El crear un Thread para atender una petición de Socket es una solución viable, siempre y cuando la carga de trabajo y el tiempo de respuesta no seam factores determinantes en el desempeño de tu aplicación. Pero en el caso de una aplicación de Alto Desempeño, en el que se requiere atender varias peticiones simultáneas en el menor tiempo posible, no es una solución que pueda funcionar.

Para el caso de C# se requiere del uso de BeginRead, EndRead, BeginWrite y EndWrite, de NetworkStream. Estos métodos permiten al Socket Clientes, en conjunto con el Socket Server, de recibir la y enviar datos de manera asincrónica, sin necesidad de crear un nuevo Thread.

La implementación fue la siguiente:


  1. Se conserva el Thread dedicado a aceptar las peticiones de conexión de los clientes. 
  2. Por cada petición de conexión se creará un nuevo objeto Request, asignándole el Socket Client que se encargará de atender la comunicación Client-Server.
  3. El objeto Request será el encargado de obtener el Request, procesarlo y enviar el Response, de manera asincrónica.
  4. El proceso de recepción inicia con la invocación del método Request::startReceiveAsync.
  5. El método startReceiveAsync invoca al método BeginRead, de NetworkStream, especificando cuantos bytes debe leer antes de invocar el método de retroalimentación Request::requestReceived, el cual se especifica en el método BeginRead.
  6. Al momento de invocar BeginRead se continua con la ejecución del programa sin esperar a que se haya recibido el Request, permitiendo con esto terminar la ejecución del método startReceiveAsync y regresando el CallStack al punto en donde se espera una nueva conexión de socket.
  7. Una vez que es invocado el método requestReceived, que es cuando se termino de leer los datos,  es importante invocar el método EndRead, del NetworkStream, para asegurarse de que se haya terminado de recibir toda la cantidad de datos solicitados en el método BeginRead. Se recomiendo utilizar EndRead en el caso de que se espera recibir grandes cantidades de información.

La secuencia de invocación entre de los métodos startReceiveAsync y requestReceived se lleva a cabo de manera asincrónica, pues el objeto NetworkStream se encarga de invocarlos a medida que los datos se encuentren disponibles para su lectura. Todo esto sin tener que utilizar un bloqueo en los sockets o en el Thread de atención al cliente.

El código para aceptar la conexión y recepción del Request es:



 

/* Metodo Server::run()*/
public void run()
{
    server = new TcpListener(IPAddress.Any, port);
    server.Start();
    
    log.Info("Iniciando espera de peticiones en el puerto [" + port + "]");

    while (keepRunning)
    {
        try
        {
            TcpClient socket = server.AcceptTcpClient();
            if (keepRunning)
                RequestManager.createRequestForEvalue(socket, idLayout);
        }
        catch (Exception ex)
        {
            log.Error("Se detecto un error en el puerto para atencion de peticiones. ERR: " + ex.Message);
            log.Error(ex.StackTrace);
        }
    }

    log.Info("Servicio deteniendo.");
}


/* Metodo RequestManager.createRequestForEvalue*/
public static bool createRequestForEvalue(TcpClient socket, int idLayout)
{
    Request req = null;
    req = new Request(socket,idLayout);

    registerRequest(req.ID,req); //Registra el Request, para su posterior uso.

    req.startReceiveAsync(); //Since V4.20
    return true;
}


/*Metodo Request.startReceiveAsync*/
public void startReceiveAsync()
{
    try
    {
        log.Info("[" + id + "] Iniciando la lectura del Request.");
        requestBuffer = new byte[BUFFER_SIZE];
        NetworkStream nst = socket.GetStream();
        nst.BeginRead(requestBuffer, 0,BUFFER_SIZE, this.requestReceived, nst);
    }catch(Exception ex)
    {
        log.Error("[" + id + "] Se presento un problema al iniciar la lectura del Request: " + ex.Message);
        closeSocket();
    }
}


/*Metodo Request.startReceiveAsync*/
public void requestReceived(IAsyncResult ar)
{

    try
    {   
        NetworkStream nst = socket.GetStream();
        nst.EndRead(ar);
        string request = Encoding.UTF8.GetString(requestBuffer, 0, BUFFER_SIZE);
        log.Info("[" + id + "] Request recibido: [" + message +"]");
        RunForIVR(request);
    }
    catch (Exception ex)
    {
        log.Error("[" + id + "] Se presento un problema al recibir el Request: " + ex.Message);
        closeSocket();
    }

}
Dentro del procede de atención se puede requerir el mantener en espera el envío del Response hasta que se tenga una condición predeterminada En este caso se sugiere crear métodos asincrónicos y evitar utilizar los métodos Wait o Sleep, por que la convinación Thread-Socket-Wait/Sleep tiene altas probabilidades de bloquear el socket socket y, por consecuencia, la recepción y escritura de otros sockets.

Para estos casos se debe hacer uso de los métodos asincrónicos (async y await) para evitar detener la ejecución de los threads y evitar bloquear los sockets. Un ejemplo:

Suponga que dentro del método RunForIVR se debe invocar un Webservices por medio de un método estático llamado setSegment, de la clase TrackingRequest, que se encuentra en otra clase. La ejecución del método setSegment no debe interrumpir la ejecución de RunForIVR, pues no es importante el resultado de la operación setSegment para la fabricación del Response, pero se debe considerar que la invocación del Webservice puede retrasar la ejecución por un tema de Timout, por ejemplo.

 En este caso se debe realizar lo sisguiente

  1. El método setSegment, de la clase TrackingRequest, debe declararse como asincrónico, usando el prefijo async.
  2. Utilizar el método asincrónico del Webservice en conjunto con la palabra await.
  3. El uso de await, del cliente de Webservice, junto con la declaración del método asincrónico seetSegment provocará que la secuencia del programa se regrese un nivel superior al CallStack, RunForIVR, para que continue su ejecución.
  4. Una vez que el Webservice haya terminado, el método setSegment continua su ejecución.
  5. Mientras se esta ejecutando el Webservice, el método RunForIVR invoca el método SendResponse.
  6. SendReposne invoca el BeginWrite, el cual invocará el método closeSocket.
  7. Sobre el método closeSocket se debe invocar el método EndWrite, el cual bloqueará el socket hasta que la transmisión de datos haya terminado.
  8. Terminada la transmisión de datos se cierra el socket, mientras que a la par se esta ejecutando el método setSegment, o ya termino de ejecutarse.

El código es el siguiente:

Clase TrackingRequest:

public static async Task setSegment(TrackingBean bean)
{
   System.Console.WriteLine("Inicia Async");
    try
    {
        BBVATrackingSoapClient ws = new BBVATrackingSoapClient("BBVATrackingSoap", url);
        ws.InnerChannel.OperationTimeout = TimeSpan.FromSeconds(1.5); // Se da un segundo de toleracncia para llevar a cabo la execucion del

        AddSegmentWS1 result = null;
        
        //Al momento de invocar el método AddSegmentAsync, se regresa el control al metodo superiro que invoco al metodo setSegment
        result = await ws.AddSegmentAsync(bean._key, bean._entity, bean._service, ACTION.OPEN_SEGMENT,CLOSE_DESC.NONE);

        if(result.AddSegmentWSResult.ERR_CODE == 0)
        {
            if (CFGManager.GetCFGManager().DebugActive)
                log.Debug("[" + bean._id + "] Operacion de segmento realizada con exito.");
        }       
        else
            log.Error("[" + bean._id + "] No se pudo registar la operacion del segmento, se obtuvo la siguiente respuesta: " + result.AddSegmentWSResult.ERR_DESC);

        result = null;
        ws = null;
    }
    catch(Exception ex)
    {
        log.Error("[" + bean._id + "] No se pudo registar la operacion del segmento. Error: " + ex.Message );
    }

    bean = null;
    
    System.Console.WriteLine("Termina Async");
    return 0;
}


El código restante para atender el Request e invocar el Response:

public void RunForIVR(string Request)
{
    string Response
 //Procesa el Request
 
 TrackingBean bean;
 //...
 TrackingRequest.setSegment(bean); //Este metodo es asincronico, por lo que no esperara a que finalice.

 //...
 //Obtiene el Response
 
 Response = "Informacion que sera enviada como respuesta";


 //Envia la respuesta
 SendResponse(Response);
}



public void SendResponse(string Response)
{
    StringBuilder sb = new StringBuilder();
    sb.Append(Response);
    sb.Append('\0', BUFFER_SIZE - Response.Length);
    string message = sb.ToString();

    log.Info("[" + id + "] ivrTrans CMD: [" + idCMD + "] RESPONSE: [" + Response + "]");

    NetworkStream nst = socket.GetStream();
    byte[] buffer = new byte[BUFFER_SIZE];
    for (int i = 0; i < BUFFER_SIZE; i++)
        buffer[i] = (byte)message.ElementAt(i);

    nst.BeginWrite(buffer, 0, BUFFER_SIZE, this.closeSocket, nst);// (new AsyncCallback(this.closeSocket));
}

public void closeSocket(IAsyncResult ar = null)
{
    
    try
    {
        if (ar != null) //Since 4.24
        {
            NetworkStream nst = socket.GetStream();
            nst.EndWrite(ar);
        }

        socket.Close();
        socket = null;
    }catch(Exception ex)
    {
        log.Warn("[" + id + "] Se presento un problema al cerrar el socket. Error: " + ex.Message + Environment.NewLine + ex.StackTrace);
    }
    log.Info("[" + id + "] Socket cerrado.");
}