How to add a finalizer on a TVar

Question

Background

In response to a question, I built and uploaded a bounded-tchan (wouldn't have been right for me to upload jnb's version). If the name isn't enough, a bounded-tchan (BTChan) is an STM channel that has a maximum capacity (writes block if the channel is at capacity).

Recently, I've received a request to add a dup feature like in the regular TChan's. And thus begins the problem.

How the BTChan looks

A simplified (and actually non-functional) view of BTChan is below.

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

Every time you write to the channel you include the number of dups (nrDups) in the tuple - this is an 'individual element counter' which indicates how many readers have gotten this element.

Every reader will decrement the counter for the element it reads then move it's read-pointer to then next element in the list. If the reader decrements the counter to zero then the value of count is decremented to properly reflect available capacity on the channel.

To be clear on the desired semantics: A channel capacity indicates the maximum number of elements queued in the channel. Any given element is queued until a reader of each dup has received the element. No elements should remain queued for a GCed dup (this is the main problem).

For example, let there be three dups of a channel (c1, c2, c3) with capacity of 2, where 2 items were written into the channel then all items were read out of c1 and c2. The channel is still full (0 remaining capacity) because c3 hasn't consumed its copies. At any point in time if all references toc3 are dropped (so c3 is GCed) then the capacity should be freed (restored to 2 in this case).

Here's the issue: let's say I have the following code

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

Causing the BTChan to look like:

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!

Notice at the end the read count for "hello" is still 1? That means the message is not considered gone (even though it will get GCed in the real implementation) and our count will never decrement. Because the channel is at capacity (1 element maximum) the writers will always block.

I want a finalizer created each time dupBTChan is called. When a dupped (or original) channel is collected all elements remaining to be read on that channel will get the per-element count decremented, also the nrDups variable will be decremented. As a result, future writes will have the correct count (a count that doesn't reserve space for variables not-read by GCed channels).

Solution 1 - Manual Resource Management (what I want to avoid)

JNB's bounded-tchan actually has manual resource management for this reason. See the cancelBTChan. I'm going for something harder for the user to get wrong (not that manual management isn't the right way to go in many cases).

Solution 2 - Use exceptions by blocking on TVars (GHC can't do this how I want)

EDIT this solution, and solution 3 which is just a spin-off, does not work! Due to bug 5055 (WONTFIX) the GHC compiler sends exceptions to both blocked threads, even though one is sufficient (which is theoretically determinable, but not practical with the GHC GC).

If all the ways to get a BTChan are IO, we can forkIO a thread that reads/retries on an extra (dummy) TVar field unique to the given BTChan. The new thread will catch an exception when all other references to the TVar are dropped, so it will know when to decrement the nrDups and individual element counters. This should work but forces all my users to use IO to get their BTChans:

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

EDIT: Yes, this is a poor mans finalizer and I don't have any particular reason to avoid using addFinalizer. That would be the same solution, still forcing use of IO afaict.

Solution 3: A cleaner API than solution 2, but GHC still doesn't support it

Users start a manager thread by calling initBTChanCollector, which will monitor a set of these dummy TVars (from solution 2) and do the needed clean-up. Basically, it shoves the IO into another thread that knows what to do via a global (unsafePerformIOed) TVar. Things work basically like solution 2, but the creation of BTChan's can still be STM. Failure to run initBTChanCollector would result in an ever-growing list (space leak) of tasks as the process runs.

Solution 4: Never allow discarding BTChans

This is akin to ignoring the problem. If the user never drops a dupped BTChan then the issue disappears.

Solution 5 I see ezyang's answer (totally valid and appreciated), but really would like to keep the current API just with a 'dup' function.

** Solution 6** Please tell me there's a better option.

EDIT: I implemented solution 3 (totally untested alpha release) and handled the potential space leak by making the global itself a BTChan - that chan should probably have a capacity of 1 so forgetting to run init shows up really quick, but that's a minor change. This works in GHCi (7.0.3) but that seems to be incidental. GHC throws exceptions to both blocked threads (the valid one reading the BTChan and the watching thread) so my if you are blocked reading a BTChan when another thread discards it's reference then you die.

Circumscription

I don't understand what exactly you have in mind. What should the semantics for the duplicated channels be with respect to resources? A channel blocks if both it and the duplicate are full? If one of them is full? – Amalea

Right, these semantics need to be clarified. If you try implementing "A channel blocks if both it and the duplicate are full", then you need to ask, am I allowed to drop elements from the queue? If the answer is no, then you've got an unbounded channel again. – Insider

(It would also be useful to ask the person who requested dup what they were planning on using it for.) – Insider

@Heinrich, ezyang Dups of a channel all share resources, the remaining capacity of a channel equals the maximum capacity of the channel minus the number of elements remaining to be read in the most populous duplicate. If all references to a duplicate are dropped then that dup should never cause reduced capacity (i.e. receives population of zero). I tried to say as much in different words in an edit just now. – Circumscription

In that case, you seem to want one bounded channel and a multiplexer. So it sounds like you could probably just use an MVar to control access to the queue (they'll block attempting to grab the variable, but if a thread blocked on an MVar dies, it will release itself) – Insider

Erm, you don't want an MVar, because you're using STM. But it should work OK with another transaction variable. The key, really, is that you should only have one underlying channel. – Insider

@ezyang Yes, I only have one underlying channel and am using the solution #3, which I think is basically what you're saying, but have ran into a bug so I filed #5055. – Circumscription

Answer 1

Here is another solution: require all accesses to the the bounded channel duplicate to be bracketed by a function that releases its resources on exit (by an exception or normally). You can use a monad with a rank-2 runner to prevent duplicated channels from leaking out. It's still manual, but the type system makes it a lot harder to do naughty things.

You really don't want to rely on true IO finalizers, because GHC gives no guarantees about when a finalizer may be run: for all you know it may wait until the end of the program before running the finalizer, which means you're deadlocked until then.

Insider

Having thought about this for a while, I agree. It is simply wrong to attempt to use finalizers for anything other than memory management --i.e. effects which are not semantically "observable" from within Haskell. The idiomatic Haskell equivalent of RAII is not finalizers, but with functions. – Defile