git.iko.soy/distributed-process
on 2ffa2bc help sign in
distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs
301 lines 12 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 , decodeProcessIdStatic
20 , cpLink
21 , cpUnlink
22 , cpRelay
23 , cpSend
24 , cpExpect
25 , cpNewChan
26 -- * Support for some CH operations
27 , cpDelay
28 , cpEnableTraceRemote
29 ) where
31import Data.ByteString.Lazy (ByteString)
32import Data.Binary (decode, encode)
33import Data.Rank1Typeable (Typeable, ANY, ANY1, ANY2, ANY3, ANY4)
34import Data.Rank1Dynamic (toDynamic)
35import Control.Distributed.Static
36 ( RemoteTable
37 , registerStatic
38 , Static
39 , staticLabel
40 , staticApply
41 , Closure
42 , closure
43 , closureApplyStatic
44 , closureApply
45 , staticCompose
46 , staticClosure
47 )
48import Control.Distributed.Process.Serializable
49 ( SerializableDict(..)
50 , Serializable
51 )
52import Control.Distributed.Process.Internal.Types
53 ( Process
54 , ProcessId
55 , SendPort
56 , ReceivePort
57 , ProcessMonitorNotification(ProcessMonitorNotification)
58 )
59import Control.Distributed.Process.Internal.Primitives
60 ( link
61 , unlink
62 , relay
63 , send
64 , expect
65 , newChan
66 , monitor
67 , unmonitor
68 , match
69 , matchIf
70 , receiveWait
71 )
73--------------------------------------------------------------------------------
74-- Remote table --
75--------------------------------------------------------------------------------
77remoteTable :: RemoteTable -> RemoteTable
78remoteTable =
79 registerStatic "$decodeDict" (toDynamic (decodeDict :: SerializableDict ANY -> ByteString -> ANY))
80 . registerStatic "$sdictUnit" (toDynamic (SerializableDict :: SerializableDict ()))
81 . registerStatic "$sdictProcessId" (toDynamic (SerializableDict :: SerializableDict ProcessId))
82 . registerStatic "$sdictSendPort_" (toDynamic (sdictSendPort_ :: SerializableDict ANY -> SerializableDict (SendPort ANY)))
83 . registerStatic "$returnProcess" (toDynamic (return :: ANY -> Process ANY))
84 . registerStatic "$seqProcess" (toDynamic ((>>) :: Process ANY1 -> Process ANY2 -> Process ANY2))
85 . registerStatic "$bindProcess" (toDynamic ((>>=) :: Process ANY1 -> (ANY1 -> Process ANY2) -> Process ANY2))
86 . registerStatic "$decodeProcessId" (toDynamic (decode :: ByteString -> ProcessId))
87 . registerStatic "$link" (toDynamic link)
88 . registerStatic "$unlink" (toDynamic unlink)
89 . registerStatic "$relay" (toDynamic relay)
90 . registerStatic "$sendDict" (toDynamic (sendDict :: SerializableDict ANY -> ProcessId -> ANY -> Process ()))
91 . registerStatic "$expectDict" (toDynamic (expectDict :: SerializableDict ANY -> Process ANY))
92 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))
93 . registerStatic "$cpSplit" (toDynamic (cpSplit :: (ANY1 -> Process ANY3) -> (ANY2 -> Process ANY4) -> (ANY1, ANY2) -> Process (ANY3, ANY4)))
94 . registerStatic "$snd" (toDynamic (snd :: (ANY1, ANY2) -> ANY2))
95 . registerStatic "$delay" (toDynamic delay)
96 where
97 decodeDict :: forall a. SerializableDict a -> ByteString -> a
98 decodeDict SerializableDict = decode
100 sdictSendPort_ :: forall a. SerializableDict a -> SerializableDict (SendPort a)
101 sdictSendPort_ SerializableDict = SerializableDict
103 sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
104 sendDict SerializableDict = send
106 expectDict :: forall a. SerializableDict a -> Process a
107 expectDict SerializableDict = expect
109 newChanDict :: forall a. SerializableDict a -> Process (SendPort a, ReceivePort a)
110 newChanDict SerializableDict = newChan
112 cpSplit :: forall a b c d. (a -> Process c) -> (b -> Process d) -> (a, b) -> Process (c, d)
113 cpSplit f g (a, b) = do
114 c <- f a
115 d <- g b
116 return (c, d)
118--------------------------------------------------------------------------------
119-- Static dictionaries and associated operations --
120--------------------------------------------------------------------------------
122-- | Static decoder, given a static serialization dictionary.
123--
124-- See module documentation of "Control.Distributed.Process.Closure" for an
125-- example.
126staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a)
127staticDecode dict = decodeDictStatic `staticApply` dict
128 where
129 decodeDictStatic :: Typeable a => Static (SerializableDict a -> ByteString -> a)
130 decodeDictStatic = staticLabel "$decodeDict"
132-- | Serialization dictionary for '()'
133sdictUnit :: Static (SerializableDict ())
134sdictUnit = staticLabel "$sdictUnit"
136-- | Serialization dictionary for 'ProcessId'
137sdictProcessId :: Static (SerializableDict ProcessId)
138sdictProcessId = staticLabel "$sdictProcessId"
140-- | Serialization dictionary for 'SendPort'
141sdictSendPort :: Typeable a
142 => Static (SerializableDict a) -> Static (SerializableDict (SendPort a))
143sdictSendPort = staticApply (staticLabel "$sdictSendPort_")
145--------------------------------------------------------------------------------
146-- Static values --
147--------------------------------------------------------------------------------
149sndStatic :: Static ((a, b) -> b)
150sndStatic = staticLabel "$snd"
152--------------------------------------------------------------------------------
153-- The CP type and associated combinators --
154--------------------------------------------------------------------------------
156-- | @CP a b@ is a process with input of type @a@ and output of type @b@
157type CP a b = Closure (a -> Process b)
159returnProcessStatic :: Typeable a => Static (a -> Process a)
160returnProcessStatic = staticLabel "$returnProcess"
162-- | 'CP' version of 'Control.Category.id'
163idCP :: Typeable a => CP a a
164idCP = staticClosure returnProcessStatic
166-- | 'CP' version of ('Control.Arrow.***')
167splitCP :: (Typeable a, Typeable b, Typeable c, Typeable d)
168 => CP a c -> CP b d -> CP (a, b) (c, d)
169splitCP p q = cpSplitStatic `closureApplyStatic` p `closureApply` q
170 where
171 cpSplitStatic :: Static ((a -> Process c) -> (b -> Process d) -> (a, b) -> Process (c, d))
172 cpSplitStatic = staticLabel "$cpSplit"
174-- | 'CP' version of 'Control.Monad.return'
175returnCP :: forall a. Serializable a
176 => Static (SerializableDict a) -> a -> Closure (Process a)
177returnCP dict x = closure decoder (encode x)
178 where
179 decoder :: Static (ByteString -> Process a)
180 decoder = returnProcessStatic
181 `staticCompose`
182 staticDecode dict
184-- | 'CP' version of ('Control.Monad.>>')
185seqCP :: (Typeable a, Typeable b)
186 => Closure (Process a) -> Closure (Process b) -> Closure (Process b)
187seqCP p q = seqProcessStatic `closureApplyStatic` p `closureApply` q
188 where
189 seqProcessStatic :: (Typeable a, Typeable b)
190 => Static (Process a -> Process b -> Process b)
191 seqProcessStatic = staticLabel "$seqProcess"
193-- | (Not quite the) 'CP' version of ('Control.Monad.>>=')
194bindCP :: forall a b. (Typeable a, Typeable b)
195 => Closure (Process a) -> CP a b -> Closure (Process b)
196bindCP x f = bindProcessStatic `closureApplyStatic` x `closureApply` f
197 where
198 bindProcessStatic :: (Typeable a, Typeable b)
199 => Static (Process a -> (a -> Process b) -> Process b)
200 bindProcessStatic = staticLabel "$bindProcess"
202--------------------------------------------------------------------------------
203-- CP versions of Cloud Haskell primitives --
204--------------------------------------------------------------------------------
206decodeProcessIdStatic :: Static (ByteString -> ProcessId)
207decodeProcessIdStatic = staticLabel "$decodeProcessId"
209-- | 'CP' version of 'link'
210cpLink :: ProcessId -> Closure (Process ())
211cpLink = closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode
212 where
213 linkStatic :: Static (ProcessId -> Process ())
214 linkStatic = staticLabel "$link"
216-- | 'CP' version of 'unlink'
217cpUnlink :: ProcessId -> Closure (Process ())
218cpUnlink = closure (unlinkStatic `staticCompose` decodeProcessIdStatic) . encode
219 where
220 unlinkStatic :: Static (ProcessId -> Process ())
221 unlinkStatic = staticLabel "$unlink"
223-- | 'CP' version of 'send'
224cpSend :: forall a. Typeable a
225 => Static (SerializableDict a) -> ProcessId -> CP a ()
226cpSend dict pid = closure decoder (encode pid)
227 where
228 decoder :: Static (ByteString -> a -> Process ())
229 decoder = (sendDictStatic `staticApply` dict)
230 `staticCompose`
231 decodeProcessIdStatic
233 sendDictStatic :: Typeable a
234 => Static (SerializableDict a -> ProcessId -> a -> Process ())
235 sendDictStatic = staticLabel "$sendDict"
237-- | 'CP' version of 'expect'
238cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a)
239cpExpect dict = staticClosure (expectDictStatic `staticApply` dict)
240 where
241 expectDictStatic :: Typeable a => Static (SerializableDict a -> Process a)
242 expectDictStatic = staticLabel "$expectDict"
244-- | 'CP' version of 'newChan'
245cpNewChan :: Typeable a
246 => Static (SerializableDict a)
247 -> Closure (Process (SendPort a, ReceivePort a))
248cpNewChan dict = staticClosure (newChanDictStatic `staticApply` dict)
249 where
250 newChanDictStatic :: Typeable a
251 => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))
252 newChanDictStatic = staticLabel "$newChanDict"
254-- | 'CP' version of 'relay'
255cpRelay :: ProcessId -> Closure (Process ())
256cpRelay = closure (relayStatic `staticCompose` decodeProcessIdStatic) . encode
257 where
258 relayStatic :: Static (ProcessId -> Process ())
259 relayStatic = staticLabel "$relay"
261-- TODO: move cpEnableTraceRemote into Trace/Primitives.hs
263cpEnableTraceRemote :: ProcessId -> Closure (Process ())
264cpEnableTraceRemote =
265 closure (enableTraceStatic `staticCompose` decodeProcessIdStatic) . encode
266 where
267 enableTraceStatic :: Static (ProcessId -> Process ())
268 enableTraceStatic = staticLabel "$enableTraceRemote"
270--------------------------------------------------------------------------------
271-- Support for spawn --
272--------------------------------------------------------------------------------
274-- | @delay them p@ is a process that waits for a signal (a message of type @()@)
275-- from 'them' (origin is not verified) before proceeding as @p@. In order to
276-- avoid waiting forever, @delay them p@ monitors 'them'. If it receives a
277-- monitor message instead it simply terminates.
278delay :: ProcessId -> Process () -> Process ()
279delay them p = do
280 ref <- monitor them
281 let sameRef (ProcessMonitorNotification ref' _ _) = ref == ref'
282 receiveWait [
283 match $ \() -> unmonitor ref >> p
284 , matchIf sameRef $ \_ -> return ()
285 ]
287-- | 'CP' version of 'delay'
288cpDelay :: ProcessId -> Closure (Process ()) -> Closure (Process ())
289cpDelay = closureApply . cpDelay'
290 where
291 cpDelay' :: ProcessId -> Closure (Process () -> Process ())
292 cpDelay' pid = closure decoder (encode pid)
294 decoder :: Static (ByteString -> Process () -> Process ())
295 decoder = delayStatic `staticCompose` decodeProcessIdStatic
297 delayStatic :: Static (ProcessId -> Process () -> Process ())
298 delayStatic = staticLabel "$delay"