061a8c4 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Introduce unsafe variants of basic messaging primitives

061a8c4Tim Watson13 years ago

distributed-process.cabal+6−4

compare with master

unifiedsplit

@@ -59,13 +59,15 @@ Library

59 rank1dynamic >= 0.1 && < 0.2,

60 syb >= 0.3 && < 0.4

61 Exposed-modules: Control.Distributed.Process,

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

63− Control.Distributed.Process.Serializable,

64 Control.Distributed.Process.Closure,

62 Control.Distributed.Process.Closure,

65 Control.Distributed.Process.Debug,

63 Control.Distributed.Process.Debug,

66− Control.Distributed.Process.Node

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

65+ Control.Distributed.Process.Node,

66+ Control.Distributed.Process.Serializable,

67+ Control.Distributed.Process.UnsafePrimitives

67 Other-modules: Control.Distributed.Process.Internal.Primitives,

68 Other-modules: Control.Distributed.Process.Internal.Primitives,

68 Control.Distributed.Process.Internal.Spawn,

69 Control.Distributed.Process.Internal.Spawn,

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

69 Control.Distributed.Process.Internal.CQueue,

71 Control.Distributed.Process.Internal.CQueue,

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

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

71 Control.Distributed.Process.Internal.Messaging,

73 Control.Distributed.Process.Internal.Messaging,

@@ -75,7 +77,7 @@ Library

75 Control.Distributed.Process.Internal.StrictContainerAccessors,

77 Control.Distributed.Process.Internal.StrictContainerAccessors,

76 Control.Distributed.Process.Internal.Trace.Types,

78 Control.Distributed.Process.Internal.Trace.Types,

77 Control.Distributed.Process.Internal.Trace.Primitives,

79 Control.Distributed.Process.Internal.Trace.Primitives,

78− Control.Distributed.Process.Internal.Trace.Tracer

80+ Control.Distributed.Process.Internal.Trace.Tracer,

79 Control.Distributed.Process.Internal.Trace.Remote

81 Control.Distributed.Process.Internal.Trace.Remote

80 Extensions: RankNTypes,

82 Extensions: RankNTypes,

81 ScopedTypeVariables,

83 ScopedTypeVariables,

src/Control/Distributed/Process.hs+7

@@ -36,6 +36,10 @@ module Control.Distributed.Process

36 , receiveChanTimeout

37 , mergePortsBiased

38 , mergePortsRR

39+ -- * Unsafe messaging variants

40+ , unsafeSend

41+ , unsafeSendChan

42+ , unsafeNSend

39 -- * Advanced messaging

43 -- * Advanced messaging

40 , Match

44 , Match

41 , receiveWait

45 , receiveWait

@@ -187,6 +191,9 @@ import Control.Distributed.Process.Internal.Primitives

187 , receiveChan

191 , receiveChan

188 , mergePortsBiased

192 , mergePortsBiased

189 , mergePortsRR

193 , mergePortsRR

194+ , unsafeSend

195+ , unsafeSendChan

196+ , unsafeNSend

190 -- Advanced messaging

197 -- Advanced messaging

191 , Match

198 , Match

192 , receiveWait

199 , receiveWait

src/Control/Distributed/Process/Internal/Messaging.hs+31−2

@@ -5,6 +5,7 @@ module Control.Distributed.Process.Internal.Messaging

5 , disconnect

6 , closeImplicitReconnections

7 , impliesDeathOf

8+ , sendCtrlMsg

8 ) where

9 ) where

10import Data.Accessor ((^.), (^=))

11import Data.Accessor ((^.), (^=))

@@ -13,8 +14,12 @@ import qualified Data.Map as Map (partitionWithKey, elems)

13import qualified Data.ByteString.Lazy as BSL (toChunks)

14import qualified Data.ByteString.Lazy as BSL (toChunks)

14import qualified Data.ByteString as BSS (ByteString)

15import qualified Data.ByteString as BSS (ByteString)

15import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)

16import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)

17+import Control.Distributed.Process.Serializable ()

18+

16import Control.Concurrent.Chan (writeChan)

19import Control.Concurrent.Chan (writeChan)

17import Control.Monad (unless)

20import Control.Monad (unless)

21+import Control.Monad.IO.Class (MonadIO, liftIO)

