haskell - 如何在 TVar 上添加终结器

Question

背景

回复 question ，我建和uploaded a bounded-tchan (我不适合上传 jnb's version )。如果名称不够，则有界 tchan (BTChan) 是具有最大容量的 STM channel (如果 channel 已满则写入块)。

最近，我收到了添加像 regular TChan's 中的 dup 功能的请求。 .问题就这样开始了。

BTChan 的外观

下面是 BTChan 的简化(实际上是非功能性的) View 。

data BTChan a = BTChan
    { max :: Int
    , count :: TVar Int
    , channel :: TVar [(Int, a)]
    , nrDups  :: TVar Int
    }

每次写入 channel 时，都会在元组中包含重复次数( nrDups ) - 这是一个“单个元素计数器”，指示有多少读者获得了该元素。

每个读取器都会减少它读取的元素的计数器，然后将它的读取指针移动到列表中的下一个元素。如果阅读器将计数器递减为零，则 count 的值递减以正确反射(reflect)信道上的可用容量。

明确所需的语义: channel 容量表示在 channel 中排队的最大元素数。任何给定的元素都会排队，直到每个 dup 的阅读器都收到了该元素。没有元素应该保持排队等待 GCed dup(这是主要问题)。

例如，假设容量为 2 的 channel (c1, c2, c3) 有 3 个副本，其中将 2 个项目写入 channel ，然后从 c1 中读出所有项目。和 c2 .由于 c3 channel 仍然已满(剩余容量为 0)没有消耗它的副本。在任何时间点，如果所有对 c3 的引用被丢弃(所以 c3 被垃圾回收)然后容量应该被释放(在这种情况下恢复到 2)。

这是问题所在:假设我有以下代码

c <- newBTChan 1
_ <- dupBTChan c  -- This represents what would probably be a pathological bug or terminated reader
writeBTChan c "hello"
_ <- readBTChan c

导致 BTChan 看起来像:

BTChan 1 (TVar 0) (TVar []) (TVar 1)             -->   -- newBTChan
BTChan 1 (TVar 0) (TVar []) (TVar 2)             -->   -- dupBTChan
BTChan 1 (TVar 1) (TVar [(2, "hello")]) (TVar 2) -->   -- readBTChan c
BTChan 1 (TVar 1) (TVar [(1, "hello")]) (TVar 2)       -- OH NO!

注意最后 "hello" 的读取计数还在 1 ?这意味着消息不会被认为消失(即使它会在实际实现中被 GCed)和我们的 count永远不会递减。由于 channel 已满(最多 1 个元素)，因此编写器将始终阻塞。

我想要每次都创建一个终结器 dupBTChan叫做。当收集复制(或原始) channel 时，该 channel 上剩余要读取的所有元素将减少每个元素的计数， nrDups 也是如此。变量将减少。因此， future 的写入将具有正确的 count (a count 不会为 GCed channel 未读取的变量保留空间)。

解决方案 1 - 手动资源管理(我想避免的)

JNB的bounded-tchan实际上就是因为这个原因有手动资源管理。见 cancelBTChan .我要让用户更难出错(并不是说手动管理在很多情况下都不是正确的方法)。

解决方案 2 - 通过阻止 TVar 来使用异常(GHC 无法按照我的方式执行此操作)

编辑此解决方案，而解决方案 3 只是衍生产品，不起作用!由于 bug 5055 (WONTFIX) GHC 编译器向两个阻塞的线程发送异常，即使一个就足够了(理论上可以确定，但对于 GHC GC 不实用)。

如果所有的方法都得到了 BTChan是IO，我们可以 forkIO在给定 BTChan 唯一的额外(虚拟)TVar 字段上读取/重试的线程.当对 TVar 的所有其他引用都被删除时，新线程将捕获异常，因此它将知道何时减少 nrDups和单个元素计数器。这应该可以工作，但会强制我所有的用户使用 IO 来获取他们的 BTChan s:

data BTChan = BTChan { ... as before ..., dummyTV :: TVar () }

dupBTChan :: BTChan a -> IO (BTChan a)
dupBTChan c = do
       ... as before ...
       d <- newTVarIO ()
       let chan = BTChan ... d
       forkIO $ watchChan chan
       return chan

watchBTChan :: BTChan a -> IO ()
watchBTChan b = do
    catch (atomically (readTVar (dummyTV b) >> retry)) $ \e -> do
    case fromException e of
        BlockedIndefinitelyOnSTM -> atomically $ do -- the BTChan must have gotten collected
            ls <- readTVar (channel b)
            writeTVar (channel b) (map (\(a,b) -> (a-1,b)) ls)
            readTVar (nrDup b) >>= writeTVar (nrDup b) . (-1)
        _ -> watchBTChan b

编辑:是的，这是一个穷人的终结者，我没有任何特别的理由避免使用 addFinalizer .那将是相同的解决方案，仍然强制使用 IO afaict。

解决方案 3:比解决方案 2 更简洁的 API，但 GHC 仍然不支持它

用户通过调用 initBTChanCollector 启动管理器线程，它将监视一组这些虚拟 TVar(来自解决方案 2)并进行所需的清理。基本上，它将 IO 推送到另一个线程中，该线程知道通过全局( unsafePerformIO ed) TVar 做什么.事情基本上像解决方案2一样工作，但BTChan的创建仍然可以是STM。运行失败 initBTChanCollector随着进程的运行，将导致不断增长的任务列表(空间泄漏)。

解决方案 4:永远不允许丢弃 BTChan s

这类似于忽略问题。如果用户从不丢弃被欺骗的 BTChan那么问题就消失了。

解决方案 5
我看到 ezyang 的回答(完全有效且值得赞赏)，但我真的很想保留当前的 ​​API 只使用“dup”功能。

** 解决方案 6**
请告诉我有更好的选择。

编辑:
我 implemented solution 3 (完全未经测试的 alpha 版本)并通过使全局本身成为 BTChan 来处理潜在的空间泄漏。 -那个chan应该有1的容量所以忘记运行 init出现得非常快，但这是一个很小的变化。这适用于 GHCi (7.0.3) 但这似乎是偶然的。 GHC 向两个被阻塞的线程(读取 BTChan 和观察线程的有效线程)抛出异常，所以如果当另一个线程丢弃它的引用时你被阻塞读取 BTChan，那么你就会死。

Answer 1

这是另一种解决方案:要求对有界 channel 副本的所有访问都被一个在退出时释放其资源的函数括起来(通过异常或正常)。您可以使用带有 rank-2 runner 的 monad 来防止重复的 channel 泄漏。它仍然是手动的，但类型系统使得做淘气的事情变得更加困难。

您真的不想依赖真正的 IO 终结器，因为 GHC 无法保证终结器何时可以运行:据您所知，它可能会等到程序结束后再运行终结器，这意味着您已陷入僵局直到那时。