TMVar，但没有缓冲区？答案

【问题标题】：TMVar, but without the buffer?TMVar，但没有缓冲区？
【发布时间】：2015-06-13 22:08:53
【问题描述】：

我正在尝试在 Haskell 轻量级线程之间进行通信。线程希望相互发送消息以进行通信和同步。

我最初为此使用 TMVar，但我刚刚意识到语义是错误的：TMVar 将在内部存储一条消息，因此将消息定位到空的 TMVar 不会堵塞。仅当您将消息发布到 full TMVar 时，它才会被阻止。

谁能推荐一个类似的STM IPC 结构：

将导致所有写入阻塞，直到消息被消耗；
在提供消息之前会导致所有读取被阻塞？

即零长度管道是理想的；但我不认为BoundedChan 的容量为 0 会很高兴。（另外，它不是 STM。）

【问题讨论】：

你的外部问题是什么？你为什么要这个？
您是否愿意考虑一种解决方案，使发送成为IO 操作而不是STM 操作（但接收仍然是STM 操作）？反之亦然（发送是STM 操作，但接收是IO 操作）？
我需要能够组合操作（以便能够一次读取其中的几个内容，并在第一件事变得可读时立即解除阻塞）。 AFAIK 这只能通过 STM 来完成。
@Daniel Wagner：事实上，是的。见下文。

标签： multithreading haskell stm

【解决方案1】：

如果我正确理解您的问题，我认为您不能，因为事务保证意味着事务 A 在事务 B 提交之前无法从事务 B 的写入中读取，此时它不能再阻塞。

TMVar 是您使用 STM 时最接近的结果。使用 IO，您可以构建一个仅在读取器可用时才完成写入的结构（此结构可能已经存在，但我不知道）。

【讨论】：

在其他系统中，我会通过将信号量连接在一起来构建它。我需要坐下来好好考虑一下，但是写入有两个阻塞阶段（等待空，然后发布消息，然后再次等待空），而读取只有一个（等待完整，然后读取消息） .因此，将写入作为单个事务可能没有意义，即使这失去了我的可组合性。 TMVar 源代码非常简单；我可能会尝试添加一个“阻止直到空”操作。如果我尝试在事务中做一些不可能的事情，STM 会出错还是死锁？
两者：如果死锁，STM 会出错。您不能在与“put”相同的事务中添加“block until empty”，因为它只会死锁，因为事务永远不会被提交（也因为 STM 没有阻塞的概念）。
你，我需要在两个事务中完成它并依靠更高级别的东西来保持它的一致性。 ...我很惊讶这样的事情还不存在；基本上我想要的是琳达。肯定有人这样做了？
如果您将其设为两个事务，则它不再是可组合的，并且您将失去 STM 的事务保证。在 IO 中使用 MVar 实现这一点会更容易、更明智。
是的，我必须改变我的设计。我唯一需要可组合性的就是等待多个渠道的变化；鉴于等待直到消耗阶段仅在一个通道变为可写之后发生，这可能发生在不同的事务中。我还没有找到编写 MVar 的方法。

【解决方案2】：

我建议重新制定这两个要求：

将导致所有写入阻塞，直到消息被消耗；

将导致所有读取被阻止，直到提供消息。

问题在于术语block 和consumed/provided。对于 STM，没有 block 的概念，只有 retry，它具有不同的语义：它重新启动当前事务 - 它不会等到某事发生（这可能导致死锁）。所以我们不能说“block until ...”，我们只能说“事务只有在...时才成功”。

同样，“直到消息被消费/提供”是什么意思？由于事务是原子的，因此只能“直到消费/提供消息的事务成功”。

所以让我们尝试重新表述：

将导致所有写入重试，直到使用该消息的事务成功；

将导致所有读取重试，直到提供消息的事务成功为止。

但是现在第一点没有意义：如果写入重试，则没有消息可以使用，事务没有暂停，它被丢弃并重新开始 - 可能会产生不同的消息！

换句话说：任何数据只有在成功（完成）时才能离开 STM 事务。这是设计使然 - 从外部世界/其他交易的角度来看，交易始终是原子 - 您永远无法仅观察交易的一部分的结果。您永远无法观察到两个事务交互。

因此，长度为 0 的队列是一个不好的类比——尽管它永远不允许传递任何数据。在任何事务结束时，它必须是空的，因此不会有任何数据通过。

尽管如此，我相信可以根据您的目标重新制定要求并随后找到解决方案。

【讨论】：

【解决方案3】：

你说你会喜欢在IO 而不是STM 的一方或另一方。因此，编写此代码并不难。让我们从在IO 中接收的版本开始。为此，接收者必须发起握手。

type SynchronousVar a = TChan (TMVar a)

send :: SynchronousVar a -> a -> STM a
receive :: SynchronousVar a -> IO a

send svar a = do
    tmvar <- readTChan svar
    putTMVar tmvar a

receive svar = do
    tmvar <- newEmptyTMVarIO
    atomically $ writeTChan svar tmvar
    atomically $ takeTMVar tmvar

可以编写一个类似的协议，让发送开始握手。

type SynchronousVar a = TChan (a, TMVar ())

send :: SynchronousVar a -> a -> IO a
receive :: SynchronousVar a -> STM a

send svar a = do
    tmvar <- newEmptyTMVarIO
    atomically $ writeTChan svar (a, tmvar)
    atomically $ takeTMVar tmvar

receive svar = do
    (a, tmvar) <- readTChan svar
    putTMvar tmvar ()
    return a

可能，如果您真的需要同步通信，这是因为您需要双向通信（即在 IO 中运行的操作想要了解与它同步的线程的一些信息）。扩展上述协议以传递更多关于同步的信息并不难（在前一种情况下通过将其添加到单元组或在后一种情况下添加到TMVar）。

【讨论】：