git.iko.soy/distributed-process
on 42ba2b9 help sign in
distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs
277 lines 11 kB history raw
1module Control.Distributed.Process.Internal.Closure.BuiltIn
2 ( -- * Remote table
3 remoteTable
4 -- * Static dictionaries and associated operations
5 , staticDecode
6 , sdictUnit
7 , sdictProcessId
8 , sdictSendPort
9 -- * Some static values
10 , sndStatic
11 -- * The CP type and associated combinators
12 , CP
13 , idCP
14 , splitCP
15 , returnCP
16 , bindCP
17 , seqCP
18 -- * CP versions of Cloud Haskell primitives
19 , cpLink
20 , cpUnlink
21 , cpSend
22 , cpExpect
23 , cpNewChan
24 -- * Support for some CH operations
25 , cpDelay
26 ) where
28import Data.ByteString.Lazy (ByteString)
29import Data.Binary (decode, encode)
30import Data.Rank1Typeable (Typeable, ANY, ANY1, ANY2, ANY3, ANY4)
31import Data.Rank1Dynamic (toDynamic)
32import Control.Distributed.Static
33 ( RemoteTable
34 , registerStatic
35 , Static
36 , staticLabel
37 , staticApply
38 , Closure
39 , closure
40 , closureApplyStatic
41 , closureApply
42 , staticCompose
43 , staticClosure
44 )
45import Control.Distributed.Process.Serializable
46 ( SerializableDict(..)
47 , Serializable
48 )
49import Control.Distributed.Process.Internal.Types
50 ( Process
51 , ProcessId
52 , SendPort
53 , ReceivePort
54 , ProcessMonitorNotification(ProcessMonitorNotification)
55 )
56import Control.Distributed.Process.Internal.Primitives
57 ( link
58 , unlink
59 , send
60 , expect
61 , newChan
62 , monitor
63 , unmonitor
64 , match
65 , matchIf
66 , receiveWait
67 )
69--------------------------------------------------------------------------------
70-- Remote table --
71--------------------------------------------------------------------------------
73remoteTable :: RemoteTable -> RemoteTable
74remoteTable =
75 registerStatic "$decodeDict" (toDynamic (decodeDict :: SerializableDict ANY -> ByteString -> ANY))
76 . registerStatic "$sdictUnit" (toDynamic (SerializableDict :: SerializableDict ()))
77 . registerStatic "$sdictProcessId" (toDynamic (SerializableDict :: SerializableDict ProcessId))
78 . registerStatic "$sdictSendPort_" (toDynamic (sdictSendPort_ :: SerializableDict ANY -> SerializableDict (SendPort ANY)))
79 . registerStatic "$returnProcess" (toDynamic (return :: ANY -> Process ANY))
80 . registerStatic "$seqProcess" (toDynamic ((>>) :: Process ANY1 -> Process ANY2 -> Process ANY2))
81 . registerStatic "$bindProcess" (toDynamic ((>>=) :: Process ANY1 -> (ANY1 -> Process ANY2) -> Process ANY2))
82 . registerStatic "$decodeProcessId" (toDynamic (decode :: ByteString -> ProcessId))
83 . registerStatic "$link" (toDynamic link)
84 . registerStatic "$unlink" (toDynamic unlink)
85 . registerStatic "$sendDict" (toDynamic (sendDict :: SerializableDict ANY -> ProcessId -> ANY -> Process ()))
86 . registerStatic "$expectDict" (toDynamic (expectDict :: SerializableDict ANY -> Process ANY))
87 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))
88 . registerStatic "$cpSplit" (toDynamic (cpSplit :: (ANY1 -> Process ANY3) -> (ANY2 -> Process ANY4) -> (ANY1, ANY2) -> Process (ANY3, ANY4)))
89 . registerStatic "$snd" (toDynamic (snd :: (ANY1, ANY2) -> ANY2))
90 . registerStatic "$delay" (toDynamic delay)
91 where
92 decodeDict :: forall a. SerializableDict a -> ByteString -> a
93 decodeDict SerializableDict = decode
95 sdictSendPort_ :: forall a. SerializableDict a -> SerializableDict (SendPort a)
96 sdictSendPort_ SerializableDict = SerializableDict
98 sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
99 sendDict SerializableDict = send
101 expectDict :: forall a. SerializableDict a -> Process a
102 expectDict SerializableDict = expect
104 newChanDict :: forall a. SerializableDict a -> Process (SendPort a, ReceivePort a)
105 newChanDict SerializableDict = newChan
107 cpSplit :: forall a b c d. (a -> Process c) -> (b -> Process d) -> (a, b) -> Process (c, d)
108 cpSplit f g (a, b) = do
109 c <- f a
110 d <- g b
111 return (c, d)
113--------------------------------------------------------------------------------
114-- Static dictionaries and associated operations --
115--------------------------------------------------------------------------------
117-- | Static decoder, given a static serialization dictionary.
118--
119-- See module documentation of "Control.Distributed.Process.Closure" for an
120-- example.
121staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a)
122staticDecode dict = decodeDictStatic `staticApply` dict
123 where
124 decodeDictStatic :: Typeable a => Static (SerializableDict a -> ByteString -> a)
125 decodeDictStatic = staticLabel "$decodeDict"
127-- | Serialization dictionary for '()'
128sdictUnit :: Static (SerializableDict ())
129sdictUnit = staticLabel "$sdictUnit"
131-- | Serialization dictionary for 'ProcessId'
132sdictProcessId :: Static (SerializableDict ProcessId)
133sdictProcessId = staticLabel "$sdictProcessId"
135-- | Serialization dictionary for 'SendPort'
136sdictSendPort :: Typeable a
137 => Static (SerializableDict a) -> Static (SerializableDict (SendPort a))
138sdictSendPort = staticApply (staticLabel "$sdictSendPort_")
140--------------------------------------------------------------------------------
141-- Static values --
142--------------------------------------------------------------------------------
144sndStatic :: Static ((a, b) -> b)
145sndStatic = staticLabel "$snd"
147--------------------------------------------------------------------------------
148-- The CP type and associated combinators --
149--------------------------------------------------------------------------------
151-- | @CP a b@ is a process with input of type @a@ and output of type @b@
152type CP a b = Closure (a -> Process b)
154returnProcessStatic :: Typeable a => Static (a -> Process a)
155returnProcessStatic = staticLabel "$returnProcess"
157-- | 'CP' version of 'Control.Category.id'
158idCP :: Typeable a => CP a a
159idCP = staticClosure returnProcessStatic
161-- | 'CP' version of ('Control.Arrow.***')
162splitCP :: (Typeable a, Typeable b, Typeable c, Typeable d)
163 => CP a c -> CP b d -> CP (a, b) (c, d)
164splitCP p q = cpSplitStatic `closureApplyStatic` p `closureApply` q
165 where
166 cpSplitStatic :: Static ((a -> Process c) -> (b -> Process d) -> (a, b) -> Process (c, d))
167 cpSplitStatic = staticLabel "$cpSplit"
169-- | 'CP' version of 'Control.Monad.return'
170returnCP :: forall a. Serializable a
171 => Static (SerializableDict a) -> a -> Closure (Process a)
172returnCP dict x = closure decoder (encode x)
173 where
174 decoder :: Static (ByteString -> Process a)
175 decoder = returnProcessStatic
176 `staticCompose`
177 staticDecode dict
179-- | 'CP' version of ('Control.Monad.>>')
180seqCP :: (Typeable a, Typeable b)
181 => Closure (Process a) -> Closure (Process b) -> Closure (Process b)
182seqCP p q = seqProcessStatic `closureApplyStatic` p `closureApply` q
183 where
184 seqProcessStatic :: (Typeable a, Typeable b)
185 => Static (Process a -> Process b -> Process b)
186 seqProcessStatic = staticLabel "$seqProcess"
188-- | (Not quite the) 'CP' version of ('Control.Monad.>>=')
189bindCP :: forall a b. (Typeable a, Typeable b)
190 => Closure (Process a) -> CP a b -> Closure (Process b)
191bindCP x f = bindProcessStatic `closureApplyStatic` x `closureApply` f
192 where
193 bindProcessStatic :: (Typeable a, Typeable b)
194 => Static (Process a -> (a -> Process b) -> Process b)
195 bindProcessStatic = staticLabel "$bindProcess"
197--------------------------------------------------------------------------------
198-- CP versions of Cloud Haskell primitives --
199--------------------------------------------------------------------------------
201decodeProcessIdStatic :: Static (ByteString -> ProcessId)
202decodeProcessIdStatic = staticLabel "$decodeProcessId"
204-- | 'CP' version of 'link'
205cpLink :: ProcessId -> Closure (Process ())
206cpLink = closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode
207 where
208 linkStatic :: Static (ProcessId -> Process ())
209 linkStatic = staticLabel "$link"
211-- | 'CP' version of 'unlink'
212cpUnlink :: ProcessId -> Closure (Process ())
213cpUnlink = closure (unlinkStatic `staticCompose` decodeProcessIdStatic) . encode
214 where
215 unlinkStatic :: Static (ProcessId -> Process ())
216 unlinkStatic = staticLabel "$unlink"
218-- | 'CP' version of 'send'
219cpSend :: forall a. Typeable a
220 => Static (SerializableDict a) -> ProcessId -> CP a ()
221cpSend dict pid = closure decoder (encode pid)
222 where
223 decoder :: Static (ByteString -> a -> Process ())
224 decoder = (sendDictStatic `staticApply` dict)
225 `staticCompose`
226 decodeProcessIdStatic
228 sendDictStatic :: Typeable a
229 => Static (SerializableDict a -> ProcessId -> a -> Process ())
230 sendDictStatic = staticLabel "$sendDict"
232-- | 'CP' version of 'expect'
233cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a)
234cpExpect dict = staticClosure (expectDictStatic `staticApply` dict)
235 where
236 expectDictStatic :: Typeable a => Static (SerializableDict a -> Process a)
237 expectDictStatic = staticLabel "$expectDict"
239-- | 'CP' version of 'newChan'
240cpNewChan :: Typeable a
241 => Static (SerializableDict a)
242 -> Closure (Process (SendPort a, ReceivePort a))
243cpNewChan dict = staticClosure (newChanDictStatic `staticApply` dict)
244 where
245 newChanDictStatic :: Typeable a
246 => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))
247 newChanDictStatic = staticLabel "$newChanDict"
249--------------------------------------------------------------------------------
250-- Support for spawn --
251--------------------------------------------------------------------------------
253-- | @delay them p@ is a process that waits for a signal (a message of type @()@)
254-- from 'them' (origin is not verified) before proceeding as @p@. In order to
255-- avoid waiting forever, @delay them p@ monitors 'them'. If it receives a
256-- monitor message instead it simply terminates.
257delay :: ProcessId -> Process () -> Process ()
258delay them p = do
259 ref <- monitor them
260 let sameRef (ProcessMonitorNotification ref' _ _) = ref == ref'
261 receiveWait [
262 match $ \() -> unmonitor ref >> p
263 , matchIf sameRef $ \_ -> return ()
264 ]
266-- | 'CP' version of 'delay'
267cpDelay :: ProcessId -> Closure (Process ()) -> Closure (Process ())
268cpDelay = closureApply . cpDelay'
269 where
270 cpDelay' :: ProcessId -> Closure (Process () -> Process ())
271 cpDelay' pid = closure decoder (encode pid)
273 decoder :: Static (ByteString -> Process () -> Process ())
274 decoder = delayStatic `staticCompose` decodeProcessIdStatic
276 delayStatic :: Static (ProcessId -> Process () -> Process ())
277 delayStatic = staticLabel "$delay"