2cfa987 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Deliver intra-node messages directly to the destination process' mailbox

2cfa987Tim Watson13 years ago

distributed-process.cabal+6−6

compare with master

unifiedsplit

@@ -59,19 +59,19 @@ Library

59 rank1dynamic >= 0.1 && < 0.2,

60 syb >= 0.3 && < 0.4

61 Exposed-modules: Control.Distributed.Process,

62− Control.Distributed.Process.Serializable,

63 Control.Distributed.Process.Closure,

62 Control.Distributed.Process.Closure,

64 Control.Distributed.Process.Node,

63 Control.Distributed.Process.Node,

65− Control.Distributed.Process.Internal.Primitives,

64+ Control.Distributed.Process.Serializable

65+ Other-modules: Control.Distributed.Process.Internal.Closure.BuiltIn,

66 Control.Distributed.Process.Internal.CQueue,

67− Control.Distributed.Process.Internal.Types,

68− Control.Distributed.Process.Internal.Trace,

69− Control.Distributed.Process.Internal.Closure.BuiltIn,

70 Control.Distributed.Process.Internal.Messaging,

67 Control.Distributed.Process.Internal.Messaging,

68+ Control.Distributed.Process.Internal.Primitives,

69+ Control.Distributed.Process.Internal.StrictContainerAccessors,

71 Control.Distributed.Process.Internal.StrictList,

70 Control.Distributed.Process.Internal.StrictList,

72 Control.Distributed.Process.Internal.StrictMVar,

71 Control.Distributed.Process.Internal.StrictMVar,

72+ Control.Distributed.Process.Internal.Trace,

73+ Control.Distributed.Process.Internal.Types,

73 Control.Distributed.Process.Internal.WeakTQueue

74 Control.Distributed.Process.Internal.WeakTQueue

74− Control.Distributed.Process.Internal.StrictContainerAccessors

75 Extensions: RankNTypes,

76 ScopedTypeVariables,

77 FlexibleInstances,

src/Control/Distributed/Process.hs+6−1

@@ -15,7 +15,7 @@ please see the distributed-process wiki page on github:

15module Control.Distributed.Process

16 ( -- * Basic types

17 ProcessId

18− , NodeId

18+ , NodeId(..)

19 , Process

20 , SendPortId

21 , processNodeId

@@ -81,6 +81,8 @@ module Control.Distributed.Process

81 , unmonitor

82 , withMonitor

83 , MonitorRef -- opaque

84+ , ProcessExitException()

85+ , exitSource

84 , ProcessLinkException(..)

86 , ProcessLinkException(..)

85 , NodeLinkException(..)

87 , NodeLinkException(..)

86 , PortLinkException(..)

88 , PortLinkException(..)

@@ -127,6 +129,7 @@ module Control.Distributed.Process

127 , spawnMonitor

129 , spawnMonitor

128 , spawnChannel

130 , spawnChannel

129 , DidSpawn(..)

131 , DidSpawn(..)

132+ , nullProcessId

130 -- * Local versions of 'spawn'

133 -- * Local versions of 'spawn'

131 , spawnLocal

134 , spawnLocal

132 , spawnChannelLocal

135 , spawnChannelLocal

@@ -160,6 +163,8 @@ import Control.Distributed.Process.Internal.Types

160 , ProcessMonitorNotification(..)

163 , ProcessMonitorNotification(..)

161 , NodeMonitorNotification(..)

164 , NodeMonitorNotification(..)

162 , PortMonitorNotification(..)

165 , PortMonitorNotification(..)

166+ , ProcessExitException()

167+ , exitSource

163 , ProcessLinkException(..)

168 , ProcessLinkException(..)

164 , NodeLinkException(..)

169 , NodeLinkException(..)

165 , PortLinkException(..)

170 , PortLinkException(..)

src/Control/Distributed/Process/Internal/Primitives.hs+66−14

@@ -39,6 +39,7 @@ module Control.Distributed.Process.Internal.Primitives

39 -- keep the exception constructor hidden, so that handling exit

40 -- reasons /must/ take place via the 'catchExit' family of primitives

41 , ProcessExitException()

42+ , exitSource

42 , getSelfPid

43 , getSelfPid

43 , getSelfNode

44 , getSelfNode

44 , ProcessInfo(..)

45 , ProcessInfo(..)

@@ -96,11 +97,6 @@ module Control.Distributed.Process.Internal.Primitives

96import Prelude hiding (catch)

97import Prelude hiding (catch)

97#endif

98#endif

99−import Data.Binary (decode)

100−import Data.Time.Clock (getCurrentTime)

101−import Data.Time.Format (formatTime)

102−import System.Locale (defaultTimeLocale)

103−import System.Timeout (timeout)

104import Control.Monad (when)

100import Control.Monad (when)

105import Control.Monad.Reader (ask)

101import Control.Monad.Reader (ask)

106import Control.Monad.IO.Class (MonadIO, liftIO)

102import Control.Monad.IO.Class (MonadIO, liftIO)

@@ -111,6 +107,7 @@ import Control.Distributed.Process.Internal.StrictMVar

111 ( StrictMVar

107 ( StrictMVar

112 , modifyMVar

108 , modifyMVar

113 , modifyMVar_

109 , modifyMVar_

110+ , withMVar

114 )

111 )

115import Control.Concurrent.Chan (writeChan)

112import Control.Concurrent.Chan (writeChan)

116import Control.Concurrent.STM

113import Control.Concurrent.STM

@@ -124,13 +121,12 @@ import Control.Concurrent.STM

124 )

