问题标签 [mailboxprocessor]

问题总结：

1. Stream.WriteAsync 和 Stream.ReadAsync 是否不会相互阻塞并且可以在 2 个 cpu 内核上完全并行运行？
2. 有没有办法通知等待外部源异步通知？
3. MailboxProcessor 作为 TcpClient 的包装器对性能有何影响？

一些解释：

正如我从 Stream.cs 异步操作的源代码中了解到的那样，同步完成而不是完全并行（如果您读取某些内容，它会阻止其他读取尝试直到完成，并且还会阻止写入尝试，以其他方式相同 - 写入阻止读取和写入尝试） ，唯一的好处是它不会阻塞调用 writeAsync/readAsync 的线程

所以我已经将 TcpClient 包装到接受的 MailboxProcessor 中

它的目的是在连接失败时重新连接客户端，在此期间我们不想尝试读取或写入任何内容。这也可以通过线程同步技术实现，但这需要更多的代码。

当我可以对解决方案进行基准测试时，我没有找到合适的位置，但是是否有任何 MailboxProcessor 内部基准测试，以查看它将对 tcp 连接产生什么影响（如果就整个请求/响应时间而言性能影响很小）

这也是为了向服务器发送请求以保证响应顺序与收到的请求相同但我不能依赖Write requestA->Read responseA读取正确的响应：tread1: Write requestA Read responseA tram2: Write requestB Read responseB

队列：['写请求A'; '写请求B'; '阅读响应B'; 'read responseA'] 这将导致responseA 将返回给thread2，responseB 将返回给thread1。

好在请求-响应与请求中设置的 id 相关联。所以这个协议有一些解决方案来存储Dictionary<id, TaskCompletionSource>. 这使得可以等到 Task 完成（TaskCompletionSource 设置结果）然后继续处理结果。结果集由单独的单线程持续读取来自 tcp 流的响应并通过 id 映射它们。

在 F# 中，使用异步而不是任务非常漂亮，所以我看到如何将响应映射到正确请求的方式是存储Dictionary<id, response -> unit> 或者我只能为 MailboxProcessor 提供 1 条消息，Send request并且不将写入请求与读取响应分开。

是否有任何其他类似于 TaskCompletionSource 但用于异步的方式，所以我可以：

我如何将 test.txt 的内容发送到电子邮件，任何人都可以。

我如何将 test.txt 的内容发送到电子邮件，任何人都可以。

我对下面的代码示例和人们的想法有点好奇。这个想法是从 NetworkStream (~20 msg/s) 中读取数据，而不是在 main 中工作，而是将内容传递给 MainboxProcessor 以在完成后处理并取回内容以进行绑定。

通常的方法是使用 PostAndReply，但我想在 C# 中绑定到 ListView 或其他控件。无论如何，必须对 LastN 项和过滤做魔术。另外，Rx 有一些错误处理。

下面的示例观察2..10中的数字并返回“hello X”。在8时，它像 EOF 一样停止。将其设置为 ToEnumerable 是因为其他线程在其他线程之前完成，但它也适用于订阅。

困扰我的是：

1. 以递归方式传递 Subject(obj)。我认为其中大约 3-4 个没有任何问题。好主意？
2. 对象的生命周期。

基本上，我想将以下内容更改为有限线程解决方案，因为在我的情况下，计算列表太大，产生了太多线程，我想用更少的线程进行实验和测量性能。

我的新方法，这段代码是由 Tomas Petricek 的这个在线片段借用/启发的（我测试过，它有效，但我需要它返回一个值，而不是单位）。

现在，有了这个，原始代码将变为：

总而言之，输出是正确的（它确实计算并传播回回复），但它在（有限的）线程中没有这样做。

我一直在玩，但认为我错过了明显的东西（此外，谁知道，有人可能喜欢有限线程邮箱处理器的想法，它可以按顺序返回其计算）。