22+import Control.Monad.Reader (ask)

18import qualified Network.Transport as NT

23import qualified Network.Transport as NT

19 ( Connection

24 ( Connection

20 , send

25 , send

@@ -36,8 +41,11 @@ import Control.Distributed.Process.Internal.Types

36 , ProcessSignal(Died)

41 , ProcessSignal(Died)

37 , DiedReason(DiedDisconnect)

42 , DiedReason(DiedDisconnect)

38 , ImplicitReconnect(WithImplicitReconnect)

43 , ImplicitReconnect(WithImplicitReconnect)

39− , NodeId

44+ , NodeId(..)

40− , ProcessId(processNodeId)

45+ , ProcessId(..)

46+ , LocalNode(..)

47+ , LocalProcess(..)

48+ , Process(..)

41 , SendPortId(sendPortProcessId)

49 , SendPortId(sendPortProcessId)

42 , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier)

50 , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier)

43 )

51 )

@@ -167,3 +175,24 @@ SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' =

167 cid' == cid

175 cid' == cid

168_ `impliesDeathOf` _ =

176_ `impliesDeathOf` _ =

169 False

177 False

178+

179+

180+-- Send a control message

181+sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node

182+ -> ProcessSignal -- ^ Message to send

183+ -> Process ()

184+sendCtrlMsg mNid signal = do

185+ proc <- ask

186+ let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)

187+ , ctrlMsgSignal = signal

188+ }

189+ case mNid of

190+ Nothing -> do

191+ liftIO $ writeChan (localCtrlChan (processNode proc)) msg

192+ Just nid ->

193+ liftIO $ sendBinary (processNode proc)

194+ (ProcessIdentifier (processId proc))

195+ (NodeIdentifier nid)

196+ WithImplicitReconnect

197+ msg

198+

src/Control/Distributed/Process/Internal/Primitives.hs+29−19

@@ -14,6 +14,10 @@ module Control.Distributed.Process.Internal.Primitives

14 , receiveChan

15 , mergePortsBiased

16 , mergePortsRR

17+ -- * Unsafe messaging variants

18+ , unsafeSend

19+ , unsafeSendChan

20+ , unsafeNSend

17 -- * Advanced messaging

21 -- * Advanced messaging

18 , Match

22 , Match

19 , receiveWait

23 , receiveWait

@@ -93,6 +97,8 @@ module Control.Distributed.Process.Internal.Primitives

93 -- * Reconnecting

97 -- * Reconnecting

94 , reconnect

98 , reconnect

95 , reconnectPort

99 , reconnectPort

100+ -- * Internal Exports

101+ , sendCtrlMsg

96 ) where

102 ) where

103

98#if ! MIN_VERSION_base(4,6,0)

104#if ! MIN_VERSION_base(4,6,0)

@@ -138,6 +144,7 @@ import Control.Distributed.Static

138 )

144 )

139import Data.Rank1Typeable (Typeable)

145import Data.Rank1Typeable (Typeable)

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

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

147+import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe

141import Control.Distributed.Process.Internal.Types

148import Control.Distributed.Process.Internal.Types

142 ( NodeId(..)

149 ( NodeId(..)

143 , ProcessId(..)

150 , ProcessId(..)

@@ -181,6 +188,7 @@ import Control.Distributed.Process.Internal.Messaging

181 , sendBinary

188 , sendBinary

182 , sendPayload

189 , sendPayload

183 , disconnect

190 , disconnect

191+ , sendCtrlMsg

184 )

192 )

185import Control.Distributed.Process.Internal.Trace.Types

193import Control.Distributed.Process.Internal.Trace.Types

