c# - 在热路径上使用异步 API 包装基于回调的 API 时避免分配并保持并发

Question

我在 StackOverflow 上阅读了很多关于使用 TaskCompletionSource 包装基于回调的 API 和基于 Task 的 API 的文章和问题，我正在尝试在与 Solace PubSub+ 消息代理通信时使用这种技术。

我最初的观察是这种技术似乎转移了并发的责任。例如，Solace 代理库有一个 Send() 方法，它可能会阻塞，然后我们在网络通信完成后得到一个回调，以指示“真正的”成功或失败。所以这个Send()方法可以被非常快速的调用，并且vendor库在内部限制了并发。

当你围绕它放置一个任务时，你似乎要么序列化操作(foreach message await SendWrapperAsync(message))，要么通过决定启动多少任务来自己接管并发责任(例如，使用 TPL 数据流)。

无论如何，我决定用保证器包装 Send 调用，保证器将永远重试，直到回调指示成功，并负责并发。这是一个“有保证”的消息传递系统。失败不是一种选择。这要求担保人可以施加背压，但这不在这个问题的范围内。我在下面的示例代码中对此有一些评论。

它的意思是我的热路径(包含发送 + 回调)由于重试逻辑而“特别热”。因此这里有很多 TaskCompletionSource 创建。

供应商自己的文档建议尽可能重用他们的 Message 对象，而不是为每个 Send 重新创建它们。为此，我决定使用 Channel 作为环形缓冲区。但这让我想知道 - 是否有一些替代 TaskCompletionSource 方法的方法 - 也许其他一些对象也可以缓存在环形缓冲区中并重复使用，从而实现相同的结果？

我意识到这可能是对微优化的过分热心尝试，老实说，我正在探索 C# 的几个方面，这些方面超出了我的薪水等级(我是 SQL 专家，真的)，所以我可能会遗漏一些东西明显的。如果答案是“你实际上不需要这种优化”，那我就不会放心了。如果答案是“这确实是唯一明智的方法”，我的好奇心就可以得到满足。

这是一个功能齐全的控制台应用程序，它模拟了 MockBroker 对象中 Solace 库的行为，以及我对它进行包装的尝试。我的热路径是 Guarantor 类中的 SendOneAsync 方法。代码对于 SO 来说可能有点太长，但它是我可以创建的最小演示，它捕获了所有重要元素。

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

internal class Message { public bool sent; public int payload; public object correlator; }

// simulate third party library behaviour
internal class MockBroker
{
    public bool TrySend(Message m, Action<Message> callback)
    {
        if (r.NextDouble() < 0.5) return false; // simulate chance of immediate failure / "would block" response
        Task.Run(() => { Thread.Sleep(100); m.sent = r.NextDouble() < 0.5; callback(m); }); // simulate network call
        return true;
    }

    private Random r = new();
}

// Turns MockBroker into a "guaranteed" sender with an async concurrency limit
internal class Guarantor
{
    public Guarantor(int maxConcurrency)
    {
        _broker = new MockBroker();
        // avoid message allocations in SendOneAsync
        _ringBuffer = Channel.CreateBounded<Message>(maxConcurrency);
        for (int i = 0; i < maxConcurrency; i++) _ringBuffer.Writer.TryWrite(new Message());
    }

    // real code pushing into a T.T.T.DataFlow block with bounded capacity and parallelism
    // execution options both equal to maxConcurrency here, providing concurrency and backpressure
    public async Task Post(int payload) => await SendOneAsync(payload);

    private async Task SendOneAsync(int payload)
    {
        Message msg = await _ringBuffer.Reader.ReadAsync();
        msg.payload = payload;
        // send must eventually succeed
        while (true)
        {
            // *** can this allocation be avoided? ***
            var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
            msg.correlator = tcs;
            // class method in real code, inlined here to make the logic more apparent
            Action<Message> callback = (msg) => (msg.correlator as TaskCompletionSource<bool>).SetResult(msg.sent);
            if (_broker.TrySend(msg, callback) && await tcs.Task) break;
            else
            {
                // simple demo retry logic
                Console.WriteLine($"retrying {msg.payload}");
                await Task.Delay(500);
            }
        }
        // real code raising an event here to indicate successful delivery
        await _ringBuffer.Writer.WriteAsync(msg);
        Console.WriteLine(payload);
    }