我正在寻找 F# 或 C# 中的生产者/消费者通信机制，该机制允许多个消费者同时处理传入消息并生成对多个生产者的回复。System.Collections.Concurrent.BlockingCollection 还不错，但不像 F# 邮箱处理器那样面向消息。非常需要基于消息的抽象（基于请求/回复）。与 F# 邮箱处理器内部的可变状态不同，它不是必需的。如果从邮箱（消息队列）读取会阻塞，那么线程应该被释放到线程池中，就像在 F# 中一样。节流阀（降低生产者速度的流量控制）将是理想的，但不是必需的。有任何想法吗？

我正在尝试在 F# 中测试 MailboxProcessor。我想测试我给出的函数 f 在发布消息时是否实际执行。

原始代码使用 Xunit，但我制作了一个 fsx，可以使用 fsharpi 执行。

到目前为止，我正在这样做：

这段代码有效，但对我来说看起来不太好。

1) 在 f# 中 TaskCompletionSource 的使用是正常的，还是有一些专门的东西可以让我等待完成？

2）我在 waitingFor 函数中使用第二个参数来约束它，我知道我可以使用类型 MyType<'a>() 来做到这一点，还有其他选择吗？我宁愿不使用我觉得很麻烦的新 MyType。

3）除了这样做之外，还有其他选择来测试我的代理吗？到目前为止，我发现的唯一一篇关于该主题的文章是 2009 年的这篇博文http://www.markhneedham.com/blog/2009/05/30/f-testing-asynchronous-calls-to-mailboxprocessor/

我正在使用代理（MailboxProcessor）在需要响应的地方进行一些有状态的处理。

• 来电者使用MailboxProcessor.PostAndAsyncReply
• 在代理内部，会给出一个响应AsyncReplyChannel.Reply

但是，通过查看 f# 源代码，我发现在响应传递之前，代理不会处理下一条消息。总的来说，这是一件好事。但在我的情况下，代理更希望继续处理消息而不是等待响应传递。

做这样的事情来传递响应是否有问题？（或者有更好的选择吗？）

我意识到这种方法并不能保证响应将按顺序传递。我没关系。

参考示例

我试图将其减少到尽可能小的复制，但它仍然有点长，我很抱歉。

我有一个 F# 项目，它引用了一个 C# 项目，代码如下。

这是 F# 代码。

问题是，取消后，控制永远不会返回到异步块。我不确定它是挂在里面AwaitTask还是Catch. 直觉告诉我它在尝试返回之前的同步上下文时被阻塞，但我不确定如何确认这一点。我正在寻找有关如何解决此问题的想法，或者也许在这里有更深入了解的人可以发现问题。

可能的解决方案

我正在研究具有多种状态的 F# 代理，即使用“let rec/and”关键字组合（根据 Expert F# 3.0 的“消息处理和状态机”）来提供多个异步块。到目前为止，我能找到的唯一示例是此处讨论的“节流代理” （也是 Fssnip.net）。是否有任何其他资源可用于学习此模式？

编辑：我的特定应用程序是具有两种状态的代理，

我想提供一些方法来传递进给率乘数值，该值通过在“馈送”异步状态下传入的乘数增加/减少延迟，而不会中断值的馈送。我想这个问题可以归结为“你如何保持异步状态块主动循环，同时仍然知道新消息？” 几乎就像跳过收件箱一样。接收异步等待，除非确实有消息进来？收件箱.scan？

编辑 2：鉴于 MailboxProcessor 的消息队列方面，我可以看到代理接收并放入队列的外部消息（具有不同的 rateMultiplier 值）将成功更改速率，而不会中断数据值的流出。任何关于“暂停”的建议仍将不胜感激。

我在这里有一个多状态 F# MailboxProcessor 示例，只是想知道为什么它可以编译但行为出乎意料 - F# 代理可以在传入的 lambda 函数中只有一个 inbox.Receive() 语句吗？我正在尝试遵循“Expert F# 3.0”第 284 页中提供的一般示例模式，其中使用多个 async {} 主体允许多个状态，但没有具体说明是否可以使用 inbox.Receive()在每个异步？

产量

所以让！跳过循环 1 中的 msg = inbox.Receive()？我原以为loop2是由return完成的！loop1 和那个让！inbox.Receive() 的分配特定于它使用的异步块。