git.iko.soy/distributed-process
on 39feb45 help sign in
distributed-process/src/Control/Distributed/Process/UnsafePrimitives.hs
114 lines 4.6 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' 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--
42module Control.Distributed.Process.UnsafePrimitives
43 ( -- * Unsafe Basic Messaging
44 send
45 , sendChan
46 , nsend
47 ) where
49import Control.Distributed.Process.Internal.Messaging
50 ( sendMessage
51 , sendBinary
52 , sendCtrlMsg
53 )
55import 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 )
68import Control.Distributed.Process.Serializable (Serializable)
70import Control.Monad.IO.Class (MonadIO, liftIO)
71import Control.Monad.Reader (ask)
73-- | Named send to a process in the local registry (asynchronous)
74nsend :: Serializable a => String -> a -> Process ()
75nsend label msg =
76 sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg))
78-- | Send a message
79send :: Serializable a => ProcessId -> a -> Process ()
80send 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
92-- | Send a message on a typed channel
93sendChan :: Serializable a => SendPort a -> a -> Process ()
94sendChan (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
107unsafeSendLocal :: (Serializable a) => ProcessId -> a -> Process ()
108unsafeSendLocal pid msg =
109 sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg)
111unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
112unsafeSendChanLocal spId msg =
113 sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg)