9db1e1c · distributed-process

git.iko.soy/distributed-process

on master help sign in

Start using weak references to clean up

9db1e1cEdsko de Vries14 years ago

distributed-process/distributed-process.cabal+2−1

compare with master

unifiedsplit

@@ -63,7 +63,8 @@ Library

63 Control.Distributed.Process.Internal.Types,

64 Control.Distributed.Process.Internal.Closure.BuiltIn,

65 Control.Distributed.Process.Internal.Messaging,

66− Control.Distributed.Process.Internal.StrictMVar

66+ Control.Distributed.Process.Internal.StrictMVar,

67+ Control.Distributed.Process.Internal.WeakTQueue

67 Extensions: RankNTypes,

68 Extensions: RankNTypes,

68 ScopedTypeVariables,

69 ScopedTypeVariables,

69 FlexibleInstances,

70 FlexibleInstances,

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+33−19

@@ -77,14 +77,16 @@ import Control.Monad.IO.Class (MonadIO, liftIO)

77import Control.Applicative ((<$>))

78import Control.Exception (Exception, throwIO, SomeException)

79import qualified Control.Exception as Ex (catch, mask)

80−import Control.Distributed.Process.Internal.StrictMVar (modifyMVar)

80+import Control.Distributed.Process.Internal.StrictMVar

81+ ( StrictMVar

82+ , modifyMVar

83+ , modifyMVar_

84+ )

81import Control.Concurrent.Chan (writeChan)

85import Control.Concurrent.Chan (writeChan)

82import Control.Concurrent.STM

86import Control.Concurrent.STM

83 ( STM

87 ( STM

84 , atomically

88 , atomically

85 , orElse

89 , orElse

86− , newTChan

87− , readTChan

88 , newTVar

90 , newTVar

89 , readTVar

91 , readTVar

90 , writeTVar

92 , writeTVar

@@ -123,12 +125,19 @@ import Control.Distributed.Process.Internal.Types

123 , createMessage

125 , createMessage

124 , runLocalProcess

126 , runLocalProcess

125 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

127 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

128+ , LocalProcessState

129+ , LocalSendPortId

126 )

130 )

127import Control.Distributed.Process.Internal.Messaging

131import Control.Distributed.Process.Internal.Messaging

128 ( sendMessage

132 ( sendMessage

129 , sendBinary

133 , sendBinary

130 , disconnect

134 , disconnect

131 )

135 )

136+import Control.Distributed.Process.Internal.WeakTQueue

137+ ( newTQueueIO

138+ , readTQueue

139+ , mkWeakTQueue

140+ )

132

141

133--------------------------------------------------------------------------------

142--------------------------------------------------------------------------------

134-- Basic messaging --

143-- Basic messaging --

@@ -157,21 +166,26 @@ expect = receiveWait [match return]

157-- | Create a new typed channel

166-- | Create a new typed channel

158newChan :: Serializable a => Process (SendPort a, ReceivePort a)

167newChan :: Serializable a => Process (SendPort a, ReceivePort a)

159newChan = do

168newChan = do

160− proc <- ask

169+ proc <- ask

161− liftIO . modifyMVar (processState proc) $ \st -> do

170+ liftIO . modifyMVar (processState proc) $ \st -> do

162− chan <- liftIO . atomically $ newTChan

163− let lcid = st ^. channelCounter

171+ let lcid = st ^. channelCounter

164− cid = SendPortId { sendPortProcessId = processId proc

172+ let cid = SendPortId { sendPortProcessId = processId proc

165− , sendPortLocalId = lcid

173+ , sendPortLocalId = lcid

166− }

174+ }

167− sport = SendPort cid

175+ let sport = SendPort cid

176+ chan <- liftIO newTQueueIO

177+ chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid

168− rport = ReceivePortSingle chan

178+ let rport = ReceivePortSingle chan

169− tch = TypedChannel chan

179+ let tch = TypedChannel chan'

170− return ( (channelCounter ^: (+ 1))

180+ return ( (channelCounter ^: (+ 1))

171− . (typedChannelWithId lcid ^= Just tch)

181+ . (typedChannelWithId lcid ^= Just tch)

172− $ st

182+ $ st

173− , (sport, rport)

183+ , (sport, rport)

174− )

184+ )

185+ where

186+ finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()

187+ finalizer processState lcid = modifyMVar_ processState $

188+ return . (typedChannelWithId lcid ^= Nothing)

175

189

176-- | Send a message on a typed channel

190-- | Send a message on a typed channel

177sendChan :: Serializable a => SendPort a -> a -> Process ()

191sendChan :: Serializable a => SendPort a -> a -> Process ()

@@ -189,7 +203,7 @@ receiveChan = liftIO . atomically . receiveSTM

189 where

203 where

190 receiveSTM :: ReceivePort a -> STM a

204 receiveSTM :: ReceivePort a -> STM a

191 receiveSTM (ReceivePortSingle c) =

205 receiveSTM (ReceivePortSingle c) =

192− readTChan c

206+ readTQueue c

193 receiveSTM (ReceivePortBiased ps) =

207 receiveSTM (ReceivePortBiased ps) =

194 foldr1 orElse (map receiveSTM ps)

208 foldr1 orElse (map receiveSTM ps)

195 receiveSTM (ReceivePortRR psVar) = do

209 receiveSTM (ReceivePortRR psVar) = do

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+5−3

@@ -64,6 +64,7 @@ module Control.Distributed.Process.Internal.Types

64 , typedChannelWithId

65 ) where

67+import System.Mem.Weak (Weak)

67import Data.Map (Map)

68import Data.Map (Map)

68import Data.Int (Int32)

69import Data.Int (Int32)

69import Data.Typeable (Typeable)

70import Data.Typeable (Typeable)

