Inicio de sesión asincrónico: ¿cómo se debe hacer?

En muchos de los servicios en los que trabajo, se realizan muchos registros. Los servicios son servicios WCF (principalmente) que usan la clase .NET EventLogger.

Estoy en el proceso de mejorar el rendimiento de estos servicios, y tuve que pensar que el registro asíncrono beneficiaría el rendimiento.

No estoy al tanto de lo que sucede cuando varios subprocesos solicitan iniciar sesión, y si crea un cuello de botella, pero incluso si no lo hace, sigo pensando que no debería interferir con el proceso real que se está ejecutando.

Creo que debería invocar el mismo método de registro que llamo ahora, pero hacerlo con un nuevo hilo, mientras continúo con el proceso real.

Algunas preguntas sobre eso:

¿Está bien?

¿Hay alguna desventaja?

¿Debería hacerse de otra manera?

¿Quizás es tan rápido que ni siquiera vale la pena el esfuerzo?

El hilo separado para la operación de E / S suena razonable.

Por ejemplo, no sería bueno registrar qué botones ha presionado el usuario en el mismo hilo de la interfaz de usuario. Dicha interfaz de usuario se bloqueará al azar y tendrá un rendimiento lento percibido .

La solución es desacoplar el evento de su procesamiento.

Aquí hay mucha información sobre el problema del productor-consumidor y la cola de eventos del mundo de desarrollo de juegos

A menudo hay un código como

///Never do this!!!
public void WriteLog_Like_Bastard(string msg)
{
    lock (_lockBecauseILoveThreadContention)
    {
        File.WriteAllText("c:\\superApp.log", msg);
    }
}

Este enfoque conducirá a la contención de hilos. Todos los subprocesos de procesamiento lucharán para poder bloquear y escribir en el mismo archivo a la vez.

Algunos pueden intentar quitar las cerraduras.

public void Log_Like_Dumbass(string msg)
{
      try 
      {  File.Append("c:\\superApp.log", msg); }
        catch (Exception ex) 
        {
            MessageBox.Show("Log file may be locked by other process...")
        }
      }    
}

No es posible predecir el resultado si 2 hilos ingresarán al método al mismo tiempo.

Entonces, los desarrolladores eventualmente deshabilitarán el registro ...

¿Es posible arreglarlo?

Si.

Digamos que tenemos interfaz:

 public interface ILogger
 {
    void Debug(string message);
    // ... etc
    void Fatal(string message);
 }

En lugar de esperar a que la cerradura y la realización de bloqueo de operación de archivo cada vez que ILoggerse llama Vamos a añadir nueva Mensaje de registro a la Penging Cola de mensajes y el retorno a las cosas más importantes:

public class AsyncLogger : ILogger
{
    private readonly BlockingCollection<LogMessage> _pendingMessages;
    private readonly Type _loggerFor;
    private readonly IThreadAdapter _threadAdapter;

    public AsyncLogger(BlockingCollection<LogMessage> pendingMessages, Type loggerFor, IThreadAdapter threadAdapter)
    {
        _pendingMessages = pendingMessages;
        _loggerFor = loggerFor;
        _threadAdapter = threadAdapter;
    }

    public void Debug(string message)
    {
        Push(LoggingLevel.Debug, message);
    }

    public void Fatal(string message)
    {
        Push(LoggingLevel.Fatal, message);
    }

    private void Push(LoggingLevel importance, string message)
    {
        // since we do not know when our log entry will be written to disk, remember current time
        var timestamp = DateTime.Now;
        var threadId = _threadAdapter.GetCurrentThreadId();

        // adds message to the queue in lock-free manner and immediately returns control to caller
        _pendingMessages.Add(LogMessage.Create(timestamp, importance, message, _loggerFor, threadId));
    }
}

Hemos terminado con este simple registrador asincrónico .

El siguiente paso es procesar los mensajes entrantes.

Para simplificar, comencemos un nuevo subproceso y esperemos para siempre hasta que la aplicación salga o Asynchronous Logger agregará un nuevo mensaje a la cola pendiente .

public class LoggingQueueDispatcher : IQueueDispatcher
{
    private readonly BlockingCollection<LogMessage> _pendingMessages;
    private readonly IEnumerable<ILogListener> _listeners;
    private readonly IThreadAdapter _threadAdapter;
    private readonly ILogger _logger;
    private Thread _dispatcherThread;

    public LoggingQueueDispatcher(BlockingCollection<LogMessage> pendingMessages, IEnumerable<ILogListener> listeners, IThreadAdapter threadAdapter, ILogger logger)
    {
        _pendingMessages = pendingMessages;
        _listeners = listeners;
        _threadAdapter = threadAdapter;
        _logger = logger;
    }

