git.iko.soy/distributed-process
on d2b5d82 help sign in
distributed-process/src/Control/Distributed/Process/UnsafePrimitives.hs
208 lines 8.9 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. In addition, messages passed between Cloud
36-- Haskell processes are written to a tracing infrastructure on the local node,
37-- to provide improved introspection and debugging facilities for complex actor
38-- based systems. This module makes no attempt to force evaluation in these
39-- cases either, thus evaluation problems in passed data structures could not
40-- only crash your processes, but could also bring down critical internal
41-- services on which the node relies to function correctly.
42--
43-- If you wish to repudiate such issues, you are advised to consider the use
44-- of NFSerialisable in the distributed-process-extras package, which type
45-- class brings NFData into scope along with Serializable, such that we can
46-- force evaluation. Intended for use with modules such as this one, this
47-- approach guarantees correct evaluatedness in terms of @NFData@. Please note
48-- however, that we /cannot/ guarantee that an @NFData@ instance will behave the
49-- same way as a @Binary@ one with regards evaluation, so it is still possible
50-- to introduce unexpected behaviour by using /unsafe/ primitives in this way.
51--
52-- You have been warned!
53--
54-- This module is exported so that you can replace the use of Cloud Haskell's
55-- /safe/ messaging primitives. If you want to use both variants, then you can
56-- take advantage of qualified imports, however "Control.Distributed.Process"
57-- also re-exports these functions under different names, using the @unsafe@
58-- prefix.
59--
60module Control.Distributed.Process.UnsafePrimitives
61 ( -- * Unsafe Basic Messaging
62 send
63 , sendChan
64 , nsend
65 , nsendRemote
66 , usend
67 , wrapMessage
68 ) where
70import Control.Distributed.Process.Internal.Messaging
71 ( sendMessage
72 , sendBinary
73 , sendCtrlMsg
74 )
75import Control.Distributed.Process.Management.Internal.Types
76 ( MxEvent(..)
77 )
78import Control.Distributed.Process.Management.Internal.Trace.Types
79 ( traceEvent
80 )
81import Control.Distributed.Process.Internal.Types
82 ( ProcessId(..)
83 , NodeId(..)
84 , LocalNode(..)
85 , LocalProcess(..)
86 , Process(..)
87 , SendPort(..)
88 , ProcessSignal(..)
89 , Identifier(..)
90 , ImplicitReconnect(..)
91 , SendPortId(..)
92 , Message
93 , createMessage
94 , sendPortProcessId
95 , unsafeCreateUnencodedMessage
96 )
97import Control.Distributed.Process.Serializable (Serializable)
99import Control.Monad.IO.Class (liftIO)
100import Control.Monad.Reader (ask)
102-- | Named send to a process in the local registry (asynchronous)
103nsend :: Serializable a => String -> a -> Process ()
104nsend label msg = do
105 proc <- ask
106 let us = processId proc
107 let node = localNodeId (processNode proc)
108 let msg' = wrapMessage msg
109 -- see [note: tracing]
110 liftIO $ traceEvent (localEventBus (processNode proc))
111 (MxSentToName label us msg')
112 sendCtrlMsg Nothing (NamedSend label msg')
114-- [note: tracing]
115-- In the remote case, we do not fire a trace event until after sending is
116-- complete, since 'sendMessage' can block in the networking stack.
117--
118-- In addition, MxSent trace messages are dispatched by the node controller's
119-- thread for local sends, which also covers local nsend, and local usend.
120--
121-- Also note that tracing writes to the local node's control channel, and this
122-- module explicitly specifies to its clients that it does unsafe message
123-- encoding. The same is true for the messages it puts onto the Management
124-- event bus, however we do *not* want unevaluated thunks hitting the event
125-- bus control thread. Hence the word /Unsafe/ in this module's name!
126--
128-- | Named send to a process in a remote registry (asynchronous)
129nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
130nsendRemote nid label msg = do
131 proc <- ask
132 let us = processId proc
133 let node = processNode proc
134 if localNodeId (processNode proc) == nid
135 then nsend label msg
136 else do -- see [note: tracing] NB: this is a remote call to another NC...
137 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
138 liftIO $ traceEvent (localEventBus node)
139 (MxSentToName label us (wrapMessage msg))
141-- | Send a message
142send :: Serializable a => ProcessId -> a -> Process ()
143send them msg = do
144 proc <- ask
145 let node = localNodeId (processNode proc)
146 destNode = (processNodeId them)
147 us = (processId proc) in do
148 case destNode == node of
149 True -> unsafeSendLocal them msg
150 False -> liftIO $ do sendMessage (processNode proc)
151 (ProcessIdentifier (processId proc))
152 (ProcessIdentifier them)
153 NoImplicitReconnect
154 msg
155 -- see [note: tracing]
156 liftIO $ traceEvent (localEventBus (processNode proc))
157 (MxSent them us (wrapMessage msg))
159-- | Send a message unreliably.
160--
161-- Unlike 'send', this function is insensitive to 'reconnect'. It will
162-- try to send the message regardless of the history of connection failures
163-- between the nodes.
164--
165-- Message passing with 'usend' is ordered for a given sender and receiver
166-- if the messages arrive at all.
167--
168usend :: Serializable a => ProcessId -> a -> Process ()
169usend them msg = do
170 proc <- ask
171 let there = processNodeId them
172 let (us, node) = (processId proc, processNode proc)
173 if localNodeId (processNode proc) == there
174 then unsafeSendLocal them msg
175 else do sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
176 (createMessage msg)
177 -- I would assert that is it *still* cheaper to not encode here...
178 liftIO $ traceEvent (localEventBus node)
179 (MxSent them us (wrapMessage msg))
181unsafeSendLocal :: (Serializable a) => ProcessId -> a -> Process ()
182unsafeSendLocal pid msg =
183 let msg' = wrapMessage msg in do
184 sendCtrlMsg Nothing $ LocalSend pid msg'
185 -- see [note: tracing]
187-- | Send a message on a typed channel
188sendChan :: Serializable a => SendPort a -> a -> Process ()
189sendChan (SendPort cid) msg = do
190 proc <- ask
191 let node = localNodeId (processNode proc)
192 destNode = processNodeId (sendPortProcessId cid) in do
193 case destNode == node of
194 True -> unsafeSendChanLocal cid msg
195 False -> do
196 liftIO $ sendBinary (processNode proc)
197 (ProcessIdentifier (processId proc))
198 (SendPortIdentifier cid)
199 NoImplicitReconnect
200 msg
201 where
202 unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
203 unsafeSendChanLocal spId msg' =
204 sendCtrlMsg Nothing $ LocalPortSend spId (wrapMessage msg')
206-- | Create an unencoded @Message@ for any @Serializable@ type.
207wrapMessage :: Serializable a => a -> Message
208wrapMessage = unsafeCreateUnencodedMessage