当超时发生时，网络绑定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

PLINQ，是并行LINQ的缩写，你猜到了，并行LINQ查询。例如，如果编写collection.AsParallel().Where(/* some condition */).Select(/* some projection */).ToList()，则Where()和Select()将并行执行。

但是如果不这样做，则调用AsParallel()，说“下面的LINQ查询应该并行执行”。然后，通过调用WithExecutionMode()和WithDegreeOfParallelism()来配置即将出现的查询的并行性。然后，您实际上没有任何LINQ查询，而是使用foreach，它将对集合进行串行迭代。

如果你想并行执行一个foreach，你不想要Parallel.ForEach()，你想要Parallel.ForEach()

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

Answer 2

更新：

我明白了，问题是前程循环。您可能已经假设，通过使查询AsParallel，然后在foreach中执行它，它将使其并行。那当然不会发生。此外，使用PLINQ可以实现与在斯维克的回答中演示的相同的功能。

但是，这里还有一种并行化代码的方法，下面我将提到这一点，因为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()是，

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

...and --我的想法是，由于并发执行任务的最小数目不是固定的，所以它可能是1。

从本质上说，您的猜测可能是正确的，即这种执行不是并行进行的，因此是超时的。此外，即使它是并行的，其并行度等于20，也不能保证总是这样。

我的建议是，将“计算机扫描”放在一个BlockingCollection中，然后生成20个任务，每个任务从这个BlockingCollection中读取一台计算机，然后扫描它。这种实现自然会成为生产者消费者，因为这是问题设计的内在品质。