如何確保任務同時執行？

Question

我正在實施一些多線程單元測試，並提出了一個難以確保兩個作業實際並行執行的問題 - 一個總是先啓動另一個。讓我們考慮一下我的初步實施測試​​場景來演示行爲：

static void Main(string[] args) 
{ 
    var repeats = 1000000; 
    var firstWinCount = 0; 
    var secondWinCount = 0; 
    int x = 0; 

    long time1 = 0; 
    long time2 = 0; 

    long totalTimeDiff = 0; 

    var sw = new Stopwatch(); 
    sw.Start(); 

    for (int i = 0; i < repeats; i++) 
    { 
     x = 0; 
     var task1 = new Task(() => 
     { 
      Interlocked.CompareExchange(ref x, 1, 0); 
      time1 = sw.ElapsedMilliseconds; 
     }); 
     var task2 = new Task(() => 
     { 
      Interlocked.CompareExchange(ref x, 2, 0); 
      time2 = sw.ElapsedMilliseconds; 
     }); 
     task1.Start(); 
     task2.Start(); 
     Task.WaitAll(task1, task2); 

     totalTimeDiff += Math.Abs(time1 - time2); 

     if (x == 1) 
     { 
      firstWinCount++; 
     } 
     else 
     { 
      if (x == 2) 
      { 
       secondWinCount++; 
      } 
     } 
    } 
    Console.WriteLine("First win count: {0}, percentage: {1}", firstWinCount, firstWinCount/(double)repeats * 100); 
    Console.WriteLine("Second win count: {0}, percentage: {1}", secondWinCount, secondWinCount/(double)repeats * 100); 

    Console.WriteLine("Avg sync diff: {0}ns", totalTimeDiff * 1000000/repeats); 
} 

輸出是：

First win count: 950538, percentage: 95,0538 
Second win count: 49462, percentage: 4,9462 
Avg sync diff: 1012ns 

我們可以看到，大部分的時間第一個任務開始執行早些時候然後第二個，因爲它得到到線程池第一：

task1.Start(); 
task2.Start(); 

由於線程池是在某種程度上任務安排非常難以預測，也絕對沒有保證日首先任務將不會完成，直到第二個任務開始。所以很難確保我們正在測試多線程場景。

令人驚訝的是，我在網上找不到類似的問題。

我自己的考慮和想法AutoResetEvents，鎖和互鎖同步建設導致任務同步的以下解決方案：

int sync = 0; 
var task1 = new Task(() => 
{ 
    Interlocked.Increment(ref sync); 
    while (Interlocked.CompareExchange(ref sync, 3, 2) != 2) ; 

    Interlocked.CompareExchange(ref x, 1, 0); 
    time1 = sw.ElapsedMilliseconds; 
}); 
var task2 = new Task(() => 
{ 
    while (Interlocked.CompareExchange(ref sync, 2, 1) != 1) ; 

    Interlocked.CompareExchange(ref x, 2, 0); 
    time2 = sw.ElapsedMilliseconds; 
}); 

基本上這個想法確保兩個線程不會被阻止（所以很可能有處理器時間），同時等待其他任務開始處理。其結果是，我設法減少同步時間差〜1000納秒至〜130納秒和大大增加的短運行的任務的概率在並行執行：

First win count: 23182, percentage: 2,3182 
Second win count: 976818, percentage: 97,6818 
Avg sync diff: 128ns 

剩餘缺點是任務的那排序是仍然相當明確：第一個任務總是等待第二個完成，第二個，一旦第一個知道第一個等待它，不再等待並開始執行它的工作。所以第二份工作可能首先開始。據我所知，由於[相對較少]的線程切換，exclusins（2.3％）是可能的。我可以用隨機同步順序來解決它，但這是另一個複雜因素。

我想知道我是否在重新發明輪子，是否有更好的方法來最大限度地提高兩個任務同時執行的概率，並且每個任務的啓動稍微早一些。

PS：據我所知，多線程情景通常是遠慢然後100納秒（在同步施工慢是至少1000倍任何線程切換或塊），所以該延遲同步並不重要在大多數案例。但是在測試非阻塞的高性能代碼時，這可能是至關重要的。

Answer 1

因此，似乎沒有更好的解決辦法，然後我用互鎖同步的想法，所以我實現它作爲可重複使用的類，並添加啓動，以確保啓動順序的機會均等的隨機化：

public class Operations 
{ 
    private static int _runId = 0; 

    public static void ExecuteSimultaneously(Action action1, Action action2) 
    { 
     Action slightlyEarlierStartingAction; 
     Action slightlyLaterStartingAction; 

     if (Interlocked.Increment(ref _runId) % 2 == 0) 
     { 
      slightlyEarlierStartingAction = action1; 
      slightlyLaterStartingAction = action2; 
     } 
     else 
     { 
      slightlyEarlierStartingAction = action2; 
      slightlyLaterStartingAction = action1; 
     } 

     int sync = 0; 

     var taskA = new Task(() => 
     { 
      Interlocked.Increment(ref sync); 
      while (Interlocked.CompareExchange(ref sync, 3, 2) != 2) ; 

      slightlyLaterStartingAction(); 
     }); 

     var taskB = new Task(() => 
     { 
      while (Interlocked.CompareExchange(ref sync, 2, 1) != 1) ; 

      slightlyEarlierStartingAction(); 
     }); 

     taskA.Start(); 
     taskB.Start(); 

     Task.WaitAll(taskA, taskB); 
    } 
} 

