如何确保任务同时执行？

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