git.iko.soy/distributed-process
on fix-spawnChan help sign in
distributed-process/src/Control/Distributed/Process/Internal/WeakTQueue.hs
101 lines 3.5 kB history raw
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 #-}
8module 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
21import Prelude hiding (read)
22import GHC.Conc
23import Data.Typeable (Typeable)
24import GHC.IO (IO(IO))
25import GHC.Prim (mkWeak#)
26import GHC.Weak (Weak(Weak))
28--------------------------------------------------------------------------------
29-- Original functionality --
30--------------------------------------------------------------------------------
32-- | 'TQueue' is an abstract type representing an unbounded FIFO channel.
33data TQueue a = TQueue {-# UNPACK #-} !(TVar [a])
34 {-# UNPACK #-} !(TVar [a])
35 deriving Typeable
37instance Eq (TQueue a) where
38 TQueue a _ == TQueue b _ = a == b
40-- |Build and returns a new instance of 'TQueue'
41newTQueue :: STM (TQueue a)
42newTQueue = do
43 read <- newTVar []
44 write <- newTVar []
45 return (TQueue read write)
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.
51newTQueueIO :: IO (TQueue a)
52newTQueueIO = do
53 read <- newTVarIO []
54 write <- newTVarIO []
55 return (TQueue read write)
57-- |Write a value to a 'TQueue'.
58writeTQueue :: TQueue a -> a -> STM ()
59writeTQueue (TQueue _read write) a = do
60 listend <- readTVar write
61 writeTVar write (a:listend)
63-- |Read the next value from the 'TQueue'.
64readTQueue :: TQueue a -> STM a
65readTQueue (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
79-- | A version of 'readTQueue' which does not retry. Instead it
80-- returns @Nothing@ if no value is available.
81tryReadTQueue :: TQueue a -> STM (Maybe a)
82tryReadTQueue c = fmap Just (readTQueue c) `orElse` return Nothing
84-- |Returns 'True' if the supplied 'TQueue' is empty.
85isEmptyTQueue :: TQueue a -> STM Bool
86isEmptyTQueue (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
95--------------------------------------------------------------------------------
96-- New functionality --
97--------------------------------------------------------------------------------
99mkWeakTQueue :: TQueue a -> IO () -> IO (Weak (TQueue a))
100mkWeakTQueue q@(TQueue _read (TVar write#)) f = IO $ \s ->
101 case mkWeak# write# q f s of (# s', w #) -> (# s', Weak w #)