Cómo realizar tareas en la tarea paralela Biblioteca afecta ActivityID?

Question

Antes de utilizar la tarea paralela Biblioteca, he utilizado a menudo CorrelationManager.ActivityId hacer un seguimiento de rastreo de informes / error con varios subprocesos.

ActivityId se almacena en almacenamiento local de subprocesos, por lo que cada uno recibe el hilo de su propia copia. La idea es que cuando el fuego de un hilo (actividad), se asigna un nuevo ActivityId. El ActivityId se escribe en los registros con cualquier otra información de rastreo, por lo que es posible singularizar la información de rastreo para un solo 'Actividad'. Esto es realmente útil con WCF como el ActivityId puede ser arrastrado a la componente de servicio.

Este es un ejemplo de lo que estoy hablando:

static void Main(string[] args)
{
    ThreadPool.QueueUserWorkItem(new WaitCallback((o) =>
    {
        DoWork();
    }));
}

static void DoWork()
{
    try
    {
        Trace.CorrelationManager.ActivityId = Guid.NewGuid();
        //The functions below contain tracing which logs the ActivityID.
        CallFunction1();
        CallFunction2();
        CallFunction3();
    }
    catch (Exception ex)
    {
        Trace.Write(Trace.CorrelationManager.ActivityId + " " + ex.ToString());
    }
}

Ahora, con el TPL, mi opinión es que múltiples tareas comparten Hilos. ¿Quiere decir esto que ActivityId es propenso a ser reinstaladas mediados de tarea (por otra tarea)? ¿Hay un nuevo mecanismo para hacer frente a la actividad de rastreo?

Answer 1

Me corrió algunos experimentos y resulta que el supuesto en mi pregunta es incorrecta -. Múltiples tareas creadas con el TPL no se ejecutan en el mismo hilo, al mismo tiempo

ThreadLocalStorage es seguro de usar con TPL en .NET 4.0, ya que un hilo sólo puede ser utilizado por una tarea a la vez.

La suposición de que las tareas se pueden compartir simultáneamente hilos se basó en una entrevista que oí hablar C # 5.0 en (lo siento, no puedo recordar que muestran que era) - así que mi pregunta puede (o no) a ser pronto relevante

.

Mi experimento se inicia una serie de tareas, y registra cómo corrieron muchas tareas, el tiempo que tomaron, y cómo se consume muchos hilos. El código está por debajo si alguien quisiera repetirla.

class Program
{
    static void Main(string[] args)
    {
        int totalThreads = 100;
        TaskCreationOptions taskCreationOpt = TaskCreationOptions.None;
        Task task = null;
        Stopwatch stopwatch = new Stopwatch();
        stopwatch.Start();
        Task[] allTasks = new Task[totalThreads];
        for (int i = 0; i < totalThreads; i++)
        {
            task = Task.Factory.StartNew(() =>
           {
               DoLongRunningWork();
           }, taskCreationOpt);

            allTasks[i] = task;
        }

        Task.WaitAll(allTasks);
        stopwatch.Stop();

        Console.WriteLine(String.Format("Completed {0} tasks in {1} milliseconds", totalThreads, stopwatch.ElapsedMilliseconds));
        Console.WriteLine(String.Format("Used {0} threads", threadIds.Count));
        Console.ReadKey();
    }


    private static List<int> threadIds = new List<int>();
    private static object locker = new object();
    private static void DoLongRunningWork()
    {
        lock (locker)
        {
            //Keep a record of the managed thread used.
            if (!threadIds.Contains(Thread.CurrentThread.ManagedThreadId))
                threadIds.Add(Thread.CurrentThread.ManagedThreadId);
        }
        Guid g1 = Guid.NewGuid();
        Trace.CorrelationManager.ActivityId = g1;
        Thread.Sleep(3000);
        Guid g2 = Trace.CorrelationManager.ActivityId;
        Debug.Assert(g1.Equals(g2));
    }
}

La salida (por supuesto, esto dependerá de la máquina) era:

Completed 100 tasks in 23097 milliseconds
Used 23 threads

Cambiar taskCreationOpt a TaskCreationOptions.LongRunning dio resultados diferentes:

