كيف تؤثر المهام في المكتبة الموازية للمهمة على ActivityId؟

Question

قبل استخدام المكتبة الموازية للمهمة ، غالبًا ما استخدمت ConsleationManager.ActivityID لتتبع الإبلاغ عن التتبع/الخطأ مع مؤشرات ترابط متعددة.

يتم تخزين ActivityId في سلسلة التخزين المحلية ، لذلك يحصل كل مؤشر ترابط على نسخته الخاصة. الفكرة هي أنه عندما تقوم بإطلاق سلسلة (نشاط) ، يمكنك تعيين ActivityId جديد. سيتم كتابة ActivityId إلى السجلات مع أي معلومات تتبع أخرى ، مما يجعل من الممكن تخصيص معلومات التتبع لـ "نشاط" واحد. هذا مفيد حقًا مع WCF حيث يمكن نقل ActivityId إلى مكون الخدمة.

إليك مثال على ما أتحدث عنه:

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

الآن ، مع TPL ، أفهم أن المهام المتعددة تشارك المواضيع. هل هذا يعني أن ActivityId عرضة للتصنيع في المهمة المتوسطة (بمهمة أخرى)؟ هل هناك آلية جديدة للتعامل مع تتبع النشاط؟

Answer 1

قمت بإجراء بعض التجارب واتضح أن الافتراض في سؤالي غير صحيح - لا يتم تشغيل مهام متعددة تم إنشاؤها باستخدام TPL على نفس الخيط في نفس الوقت.

يعد ThreadLocalStorage آمنًا للاستخدام مع TPL في .NET 4.0 ، حيث لا يمكن استخدام مؤشر ترابط إلا من خلال مهمة واحدة في كل مرة.

إن الافتراض بأن المهام يمكن أن تشارك المواضيع بشكل متزامن كانت تستند إلى مقابلة سمعت عنها ج# 5.0 على dotnetrocks (آسف ، لا أستطيع أن أتذكر ما الذي يظهر أنه) - لذا قد يصبح سؤالي (أو قد لا) ذا صلة قريبًا.

تبدأ تجربتي بعدد من المهام ، وتسجل عدد المهام التي تم تشغيلها ، والمدة التي استغرقتها ، وعدد المواضيع التي تم استهلاكها. الكود أدناه إذا كان أي شخص يرغب في تكراره.

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

كان الإخراج (بالطبع هذا يعتمد على الجهاز) هو:

Completed 100 tasks in 23097 milliseconds
Used 23 threads

تغيير TaskCreationOpt إلى TaskCreationOption. أعطى longrunning نتائج مختلفة:

Completed 100 tasks in 3458 milliseconds 
Used 100 threads

Answer 2

يرجى تسامح نشر هذا كإجابة لأنه لا يجيب حقًا على سؤالك ، ومع ذلك ، فهو مرتبط بسؤالك لأنه يتعامل مع سلوك MonleationManager وخيوط/مهام/إلخ. لقد كنت أبحث في استخدام ارتباط ماناجير LogicalOperationStack (و StartLogicalOperation/StopLogicalOperation الأساليب) لتوفير سياق إضافي في سيناريوهات Multithreading.

أخذت مثالك وقمت بتعديله قليلاً لإضافة القدرة على أداء العمل بالتوازي باستخدام Parallel. for. أيضا ، أنا أستخدم StartLogicalOperation/StopLogicalOperation إلى قوسين (داخليًا) DoLongRunningWork. من الناحية المفاهيمية ، DoLongRunningWork يفعل شيئًا كهذا في كل مرة يتم تنفيذها:

DoLongRunningWork
  StartLogicalOperation
  Thread.Sleep(3000)
  StopLogicalOperation

لقد وجدت أنه إذا قمت بإضافة هذه العمليات المنطقية إلى الكود الخاص بك (أكثر أو أقل كما هو) ، تظل جميع الأوبرا المنطقية متزامنة (دائمًا العدد المتوقع للعمليات على المكدس وقيم العمليات على المكدس دائمًا مُتوقع).

في بعض الاختبارات الخاصة بي ، وجدت أن هذا لم يكن دائمًا هو الحال. كان المكدس المنطقي "تالف". أفضل تفسير يمكن أن أتوصل إليه هو أن "دمج" معلومات CallContext في سياق مؤشر الترابط "الأصل" عندما يكون خيط "الطفل" يسبب "معلومات سياق الخيط الطفل القديم (العملية المنطقية)" ورثت "من قبل خيط طفل الأخوة.

قد تكون المشكلة مرتبطة أيضًا بحقيقة أن الموازي. على ما يبدو يستخدم الخيط الرئيسي (على الأقل في رمز المثال ، كما هو مكتوب) كواحد من "مؤشرات الترابط العامل" (أو أي شيء يجب أن يطلق عليه في المجال الموازي). كلما تم تنفيذ DolongRunningwork ، يتم بدء تشغيل منطقي جديد (في البداية) وتوقف (في النهاية) (أي ، يتم دفعه على strivicaloperationstack وتراجعت عن ذلك). إذا كان للخيط الرئيسي بالفعل عملية منطقية سارية ، وإذا تم تنفيذ DolongRunningwork على الخيط الرئيسي ، فسيتم بدء تشغيل منطقي جديد ، لذا فإن برنامج LogicalOperationstack الرئيسي للمعلومات الرئيسية لديه عمليتان. أي عمليات إعدام لاحقة من DolongRunningwork (طالما أن هذا "التكرار" من DolongRunningwork يتم تنفيذه على الخيط الرئيسي) سوف يرث (على ما يبدو) (على ما يبدو) ترث المنطق المنطقي للموضوع الرئيسي (الذي لديه الآن عمليتان عليها ، بدلاً من العملية المتوقعة فقط).