121 )

125import Control.Distributed.Process.Internal.CQueue

122import Control.Distributed.Process.Internal.CQueue

126 ( dequeue

123 ( dequeue

124+ , enqueue

127 , BlockSpec(..)

125 , BlockSpec(..)

128 , MatchOn(..)

126 , MatchOn(..)

129 )

127 )

130import Control.Distributed.Process.Serializable (Serializable, fingerprint)

128import Control.Distributed.Process.Serializable (Serializable, fingerprint)

131−import Data.Accessor ((^.), (^:), (^=))

132import Control.Distributed.Static (Closure, Static)

129import Control.Distributed.Static (Closure, Static)

133−import Data.Rank1Typeable (Typeable)

134import qualified Control.Distributed.Static as Static (unstatic, unclosure)

130import qualified Control.Distributed.Static as Static (unstatic, unclosure)

135import Control.Distributed.Process.Internal.Types

131import Control.Distributed.Process.Internal.Types

136 ( NodeId(..)

132 ( NodeId(..)

@@ -149,10 +145,12 @@ import Control.Distributed.Process.Internal.Types

149 , ReceivePort(..)

145 , ReceivePort(..)

150 , channelCounter

146 , channelCounter

151 , typedChannelWithId

147 , typedChannelWithId

148+ , localProcessWithId

152 , TypedChannel(..)

149 , TypedChannel(..)

153 , SendPortId(..)

150 , SendPortId(..)

154 , Identifier(..)

151 , Identifier(..)

155 , ProcessExitException(..)

152 , ProcessExitException(..)

153+ , exitSource

156 , DidUnmonitor(..)

154 , DidUnmonitor(..)

157 , DidUnlinkProcess(..)

155 , DidUnlinkProcess(..)

158 , DidUnlinkNode(..)

156 , DidUnlinkNode(..)

@@ -166,7 +164,7 @@ import Control.Distributed.Process.Internal.Types

166 , createUnencodedMessage

164 , createUnencodedMessage

167 , runLocalProcess

165 , runLocalProcess

168 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

166 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

169− , LocalProcessState

167+ , LocalProcessState(..)

170 , LocalSendPortId

168 , LocalSendPortId

171 , messageToPayload

169 , messageToPayload

172 )

170 )

@@ -178,10 +176,21 @@ import Control.Distributed.Process.Internal.Messaging