Completed 100 tasks in 3458 milliseconds 
Used 100 threads

Answer 2

Por favor, perdona mi publicar esto como una respuesta, ya que no es realmente responder a su pregunta, sin embargo, se relaciona a su pregunta, ya que se ocupa de la conducta y CorrelationManager hilos / tareas / etc. He estado buscando en el uso de LogicalOperationStack (y StartLogicalOperation/StopLogicalOperation métodos) de la CorrelationManager para proporcionar un contexto adicional en escenarios multi-hilo.

Tomé tu ejemplo y modificado ligeramente para añadir la capacidad de realizar un trabajo en paralelo usando Parallel.For. Además, utilizo StartLogicalOperation/StopLogicalOperation al soporte DoLongRunningWork (internamente). Conceptualmente, DoLongRunningWork hace algo como esto cada vez que se ejecuta:

DoLongRunningWork
  StartLogicalOperation
  Thread.Sleep(3000)
  StopLogicalOperation

He encontrado que si añado estas operaciones lógicas a su código (más o menos como es), todos los operatins lógicas permanecen en sincronía (siempre el número esperado de operaciones en la pila y los valores de las operaciones en la pila son siempre como se esperaba).

En algunos de mis propias pruebas he encontrado que esto no fue siempre el caso. La pila de operación lógica se estaba "corrompido". La mejor explicación que podría llegar a es que la "fusión" de vuelta de la información CallContext en el contexto del subproceso "padre" cuando el hilo termina "hijo" estaba causando la información de contexto hilo hijo "viejo" (operación lógica) para ser " heredado" por otro subproceso hermano niño.

El problema podría también estar relacionado con el hecho de que al parecer Parallel.For utiliza el hilo principal (al menos en el código de ejemplo, como está escrito) como uno de los "hilos de trabajo" (o lo que se debe llamar en el paralelo dominio). Siempre que se ejecuta DoLongRunningWork, una nueva operación lógica se pone en marcha (al principio) y se detuvo (al final) (es decir, inserta en la LogicalOperationStack y la espalda se desprendió de ella). Ahora bien, si el hilo principal ya tiene una operación lógica, en efecto, y si DoLongRunningWork se ejecuta en el hilo principal, y luego una nueva operación lógica se inicia de manera LogicalOperationStack del hilo principal tiene dos operaciones. Todas las ejecuciones posteriores de DoLongRunningWork (siempre y cuando este "iteración" de DoLongRunningWork está ejecutando en el hilo principal) será (al parecer) hereda LogicalOperationStack del hilo principal (que ahora tiene dos operaciones en él, en lugar de sólo la operación esperada).

Me tomó mucho tiempo para averiguar por qué el comportamiento de la LogicalOperationStack era diferente en mi ejemplo que en mi versión modificada de su ejemplo. Finalmente vi que en mi código que había entre corchetes todo el programa en una operación lógica, mientras que en mi versión modificada de su programa de pruebas no lo hice. La implicación es que en mi programa de pruebas, cada vez que se llevó a cabo mi "trabajo" (análogo a DoLongRunningWork), ya había una operación lógica en vigor. En mi versión modificada de su programa de pruebas, no había corchetes todo el programa en una operación lógica.

Así que, cuando lo modifico su programa de pruebas en el soporte de todo el programa en una operación lógica Y si estoy usando Parallel.For, me encontré con exactamente el mismo problema.

Usando el modelo conceptual anterior, este se ejecutará correctamente:

Parallel.For
  DoLongRunningWork
    StartLogicalOperation
    Sleep(3000)
    StopLogicalOperation

Si bien esto con el tiempo se afirme debido a una aparentemente por LogicalOperationStack sincronización:

StartLogicalOperation
Parallel.For
  DoLongRunningWork
    StartLogicalOperation
    Sleep(3000)
    StopLogicalOperation
StopLogicalOperation

