IO 绑定操作和 Task.Run()

Question

我对并发（实际上是 C#）还很陌生。我在两个单独的目录中有一堆 csv 文件要读取，然后我想在读取文件后进行一些处理。该处理独立于其他数据读取和处理操作。完成所有处理后，我想更新 UI。同时 UI 也需要响应，因为我需要显示一个进度条。目前我有这样的事情：

private string _directoryA;
private string _directoryB;

// The user clicks the button
private void ButtonPressed()
{
    Task.Run(() => DoJob());
} 

private void DoJob()
{
    var tasks = new List<Task>();
    var watch = Stopwatch.StartNew();

    tasks.Add(Task.Run(() => DoJobForDirectory(_directoryA).ContinueWith(t => Console.WriteLine("First Half");
    tasks.Add(Task.Run(() => DoJobForDirectory(_directoryB).ContinueWith(t => Console.WriteLine("Second Half");
    
    Task.WaitAll(tasks.ToArray());

    watch.Stop();
    Console.WriteLine($"Time Taken : {watch.ElapsedMilliseconds} ms.");
    UpdateUI();
}

private void DoJobForDirectory(string directory)
{
    var files = Directory.EnumerateFiles(directory, "*.csv");
    var tasks = new List<Task>();
    foreach (var file in files)
    {
        // Update the progress bar in the UI when a file has finished processing
        tasks.Add(Task.Run(() => DoJobForFile(file)).ContinueWith(t => UpdateCounter++));
    }

    Task.WaitAll(tasks.ToArray());
}

private void DoJobForFile(string filePath)
{
    ReadCSV();
    ProcessData();
    ...
}

我觉得我在这里遗漏了一些东西。根据我的阅读，这个操作应该是 I/O 绑定的，因为之后的处理非常轻量级（一些 for 循环和分配）。所以我真的应该只使用异步等待，而不是 Task.Run()...？但是我想不出更好的方法来做到这一点。 ReadCSV() 来自一些没有异步版本的库。使用 Parallel.ForEach 也不会提高性能。有没有更好的方法来做到这一点（提高资源效率并获得更好的性能）？

另外，当我尝试只在一个目录上运行时，所用时间几乎是两个目录所需时间的一半。由于这些操作都是独立的，我想并行运行它们，因此处理两个目录的时间应该与处理单个目录的时间大致相同（或稍微多一点），但不会慢两倍。似乎无论我做了多少 Task.Run() ，我都会同时运行有限数量的线程（一些瓶颈）。我尝试将所有 Task.Run() 更改为 new Thread()，并观察到更多线程同时处于活动状态，但最终导致性能更差。这是为什么呢？

Answer 1

Task.Run 调度在ThreadPool 上工作，这是一种保守的机制，它会根据需要立即创建多少线程（它会创建与机器的可用内核一样多的线程），以及它创建新线程的频率当工作需求很高时线程（每秒一个新线程）。您可以尝试使用影响ThreadPool 行为的ThreadPool.SetMinThreads 方法。例如：

ThreadPool.SetMinThreads(100, 100);

这样ThreadPool 将在切换到保守算法之前立即按需创建 100 个线程。

您的目录处理应用程序的性能可能不会有所改善。这是因为您的 I/O 绑定工作负载受到存储设备功能的限制。无论您如何处理代码，硬件对每个时间单位可以存储或检索的数据量都有限制。当您达到此限制时，提高性能的唯一方法就是升级您的硬件。

关于使用 Task.Run 和同步 API 来执行 I/O 绑定工作的适用性，令人惊讶的是，在许多情况下它是完成工作的最高效方式。特别是同步文件系统 API 是 significantly faster，而不是它们的异步对应物。使用同步 API 会失去内存效率。每个线程的堆栈至少需要1 MB 的内存，因此如果您一次启动 1,000 个线程，您将剥夺您的系统 1 GB 或更多的内存，这会间接影响您的应用程序的性能。

为了并行化的目的，使用Task.Run 手动启动任务是并行化工作的一种低杠杆方法。 TPL 提供了更高级别的基于 Task 的工具，例如 Parallel 类、PLINQ 库 (.AsParallel) 和 TPL Dataflow 库。

为了在后台工作期间使用进度信息更新 UI，现代方法是 IProgress<T> 接口和 Progress<T> 类。您可以找到一个示例here，作为Task.Run 和BackgroundWorker 类之间比较的一部分。

Answer 2

Task.Run(() => DoJob()); 和使用 Task.WaitAll() 是在浪费线程。

我会改成这样：

private string _directoryA;
private string _directoryB;

// The user clicks the button
private async void ButtonPressed()
{
    // disable UI controls
    try
    {
        await DoJob();
    }
    finally
    {
        // enable UI controls.
    }
} 

private async Task DoJob()
{
    var tasks = new List<Task>();
    var watch = Stopwatch.StartNew();

    tasks.Add(Task.Run(async () => DoJobForDirectory(_directoryA).ContinueWith(t => Console.WriteLine("First Half");
    tasks.Add(Task.Run(async () => DoJobForDirectory(_directoryB).ContinueWith(t => Console.WriteLine("Second Half");
    
    await Task.WhenAll(tasks.ToArray());

    watch.Stop();
    Console.WriteLine($"Time Taken : {watch.ElapsedMilliseconds} ms.");
    UpdateUI();
}

private async Task DoJobForDirectory(string directory)
{
    var files = Directory.EnumerateFiles(directory, "*.csv");
    var tasks = new List<Task>();
    foreach (var file in files)
    {
        // Update the progress bar in the UI when a file has finished processing
        tasks.Add(Task.Run(() => DoJobForFile(file)).ContinueWith(t => UpdateCounter++));
    }
    
    await Task.WhenAll(tasks.ToArray());
}

private void DoJobForFile(string filePath)
{
    ReadCSV();
    ProcessData();
    ...
}

如果你想限制线程，你可以使用 SemaphoreSlim。这是一个很好的例子作为公认的答案：

How to limit the Maximum number of parallel tasks in c#