如何在TVar上添加终结器

Question

背景

作为对问题的回应，我构建了和上传了一个 (对我来说上传jnb版本是不对的)。如果名称不够，有界的-tchan (BTChan)是一个具有最大容量的STM通道(如果通道具有容量，则写块)。

最近，我收到了一个请求，要求添加一个与常规TChan类似的dup功能。于是问题就开始了。

BTChan看上去如何

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

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

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

每个阅读器都会减少它读取的元素的计数器，然后将它的读指针移动到列表中的下一个元素。如果读取器将计数器减少到零，那么count的值就会减少，以正确地反映信道上的可用容量。

要明确所需的语义:通道容量表示在通道中排队的最大元素数。任何给定的元素都会排队，直到每个dup的读取器收到该元素为止。任何元素都不应该为GCed dup排队(这是主要问题)。

例如，假设有三个容量为2的通道(c1、c2、c3)，其中两个项被写入通道，然后从c1和c2中读取所有条目。因为c3还没有消耗它的副本，所以通道仍然是满的(剩余容量为0)。在任何时候，如果删除了对c3的所有引用(所以c3是GCed)，那么应该释放容量(在本例中恢复为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也不会减少。由于通道处于容量(最多为1元素)，写入器将始终阻塞。

我希望每次调用dupBTChan时都创建一个终结器。当收集一个被欺骗的(或原始的)通道时，在该通道上待读取的所有元素都会减少每个元素的计数，同时nrDups变量也会减少。因此，未来的写操作将具有正确的count (不为GCed通道未读取的变量保留空间的count )。

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

由于这个原因，JNB的边界-tchan实际上有手动资源管理。见cancelBTChan。我要找一些更难让用户出错的东西(不是说手动管理在很多情况下都不是正确的方法)。

解决方案2-通过阻塞TVars来使用异常(GHC不能以我想要的方式完成)

编辑这个解决方案，和解决方案3，这只是一个副产品，不工作！由于虫5055 (WONTFIX)，GHC编译器向两个阻塞线程发送异常，尽管其中一个是足够的(理论上是可确定的，但在GHC中不实用)。

如果获取BTChan的所有方法都是IO，那么我们可以在特定BTChan特有的额外(虚拟) TVar字段上读取/重试线程。当删除对TVar的所有其他引用时，新线程将捕获异常，因此它将知道何时减少nrDups和单个元素计数器。这应该有效，但强制我的所有用户使用IO获取他们的BTChan：

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，但仍然不支持

用户通过调用initBTChanCollector启动管理线程，该线程将监视一组虚拟TVars (来自解决方案2)，并进行所需的清理。基本上，它将IO推入另一个线程，该线程知道如何通过全局(unsafePerformIOed) TVar执行操作。事情基本像解决方案2，但是BTChan的创造仍然可以是STM。运行initBTChanCollector失败将导致进程运行时不断增加的任务列表(空间泄漏)。

BTChan**s**解决方案4:绝不允许丢弃

这类似于忽视这个问题。如果用户从未丢弃过一个被欺骗的BTChan，那么问题就会消失。

解决方案5 --我看到了ezyang的答案(完全有效，很受欢迎)，但是我真的很想用‘’函数来保持当前的API。

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

编辑:我已实施的解决方案3 (完全未经测试的alpha版本)，并通过使全局本身成为一个BTChan来处理潜在的空间泄漏-- chan可能应该有1的容量，所以忘记运行init会很快出现，但这是一个小变化。这在GHCi (7.0.3)中是可行的，但这似乎是偶然的。GHC向两个阻塞线程(读取BTChan和监视线程的有效线程)抛出异常，因此，如果您被阻塞，读取BTChan时，另一个线程丢弃它的引用，那么您就死定了。

Answer 1

下面是另一种解决方案:要求对有界通道副本的所有访问都由一个函数括起来，该函数在退出时释放其资源(异常或正常情况下)。你可以使用一个带有2级跑步者的单通道来防止重复的通道泄漏.它仍然是手动的，但类型系统使它更难做顽皮的事情。

您确实不希望依赖真正的IO终结器，因为GHC无法保证何时运行终结器:据您所知，在运行终结器之前，它可能要等到程序结束时才能运行，这意味着您在此之前处于僵局。