استغرق الأمر مني وقتًا طويلاً لمعرفة سبب اختلاف سلوك charicoperationstack في المثال الخاص بي عن الإصدار المعدل من مثالك. أخيرًا ، رأيت أنه في الكود الخاص بي ، قمت بتقسيم البرنامج بأكمله في عملية منطقية ، بينما في نسختي المعدلة من برنامج الاختبار الخاص بك لم أفعل. المعنى الضمني هو أنه في برنامج الاختبار الخاص بي ، في كل مرة يتم فيها "عملي" (مماثل لـ Dolongrunningwork) ، كانت هناك بالفعل عملية منطقية في الواقع. في نسختي المعدلة من برنامج الاختبار الخاص بك ، لم أقم بتخطيط البرنامج بأكمله في عملية منطقية.

لذلك ، عندما قمت بتعديل برنامج الاختبار الخاص بك لتسوية البرنامج بأكمله في عملية منطقية ، وإذا كنت أستخدم Parallel.for ، فقد واجهت نفس المشكلة تمامًا.

باستخدام النموذج المفاهيمي أعلاه ، سيتم تشغيل هذا بنجاح:

Parallel.For
  DoLongRunningWork
    StartLogicalOperation
    Sleep(3000)
    StopLogicalOperation

في حين أن هذا سيؤكد في النهاية بسبب الخروج من Sync LogicalOperationStack:

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

ها هو برنامج العينة الخاص بي. إنه مشابه لك من حيث أنه يحتوي على طريقة dolongrunningwork التي تتلاعب بـ ActivityId وكذلك المنطقي. لدي أيضًا نكهات من الركل من Dolongrunningwork. نكهة واحدة تستخدم المهام يستخدم المرء بالتوازي. يمكن أيضًا تنفيذ كل نكهة بحيث يتم إرفاق العملية المتوازية بأكملها في عملية منطقية أم لا. لذلك ، هناك ما مجموعه 4 طرق لتنفيذ العملية الموازية. لتجربة كل واحدة ، ما عليك سوى فك "الاستخدام ..." المطلوب ، إعادة الترجمة ، وتشغيله. UseTasks, UseTasks(true), ، و UseParallelFor يجب أن يركض إلى الانتهاء. UseParallelFor(true) سوف يؤكد في مرحلة ما لأن logicaloperationstack لا يحتوي على العدد المتوقع من الإدخالات.

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

  } 
}

هذه المسألة بأكملها إذا كان يمكن استخدام LogicalOperationStack مع التوازي. بالنسبة (و/أو بنيات الخيوط/المهام الأخرى) أو كيف يمكن استخدامها على الأرجح يستحق سؤالها الخاص. ربما سأقوم بنشر سؤال. في غضون ذلك ، أتساءل عما إذا كان لديك أي أفكار حول هذا (أو ، أتساءل عما إذا كنت قد فكرت في استخدام LogicalOperationStack لأن ActivityID يبدو آمنًا).

تعديل

انظر إجابتي على هذا السؤال لمزيد من المعلومات حول استخدام LogicalOperationStack و/أو callContext.LogicalSetData مع بعض الناطدات الخيط/Threadpool/Task/Parallel المختلفة.

انظر أيضًا سؤالي هنا حول SOLONTOPERATIONSTACK والملحقات المتوازية:هو الارتباطان.

أخيرًا ، انظر أيضًا سؤالي هنا في منتدى Microsoft Explaly Extensions:http://social.msdn.microsoft.com/forums/en-us/parallextensions/thread/7c5c3051-133b-4814-9db0-fc0039b4f9d9

في اختباري ، يبدو الأمر وكأنه Trace.CorrelationManager.LogicalOperationstack يمكن أن يفسد عند استخدام Parallel. for Or Parallel.invoke إذا بدأت عملية منطقية في الخيط الرئيسي ثم ابدأ/إيقاف العمليات المنطقية في المندوب. في اختباراتي (انظر أي من الرابطين أعلاه) ، يجب أن يكون لدى LogicalOperationStack دائمًا إدخالين تمامًا عند تنفيذ DolongRunningwork (إذا بدأت عملية منطقية في الخيط الرئيسي قبل ركل DolongRunningwork باستخدام تقنيات مختلفة). لذلك ، من خلال "تالفة" أعني أن المنطق سيكون في النهاية أكثر من إدخالات.

من ما يمكنني قوله ، ربما يكون هذا على الأرجح موازيًا. للموازنة. استخدم الخيط الرئيسي كواحد من مؤشرات الترابط "العامل" لأداء عمل dolongrunningwork.

باستخدام مكدس مخزّن في callContext.logicalsetData لتقليد سلوك LogicalOperationStack (على غرار Log4Net's LogicalThreadContext.stacks التي يتم تخزينها عبر callContext.setData) ، فإن نتائج أسوأ. إذا كنت أستخدم مثل هذا المكدس للحفاظ على السياق ، يصبح تالفًا (أي لا يحتوي على العدد المتوقع من الإدخالات) في جميع السيناريوهات تقريبًا حيث لدي "عملية منطقية" في الخيط الرئيسي وعملية منطقية في كل تكرار /تنفيذ مندوب DolongRunningwork.