    public void Start()
    {
        //  Here I use 'new' operator, only to simplify example. Should be using interface  '_threadAdapter.CreateBackgroundThread' to allow unit testing
        Thread thread = new Thread(MessageLoop);
        thread.Name = "LoggingQueueDispatcher Thread";
        thread.IsBackground = true;

        thread.Start();
        _logger.Debug("Asked to start log message Dispatcher ");

        _dispatcherThread = thread;
    }

    public bool WaitForCompletion(TimeSpan timeout)
    {
        return _dispatcherThread.Join(timeout);
    }

    private void MessageLoop()
    {
        _logger.Debug("Entering dispatcher message loop...");
        var cancellationToken = new CancellationTokenSource();
        LogMessage message;

        while (_pendingMessages.TryTake(out message, Timeout.Infinite, cancellationToken.Token))
        {
            // !!!!! Now it is safe to use File.AppendAllText("c:\\my.log") without ever using lock or forcing important threads to wait.
            // this is example, do not use in production
            foreach (var listener in _listeners)
            {
                listener.Log(message);
            }
        }

    }
}

Estoy pasando una cadena de oyentes personalizados. Probablemente quiera enviar un marco de registro de llamadas ( log4net, etc.)

Aquí está el resto del código:

public enum LoggingLevel
{
    Debug,
    // ... etc
    Fatal,
}


public class LogMessage
{
    public DateTime Timestamp { get; private set; }
    public LoggingLevel Importance { get; private set; }
    public string Message { get; private set; }
    public Type Source { get; private set; }
    public int ThreadId { get; private set; }

    private LogMessage(DateTime timestamp, LoggingLevel importance, string message, Type source, int threadId)
    {
        Timestamp = timestamp;
        Message = message;
        Source = source;
        ThreadId = threadId;
        Importance = importance;
    }

    public static LogMessage Create(DateTime timestamp, LoggingLevel importance, string message, Type source, int threadId)
    {
        return  new LogMessage(timestamp, importance, message, source, threadId);
    }

    public override string ToString()
    {
        return string.Format("{0}  [TID:{4}] {1:h:mm:ss} ({2})\t{3}", Importance, Timestamp, Source, Message, ThreadId);
    }
}

public class LoggerFactory : ILoggerFactory
{
    private readonly BlockingCollection<LogMessage> _pendingMessages;
    private readonly IThreadAdapter _threadAdapter;

    private readonly ConcurrentDictionary<Type, ILogger> _loggersCache = new ConcurrentDictionary<Type, ILogger>();


    public LoggerFactory(BlockingCollection<LogMessage> pendingMessages, IThreadAdapter threadAdapter)
    {
        _pendingMessages = pendingMessages;
        _threadAdapter = threadAdapter;
    }

    public ILogger For(Type loggerFor)
    {
        return _loggersCache.GetOrAdd(loggerFor, new AsyncLogger(_pendingMessages, loggerFor, _threadAdapter));
    }
}

public class ThreadAdapter : IThreadAdapter
{
    public int GetCurrentThreadId()
    {
        return Thread.CurrentThread.ManagedThreadId;
    }
}

public class ConsoleLogListener : ILogListener
{
    public void Log(LogMessage message)
    {
        Console.WriteLine(message.ToString());
        Debug.WriteLine(message.ToString());
    }
}

public class SimpleTextFileLogger : ILogListener
{
    private readonly IFileSystem _fileSystem;
    private readonly string _userRoamingPath;
    private readonly string _logFileName;
    private FileStream _fileStream;

    public SimpleTextFileLogger(IFileSystem fileSystem, string userRoamingPath, string logFileName)
    {
        _fileSystem = fileSystem;
        _userRoamingPath = userRoamingPath;
        _logFileName = logFileName;
    }

    public void Start()
    {
        _fileStream = new FileStream(_fileSystem.Path.Combine(_userRoamingPath, _logFileName), FileMode.Append);
    }

    public void Stop()
    {
        if (_fileStream != null)
        {
            _fileStream.Dispose();
        }
    }

    public void Log(LogMessage message)
    {
        var bytes = Encoding.UTF8.GetBytes(message.ToString() + Environment.NewLine);
        _fileStream.Write(bytes, 0, bytes.Length);
    }
}

public interface ILoggerFactory
{
    ILogger For(Type loggerFor);
}

public interface ILogListener
{
    void Log(LogMessage message);
}

public interface IThreadAdapter
{
    int GetCurrentThreadId();
}

public interface IQueueDispatcher
{
    void Start();
}

Punto de entrada:

