在热路径上使用异步 API 包装基于回调的 API 时避免分配并保持并发
Avoiding allocations and maintaining concurrency when wrapping a callback-based API with an async API on a hot path
我在 Whosebug 上阅读了很多关于使用 TaskCompletionSource 将基于 API 的回调与基于 Task 的回调包装在一起的文章和问题，我正在尝试在与 Solace PubSub+ 消息代理通信时使用这种技术。
我最初的观察是这种技术似乎转移了并发的责任。例如，Solace 代理库有一个 Send() 方法，它可能会阻塞，然后我们在网络通信完成后得到一个回调，以指示“真正的”成功或失败。所以这个Send()方法可以很快调用，vendor库内部限制并发
当你围绕它放置一个任务时，你似乎要么序列化操作（foreach message await SendWrapperAsync(message)），要么通过决定启动多少任务（例如，使用 TPL 数据流）自己接管并发责任。
无论如何，我决定用保证器包装 Send 调用，保证器将永远重试，直到回调指示成功，并负责并发。这是一个“有保证”的消息传递系统。失败不是一种选择。这要求担保人可以施加背压，但这不在这个问题的范围内。我在下面的示例代码中对此有一些评论。
它的意思是我的热路径（包含发送 + 回调）由于重试逻辑而“特别热”。所以这里有很多 TaskCompletionSource 创作。
供应商自己的文档建议尽可能重复使用他们的 Message 对象，而不是为每个 Send 重新创建它们。为此，我决定使用 Channel 作为环形缓冲区。但这让我想知道 - TaskCompletionSource 方法是否有其他替代方法 - 也许其他一些对象也可以缓存在环形缓冲区中并重复使用，从而达到相同的结果？
我意识到这可能是对微优化的过度热心尝试，老实说，我正在探索 C# 的几个方面，这些方面超出了我的薪水等级（我是 SQL 人，真的），所以我可能会遗漏一些明显的东西。如果答案是“你实际上不需要这种优化”，那我就不会放心了。如果答案是“这确实是唯一明智的方法”，我的好奇心就可以得到满足。
这是一个功能齐全的控制台应用程序，它模拟了 MockBroker 对象中 Solace 库的行为，以及我对它进行包装的尝试。我的热路径是Guarantorclass中的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);
}
}
是的，您可以通过使用轻量级 ValueTasks instead of Tasks. At first you need a reusable object that can implement the IValueTaskSource<T> interface, and the Message seems like the perfect candidate. For implementing this interface you can use the ManualResetValueTaskSourceCore<T> 结构来避免分配 TaskCompletionSource 个实例。这是一个可变结构，因此 不应将其声明为 readonly。您只需要将接口方法委托给这个名称很长的结构的相应方法：
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三个成员。其他三个成员（Reset、Version 和 SetResult）将用于创建和控制 ValueTask<bool>。
现在让我们将 MockBroker class 的 TrySend 方法包装在异步方法 TrySendAsync 中，即 returns 一个 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> 将开始到处乱扔 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 分配的字节以及为在 Console.
中写东西
我在 Whosebug 上阅读了很多关于使用 TaskCompletionSource 将基于 API 的回调与基于 Task 的回调包装在一起的文章和问题，我正在尝试在与 Solace PubSub+ 消息代理通信时使用这种技术。
我最初的观察是这种技术似乎转移了并发的责任。例如，Solace 代理库有一个 Send() 方法，它可能会阻塞，然后我们在网络通信完成后得到一个回调，以指示“真正的”成功或失败。所以这个Send()方法可以很快调用，vendor库内部限制并发
当你围绕它放置一个任务时，你似乎要么序列化操作（foreach message await SendWrapperAsync(message)），要么通过决定启动多少任务（例如，使用 TPL 数据流）自己接管并发责任。
无论如何，我决定用保证器包装 Send 调用，保证器将永远重试，直到回调指示成功，并负责并发。这是一个“有保证”的消息传递系统。失败不是一种选择。这要求担保人可以施加背压，但这不在这个问题的范围内。我在下面的示例代码中对此有一些评论。
它的意思是我的热路径（包含发送 + 回调）由于重试逻辑而“特别热”。所以这里有很多 TaskCompletionSource 创作。
供应商自己的文档建议尽可能重复使用他们的 Message 对象，而不是为每个 Send 重新创建它们。为此，我决定使用 Channel 作为环形缓冲区。但这让我想知道 - TaskCompletionSource 方法是否有其他替代方法 - 也许其他一些对象也可以缓存在环形缓冲区中并重复使用，从而达到相同的结果？
我意识到这可能是对微优化的过度热心尝试，老实说，我正在探索 C# 的几个方面，这些方面超出了我的薪水等级（我是 SQL 人，真的），所以我可能会遗漏一些明显的东西。如果答案是“你实际上不需要这种优化”，那我就不会放心了。如果答案是“这确实是唯一明智的方法”，我的好奇心就可以得到满足。
这是一个功能齐全的控制台应用程序，它模拟了 MockBroker 对象中 Solace 库的行为，以及我对它进行包装的尝试。我的热路径是Guarantorclass中的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);
}
}
是的，您可以通过使用轻量级 ValueTasks instead of Tasks. At first you need a reusable object that can implement the IValueTaskSource<T> interface, and the Message seems like the perfect candidate. For implementing this interface you can use the ManualResetValueTaskSourceCore<T> 结构来避免分配 TaskCompletionSource 个实例。这是一个可变结构，因此 不应将其声明为 readonly。您只需要将接口方法委托给这个名称很长的结构的相应方法：
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三个成员。其他三个成员（Reset、Version 和 SetResult）将用于创建和控制 ValueTask<bool>。
现在让我们将 MockBroker class 的 TrySend 方法包装在异步方法 TrySendAsync 中，即 returns 一个 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> 将开始到处乱扔 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 分配的字节以及为在 Console.