Come fare i compiti nel Task Parallel Biblioteca influisce ActivityID?

Question

Prima di utilizzare il Task Parallel Biblioteca, ho spesso usato CorrelationManager.ActivityId per tenere traccia di tracciare / segnalazione degli errori con più thread.

ActivityId viene memorizzato in Thread Local Storage, in modo che ogni ottenere il filo di una propria copia. L'idea è che quando si accende un filo (attività), si assegna un nuovo ActivityId. L'ActivityId sarà scritto nei registri con qualsiasi altra informazioni di traccia, il che rende possibile individuare le informazioni di traccia per un singolo 'Activity'. Questo è veramente utile con WCF come ActivityId può essere riportato al componente di servizio.

Ecco un esempio di cosa sto parlando:

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());
    }
}

Ora, con il TPL, la mia comprensione è che più attività condividono discussioni. Questo significa che ActivityId è incline ad essere reinizializzata metà del compito (da un altro task)? Esiste un nuovo meccanismo per affrontare l'attività tracing?

Answer 1

ho incontrato alcuni esperimenti e si scopre l'assunzione in mia domanda non è corretto -. Molteplici attività create con il TPL non vengono eseguiti sullo stesso thread allo stesso tempo

ThreadLocalStorage è sicuro da usare con TPL in .NET 4.0, dal momento che un thread può essere utilizzato da un solo compito alla volta.

L'ipotesi che i compiti possono condividere le discussioni in concomitanza si è basata su un colloquio che ho sentito parlare c # 5.0 (scusate, non riesco a ricordare quale era vederlo) - quindi la mia domanda può (o non può) diventare rilevante presto

.

Il mio esperimento inizia una serie di compiti, e registra come molti compiti correvano, quanto tempo hanno preso, e quanti thread sono stati consumati. Il codice è inferiore se qualcuno volesse ripeterlo.

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));
    }
}

L'uscita (ovviamente questo dipenderà macchina) era:

Completed 100 tasks in 23097 milliseconds
Used 23 threads

La modifica taskCreationOpt a TaskCreationOptions.LongRunning ha dato risultati diversi:

Completed 100 tasks in 3458 milliseconds 
Used 100 threads

Answer 2

Si prega di perdonare il mio postando questo come una risposta in quanto non è realmente rispondere alla tua domanda, però, è legato alla tua domanda in quanto si tratta di un comportamento CorrelationManager e fili / attività / etc. Sono stato a guardare con LogicalOperationStack (e StartLogicalOperation/StopLogicalOperation metodi) del CorrelationManager di fornire contesto aggiuntivo in multithreading scenari.

Ho seguito il tuo esempio e modificato leggermente per aggiungere la possibilità di svolgere un lavoro in parallelo utilizzando Parallel.For. Inoltre, io uso StartLogicalOperation/StopLogicalOperation a staffa (internamente) DoLongRunningWork. Concettualmente, DoLongRunningWork fa qualcosa del genere ogni volta che viene eseguito:

DoLongRunningWork
  StartLogicalOperation
  Thread.Sleep(3000)
  StopLogicalOperation

Ho scoperto che se aggiungo queste operazioni logiche al codice (più o meno come è), tutti i operatins logiche rimangono in sincronia (sempre il numero atteso di operazioni su stack e dei valori delle operazioni sullo stack sono sempre come previsto).

In alcuni dei miei test ho trovato che questo non è stato sempre così. Lo stack operazione logica si stava "corrotto". La spiegazione meglio che ho potuto venire in mente è che la "fusione" di nuovo delle informazioni CallContext nel contesto del thread "padre" quando il thread "bambino" è stato la causa del "vecchio" informazioni di contesto thread figlio (operazione logica) di essere " ereditato" da un altro thread fratello del bambino.

