Thread safe StreamWriter C# ¿cómo hacerlo? 2

Question

Así que esta es una continuación de mi última pregunta - Entonces la pregunta era "¿Cuál es la mejor manera de crear un programa que sea seguro para subprocesos en términos de que necesita escribir valores dobles en un archivo? Si la función que guarda ¿los valores a través de streamwriter están siendo llamados por múltiples hilos? ¿Cuál es la mejor manera de hacerlo? "

Y modifiqué algún código encontrado en MSDN, ¿qué tal lo siguiente? Este escribe correctamente todo en el archivo.

namespace SafeThread 
{ 
    class Program 
    { 
     static void Main() 
     { 
      Threading threader = new Threading(); 

      AutoResetEvent autoEvent = new AutoResetEvent(false); 

      Thread regularThread = 
       new Thread(new ThreadStart(threader.ThreadMethod)); 
      regularThread.Start(); 

      ThreadPool.QueueUserWorkItem(new WaitCallback(threader.WorkMethod), 
       autoEvent); 

      // Wait for foreground thread to end. 
      regularThread.Join(); 

      // Wait for background thread to end. 
      autoEvent.WaitOne(); 
     } 
    } 


    class Threading 
    { 
     List<double> Values = new List<double>(); 
     static readonly Object locker = new Object(); 
     StreamWriter writer = new StreamWriter("file"); 
     static int bulkCount = 0; 
     static int bulkSize = 100000; 

     public void ThreadMethod() 
     { 
      lock (locker) 
      { 
       while (bulkCount < bulkSize) 
        Values.Add(bulkCount++); 
      } 
      bulkCount = 0; 
     } 

     public void WorkMethod(object stateInfo) 
     { 
      lock (locker) 
      { 
       foreach (double V in Values) 
       { 
        writer.WriteLine(V); 
        writer.Flush(); 
       } 
      } 
      // Signal that this thread is finished. 
      ((AutoResetEvent)stateInfo).Set(); 
     } 
    } 
} 

Answer 1

Thread y QueueUserWorkItem son las API más bajas disponibles para enhebrar. No los usaría a menos que, por fin, no tuviera otra opción. Pruebe la clase Task para una abstracción de mucho más nivel. Para detalles, see my recent blog post on the subject.

También se puede utilizar como un BlockingCollection<double> adecuada cola productor/consumidor en lugar de tratar de construir uno por su lado con las APIs más bajas disponibles para la sincronización.

Reinventar estas ruedas correctamente es sorprendentemente difícil. Recomiendo usar las clases diseñadas para este tipo de necesidad (Task y BlockingCollection, para ser específico). Están incorporados en .NET 4.0 framework y are available as an add-on for .NET 3.5.

Answer 2

El código que tiene allí está sutilmente roto - en particular, si el elemento de trabajo en cola se ejecuta en primer lugar, a continuación, se eliminará de la lista (vacío) de los valores de inmediato, antes de terminar, después del cual el trabajador va y llena la Lista (que terminará siendo ignorada). El evento de reinicio automático tampoco hace nada, ya que nada consulta o espera su estado.

Además, dado que cada hilo utiliza un diferente bloqueo, los bloqueos no tienen ningún significado! Debes asegurarte de que tienes un bloqueo compartido único cuando accedas al editor de secuencias. No necesita un bloqueo entre el código de descarga y el código de generación; solo necesitas asegurarte de que el color corre después de que termine la generación.

Probablemente estés en el camino correcto, aunque usaría una matriz de tamaño fijo en lugar de una lista, y eliminaría todas las entradas de la matriz cuando se llene. Esto evita la posibilidad de quedarse sin memoria si el hilo es de larga duración.

Answer 3

• el código tiene el escritor como var de instancia pero utilizando un casillero estático. Si tenía varias instancias escribiendo en diferentes archivos, no hay ninguna razón por la que necesitarían compartir el mismo bloqueo
• en una nota relacionada, ya que ya tiene el escritor (como una instancia privada var), puede usarlo para bloquear en su lugar de usar un objeto de casillero por separado en este caso, eso hace las cosas un poco más simples.

La "respuesta correcta" realmente depende de lo que está buscando en términos de comportamiento de bloqueo/bloqueo. Por ejemplo, lo más simple sería omitir la estructura de datos intermedia, solo tiene un método WriteValues ​​para que cada hilo 'informe' sus resultados se adelante y los escriba en el archivo.Algo así como:

StreamWriter writer = new StreamWriter("file"); 
public void WriteValues(IEnumerable<double> values) 
{ 
    lock (writer) 
    { 
     foreach (var d in values) 
     { 
      writer.WriteLine(d); 
     } 
     writer.Flush(); 
    } 
} 

Por supuesto, esto significa subprocesos de trabajo serializar durante las fases de sus de resultados del informe - en función de las características de rendimiento, que puede ser muy bien aunque (5 minutos para generar, 500 ms para escribir, por ejemplo,)