186 ( traceEvent

194 ( traceEvent

@@ -220,6 +228,13 @@ send them msg = do

220 liftIO $ traceEvent (localTracer node)

228 liftIO $ traceEvent (localTracer node)

221 (TraceEvSent them us (createUnencodedMessage msg))

229 (TraceEvSent them us (createUnencodedMessage msg))

222

230

231+-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize

232+-- and (in the case when the destination process resides on the same local

233+-- node) therefore ensure that the payload is fully evaluated before it is

234+-- delivered.

235+unsafeSend :: Serializable a => ProcessId -> a -> Process ()

236+unsafeSend = Unsafe.send

237+

223-- | Wait for a message of a specific type

238-- | Wait for a message of a specific type

224expect :: forall a. Serializable a => Process a

239expect :: forall a. Serializable a => Process a

225expect = receiveWait [match return]

240expect = receiveWait [match return]

@@ -267,6 +282,13 @@ sendChan (SendPort cid) msg = do

267 NoImplicitReconnect

282 NoImplicitReconnect

268 msg

283 msg

269

284

285+-- | Send a message on a typed channel. This function makes /no/ attempt to

286+-- serialize and (in the case when the @ReceivePort@ resides on the same local

287+-- node) therefore ensure that the payload is fully evaluated before it is

288+-- delivered.

289+unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()

290+unsafeSendChan = Unsafe.sendChan

291+

270-- | Wait for a message on a typed channel

292-- | Wait for a message on a typed channel

271receiveChan :: Serializable a => ReceivePort a -> Process a

293receiveChan :: Serializable a => ReceivePort a -> Process a

272receiveChan = liftIO . atomically . receiveSTM

294receiveChan = liftIO . atomically . receiveSTM

@@ -986,6 +1008,13 @@ nsend :: Serializable a => String -> a -> Process ()

986nsend label msg =

1008nsend label msg =

987 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))

1009 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))

988

1010

1011+-- | Named send to a process in the local registry (asynchronous).

1012+-- This function makes /no/ attempt to serialize and (in the case when the

1013+-- destination process resides on the same local node) therefore ensure that

1014+-- the payload is fully evaluated before it is delivered.

1015+unsafeNSend :: Serializable a => String -> a -> Process ()

1016+unsafeNSend = Unsafe.nsend

1017+

989-- | Named send to a process in a remote registry (asynchronous)

1018-- | Named send to a process in a remote registry (asynchronous)

990nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

1019nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

991nsendRemote nid label msg =

1020nsendRemote nid label msg =

@@ -1091,22 +1120,3 @@ monitor' ident = do

1091-- | Link to a process/node/channel

1120-- | Link to a process/node/channel

1092link' :: Identifier -> Process ()

1121link' :: Identifier -> Process ()

1093link' = sendCtrlMsg Nothing . Link

1122link' = sendCtrlMsg Nothing . Link

1094−

1095−-- Send a control message

1096−sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node

1097− -> ProcessSignal -- ^ Message to send

1098− -> Process ()

1099−sendCtrlMsg mNid signal = do

1100− proc <- ask

1101− let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)

1102− , ctrlMsgSignal = signal

1103− }

1104− case mNid of

1105− Nothing -> do

1106− liftIO $ writeChan (localCtrlChan (processNode proc)) msg

1107− Just nid ->

1108− liftIO $ sendBinary (processNode proc)

1109− (ProcessIdentifier (processId proc))

1110− (NodeIdentifier nid)

1111− WithImplicitReconnect

1112− msg

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

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

31 , Message(..)

32 , createMessage

33 , createUnencodedMessage

34+ , unsafeCreateUnencodedMessage

34 , messageToPayload

35 , messageToPayload

35 , payloadToMessage

36 , payloadToMessage

36 -- * Node controller user-visible data types

37 -- * Node controller user-visible data types

@@ -339,6 +340,11 @@ createUnencodedMessage :: Serializable a => a -> Message

339createUnencodedMessage a =

340createUnencodedMessage a =

340 let encoded = encode a in BSL.length encoded `seq` UnencodedMessage (fingerprint a) a

341 let encoded = encode a in BSL.length encoded `seq` UnencodedMessage (fingerprint a) a

341

342

343+-- | Turn any serializable term into an unencodede/local message, without

344+-- evalutaing it!

345+unsafeCreateUnencodedMessage :: Serializable a => a -> Message

346+unsafeCreateUnencodedMessage a = UnencodedMessage (fingerprint a) a

347+

342-- | Serialize a message

348-- | Serialize a message

343messageToPayload :: Message -> [BSS.ByteString]

349messageToPayload :: Message -> [BSS.ByteString]

344messageToPayload (EncodedMessage fp enc) = encodeFingerprint fp : BSL.toChunks enc