Aquí está mi programa de ejemplo. Es similar a la suya, ya que tiene un método DoLongRunningWork que manipula la ActivityId así como la LogicalOperationStack. También tengo dos sabores de patadas de DoLongRunningWork. Un sabor utiliza Tareas uno Parallel.For usos. Cada sabor también puede ser ejecutado de tal manera que toda la operación parallelized está encerrado en una operación lógica o no. Por lo tanto, hay un total de 4 maneras de ejecutar la operación en paralelo. Para probar cada uno, simplemente quite el comentario deseado "Uso ..." método, recompilación, y correr. UseTasks, UseTasks(true) y UseParallelFor todo debe funcionar a completien. UseParallelFor(true) hará valer en algún momento debido a que el LogicalOperationStack no tiene el número esperado de entradas.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;

namespace CorrelationManagerParallelTest
{
  class Program 
  {     
    static void Main(string[] args)     
    { 
      //UseParallelFor(true) will assert because LogicalOperationStack will not have expected
      //number of entries, all others will run to completion.

      UseTasks(); //Equivalent to original test program with only the parallelized
                      //operation bracketed in logical operation.
      ////UseTasks(true); //Bracket entire UseTasks method in logical operation
      ////UseParallelFor();  //Equivalent to original test program, but use Parallel.For
                             //rather than Tasks.  Bracket only the parallelized
                             //operation in logical operation.
      ////UseParallelFor(true); //Bracket entire UseParallelFor method in logical operation
    }       

    private static List<int> threadIds = new List<int>();     
    private static object locker = new object();     

    private static int mainThreadId = Thread.CurrentThread.ManagedThreadId;

    private static int mainThreadUsedInDelegate = 0;

    // baseCount is the expected number of entries in the LogicalOperationStack
    // at the time that DoLongRunningWork starts.  If the entire operation is bracketed
    // externally by Start/StopLogicalOperation, then baseCount will be 1.  Otherwise,
    // it will be 0.
    private static void DoLongRunningWork(int baseCount)     
    {
      lock (locker)
      {
        //Keep a record of the managed thread used.             
        if (!threadIds.Contains(Thread.CurrentThread.ManagedThreadId))
          threadIds.Add(Thread.CurrentThread.ManagedThreadId);

        if (Thread.CurrentThread.ManagedThreadId == mainThreadId)
        {
          mainThreadUsedInDelegate++;
        }
      }         

      Guid lo1 = Guid.NewGuid();
      Trace.CorrelationManager.StartLogicalOperation(lo1);

      Guid g1 = Guid.NewGuid();         
      Trace.CorrelationManager.ActivityId = g1;

      Thread.Sleep(3000);         

      Guid g2 = Trace.CorrelationManager.ActivityId;
      Debug.Assert(g1.Equals(g2));

      //This assert, LogicalOperation.Count, will eventually fail if there is a logical operation
      //in effect when the Parallel.For operation was started.
      Debug.Assert(Trace.CorrelationManager.LogicalOperationStack.Count == baseCount + 1, string.Format("MainThread = {0}, Thread = {1}, Count = {2}, ExpectedCount = {3}", mainThreadId, Thread.CurrentThread.ManagedThreadId, Trace.CorrelationManager.LogicalOperationStack.Count, baseCount + 1));
      Debug.Assert(Trace.CorrelationManager.LogicalOperationStack.Peek().Equals(lo1), string.Format("MainThread = {0}, Thread = {1}, Count = {2}, ExpectedCount = {3}", mainThreadId, Thread.CurrentThread.ManagedThreadId, Trace.CorrelationManager.LogicalOperationStack.Peek(), lo1));

      Trace.CorrelationManager.StopLogicalOperation();
    } 

    private static void UseTasks(bool encloseInLogicalOperation = false)
    {
      int totalThreads = 100;
      TaskCreationOptions taskCreationOpt = TaskCreationOptions.None;
      Task task = null;
      Stopwatch stopwatch = new Stopwatch();
      stopwatch.Start();

      if (encloseInLogicalOperation)
      {
        Trace.CorrelationManager.StartLogicalOperation();
      }

      Task[] allTasks = new Task[totalThreads];
      for (int i = 0; i < totalThreads; i++)
      {
        task = Task.Factory.StartNew(() =>
        {
          DoLongRunningWork(encloseInLogicalOperation ? 1 : 0);
        }, taskCreationOpt);
        allTasks[i] = task;
      }
      Task.WaitAll(allTasks);

      if (encloseInLogicalOperation)
      {
        Trace.CorrelationManager.StopLogicalOperation();
      }

      stopwatch.Stop();
      Console.WriteLine(String.Format("Completed {0} tasks in {1} milliseconds", totalThreads, stopwatch.ElapsedMilliseconds));
      Console.WriteLine(String.Format("Used {0} threads", threadIds.Count));
      Console.WriteLine(String.Format("Main thread used in delegate {0} times", mainThreadUsedInDelegate));

      Console.ReadKey();
    }