同步精度爲130納秒在每個動作贏得比賽的這個實現，概率是非常接近50％。

我發現了一種通過在高優先級的前臺線程上調度這些任務來進一步微調同步精度的方法，但是我認爲這對我來說是一種矯枉過正。不過，如果在共享有人發現它有用：

public class PriorityScheduler : TaskScheduler 
{ 
    public static PriorityScheduler Highest = new PriorityScheduler(ThreadPriority.Highest); 
    //public static PriorityScheduler AboveNormal = new PriorityScheduler(ThreadPriority.AboveNormal); 
    //public static PriorityScheduler BelowNormal = new PriorityScheduler(ThreadPriority.BelowNormal); 
    //public static PriorityScheduler Lowest = new PriorityScheduler(ThreadPriority.Lowest); 

    private BlockingCollection<Task> _tasks = new BlockingCollection<Task>(); 
    private Thread[] _threads; 
    private ThreadPriority _priority; 
    private readonly int _maximumConcurrencyLevel = 2;//Math.Max(1, Environment.ProcessorCount); 

    public PriorityScheduler(ThreadPriority priority) 
    { 
     _priority = priority; 
    } 

    public override int MaximumConcurrencyLevel 
    { 
     get { return _maximumConcurrencyLevel; } 
    } 

    protected override IEnumerable<Task> GetScheduledTasks() 
    { 
     return _tasks; 
    } 

    protected override void QueueTask(Task task) 
    { 
     _tasks.Add(task); 

     if (_threads == null) 
     { 
      _threads = new Thread[_maximumConcurrencyLevel]; 
      for (int i = 0; i < _threads.Length; i++) 
      { 
       int local = i; 
       _threads[i] = new Thread(() => 
       { 
        foreach (Task t in _tasks.GetConsumingEnumerable()) 
         base.TryExecuteTask(t); 
       }); 
       _threads[i].Name = string.Format("PriorityScheduler: ", i); 
       _threads[i].Priority = _priority; 
       _threads[i].IsBackground = false; 
       _threads[i].Start(); 
      } 
     } 
    } 

    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) 
    { 
     return false; // we might not want to execute task that should schedule as high or low priority inline 
    } 
} 

public class Operations 
{ 
    private static int _runId = 0; 

    public static void ExecuteSimultaneously(Action action1, Action action2) 
    { 
     Action slightlyEarlierStartingAction; 
     Action slightlyLaterStartingAction; 

     if (Interlocked.Increment(ref _runId) % 2 == 0) 
     { 
      slightlyEarlierStartingAction = action1; 
      slightlyLaterStartingAction = action2; 
     } 
     else 
     { 
      slightlyEarlierStartingAction = action2; 
      slightlyLaterStartingAction = action1; 
     } 

     int sync = 0; 
     var cancellationToken = new CancellationToken(); 

     var taskA = Task.Factory.StartNew(() => 
     { 
      Interlocked.Increment(ref sync); 
      while (Interlocked.CompareExchange(ref sync, 3, 2) != 2) ; 

      slightlyLaterStartingAction(); 
     }, cancellationToken, TaskCreationOptions.None, PriorityScheduler.Highest); 

     var taskB = Task.Factory.StartNew(() => 
     { 
      while (Interlocked.CompareExchange(ref sync, 2, 1) != 1) ; 

      slightlyEarlierStartingAction(); 
     }, cancellationToken, TaskCreationOptions.None, PriorityScheduler.Highest); 

     Task.WaitAll(taskA, taskB); 
    } 
} 

這讓我優化同步精度〜100納秒：

First win count: 4992559, percentage: 49,92559 
Second win count: 5007441, percentage: 50,07441 
Avg sync diff: 98ns 

警告：使用最高優先級的線程可能會限制你的電腦響應特別是當你沒有免費的處理器內核時。

Answer 2

我會使用ManualResetEvent。

喜歡的東西：

var waitEvent = new ManualResetEvent(false); 


var task1 = new Task(() => 
{ 
    waitEvent.WaitOne(); 
    Interlocked.CompareExchange(ref x, 1, 0); 
    time1 = sw.ElapsedMilliseconds; 
}); 
var task2 = new Task(() => 
{ 
    waitEvent.WaitOne(); 
    Interlocked.CompareExchange(ref x, 2, 0); 
    time2 = sw.ElapsedMilliseconds; 
}); 
task1.Start(); 
task2.Start(); 

// a startup delay? so the thread can be queued/start executing 
// but still then, you're not aware how busy the threadpool is. 
Thread.Sleep(1000); 

waitEvent.Set(); 

Task.WaitAll(task1, task2);