@@ -59,41 +59,22 @@ module Control.Distributed.Process
5959 , expectTimeout
6060 , spawnAsync
6161 , spawnSupervised
62+ , DidSpawn(..)
6263 ) where
6364
6465import Prelude hiding (catch)
65−import Data.Binary (decode)
6666import Data.Typeable (Typeable, typeOf)
67−import Control.Monad.Reader (ask)
68−import Control.Monad.IO.Class (MonadIO, liftIO)
6967import Control.Applicative ((<$>))
70−import Control.Exception (Exception, throw)
71−import qualified Control.Exception as Exception (catch)
72−import Control.Concurrent.MVar (modifyMVar)
73−import Control.Concurrent.Chan (writeChan)
74−import Control.Concurrent.STM
75− ( STM
76− , atomically
77− , orElse
78− , newTChan
79− , readTChan
80− , newTVar
81− , readTVar
82− , writeTVar
83− )
84−import Control.Distributed.Process.Internal.CQueue (dequeue, BlockSpec(..))
85−import Control.Distributed.Process.Serializable (Serializable, fingerprint)
86−import Data.Accessor ((^.), (^:), (^=))
68+import Control.Exception (throw)
69+import Control.Distributed.Process.Internal.MessageT (getLocalNode)
70+import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)
8771import Control.Distributed.Process.Internal.Types
8872 ( RemoteTable
8973 , NodeId(..)
9074 , ProcessId(..)
91− , LocalNode(..)
92− , LocalProcess(..)
9375 , Process(..)
9476 , Closure(..)
9577 , Static(..)
96− , Message(..)
9778 , MonitorRef(..)
9879 , MonitorNotification(..)
9980 , LinkException(..)
@@ -102,34 +83,52 @@ import Control.Distributed.Process.Internal.Types
10283 , DiedReason(..)
10384 , SpawnRef(..)
10485 , DidSpawn(..)
105− , NCMsg(..)
106− , ProcessSignal(..)
107− , monitorCounter
108− , spawnCounter
10986 , Closure(..)
11087 , SendPort(..)
11188 , ReceivePort(..)
112− , channelCounter
113− , typedChannelWithId
114− , TypedChannel(..)
115− , ChannelId(..)
116− , Identifier(..)
117− , procMsg
11889 , SerializableDict(..)
90+ , procMsg
91+ , LocalNode(..)
11992 )
120−import Control.Distributed.Process.Internal.MessageT
121− ( sendMessage
122− , sendBinary
123− , getLocalNode
124− )
125−import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)
126−import Control.Distributed.Process.Internal.Closure (resolveClosure)
127−import Control.Distributed.Process.Internal.Node (runLocalProcess)
128−import {-# SOURCE #-} Control.Distributed.Process.Internal.Closure.BuiltIn
93+import Control.Distributed.Process.Internal.Closure.BuiltIn
12994 ( linkClosure
13095 , sendClosure
13196 )
13297import Control.Distributed.Process.Internal.Closure.Combinators (cpSeq, cpBind)
98+import Control.Distributed.Process.Internal.Process.Primitives
99+ ( -- Basic messaging
100+ send
101+ , expect
102+ -- Channels
103+ , newChan
104+ , sendChan
105+ , receiveChan
106+ , mergePortsBiased
107+ , mergePortsRR
108+ -- Advanced messaging
109+ , Match
110+ , receiveWait
111+ , receiveTimeout
112+ , match
113+ , matchIf
114+ , matchUnknown
115+ -- Process management
116+ , spawn
117+ , terminate
118+ , ProcessTerminationException(..)
119+ , getSelfPid
120+ , getSelfNode
121+ -- Monitoring and linking
122+ , link
123+ , unlink
124+ , monitor
125+ , unmonitor
126+ -- Auxiliary API
127+ , catch
128+ , expectTimeout
129+ , spawnAsync
130+ )
131+import Control.Distributed.Process.Internal.Closure (resolveClosure)
133132
134133-- INTERNAL NOTES
135134--
@@ -191,134 +190,10 @@ import Control.Distributed.Process.Internal.Closure.Combinators (cpSeq, cpBind)
191190-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
192191
193192--------------------------------------------------------------------------------
194−-- Basic messaging --
195−--------------------------------------------------------------------------------
196−
197−-- | Send a message
198−send :: Serializable a => ProcessId -> a -> Process ()
199−-- This requires a lookup on every send. If we want to avoid that we need to
200−-- modify serializable to allow for stateful (IO) deserialization
201−-- Warning: if we change how 'send' is implemented, might also want to change
202−-- the implementation of 'Internal.Closure.TH.generateSender'
203−send them msg = procMsg $ sendMessage (ProcessIdentifier them) msg
204−
205−-- | Wait for a message of a specific type
206−expect :: forall a. Serializable a => Process a
207−expect = receiveWait [match return]
208−
209−--------------------------------------------------------------------------------
210−-- Channels --
211−--------------------------------------------------------------------------------
212−
213−-- | Create a new typed channel
214−newChan :: Serializable a => Process (SendPort a, ReceivePort a)
215−newChan = do
216− proc <- ask
217− liftIO . modifyMVar (processState proc) $ \st -> do
218− chan <- liftIO . atomically $ newTChan
219− let lcid = st ^. channelCounter
220− cid = ChannelId { channelProcessId = processId proc
221− , channelLocalId = lcid
222− }
223− sport = SendPort cid
224− rport = ReceivePortSingle chan
225− tch = TypedChannel chan
226− return ( (channelCounter ^: (+ 1))
227− . (typedChannelWithId lcid ^= Just tch)
228− $ st
229− , (sport, rport)
230− )
231−
232−-- | Send a message on a typed channel
233−sendChan :: Serializable a => SendPort a -> a -> Process ()
234−sendChan (SendPort them) msg = procMsg $ sendBinary (ChannelIdentifier them) msg
235−
236−-- | Wait for a message on a typed channel
237−receiveChan :: Serializable a => ReceivePort a -> Process a
238−receiveChan = liftIO . atomically . receiveSTM
239− where
240− receiveSTM :: ReceivePort a -> STM a
241− receiveSTM (ReceivePortSingle c) =
242− readTChan c
243− receiveSTM (ReceivePortBiased ps) =
244− foldr1 orElse (map receiveSTM ps)
245− receiveSTM (ReceivePortRR psVar) = do
246− ps <- readTVar psVar
247− a <- foldr1 orElse (map receiveSTM ps)
248− writeTVar psVar (rotate ps)
249− return a
250−
251− rotate :: [a] -> [a]
252− rotate [] = []
253− rotate (x:xs) = xs ++ [x]
254−
255−-- | Merge a list of typed channels.
256−--
257−-- The result port is left-biased: if there are messages available on more
258−-- than one port, the first available message is returned.
259−mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
260−mergePortsBiased = return . ReceivePortBiased
261−
262−-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
263−-- left-biased)
264−mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
265−mergePortsRR ps = liftIO . atomically $ ReceivePortRR <$> newTVar ps
266−
267−--------------------------------------------------------------------------------
268−-- Advanced messaging --
193+-- Primitives are defined in a separate module; here we only define derived --
194+-- constructs --
269195--------------------------------------------------------------------------------
270196
271−-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
272−newtype Match b = Match { unMatch :: Message -> Maybe (Process b) }
273−
274−-- | Test the matches in order against each message in the queue
275−receiveWait :: [Match b] -> Process b
276−receiveWait ms = do
277− queue <- processQueue <$> ask
278− Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
279− proc
280−
281−-- | Like 'receiveWait' but with a timeout.
282−--
283−-- If the timeout is zero do a non-blocking check for matching messages.
284−receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
285−receiveTimeout t ms = do
286− queue <- processQueue <$> ask
287− let blockSpec = if t == 0 then NonBlocking else Timeout t
288− mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
289− case mProc of
290− Nothing -> return Nothing
291− Just proc -> Just <$> proc
292−
293−-- | Match against any message of the right type
294−match :: forall a b. Serializable a => (a -> Process b) -> Match b
295−match = matchIf (const True)
296−
297−-- | Match against any message of the right type that satisfies a predicate
298−matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
299−matchIf c p = Match $ \msg ->
300− let decoded :: a
301− decoded = decode . messageEncoding $ msg in
302− if messageFingerprint msg == fingerprint (undefined :: a) && c decoded
303− then Just $ p decoded
304− else Nothing
305−
306−-- | Remove any message from the queue
307−matchUnknown :: Process b -> Match b
308−matchUnknown = Match . const . Just
309−
310−--------------------------------------------------------------------------------
311−-- Process management --
312−--------------------------------------------------------------------------------
313−
314−-- | Spawn a process
315−spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
316−spawn nid proc = do
317− ref <- spawnAsync nid proc
318− receiveWait [ matchIf (\(DidSpawn ref' _) -> ref == ref')
319− (\(DidSpawn _ pid) -> return pid)
320− ]
321−
322197-- | Run a process remotely and wait for it to reply
323198call :: SerializableDict a -> NodeId -> Closure (Process a) -> Process a
324199call sdict@SerializableDict nid proc = do
@@ -326,90 +201,6 @@ call sdict@SerializableDict nid proc = do
326201 spawn nid (proc `cpBind` sendClosure sdict us)
327202 expect
328203
329−-- | Thrown by 'terminate'
330−data ProcessTerminationException = ProcessTerminationException
331− deriving (Show, Typeable)
332−
333−instance Exception ProcessTerminationException
334−
335−-- | Terminate (throws a ProcessTerminationException)
336−terminate :: Process a
337−terminate = liftIO $ throw ProcessTerminationException
338−
339−-- | Our own process ID
340−getSelfPid :: Process ProcessId
341−getSelfPid = processId <$> ask
342−
343−-- | Get the node ID of our local node
344−getSelfNode :: Process NodeId
345−getSelfNode = localNodeId <$> procMsg getLocalNode
346−
347−--------------------------------------------------------------------------------
348−-- Monitoring and linking --
349−--------------------------------------------------------------------------------
350−
351−-- | Link to a remote process (asynchronous)
352−--
353−-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
354−link :: ProcessId -> Process ()
355−link = sendCtrlMsg Nothing . Link
356−
357−-- | Remove a link (asynchronous)
358−unlink :: ProcessId -> Process ()
359−unlink = sendCtrlMsg Nothing . Unlink
360−
361−-- | Monitor another process (asynchronous)
362−monitor :: ProcessId -> Process MonitorRef
363−monitor them = do
364− monitorRef <- getMonitorRefFor them
365− sendCtrlMsg Nothing $ Monitor them monitorRef
366− return monitorRef
367−
368−-- | Remove a monitor (asynchronous)
369−unmonitor :: MonitorRef -> Process ()
370−unmonitor = sendCtrlMsg Nothing . Unmonitor
371−
372−--------------------------------------------------------------------------------
373−-- Closures --
374−--------------------------------------------------------------------------------
375−
376−-- | Deserialize a closure
377−unClosure :: forall a. Typeable a => Closure a -> Process a
378−unClosure (Closure (Static label) env) = do
379− rtable <- remoteTable <$> procMsg getLocalNode
380− let Just dyn = resolveClosure rtable label env
381− return $ fromDyn dyn (throw (typeError dyn))
382− where
383− typeError dyn = userError $ "lookupStatic type error: "
384− ++ "cannot match " ++ show (dynTypeRep dyn)
385− ++ " against " ++ show (typeOf (undefined :: a))
386−
387−--------------------------------------------------------------------------------
388−-- Auxiliary API --
389−--------------------------------------------------------------------------------
390−
391−-- | Catch exceptions within a process
392−catch :: Exception e => Process a -> (e -> Process a) -> Process a
393−catch p h = do
394− node <- procMsg getLocalNode
395− lproc <- ask
396− let run :: Process a -> IO a
397− run proc = runLocalProcess node proc lproc
398− liftIO $ Exception.catch (run p) (run . h)
399−
400−-- | Like 'expect' but with a timeout
401−expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
402−expectTimeout timeout = receiveTimeout timeout [match return]
403−
404−-- | Asynchronous version of 'spawn'
405−--
406−-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
407−spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
408−spawnAsync nid proc = do
409− spawnRef <- getSpawnRef
410− sendCtrlMsg (Just nid) $ Spawn proc spawnRef
411− return spawnRef
412−
413204-- | Spawn a child process, have the child link to the parent and the parent
414205-- monitor the child
415206spawnSupervised :: NodeId
@@ -421,40 +212,14 @@ spawnSupervised nid proc = do
421212 ref <- monitor them
422213 return (them, ref)
423214
424−--------------------------------------------------------------------------------
425−-- Auxiliary functions --
426−--------------------------------------------------------------------------------
427−
428−getMonitorRefFor :: ProcessId -> Process MonitorRef
429−getMonitorRefFor pid = do
430− proc <- ask
431− liftIO $ modifyMVar (processState proc) $ \st -> do
432− let counter = st ^. monitorCounter
433− return ( monitorCounter ^: (+ 1) $ st
434− , MonitorRef pid counter
435− )
436−
437−getSpawnRef :: Process SpawnRef
438−getSpawnRef = do
439− proc <- ask
440− liftIO $ modifyMVar (processState proc) $ \st -> do
441− let counter = st ^. spawnCounter
442− return ( spawnCounter ^: (+ 1) $ st
443− , SpawnRef counter
444− )
215+-- | Deserialize a closure
216+unClosure :: forall a. Typeable a => Closure a -> Process a
217+unClosure (Closure (Static label) env) = do
218+ rtable <- remoteTable <$> procMsg getLocalNode
219+ let Just dyn = resolveClosure rtable label env
220+ return $ fromDyn dyn (throw (typeError dyn))
221+ where
222+ typeError dyn = userError $ "lookupStatic type error: "
223+ ++ "cannot match " ++ show (dynTypeRep dyn)
224+ ++ " against " ++ show (typeOf (undefined :: a))
445225
446−-- Send a control message
447−sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node
448− -> ProcessSignal -- ^ Message to send
449− -> Process ()
450−sendCtrlMsg mNid signal = do
451− us <- getSelfPid
452− let msg = NCMsg { ctrlMsgSender = ProcessIdentifier us
453− , ctrlMsgSignal = signal
454− }
455− case mNid of
456− Nothing -> do
457− ctrlChan <- localCtrlChan <$> procMsg getLocalNode
458− liftIO $ writeChan ctrlChan msg
459− Just nid ->
460− procMsg $ sendBinary (NodeIdentifier nid) msg