@@ -0,0 +1,101 @@
1+-- | Clone of Control.Concurrent.STM.TQueue with support for mkWeakTQueue
2+--
3+-- Not all functionality from the original module is available: unGetTQueue,
4+-- peekTQueue and tryPeekTQueue are missing. In order to implement these we'd
5+-- need to be able to touch# the write end of the queue inside unGetTQueue, but
6+-- that means we need a version of touch# that works within the STM monad.
7+{-# LANGUAGE MagicHash, UnboxedTuples #-}
8+module Control.Distributed.Process.Internal.WeakTQueue (
9+ -- * Original functionality
10+ TQueue,
11+ newTQueue,
12+ newTQueueIO,
13+ readTQueue,
14+ tryReadTQueue,
15+ writeTQueue,
16+ isEmptyTQueue,
17+ -- * New functionality
18+ mkWeakTQueue
19+ ) where
20+
21+import Prelude hiding (read)
22+import GHC.Conc
23+import Data.Typeable (Typeable)
24+import GHC.IO (IO(IO))
25+import GHC.Prim (mkWeak#)
26+import GHC.Weak (Weak(Weak))
27+
28+--------------------------------------------------------------------------------
29+-- Original functionality --
30+--------------------------------------------------------------------------------
31+
32+-- | 'TQueue' is an abstract type representing an unbounded FIFO channel.
33+data TQueue a = TQueue {-# UNPACK #-} !(TVar [a])
34+ {-# UNPACK #-} !(TVar [a])
35+ deriving Typeable
36+
37+instance Eq (TQueue a) where
38+ TQueue a _ == TQueue b _ = a == b
39+
40+-- |Build and returns a new instance of 'TQueue'
41+newTQueue :: STM (TQueue a)
42+newTQueue = do
43+ read <- newTVar []
44+ write <- newTVar []
45+ return (TQueue read write)
46+
47+-- |@IO@ version of 'newTQueue'. This is useful for creating top-level
48+-- 'TQueue's using 'System.IO.Unsafe.unsafePerformIO', because using
49+-- 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't
50+-- possible.
51+newTQueueIO :: IO (TQueue a)
52+newTQueueIO = do
53+ read <- newTVarIO []
54+ write <- newTVarIO []
55+ return (TQueue read write)
56+
57+-- |Write a value to a 'TQueue'.
58+writeTQueue :: TQueue a -> a -> STM ()
59+writeTQueue (TQueue _read write) a = do
60+ listend <- readTVar write
61+ writeTVar write (a:listend)
62+
63+-- |Read the next value from the 'TQueue'.
64+readTQueue :: TQueue a -> STM a
65+readTQueue (TQueue read write) = do
66+ xs <- readTVar read
67+ case xs of
68+ (x:xs') -> do writeTVar read xs'
69+ return x
70+ [] -> do ys <- readTVar write
71+ case ys of
72+ [] -> retry
73+ _ -> case reverse ys of
74+ [] -> error "readTQueue"
75+ (z:zs) -> do writeTVar write []
76+ writeTVar read zs
77+ return z
78+
79+-- | A version of 'readTQueue' which does not retry. Instead it
80+-- returns @Nothing@ if no value is available.
81+tryReadTQueue :: TQueue a -> STM (Maybe a)
82+tryReadTQueue c = fmap Just (readTQueue c) `orElse` return Nothing
83+
84+-- |Returns 'True' if the supplied 'TQueue' is empty.
85+isEmptyTQueue :: TQueue a -> STM Bool
86+isEmptyTQueue (TQueue read write) = do
87+ xs <- readTVar read
88+ case xs of
89+ (_:_) -> return False
90+ [] -> do ys <- readTVar write
91+ case ys of
92+ [] -> return True
93+ _ -> return False
94+
95+--------------------------------------------------------------------------------
96+-- New functionality --
97+--------------------------------------------------------------------------------
98+
99+mkWeakTQueue :: TQueue a -> IO () -> IO (Weak (TQueue a))
100+mkWeakTQueue q@(TQueue _read (TVar write#)) f = IO $ \s ->
101+ case mkWeak# write# q f s of (# s', w #) -> (# s', Weak w #)