git.iko.soy/distributed-process
on 39ed082 help sign in
distributed-process/src/Control/Distributed/Process/UnsafePrimitives.hs
120 lines 4.8 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 , wrapMessage
49 ) where
51import Control.Distributed.Process.Internal.Messaging
52 ( sendMessage
53 , sendBinary
54 , sendCtrlMsg
55 )
57import Control.Distributed.Process.Internal.Types
58 ( ProcessId(..)
59 , LocalNode(..)
60 , LocalProcess(..)
61 , Process(..)
62 , SendPort(..)
63 , ProcessSignal(..)
64 , Identifier(..)
65 , ImplicitReconnect(..)
66 , SendPortId(..)
67 , Message
68 , sendPortProcessId
69 , unsafeCreateUnencodedMessage
70 )
71import Control.Distributed.Process.Serializable (Serializable)
73import Control.Monad.IO.Class (liftIO)
74import Control.Monad.Reader (ask)
76-- | Named send to a process in the local registry (asynchronous)
77nsend :: Serializable a => String -> a -> Process ()
78nsend label msg =
79 sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg))
81-- | Send a message
82send :: Serializable a => ProcessId -> a -> Process ()
83send them msg = do
84 proc <- ask
85 let node = localNodeId (processNode proc)
86 destNode = (processNodeId them) in do
87 case destNode == node of
88 True -> unsafeSendLocal them msg
89 False -> liftIO $ sendMessage (processNode proc)
90 (ProcessIdentifier (processId proc))
91 (ProcessIdentifier them)
92 NoImplicitReconnect
93 msg
94 where
95 unsafeSendLocal :: (Serializable a) => ProcessId -> a -> Process ()
96 unsafeSendLocal pid msg' =
97 sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg')
99-- | Send a message on a typed channel
100sendChan :: Serializable a => SendPort a -> a -> Process ()
101sendChan (SendPort cid) msg = do
102 proc <- ask
103 let node = localNodeId (processNode proc)
104 destNode = processNodeId (sendPortProcessId cid) in do
105 case destNode == node of
106 True -> unsafeSendChanLocal cid msg
107 False -> do
108 liftIO $ sendBinary (processNode proc)
109 (ProcessIdentifier (processId proc))
110 (SendPortIdentifier cid)
111 NoImplicitReconnect
112 msg
113 where
114 unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
115 unsafeSendChanLocal spId msg' =
116 sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg')
118-- | Create an unencoded @Message@ for any @Serializable@ type.
119wrapMessage :: Serializable a => a -> Message
120wrapMessage = unsafeCreateUnencodedMessage