git.iko.soy/distributed-process
on master help sign in
distributed-process/src/Control/Distributed/Process/UnsafePrimitives.hs
200 lines 8.3 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 msg' = wrapMessage msg
108 -- see [note: tracing]
109 liftIO $ traceEvent (localEventBus (processNode proc))
110 (MxSentToName label us msg')
111 sendCtrlMsg Nothing (NamedSend label msg')
113-- | Named send to a process in a remote registry (asynchronous)
114nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
115nsendRemote nid label msg = do
116 proc <- ask
117 let us = processId proc
118 let node = processNode proc
119 if localNodeId node == nid
120 then nsend label msg
121 else
122 let lbl = label ++ "@" ++ show nid in do
123 -- see [note: tracing] NB: this is a remote call to another NC...
124 liftIO $ traceEvent (localEventBus node)
125 (MxSentToName lbl us (wrapMessage msg))
126 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
128-- | Send a message
129send :: Serializable a => ProcessId -> a -> Process ()
130send them msg = do
131 proc <- ask
132 let node = localNodeId (processNode proc)
133 destNode = (processNodeId them)
134 us = (processId proc)
135 msg' = wrapMessage msg in do
136 -- see [note: tracing]
137 liftIO $ traceEvent (localEventBus (processNode proc))
138 (MxSent them us msg')
139 if destNode == node
140 then sendCtrlMsg Nothing $ LocalSend them msg'
141 else liftIO $ sendMessage (processNode proc)
142 (ProcessIdentifier (processId proc))
143 (ProcessIdentifier them)
144 NoImplicitReconnect
145 msg
147-- | Send a message unreliably.
148--
149-- Unlike 'send', this function is insensitive to 'reconnect'. It will
150-- try to send the message regardless of the history of connection failures
151-- between the nodes.
152--
153-- Message passing with 'usend' is ordered for a given sender and receiver
154-- if the messages arrive at all.
155--
156usend :: Serializable a => ProcessId -> a -> Process ()
157usend them msg = do
158 proc <- ask
159 let there = processNodeId them
160 let (us, node) = (processId proc, processNode proc)
161 let msg' = wrapMessage msg
162 -- see [note: tracing]
163 liftIO $ traceEvent (localEventBus node) (MxSent them us msg')
164 if localNodeId (processNode proc) == there
165 then sendCtrlMsg Nothing $ LocalSend them msg'
166 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
167 (createMessage msg)
169-- [note: tracing]
170-- Note that tracing writes to the local node's control channel, and this
171-- module explicitly specifies to its clients that it does unsafe message
172-- encoding. The same is true for the messages it puts onto the Management
173-- event bus, however we do *not* want unevaluated thunks hitting the event
174-- bus control thread. Hence the word /Unsafe/ in this module's name!
175--
177-- | Send a message on a typed channel
178sendChan :: Serializable a => SendPort a -> a -> Process ()
179sendChan (SendPort cid) msg = do
180 proc <- ask
181 let node = processNode proc
182 pid = processId proc
183 us = localNodeId node
184 them = processNodeId (sendPortProcessId cid)
185 msg' = wrapMessage msg in do
186 liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid msg')
187 if them == us
188 then unsafeSendChanLocal cid msg' -- NB: we wrap to P.Message !!!
189 else liftIO $ sendBinary node
190 (ProcessIdentifier pid)
191 (SendPortIdentifier cid)
192 NoImplicitReconnect
193 msg
194 where
195 unsafeSendChanLocal :: SendPortId -> Message -> Process ()
196 unsafeSendChanLocal p m = sendCtrlMsg Nothing $ LocalPortSend p m
198-- | Create an unencoded @Message@ for any @Serializable@ type.
199wrapMessage :: Serializable a => a -> Message
200wrapMessage = unsafeCreateUnencodedMessage