发生超时时网络绑定 I/O 的缓慢并行化

Question

我正在并行化一个高度依赖 WinAPI NetAPI32 调用的方法。如果用户输入的主机已关闭或数百个列表中的多个主机，调用有时会超时。

int prevThreads, prevPorts;
ThreadPool.GetMinThreads(out prevThreads, out prevPorts);
ThreadPool.SetMinThreads(20, prevPorts);

var parallelScanList = computersToScan.AsParallel().WithExecutionMode(ParallelExecutionMode.ForceParallelism).WithDegreeOfParallelism(20);

Api.WinApi.AdvApi.LogonAndImpersonate(connection.UserCredential);

foreach (var computer in parallelScanList)
{
        //...
        //this takes a long time to timeout
        status = NetApi.NetUserEnum(computer.DnsHostname, 2,
                (int)NetApi.NetUserEnumFilter.FILTER_NORMAL_ACCOUNT,
                out userbufPtr, (int)LmCons.MAX_PREFERRED_LENGTH, out userEntriesRead, out totalEntries,
                out userResumeHandle);

}

我们在使用消费者/生产者的 C 客户端中具有与此类似的逻辑。启动 20 个线程并让它们读取一个列表，直到它耗尽。

function StartProcessingHosts()
{
  for 1 to 20
     StartProcessThread()
}

function ProcessHostsThread()
{
  while(moreHosts)
  {
     //obviously synchronization around here
     var host = popHost();
     DoSomething(host);
  }
}

而且速度非常快，因为这些网络调用一直在等待，并且可能无法连接到停机主机。

我目前在 C# 中的做法似乎是一次一个。

Answer 1

更新：

我明白了，问题在于 foreach 循环。您可能已经假设通过查询 AsParallel 然后在 foreach 中执行它会使其并行。当然这不会发生。此外，使用 PLINQ，您可以实现与 svick's answer 中演示的相同。

但是，这是并行化代码的另一种方法，我在下面提到它，因为 svick 的回答也受到以下事实的影响：仅通过设置 MaxDegreeOfParallelism = 20 并不能保证 20 次并行执行。它仍然只是并行执行的上限，而不是下限。如果 PLINQ 执行引擎觉得它应该只启动 5 次并行执行，它只会启动 5 次，这完全是合法的执行。

以下代码保证 20 次并行执行：

var concurrentScanList = new ConcurrentQueue<Computer>(computersToScan);
var taskFactory = new TaskFactory(TaskCreationOptions.LongRunning, TaskContinuationOptions.None);
var taskArray = new Task[20];

//Initializing the tasks
for (var index = 0; index < taskArray.Length; index++)
{
    taskArray[index] = taskFactory.StartNew(() =>
    {
        Computer host;
        while (concurrentScanList.TryDequeue(out host))
        {
            DoSomething(host);
        }
    });
}

//Wait for all tasks to finish - queue will be empty then
Task.WaitAll(baseProcessorTaskArray);

---

旧答案：

WithDegreeOfParallelism() 是，

用于处理查询的最大并发执行任务数。

...我认为由于并发执行任务的最小数量不固定，它可能是1。

基本上，您的猜测可能是正确的，即此执行不是并行发生的，因此会超时。此外，即使它 以等于 20 的并行度并行发生，也不能保证总是这样。

我的建议是您将“要扫描的计算机”放在BlockingCollection 中，然后生成 20 个任务，每个任务都从此 BlockingCollection 读取计算机，然后对其进行扫描。这个实现自然会是生产者消费者，因为这是你的问题设计的内在品质。

Answer 2

PLINQ，Parallel LINQ 的缩写，您猜对了，并行 LINQ 查询。比如你写collection.AsParallel().Where(/* some condition */).Select(/* some projection */).ToList()，那么Where()和Select()会并行执行。

但你不这样做，你打电话给AsParallel()，说“下面的 LINQ 查询应该并行执行”。然后通过调用WithExecutionMode() 和WithDegreeOfParallelism() 来配置即将到来的查询的并行度。然后您实际上没有任何 LINQ 查询，而是使用 foreach，它将串行迭代集合。

如果你想并行执行foreach，你不需要PLINQ，你需要Parallel.ForEach()：

Parallel.ForEach(computersToScan, new ParallelOptions { MaxDegreeOfParallelism = 20 },
    computer =>
    {
        //...
    });