    private static void UseParallelFor(bool encloseInLogicalOperation = false)
    {
      int totalThreads = 100;
      Stopwatch stopwatch = new Stopwatch();
      stopwatch.Start();

      if (encloseInLogicalOperation)
      {
        Trace.CorrelationManager.StartLogicalOperation();
      }

      Parallel.For(0, totalThreads, i =>
      {
        DoLongRunningWork(encloseInLogicalOperation ? 1 : 0);
      });

      if (encloseInLogicalOperation)
      {
        Trace.CorrelationManager.StopLogicalOperation();
      }

      stopwatch.Stop();
      Console.WriteLine(String.Format("Completed {0} tasks in {1} milliseconds", totalThreads, stopwatch.ElapsedMilliseconds));
      Console.WriteLine(String.Format("Used {0} threads", threadIds.Count));
      Console.WriteLine(String.Format("Main thread used in delegate {0} times", mainThreadUsedInDelegate));

      Console.ReadKey();
    }

  } 
}

Todo este asunto de si LogicalOperationStack se puede utilizar con Parallel.For (y / u otro roscado / construcciones de tareas) o la forma en que se puede utilizar probablemente merece su propia pregunta. Tal vez voy a publicar una pregunta. Mientras tanto, me pregunto si tiene alguna idea sobre esto (o, me pregunto si se había considerado el uso de LogicalOperationStack desde ActivityId parece ser seguro).

[EDIT]

Ver mi respuesta a esta pregunta para obtener más información sobre el uso LogicalOperationStack y / o CallContext.LogicalSetData con algunos de los diversos Tema / ThreadPool / tareas / contstructs paralelas.

Ver también mi pregunta aquí en lo que alrededor de LogicalOperationStack y extensiones paralelas: Is CorrelationManager.LogicalOperationStack compatible con Parallel.For, Tareas, hilos, etc.

Por último, ver también mi pregunta aquí en el foro paralelo de las extensiones de Microsoft: http: // social. msdn.microsoft.com/Forums/en-US/parallelextensions/thread/7c5c3051-133b-4814-9db0-fc0039b4f9d9

En mis pruebas, parece que Trace.CorrelationManager.LogicalOperationStack se pueden dañar cuando se utiliza Parallel.For o Parallel.Invoke si se inicia una operación lógica en el hilo principal y luego encender / apagar las operaciones lógicas en el delegado. En mis pruebas (ver cualquiera de los dos enlaces de arriba) del LogicalOperationStack siempre debe tener exactamente 2 entradas cuando DoLongRunningWork está ejecutando (si comienzo a una operación lógica en el hilo principal antes de patear de DoLongRunningWork utilizando diversas técnicas). Así, por "dañado" me refiero a que el LogicalOperationStack eventualmente tendrá muchos más de 2 entradas.

Por lo que puedo decir, esto es probablemente porque Parallel.For y Parallel.Invoke utilizan el hilo principal como uno de los temas de "trabajadores" para realizar la acción DoLongRunningWork.

Uso de una pila almacenada en CallContext.LogicalSetData para imitar el comportamiento de la LogicalOperationStack (similar a LogicalThreadContext.Stacks de log4net que se almacena a través de CallContext.SetData) produce resultados aún peores. Si estoy usando una pila tal de mantener contexto, resulta dañado (es decir, no tiene el número esperado de entradas) en casi todos los escenarios en los que tengo una "operación lógica" en el hilo principal y una operación lógica en cada iteración / ejecución del delegado DoLongRunningWork.