@@ -81,7 +82,7 @@ import Control.Category ((>>>))

81import Control.Exception (Exception)

82import Control.Exception (Exception)

82import Control.Concurrent (ThreadId)

83import Control.Concurrent (ThreadId)

83import Control.Concurrent.Chan (Chan)

84import Control.Concurrent.Chan (Chan)

84−import Control.Concurrent.STM (TChan, TVar)

85+import Control.Concurrent.STM (TVar)

85import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

86import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

86import Control.Applicative (Applicative, (<$>), (<*>))

87import Control.Applicative (Applicative, (<$>), (<*>))

87import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

88import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

@@ -97,6 +98,7 @@ import Control.Distributed.Process.Serializable

97 )

98 )

98import Control.Distributed.Process.Internal.CQueue (CQueue)

99import Control.Distributed.Process.Internal.CQueue (CQueue)

99import Control.Distributed.Process.Internal.StrictMVar (StrictMVar)

100import Control.Distributed.Process.Internal.StrictMVar (StrictMVar)

101+import Control.Distributed.Process.Internal.WeakTQueue (TQueue)

100import Control.Distributed.Static (RemoteTable, Closure)

102import Control.Distributed.Static (RemoteTable, Closure)

101

103

102--------------------------------------------------------------------------------

104--------------------------------------------------------------------------------

@@ -248,7 +250,7 @@ instance Show SendPortId where

248 show (SendPortId (ProcessId (NodeId addr) (LocalProcessId _ plid)) clid)

250 show (SendPortId (ProcessId (NodeId addr) (LocalProcessId _ plid)) clid)

249 = "cid://" ++ show addr ++ ":" ++ show plid ++ ":" ++ show clid

251 = "cid://" ++ show addr ++ ":" ++ show plid ++ ":" ++ show clid

250

252

251−data TypedChannel = forall a. Serializable a => TypedChannel (TChan a)

253+data TypedChannel = forall a. Serializable a => TypedChannel (Weak (TQueue a))

252

254

253-- | The send send of a typed channel (serializable)

255-- | The send send of a typed channel (serializable)

254newtype SendPort a = SendPort {

256newtype SendPort a = SendPort {

@@ -260,7 +262,7 @@ newtype SendPort a = SendPort {

260-- | The receive end of a typed channel (not serializable)

262-- | The receive end of a typed channel (not serializable)

261data ReceivePort a =

263data ReceivePort a =

262 -- | A single receive port

264 -- | A single receive port

263− ReceivePortSingle (TChan a)

265+ ReceivePortSingle (TQueue a)

264 -- | A left-biased combination of receive ports

266 -- | A left-biased combination of receive ports

265 | ReceivePortBiased [ReceivePort a]

267 | ReceivePortBiased [ReceivePort a]

266 -- | A round-robin combination of receive ports

268 -- | A round-robin combination of receive ports

distributed-process/src/Control/Distributed/Process/Internal/WeakTQueue.hsnew file+101

@@ -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 #)

distributed-process/src/Control/Distributed/Process/Node.hs+9−3

@@ -17,6 +17,7 @@ import Prelude hiding (catch)

17#endif

19import System.IO (fixIO, hPutStrLn, stderr)

20+import System.Mem.Weak (deRefWeak)

20import qualified Data.ByteString.Lazy as BSL (fromChunks)

21import qualified Data.ByteString.Lazy as BSL (fromChunks)

21import Data.Binary (decode)

22import Data.Binary (decode)

22import Data.Map (Map)

23import Data.Map (Map)

@@ -52,7 +53,7 @@ import Control.Distributed.Process.Internal.StrictMVar

52 , takeMVar

53 , takeMVar

53 )

54 )

54import Control.Concurrent.Chan (newChan, writeChan, readChan)

55import Control.Concurrent.Chan (newChan, writeChan, readChan)

55−import Control.Concurrent.STM (atomically, writeTChan)

56+import Control.Concurrent.STM (atomically)

56import Control.Distributed.Process.Internal.CQueue (enqueue, newCQueue)

57import Control.Distributed.Process.Internal.CQueue (enqueue, newCQueue)

57import qualified Network.Transport as NT

58import qualified Network.Transport as NT

58 ( Transport

59 ( Transport

@@ -131,6 +132,7 @@ import Control.Distributed.Process.Internal.Messaging

131 )

132 )

132import Control.Distributed.Process.Internal.Primitives (expect, register, finally)

133import Control.Distributed.Process.Internal.Primitives (expect, register, finally)

133import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)

134import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)

135+import Control.Distributed.Process.Internal.WeakTQueue (writeTQueue)

134

136

135--------------------------------------------------------------------------------

137--------------------------------------------------------------------------------

136-- Initialization --

138-- Initialization --

@@ -315,8 +317,12 @@ handleIncomingMessages node = go initConnectionState

315 let msg = payloadToMessage payload

317 let msg = payloadToMessage payload

316 enqueue (processQueue proc) msg

318 enqueue (processQueue proc) msg

317 go st

319 go st

318− Just (_, ToChan (TypedChannel chan)) -> do

320+ Just (_, ToChan (TypedChannel chan')) -> do

321+ mChan <- deRefWeak chan'

322+ -- If mChan is Nothing, the process has given up the read end of

323+ -- the channel and we simply ignore the incoming message

324+ forM_ mChan $ \chan -> atomically $

319− atomically $ writeTChan chan . decode . BSL.fromChunks $ payload

325+ writeTQueue chan . decode . BSL.fromChunks $ payload

320 go st

326 go st

321 Just (_, ToNode) -> do

327 Just (_, ToNode) -> do

322 let ctrlMsg = decode . BSL.fromChunks $ payload

328 let ctrlMsg = decode . BSL.fromChunks $ payload