350messageToPayload (EncodedMessage fp enc) = encodeFingerprint fp : BSL.toChunks enc

src/Control/Distributed/Process/UnsafePrimitives.hsnew file+114

@@ -0,0 +1,114 @@

1+-----------------------------------------------------------------------------

2+-- |

3+-- Module : Control.Distributed.Process.UnsafePrimitives

4+-- Copyright : (c) Well-Typed / Tim Watson

5+-- License : BSD3 (see the file LICENSE)

6+--

7+-- Maintainer : Tim Watson <watson.timothy@gmail.com>

8+-- Stability : experimental

9+-- Portability : non-portable (requires concurrency)

10+--

11+-- [Unsafe Variants of Cloud Haskell's Messaging Primitives]

12+--

13+-- Cloud Haskell's semantics do not mention evaluation/strictness properties

14+-- at all; Only the promise (or lack) of signal/message delivery between

15+-- processes is considered. For practical purposes, Cloud Haskell optimises

16+-- communication between (intra-node) local processes by skipping the

17+-- network-transport layer. In order to ensure the same strictness semantics

18+-- however, messages /still/ undergo binary serialisation before (internal)

19+-- transmission takes place. Thus we ensure that in both (the local and remote)

20+-- cases, message payloads are fully evaluated. Whilst this provides the user

21+-- with /unsurprising/ behaviour, the resulting performance overhead is quite

22+-- severe. Communicating data between Haskell threads /without/ forcing binary

23+-- serialisation reduces (intra-node, inter-process) communication overheads

24+-- by several orders of magnitude.

25+--

26+-- This module provides variants of Cloud Haskell's messaging primitives

27+-- ('send', 'sendChan' and 'nsend') which do /not/ force binary serialisation in

28+-- the local case, thereby offering superior intra-node messaging performance.

29+-- The /catch/ is that any evaluation problems lurking within the passed data

30+-- structure - e.g., fields set to @undefined@ and so on - will show up in the

31+-- receiver rather than in the caller (as they would with the /normal/ strategy).

32+--

33+-- Use of the functions in this module can potentially change the runtime

34+-- behaviour of your application. You have been warned!

35+--

36+-- This module is exported so that you can replace the use of Cloud Haskell's

37+-- /safe/ messaging primitives. If you want to use both variants, then you can

38+-- take advantage of qualified imports, however "Control.Distributed.Process"

39+-- also re-exports these functions under different names, using the @unsafe@

40+-- prefix.

41+--

42+module Control.Distributed.Process.UnsafePrimitives

43+ ( -- * Unsafe Basic Messaging

44+ send

45+ , sendChan

46+ , nsend

47+ ) where

48+

49+import Control.Distributed.Process.Internal.Messaging

50+ ( sendMessage

51+ , sendBinary

52+ , sendCtrlMsg

53+ )

54+

55+import Control.Distributed.Process.Internal.Types

56+ ( ProcessId(..)

57+ , LocalNode(..)

58+ , LocalProcess(..)

59+ , Process(..)

60+ , SendPort(..)

61+ , ProcessSignal(..)

62+ , Identifier(..)

63+ , ImplicitReconnect(..)

64+ , SendPortId(..)

65+ , sendPortProcessId

66+ , unsafeCreateUnencodedMessage

67+ )

68+import Control.Distributed.Process.Serializable (Serializable)

69+

70+import Control.Monad.IO.Class (MonadIO, liftIO)

71+import Control.Monad.Reader (ask)

72+

73+-- | Named send to a process in the local registry (asynchronous)

74+nsend :: Serializable a => String -> a -> Process ()

75+nsend label msg =

76+ sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg))

77+

78+-- | Send a message

79+send :: Serializable a => ProcessId -> a -> Process ()

80+send them msg = do

81+ proc <- ask

82+ let node = localNodeId (processNode proc)

83+ destNode = (processNodeId them) in do

84+ case destNode == node of

85+ True -> unsafeSendLocal them msg

86+ False -> liftIO $ sendMessage (processNode proc)

87+ (ProcessIdentifier (processId proc))

88+ (ProcessIdentifier them)

89+ NoImplicitReconnect

90+ msg

91+

92+-- | Send a message on a typed channel

93+sendChan :: Serializable a => SendPort a -> a -> Process ()

