如何为 Haskell STM 通道实现相当于 Go 的 select 语句？

Posted 2023-03-14

【中文标题】如何为 Haskell STM 通道实现相当于 Go 的 select 语句？

【英文标题】：How to implement the equivalent of Go's select statement for Haskell STM channels?

【发布时间】：2014-08-28 00:09:53

【问题描述】：

Go 语言有一个select 语句，可用于轮询多个通道并根据哪个通道首先非空来执行特定操作。

例如

select 
  case a := <- chanA:
    foo(a)
  case b := <- chanB:
    baz(b)
  case c := <- chanC:
    bar(c)

这将等到chanA、chanB 或chanC 非空，然后如果chanB 非空，它将从chanB 读取并将结果存储在@987654329 @，然后拨打baz(b)。还可以添加default: 子句，这意味着select 语句不会在通道上等待，而是会在所有通道为空时执行default 子句的任何操作。

在 Haskell 中为 STM TChans 实现这样的最佳方法是什么？它可以通过 if-else 链简单地完成：检查每个 chan isEmptyChan，如果它不为空，则从中读取并调用适当的函数，或者如果所有通道都是空的，则调用 retry。我想知道是否会有更优雅/惯用的方式来做到这一点？

请注意，Go 的 select 语句也可以在其 case 中包含 send 语句，并且只有在其通道为空时才会完成一个 send 语句。如果该功能也可以复制，那就太好了，尽管我不确定是否会有一种优雅的方式来做到这一点。

只是稍微相关，但我刚刚注意到了一些内容，但我不确定在哪里发布：retry 描述中的 Control.Monad.STM 页面上有一个错字：

“实现可能会阻塞线程，直到它已读取的 TVar 之一被更新。”

【问题讨论】：

你可能想从Control.Concurrent.Async看race。

值得注意的是，go 不会执行第一个可用的操作，而是随机选择的任何可用的操作。它特别不会仅仅因为它们是稍后定义的或在选择路径中不走运而使频道饿死。

这与 Go 的 select 完全不同。 Go 中的通道是有界的，不像 TChan（使它们真正有用），select 可以与发送操作一起使用。

【参考方案1】：

您可以使用orElse 实现select 语义（用于读取和写入）（注意：它特定于ghc。） 例如：

forever $ atomically $
  writeTChan chan1 "hello" `orElse` writeTChan chan2 "world" `orElse` ...

这个想法是，当一个动作重试时（例如，您正在写入 chan，但它已满；或者您正在读取 chan，但它是空的），执行第二个动作。 default 语句只是一个 return () 作为链中的最后一个操作。

添加： 正如@Dustin 所说，去选择随机分支是有充分理由的。可能最简单的解决方案是在每次迭代中随机播放操作，在大多数情况下应该没问题。正确复制 go 语义（仅对活动分支进行洗牌）有点困难。可能要手动检查所有分支的isEmptyChan。

【讨论】：

像达斯汀上面评论的那样，这不会有饥饿问题吗？

@Dan 我相信你是对的：（假设 Yuras 的示例使用 readTChan）readTChan chanN 只会在所有通道 N 为空时读取。因此，饥饿的可能性实际上比您最初预期的还要严重。

谢谢！我没有意识到orElse 可以像那样被锁住。我也没有考虑过饿死的可能性。

【参考方案2】：

避免饥饿

foreverK :: (a -> m a) -> a -> m ()
foreverK loop = go
 where go = loop >=> go

-- Existential, not really required, but feels more like the Go version
data ChanAct = Action (TChan a) (a -> STM ())

perform :: STM ()
perform (Action c a) = readTChan c >>= a

foreverSelectE :: [ChanAct] -> STM ()
foreverSelectE = foreverSelect . map perform

foreverSelect :: [STM ()] -> STM ()
foreverSelect = foreverK $ \xs -> first xs >> return (rotate1 xs)

-- Should only be defined for non-empty sequences, but return () is an okay default.
-- Will NOT block the thread, but might do nothing.
first :: [STM ()] -> STM ()
first = foldr orElse (return ())

-- Should only be defined for non-empty sequences, really.
-- Also, using a list with O(1) viewL and snoc could be better.
rotate1 :: [a] -> [a]
rotate1 []    = []
rotate1 (h:t) = t ++ [h]

example = foreverSelectE
    [ Action chanA foo
    , Action charB baz
    , Action chanC bar
    ]

为了永远避免，您可以改为使用 mkSelect :: [STM ()] -> STM (STM ()) 来“隐藏”一个 TVar [STM ()] 并在每次使用时旋转它，如下所示：

example1 :: STM ()
example1 = do
    select <- mkSelect [actions] -- Just set-up
    stuff1
    select -- does one of the actions
    stuff2
    select -- does one of the actions

main = OpenGL.idleCallback $= atomically example1

扩展该技术，您可以选择报告它是否执行了一个动作，或者它执行了哪个动作，甚至循环，直到所有动作都被阻塞，等等。