git.iko.soy/distributed-process
on remove-table-api help sign in
distributed-process/src/Control/Distributed/Process/UnsafePrimitives.hs
151 lines 6.0 kB history raw
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', 'nsend' and 'wrapMessage') which do /not/ force binary
28-- serialisation in the local case, thereby offering superior intra-node
29-- messaging performance. The /catch/ is that any evaluation problems lurking
30-- within the passed data structure (e.g., fields set to @undefined@ and so on)
31-- will show up in the receiver rather than in the caller (as they would with
32-- the /normal/ strategy).
33--
34-- Use of the functions in this module can potentially change the runtime
35-- behaviour of your application. You have been warned!
36--
37-- This module is exported so that you can replace the use of Cloud Haskell's
38-- /safe/ messaging primitives. If you want to use both variants, then you can
39-- take advantage of qualified imports, however "Control.Distributed.Process"
40-- also re-exports these functions under different names, using the @unsafe@
41-- prefix.
42--
43module Control.Distributed.Process.UnsafePrimitives
44 ( -- * Unsafe Basic Messaging
45 send
46 , sendChan
47 , nsend
48 , nsendRemote
49 , usend
50 , wrapMessage
51 ) where
53import Control.Distributed.Process.Internal.Messaging
54 ( sendMessage
55 , sendBinary
56 , sendCtrlMsg
57 )
59import Control.Distributed.Process.Internal.Types
60 ( ProcessId(..)
61 , NodeId(..)
62 , LocalNode(..)
63 , LocalProcess(..)
64 , Process(..)
65 , SendPort(..)
66 , ProcessSignal(..)
67 , Identifier(..)
68 , ImplicitReconnect(..)
69 , SendPortId(..)
70 , Message
71 , createMessage
72 , sendPortProcessId
73 , unsafeCreateUnencodedMessage
74 )
75import Control.Distributed.Process.Serializable (Serializable)
77import Control.Monad.IO.Class (liftIO)
78import Control.Monad.Reader (ask)
80-- | Named send to a process in the local registry (asynchronous)
81nsend :: Serializable a => String -> a -> Process ()
82nsend label msg =
83 sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg))
85-- | Named send to a process in a remote registry (asynchronous)
86nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
87nsendRemote nid label msg = do
88 proc <- ask
89 if localNodeId (processNode proc) == nid
90 then nsend label msg
91 else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
93-- | Send a message
94send :: Serializable a => ProcessId -> a -> Process ()
95send them msg = do
96 proc <- ask
97 let node = localNodeId (processNode proc)
98 destNode = (processNodeId them) in do
99 case destNode == node of
100 True -> unsafeSendLocal them msg
101 False -> liftIO $ sendMessage (processNode proc)
102 (ProcessIdentifier (processId proc))
103 (ProcessIdentifier them)
104 NoImplicitReconnect
105 msg
106 where
107 unsafeSendLocal :: (Serializable a) => ProcessId -> a -> Process ()
108 unsafeSendLocal pid msg' =
109 sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg')
111-- | Send a message unreliably.
112--
113-- Unlike 'send', this function is insensitive to 'reconnect'. It will
114-- try to send the message regardless of the history of connection failures
115-- between the nodes.
116--
117-- Message passing with 'usend' is ordered for a given sender and receiver
118-- if the messages arrive at all.
119--
120usend :: Serializable a => ProcessId -> a -> Process ()
121usend them msg = do
122 proc <- ask
123 let there = processNodeId them
124 if localNodeId (processNode proc) == there
125 then sendCtrlMsg Nothing $
126 LocalSend them (unsafeCreateUnencodedMessage msg)
127 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
128 (createMessage msg)
130-- | Send a message on a typed channel
131sendChan :: Serializable a => SendPort a -> a -> Process ()
132sendChan (SendPort cid) msg = do
133 proc <- ask
134 let node = localNodeId (processNode proc)
135 destNode = processNodeId (sendPortProcessId cid) in do
136 case destNode == node of
137 True -> unsafeSendChanLocal cid msg
138 False -> do
139 liftIO $ sendBinary (processNode proc)
140 (ProcessIdentifier (processId proc))
141 (SendPortIdentifier cid)
142 NoImplicitReconnect
143 msg
144 where
145 unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
146 unsafeSendChanLocal spId msg' =
147 sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg')
149-- | Create an unencoded @Message@ for any @Serializable@ type.
150wrapMessage :: Serializable a => a -> Message
151wrapMessage = unsafeCreateUnencodedMessage