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