在递归方法中使用 async/await 的正确方法是什么？

Question

在递归方法中使用 async/await 的正确方法是什么？这是我的方法：

public string ProcessStream(string streamPosition)
{
    var stream = GetStream(streamPosition);

    if (stream.Items.count == 0)
        return stream.NextPosition;

    foreach(var item in stream.Items) {
        ProcessItem(item);
    }

    return ProcessStream(stream.NextPosition)
}

这里是异步/等待的方法：

public async Task<string> ProcessStream(stringstreamPosition)
{
        var stream = GetStream(streamPosition);

        if (stream.Items.count == 0)
            return stream.NextPosition;

        foreach(var item in stream.Items) {
            await ProcessItem(item); //ProcessItem() is now an async method
        }

        return await ProcessStream(stream.NextPosition);
 }

Answer 1

虽然我必须提前说，该方法的意图对我来说并不完全清楚，但用一个简单的循环重新实现它是非常简单的：

public async Task<string> ProcessStream(string streamPosition)
{
    while (true)
    {
        var stream = GetStream(streamPosition);

        if (stream.Items.Count == 0)
            return stream.NextPosition;

        foreach (var item in stream.Items)
        {
            await ProcessItem(item); //ProcessItem() is now an async method
        }

        streamPosition = stream.NextPosition;
    }
}

递归不是堆栈友好的，如果您可以选择使用循环，那么在简单的同步场景（其中控制不佳的递归最终导致StackOverflowExceptions）以及异步场景中，它绝对值得研究，老实说，我什至不知道如果你把事情推得太远会发生什么（每当我尝试使用async 方法重现已知的堆栈溢出场景时，我的 VS 测试资源管理器就会崩溃）。

Recursion and the await / async Keywords 等答案表明 StackOverflowException 与 async 的问题不大，因为 async/await 状态机的工作方式，但这不是我探索过的东西，因为我倾向于避免递归尽可能。

Answer 2

当我添加代码以使您的示例更具体时，我发现递归结果很糟糕的两种可能方式。他们都假设你的数据非常大，需要特定的条件才能触发。

1. 如果ProcessItem(string) 返回一个在awaited 之前完成的Task（或者，我假设它在await 完成旋转之前完成），则继续将同步执行。在下面的代码中，我通过让ProcessItem(string) 返回Task.CompletedTask 来模拟这一点。当我这样做时，程序 very 很快以StackOverflowException 终止。这是因为 .net 的 TPL “ReleasesZalgo” by opportunistically executing continuations synchronously 不考虑当前堆栈中有多少可用空间。这意味着它会加剧使用递归算法已经存在的潜在堆栈空间问题。要查看这一点，请在下面的代码示例中注释掉 await Task.Yield();。
2. 如果您使用某种技术来阻止 TPL 同步继续（下面我使用Task.Yield()），最终程序将耗尽内存并以OutOfMemoryException 终止。如果我理解正确，如果return await 能够模拟尾调用优化，就不会发生这种情况。我想这里发生的事情是每个调用都会生成类似于簿记Task<string> 的东西，并且即使它们可以合并也会继续生成它们。要使用下面的示例重现此错误，请确保您以 32 位运行程序，禁用 Console.WriteLine() 调用（因为控制台真的很慢），并确保取消注释 await Task.Yield()。

using System;
using System.Collections.Generic;
using System.Threading.Tasks;

// Be sure to run this 32-bit to avoid making your system unstable.
class StreamProcessor
{
    Stream GetStream(string streamPosition)
    {
        var parsedStreamPosition = Convert.ToInt32(streamPosition);
        return new Stream(
            // Terminate after we reach 0.
            parsedStreamPosition > 0 ? new[] { streamPosition, } : new string[] { },
            Convert.ToString(parsedStreamPosition - 1));
    }

    Task ProcessItem(string item)
    {
        // Comment out this next line to make things go faster.
        Console.WriteLine(item);
        // Simulate the Task represented by ProcessItem finishing in
        // time to make the await continue synchronously.
        return Task.CompletedTask;
    }

    public async Task<string> ProcessStream(string streamPosition)
    {
        var stream = GetStream(streamPosition);

        if (stream.Items.Count == 0)
            return stream.NextPosition;

        foreach (var item in stream.Items)
        {
            await ProcessItem(item); //ProcessItem() is now an async method
        }

        // Without this yield (which prevents inline synchronous
        // continuations which quickly eat up the stack),
        // you get a StackOverflowException fairly quickly.
        // With it, you get an OutOfMemoryException eventually—I bet
        // that “return await” isn’t able to tail-call properly at the Task
        // level or that TPL is incapable of collapsing a chain of Tasks
        // which are all set to resolve to the value that other tasks
        // resolve to?
        await Task.Yield();

        return await ProcessStream(stream.NextPosition);
    }
}

class Program
{
    static int Main(string[] args) => new Program().Run(args).Result;
    async Task<int> Run(string[] args)
    {
        await new StreamProcessor().ProcessStream(
            Convert.ToString(int.MaxValue));
        return 0;
    }
}

class Stream
{
    public IList<string> Items { get; }
    public string NextPosition { get; }
    public Stream(
        IList<string> items,
        string nextPosition)
    {
        Items = items;
        NextPosition = nextPosition;
    }
}

所以，我想我的两个建议是：

1. 如果您不确定递归的堆栈增长是否会被其他事物中断，请使用 Task.Yield()。
2. As suggested already，如果一开始对您的问题没有意义，请避免递归。即使它是一个干净的算法，如果你的问题规模是无限的，也要避免它。