178 )

176 )

179import qualified Control.Distributed.Process.Internal.Trace as Trace

177import qualified Control.Distributed.Process.Internal.Trace as Trace

180import Control.Distributed.Process.Internal.WeakTQueue

178import Control.Distributed.Process.Internal.WeakTQueue

179+ ( TQueue

181− ( newTQueueIO

180+ , newTQueueIO

182 , readTQueue

181 , readTQueue

183 , mkWeakTQueue

182 , mkWeakTQueue

183+ , writeTQueue

184 )

185+import Data.Accessor ((^.), (^:), (^=))

186+import Data.Binary (decode)

187+import Data.Foldable (forM_)

188+import Data.Rank1Typeable (Typeable)

189+import Data.Time.Clock (getCurrentTime)

190+import Data.Time.Format (formatTime)

191+import System.Locale (defaultTimeLocale)

192+import System.Mem.Weak (deRefWeak)

193+import System.Timeout (timeout)

185import Unsafe.Coerce

194import Unsafe.Coerce

186

195

187--------------------------------------------------------------------------------

196--------------------------------------------------------------------------------

@@ -1031,20 +1040,63 @@ trace s = do

1031 node <- processNode <$> ask

1040 node <- processNode <$> ask

1032 liftIO $ Trace.trace (localTracer node) s

1041 liftIO $ Trace.trace (localTracer node) s

1033

1042

1043+--------------------------------------------------------------------------------

1044+-- Messages to local processes --

1045+--------------------------------------------------------------------------------

1046+

1047+deliverToLocalProcess :: ProcessId -> Message -> Process ()

1048+deliverToLocalProcess pid msg = do

1049+ withLocalProc pid $ \p -> enqueue (processQueue p) msg

1050+

1051+deliverToLocalPort :: SendPortId -> Message -> Process ()

1052+deliverToLocalPort spId msg =

1053+ let pid = sendPortProcessId spId

1054+ cid = sendPortLocalId spId

1055+ in do

1056+ withLocalProc pid $ \proc -> do

1057+ mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)

1058+ case mChan of

1059+ -- in the unlikely event we know nothing about this channel id,

1060+ -- see [note: missing recipients]

1061+ Nothing -> return ()

1062+ Just (TypedChannel chan') -> do

1063+ -- If ch is Nothing, the process has given up the read end of

1064+ -- the channel and we simply ignore the incoming message - this

1065+ ch <- deRefWeak chan'

1066+ forM_ ch $ \chan -> deliverChan msg chan

1067+ where deliverChan :: forall a . Message -> TQueue a -> IO ()

1068+ deliverChan (UnencodedMessage _ raw) chan' =

1069+ atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)

1070+ deliverChan (EncodedMessage _ _) _ =

1071+ -- this will /never/ happen

1072+ error "invalid local channel delivery"

1073+

1074+withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> Process ()

1075+withLocalProc pid p = do

1076+ node <- processNode <$> ask

1077+ liftIO $ do

1078+ -- see [note: missing recipients]

1079+ let lpid = processLocalId pid

1080+ mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)

1081+ forM_ mProc p

1082+

1083+-- [note: missing recipients]

1084+-- By [Unified: table 6, rule missing_process] messages to dead processes

1085+-- can silently be dropped

1086+

1034--------------------------------------------------------------------------------

1087--------------------------------------------------------------------------------

1035-- Auxiliary functions --

1088-- Auxiliary functions --

1036--------------------------------------------------------------------------------

1089--------------------------------------------------------------------------------

1037

1090

1038sendLocal :: (Serializable a) => ProcessId -> a -> Process ()

1091sendLocal :: (Serializable a) => ProcessId -> a -> Process ()

1039sendLocal pid msg =

1092sendLocal pid msg =

1040− sendCtrlMsg Nothing $ LocalSend pid (createUnencodedMessage msg)

1093+ -- sendCtrlMsg Nothing $ LocalSend pid (createUnencodedMessage msg)

