@@ -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+