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