94+sendChan (SendPort cid) msg = do

95+ proc <- ask

96+ let node = localNodeId (processNode proc)

97+ destNode = processNodeId (sendPortProcessId cid) in do

98+ case destNode == node of

99+ True -> unsafeSendChanLocal cid msg

100+ False -> do

101+ liftIO $ sendBinary (processNode proc)

102+ (ProcessIdentifier (processId proc))

103+ (SendPortIdentifier cid)

104+ NoImplicitReconnect

105+ msg

106+

107+unsafeSendLocal :: (Serializable a) => ProcessId -> a -> Process ()

108+unsafeSendLocal pid msg =

109+ sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg)

110+

111+unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

112+unsafeSendChanLocal spId msg =

113+ sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg)

114+

tests/TestCH.hs+60

@@ -1167,6 +1167,63 @@ testExitRemote transport = do

1167 takeMVar supervisedDone

1168 takeMVar supervisorDone

1169

1170+testUnsafeSend :: NT.Transport -> Assertion

1171+testUnsafeSend transport = do

1172+ serverAddr <- newEmptyMVar

1173+ clientDone <- newEmptyMVar

1174+

1175+ localNode <- newLocalNode transport initRemoteTable

1176+ void $ forkProcess localNode $ do

1177+ self <- getSelfPid

1178+ liftIO $ putMVar serverAddr self

1179+ clientAddr <- expect

1180+ unsafeSend clientAddr ()

1181+

1182+ void $ forkProcess localNode $ do

1183+ serverPid <- liftIO $ takeMVar serverAddr

1184+ getSelfPid >>= unsafeSend serverPid

1185+ () <- expect

1186+ liftIO $ putMVar clientDone ()

1187+

1188+ takeMVar clientDone

1189+

1190+testUnsafeNSend :: NT.Transport -> Assertion

1191+testUnsafeNSend transport = do

1192+ clientDone <- newEmptyMVar

1193+

1194+ localNode <- newLocalNode transport initRemoteTable

1195+

1196+ pid <- forkProcess localNode $ do

1197+ () <- expect

1198+ liftIO $ putMVar clientDone ()

1199+

1200+ void $ runProcess localNode $ do

1201+ register "foobar" pid

1202+ unsafeNSend "foobar" ()

1203+

1204+ takeMVar clientDone

1205+

1206+testUnsafeSendChan :: NT.Transport -> Assertion

1207+testUnsafeSendChan transport = do

1208+ serverAddr <- newEmptyMVar

1209+ clientDone <- newEmptyMVar

1210+

1211+ localNode <- newLocalNode transport initRemoteTable

1212+ void $ forkProcess localNode $ do

1213+ self <- getSelfPid

1214+ liftIO $ putMVar serverAddr self

1215+ sp <- expect

1216+ unsafeSendChan sp ()

1217+

1218+ void $ forkProcess localNode $ do

1219+ serverPid <- liftIO $ takeMVar serverAddr

1220+ (sp, rp) <- newChan

1221+ unsafeSend serverPid sp

1222+ () <- receiveChan rp

1223+ liftIO $ putMVar clientDone ()

1224+

1225+ takeMVar clientDone

1226+

1170tests :: (NT.Transport, TransportInternals) -> [Test]

1227tests :: (NT.Transport, TransportInternals) -> [Test]

1171tests (transport, transportInternals) = [

1228tests (transport, transportInternals) = [

1172 testGroup "Basic features" [

1229 testGroup "Basic features" [

@@ -1227,6 +1284,9 @@ tests (transport, transportInternals) = [

1227 , testCase "MonitorChannel" (testMonitorChannel transport)

1284 , testCase "MonitorChannel" (testMonitorChannel transport)

1228 -- Reconnect

1285 -- Reconnect

1229 , testCase "Reconnect" (testReconnect transport transportInternals)

1286 , testCase "Reconnect" (testReconnect transport transportInternals)

1287+ , testCase "TestUnsafeSend" (testUnsafeSend transport)

1288+ , testCase "TestUnsafeNSend" (testUnsafeNSend transport)

1289+ , testCase "TestUnsafeSendChan" (testUnsafeSendChan transport)

1230 ]

1290 ]

1231 ]

1291 ]

1232

1292