En el otro extremo del espectro, haría que los hilos de trabajo escriban en una estructura de datos. Si estás en .NET 4, te recomendaría usar un ConcurrentQueue en lugar de bloquearlo tú solo.

Además, es posible que desee hacer el archivo de E/S en lotes más grandes que los informados por los subprocesos de trabajo, por lo que puede optar por simplemente escribir en un subproceso de fondo con cierta frecuencia. Ese extremo del espectro se ve algo como lo de abajo (que le quita el Console.WriteLine llama en código real, esos son sólo existe para que pueda ver su funcionamiento en la acción)

public class ThreadSafeFileBuffer<T> : IDisposable 
{ 
    private readonly StreamWriter m_writer; 
    private readonly ConcurrentQueue<T> m_buffer = new ConcurrentQueue<T>(); 
    private readonly Timer m_timer; 

    public ThreadSafeFileBuffer(string filePath, int flushPeriodInSeconds = 5) 
    { 
     m_writer = new StreamWriter(filePath); 
     var flushPeriod = TimeSpan.FromSeconds(flushPeriodInSeconds); 
     m_timer = new Timer(FlushBuffer, null, flushPeriod, flushPeriod); 
    } 

    public void AddResult(T result) 
    { 
     m_buffer.Enqueue(result); 
     Console.WriteLine("Buffer is up to {0} elements", m_buffer.Count); 
    } 

    public void Dispose() 
    { 
     Console.WriteLine("Turning off timer"); 
     m_timer.Dispose(); 
     Console.WriteLine("Flushing final buffer output"); 
     FlushBuffer(); // flush anything left over in the buffer 
     Console.WriteLine("Closing file"); 
     m_writer.Dispose(); 
    } 

    /// <summary> 
    /// Since this is only done by one thread at a time (almost always the background flush thread, but one time via Dispose), no need to lock 
    /// </summary> 
    /// <param name="unused"></param> 
    private void FlushBuffer(object unused = null) 
    { 
     T current; 
     while (m_buffer.TryDequeue(out current)) 
     { 
      Console.WriteLine("Buffer is down to {0} elements", m_buffer.Count); 
      m_writer.WriteLine(current); 
     } 
     m_writer.Flush(); 
    } 
} 

class Program 
{ 
    static void Main(string[] args) 
    { 
     var tempFile = Path.GetTempFileName(); 
     using (var resultsBuffer = new ThreadSafeFileBuffer<double>(tempFile)) 
     { 
      Parallel.For(0, 100, i => 
      { 
       // simulate some 'real work' by waiting for awhile 
       var sleepTime = new Random().Next(10000); 
       Console.WriteLine("Thread {0} doing work for {1} ms", Thread.CurrentThread.ManagedThreadId, sleepTime); 
       Thread.Sleep(sleepTime); 
       resultsBuffer.AddResult(Math.PI*i); 
      }); 
     } 
     foreach (var resultLine in File.ReadAllLines(tempFile)) 
     { 
      Console.WriteLine("Line from result: {0}", resultLine); 
     } 
    } 
} 

Answer 4

Así que usted está diciendo ¿Quieres un montón de hilos para escribir datos en un solo archivo usando StreamWriter? Fácil. Simplemente bloquee el objeto StreamWriter.

El código aquí creará 5 hilos. Cada hilo realizará 5 "acciones" y al final de cada acción escribirá 5 líneas en un archivo llamado "archivo".

using System; 
using System.Collections.Generic; 
using System.IO; 
using System.Threading; 

namespace ConsoleApplication1 { 
    class Program { 
     static void Main() { 
      StreamWriter Writer = new StreamWriter("file"); 

      Action<int> ThreadProcedure = (i) => { 
       // A thread may perform many actions and write out the result after each action 
       // The outer loop here represents the multiple actions this thread will take 
       for (int x = 0; x < 5; x++) { 
        // Here is where the thread would generate the data for this action 
        // Well simulate work time using a call to Sleep 
        Thread.Sleep(1000); 
        // After generating the data the thread needs to lock the Writer before using it. 
        lock (Writer) { 
         // Here we'll write a few lines to the Writer 
         for (int y = 0; y < 5; y++) { 
          Writer.WriteLine("Thread id = {0}; Action id = {1}; Line id = {2}", i, x, y); 
         } 
        } 
       } 
      }; 

      //Now that we have a delegate for the thread code lets make a few instances 

      List<IAsyncResult> AsyncResultList = new List<IAsyncResult>(); 
      for (int w = 0; w < 5; w++) { 
       AsyncResultList.Add(ThreadProcedure.BeginInvoke(w, null, null)); 
      } 

      // Wait for all threads to complete 
      foreach (IAsyncResult r in AsyncResultList) { 
       r.AsyncWaitHandle.WaitOne(); 
      } 

      // Flush/Close the writer so all data goes to disk 
      Writer.Flush(); 
      Writer.Close(); 
     } 
    } 
} 

El resultado debe ser un "archivo" archivo con 125 líneas en él con todas las "acciones" realizadas simultáneamente y el resultado de cada acción por escrito de forma sincrónica en el fichero.