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