public static class Program
{
    public static void Main()
    {
        Debug.WriteLine("[Program] Entering Main ...");

        var pendingLogQueue = new BlockingCollection<LogMessage>();


        var threadAdapter = new ThreadAdapter();
        var loggerFactory = new LoggerFactory(pendingLogQueue, threadAdapter);


        var fileSystem = new FileSystem();
        var userRoamingPath = GetUserDataDirectory(fileSystem);

        var simpleTextFileLogger = new SimpleTextFileLogger(fileSystem, userRoamingPath, "log.txt");
        simpleTextFileLogger.Start();
        ILogListener consoleListener = new ConsoleLogListener();
        ILogListener[] listeners = new [] { simpleTextFileLogger , consoleListener};

        var loggingQueueDispatcher = new LoggingQueueDispatcher(pendingLogQueue, listeners, threadAdapter, loggerFactory.For(typeof(LoggingQueueDispatcher)));
        loggingQueueDispatcher.Start();

        var logger = loggerFactory.For(typeof(Console));

        string line;
        while ((line = Console.ReadLine()) != "exit")
        {
            logger.Debug("you have entered: " + line);
        }

        logger.Fatal("Exiting...");

        Debug.WriteLine("[Program] pending LogQueue will be stopped now...");
        pendingLogQueue.CompleteAdding();
        var logQueueCompleted = loggingQueueDispatcher.WaitForCompletion(TimeSpan.FromSeconds(5));

        simpleTextFileLogger.Stop();
        Debug.WriteLine("[Program] Exiting... logQueueCompleted: " + logQueueCompleted);

    }



    private static string GetUserDataDirectory(FileSystem fileSystem)
    {
        var roamingDirectory = Environment.GetFolderPath(Environment.SpecialFolder.ApplicationData);
        var userDataDirectory = fileSystem.Path.Combine(roamingDirectory, "Async Logging Sample");
        if (!fileSystem.Directory.Exists(userDataDirectory))
            fileSystem.Directory.CreateDirectory(userDataDirectory);
        return userDataDirectory;
    }
}

Los factores clave a considerar son su necesidad de confiabilidad en los archivos de registro y la necesidad de rendimiento. Consulte las desventajas. Creo que esta es una gran estrategia para situaciones de alto rendimiento.

¿Está bien? Sí

¿Hay alguna desventaja? Sí. Dependiendo de la importancia de su registro y su implementación, podría ocurrir cualquiera de los siguientes: registros escritos fuera de secuencia, acciones de hilo de registro que no se completan antes de que se completen las acciones del evento. (Imagine un escenario en el que inicia sesión "comenzando a conectarse a la base de datos" y luego bloquea el servidor; es posible que el evento de registro nunca se escriba aunque el evento haya ocurrido (!))

Si se hace de una manera diferente, es posible que desee ver el modelo Disruptor, ya que es casi ideal para este escenario

Tal vez es tan rápido que ni siquiera vale la pena el esfuerzo, no estoy de acuerdo. Si la suya es una lógica de "aplicación", y lo único que hace es escribir registros de la actividad, entonces obtendrá órdenes de magnitud de menor latencia al descargar el registro. Sin embargo, si confía en una llamada SQL DB de 5 segundos para regresar antes de registrar 1-2 declaraciones, los beneficios son mixtos.

Creo que el registro es generalmente una operación sincrónica por su naturaleza. Desea registrar cosas si suceden o si no dependen de su lógica, por lo que para registrar algo, esa cosa debe evaluarse primero.

Dicho esto, puede mejorar el rendimiento de su aplicación almacenando en caché los registros y luego creando un hilo y guardándolos en archivos cuando tiene una operación vinculada a la CPU.

Debe identificar sus puntos de control de manera inteligente para no perder su información de registro importante durante ese período de caché.

Si desea aumentar el rendimiento de sus subprocesos, debe equilibrar las operaciones de E / S y las operaciones de la CPU.

Si crea 10 hilos que hacen IO, entonces no obtendrá un aumento de rendimiento.

El registro asíncrono es la única forma de hacerlo si necesita baja latencia en los hilos de registro. La forma en que se hace para obtener el máximo rendimiento es a través del patrón disruptor para la comunicación de subprocesos sin bloqueo y sin basura. Ahora, si desea permitir que varios subprocesos se registren simultáneamente en el mismo archivo, debe sincronizar las llamadas de registro y pagar el precio en contención de bloqueo O utilizar un multiplexor sin bloqueo. Por ejemplo, CoralQueue proporciona una cola de multiplexación simple como se describe a continuación:

Puede echar un vistazo a CoralLog, que utiliza estas estrategias para el registro asincrónico.

Descargo de responsabilidad: soy uno de los desarrolladores de CoralQueue y CoralLog.