1094+ deliverToLocalProcess pid (createUnencodedMessage msg)

1041

1095

1042sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

1096sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

1043sendChanLocal spId msg =

1097sendChanLocal spId msg =

1044− -- we *must* fully serialize/encode the message here, because

1045− -- attempting to use `unsafeCoerce' in the node controller

1046− -- won't work since we know nothing about the required type

1047− sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)

1098+ -- sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)

1099+ deliverToLocalPort spId (createUnencodedMessage msg)

1048

1100

1049getMonitorRefFor :: Identifier -> Process MonitorRef

1101getMonitorRefFor :: Identifier -> Process MonitorRef

1050getMonitorRefFor ident = do

1102getMonitorRefFor ident = do

src/Control/Distributed/Process/Internal/Types.hs+4

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

39 , NodeMonitorNotification(..)

40 , PortMonitorNotification(..)

41 , ProcessExitException(..)

42+ , exitSource

42 , ProcessLinkException(..)

43 , ProcessLinkException(..)

43 , NodeLinkException(..)

44 , NodeLinkException(..)

44 , PortLinkException(..)

45 , PortLinkException(..)

@@ -411,6 +412,9 @@ instance Exception ProcessExitException

411instance Show ProcessExitException where

412instance Show ProcessExitException where

412 show (ProcessExitException pid _) = "exit-from=" ++ (show pid)

413 show (ProcessExitException pid _) = "exit-from=" ++ (show pid)

413

414

415+exitSource :: ProcessExitException -> ProcessId

416+exitSource (ProcessExitException from _) = from

417+

414instance Exception ProcessLinkException

418instance Exception ProcessLinkException

415instance Exception NodeLinkException

419instance Exception NodeLinkException

416instance Exception PortLinkException

420instance Exception PortLinkException

src/Control/Distributed/Process/Node.hs+1−1

@@ -1,7 +1,7 @@

1-- | Local nodes

2--

3module Control.Distributed.Process.Node

4− ( LocalNode

4+ ( LocalNode(localEndPoint)

5 , newLocalNode

6 , closeLocalNode

7 , forkProcess

tests/TestCH.hs+2−8

@@ -27,12 +27,6 @@ import Network.Transport.TCP

27 , defaultTCPParameters

28 )

29import Control.Distributed.Process

30−import Control.Distributed.Process.Internal.Types

31− ( NodeId(nodeAddress)

32− , LocalNode(localEndPoint)

33− , ProcessExitException(..)

34− , nullProcessId

35− )

36import Control.Distributed.Process.Node

30import Control.Distributed.Process.Node

37import Control.Distributed.Process.Serializable (Serializable)

31import Control.Distributed.Process.Serializable (Serializable)

@@ -1095,8 +1089,8 @@ testPrettyExit transport = do

1095

1089

1096 _ <- forkProcess localNode $ do

1090 _ <- forkProcess localNode $ do

1097 (die "timeout")

1091 (die "timeout")

1098− `catch` \ex@(ProcessExitException from _) ->

1092+ `catch` \(ex :: ProcessExitException) ->

1099− let expected = "exit-from=" ++ (show from)

1093+ let expected = "exit-from=" ++ (show (exitSource ex))

1100 in do

1094 in do

1101 True <- return $ (show ex) == expected

1095 True <- return $ (show ex) == expected

1102 liftIO $ putMVar done ()

1096 liftIO $ putMVar done ()

tests/TestClosure.hs−1

@@ -27,7 +27,6 @@ import Network.Socket (sClose)

27import Control.Distributed.Process

28import Control.Distributed.Process.Closure

29import Control.Distributed.Process.Node

30−import Control.Distributed.Process.Internal.Types (NodeId(nodeAddress))

31import Control.Distributed.Static (staticLabel, staticClosure)

30import Control.Distributed.Static (staticLabel, staticClosure)

33import Test.HUnit (Assertion)

32import Test.HUnit (Assertion)