Il problema potrebbe anche essere legato al fatto che a quanto pare Parallel.For utilizza il thread principale (almeno nel codice di esempio, come scritto), come uno dei "thread di lavoro" (o quello che dovrebbero essere chiamati in parallelo dominio). Ogni volta che viene eseguito DoLongRunningWork, una nuova operazione logica viene avviata (all'inizio) e fermato (alla fine) (che, spinto sul LogicalOperationStack e posteriore spuntato fuori di esso). Se il thread principale ha già un'operazione logica a tutti gli effetti e se DoLongRunningWork esegue sul thread principale, poi una nuova operazione logica viene avviato in modo LogicalOperationStack del thread principale ora ha due operazioni. Eventuali successive esecuzioni di DoLongRunningWork (purché questo "iterazione" di DoLongRunningWork è in esecuzione sul thread principale) (apparentemente) ereditano LogicalOperationStack del thread principale (che ora ha due operazioni su di esso, piuttosto che l'operazione si aspettava).

Mi c'è voluto molto tempo per capire perché il comportamento del LogicalOperationStack era diverso nel mio esempio che nella mia versione modificata del vostro esempio. Finalmente ho visto che nel mio codice che avevo tra parentesi l'intero programma in un'operazione logica, mentre nella mia versione modificata del programma di test non l'ho fatto. L'implicazione è che nel mio programma di test, ogni volta che è stato eseguito il mio "lavoro" (analogo a DoLongRunningWork), c'era già un'operazione logica a tutti gli effetti. Nella mia versione modificata del programma di test, non avevo tra parentesi l'intero programma in un'operazione logica.

Così, quando ho modificato il programma di test a staffa l'intero programma in un'operazione logica e se sto usando Parallel.For, mi sono imbattuto in esattamente lo stesso problema.

Utilizzando il modello concettuale di cui sopra, questo verrà eseguito correttamente:

Parallel.For
  DoLongRunningWork
    StartLogicalOperation
    Sleep(3000)
    StopLogicalOperation

Anche se questo finirà per affermare a causa di un apparentemente fuori LogicalOperationStack sincronia:

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

Ecco il mio programma di esempio. E 'simile al tuo, in quanto ha un metodo DoLongRunningWork che manipola l'ActivityId così come il LogicalOperationStack. Ho anche due sapori di calci di DoLongRunningWork. Un sapore utilizza Tasks si usa Parallel.For. Ogni sapore può anche essere eseguito in modo tale che l'intera operazione parallelizzato è racchiuso in un'operazione logica o meno. Quindi, ci sono un totale di 4 modi per eseguire l'operazione in parallelo. Per provare ciascuno, semplicemente rimuovere il commento il desiderato "Usa ..." metodo, ricompilazione, ed eseguire. UseTasks, UseTasks(true), e UseParallelFor dovrebbe correre a tutto Completisu. UseParallelFor(true) affermerà a un certo punto, perché il LogicalOperationStack non ha il numero previsto di iscrizioni.

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();
    }

  } 
}

L'intera questione del se LogicalOperationStack può essere utilizzato con Parallel.For (e / o di altri threading / costrutti Task), o come può essere utilizzato probabilmente merita la propria domanda. Forse Io posto una domanda. Nel frattempo, mi chiedo se avete qualche idea su questo (o, mi chiedo se si fosse pensato di utilizzare LogicalOperationStack dal ActivityId sembra essere sicuro).

[EDIT]

Vedere la mia risposta a questo domanda per ulteriori informazioni sull'utilizzo LogicalOperationStack e / o CallContext.LogicalSetData con alcuni dei vari Discussione / ThreadPool / Task / contstructs parallele.

Vedere anche la mia domanda qui su SO su LogicalOperationStack e le estensioni parallele: Is CorrelationManager.LogicalOperationStack compatibile con Parallel.For, Attività, discussioni, ecc

Infine, puoi anche consultare la mia domanda qui sul forum Parallel Extensions di Microsoft: http: // sociale. msdn.microsoft.com/Forums/en-US/parallelextensions/thread/7c5c3051-133b-4814-9db0-fc0039b4f9d9

Nel mio test sembra Trace.CorrelationManager.LogicalOperationStack può essere danneggiato quando si utilizza Parallel.For o Parallel.Invoke se si avvia un'operazione logica nel thread principale e poi START / STOP operazioni logiche nel delegato. Nel mio test (visualizzare uno dei due link sopra) il LogicalOperationStack deve sempre avere esattamente 2 voci quando DoLongRunningWork è in esecuzione (se inizio un'operazione logica nel thread principale prima di calciare DoLongRunningWork utilizzando varie tecniche). Così, "danneggiato" Voglio dire che il LogicalOperationStack alla fine hanno molti più di 2 voci.

Da quello che posso dire, questo è probabilmente perché Parallel.For e Parallel.Invoke usano il filo conduttore come uno dei fili "lavoratore" per eseguire l'azione DoLongRunningWork.

Utilizzando una pila memorizzata in CallContext.LogicalSetData per simulare il comportamento del LogicalOperationStack (simile a quella di log4net LogicalThreadContext.Stacks che è memorizzato tramite CallContext.SetData) produce risultati ancora peggiori. Se sto usando una pila da mantenere contesto, viene danneggiato (cioè non ha il numero previsto di iscrizioni) in quasi tutti gli scenari dove ho un "operazione logica" nel thread principale e un'operazione logica in ogni iterazione / esecuzione del delegato DoLongRunningWork.