    private Channel<Message> _ringBuffer;
    private MockBroker _broker;
}

internal class Program
{
    private static async Task Main(string[] args)
    {
        // at most 10 concurrent sends
        Guarantor g = new(10);
        // hacky simulation since in this demo there's nothing generating continuous events,
        // no DataFlowBlock providing concurrency (it will be limited by the Channel instead),
        // and nobody to notify when messages are successfully sent
        List<Task> sends = new(100);
        for (int i = 0; i < 100; i++) sends.Add(g.Post(i));
        await Task.WhenAll(sends);
    }
}

Answer 1

是的，你可以避免分配TaskCompletionSource实例，通过使用轻量级 ValueTask s 而不是 Task秒。首先，您需要一个可以实现 IValueTaskSource<T> 的可重用对象接口(interface)，以及 Message似乎是完美的候选人。要实现此接口(interface)，您可以使用 ManualResetValueTaskSourceCore<T> 结构。这是一个可变结构，因此不应将其声明为readonly。 。您只需要将接口(interface)方法委托(delegate)给这个名称很长的结构的相应方法:

using System.Threading.Tasks.Sources;

internal class Message : IValueTaskSource<bool>
{
    public bool sent; public int payload; public object correlator;

    private ManualResetValueTaskSourceCore<bool> _source; // Mutable struct, not readonly

    public void Reset() => _source.Reset();
    public short Version => _source.Version;
    public void SetResult(bool result) => _source.SetResult(result);

    ValueTaskSourceStatus IValueTaskSource<bool>.GetStatus(short token)
        => _source.GetStatus(token);
    void IValueTaskSource<bool>.OnCompleted(Action<object> continuation,
        object state, short token, ValueTaskSourceOnCompletedFlags flags)
            => _source.OnCompleted(continuation, state, token, flags);
    bool IValueTaskSource<bool>.GetResult(short token) => _source.GetResult(token);
}

三个成员GetStatus , OnCompleted和 GetResult是实现接口(interface)所必需的。其他三个成员( Reset 、 Version 和 SetResult )将用于创建和控制 ValueTask<bool>

现在让我们包装 TrySend MockBroker 的方法异步方法中的类 TrySendAsync , 返回 ValueTask<bool>

static class MockBrokerExtensions
{
    public static ValueTask<bool> TrySendAsync(this MockBroker source, Message message)
    {
        message.Reset();
        bool result = source.TrySend(message, m => m.SetResult(m.sent));
        if (!result) message.SetResult(false);
        return new ValueTask<bool>(message, message.Version);
    }
}

message.Reset();重置 IValueTaskSource<bool> , 并声明前面的异步操作已经完成。 IValueTaskSource<T>一次只支持一个异步操作，生成的 ValueTask<T> 只能等待一次，在下一次Reset()之后就不能再等待了.这就是您为避免分配对象而必须付出的代价:您必须遵循更严格的规则。如果你试图违反规则(有意或无意)，ManualResetValueTaskSourceCore<T>将开始 throw InvalidOperationException到处都是。

现在让我们使用 TrySendAsync扩展方法:

while (true)
{
    if (await _broker.TrySendAsync(msg)) break;

    // simple demo retry logic
    Console.WriteLine($"retrying {msg.payload}");
    await Task.Delay(500);
}

您可以在 Console 中打印GC.GetTotalAllocatedBytes(true)整个操作前后，看区别。确保在 Release模式下运行应用程序，以查看真实情况。您可能会发现差异并不那么令人印象深刻，因为 TaskCompletionSource 的大小与 Task.Delay 分配的字节相比，实例非常小，以及所有 string s 是为在 Console 中编写内容而生成的.