git.iko.soy/distributed-process
on fix-spawnChan help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1020 lines 40 kB history raw
1-- | Local nodes
2--
3module Control.Distributed.Process.Node
4 ( LocalNode
5 , newLocalNode
6 , closeLocalNode
7 , forkProcess
8 , runProcess
9 , initRemoteTable
10 , localNodeId
11 ) where
13-- TODO: Calls to 'sendBinary' and co (by the NC) may stall the node controller.
15#if ! MIN_VERSION_base(4,6,0)
16import Prelude hiding (catch)
17#endif
19import System.IO (fixIO, hPutStrLn, stderr)
20import System.Mem.Weak (Weak, deRefWeak)
21import qualified Data.ByteString.Lazy as BSL (fromChunks)
22import Data.Binary (decode)
23import Data.Map (Map)
24import qualified Data.Map as Map
25 ( empty
26 , toList
27 , fromList
28 , filter
29 , partitionWithKey
30 , elems
31 , filterWithKey
32 , foldlWithKey
33 )
34import Data.Set (Set)
35import qualified Data.Set as Set
36 ( empty
37 , insert
38 , delete
39 , member
40 , toList
41 )
42import Data.Foldable (forM_)
43import Data.Maybe (isJust, isNothing, catMaybes)
44import Data.Typeable (Typeable)
45import Control.Category ((>>>))
46import Control.Applicative ((<$>))
47import Control.Monad (void, when)
48import Control.Monad.IO.Class (MonadIO, liftIO)
49import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)
50import qualified Control.Monad.State.Strict as StateT (get, put)
51import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
52import Control.Exception (throwIO, SomeException, Exception, throwTo)
53import qualified Control.Exception as Exception (catch)
54import Control.Concurrent (forkIO)
55import Control.Distributed.Process.Internal.StrictMVar
56 ( newMVar
57 , withMVar
58 , modifyMVar
59 , modifyMVar_
60 , newEmptyMVar
61 , putMVar
62 , takeMVar
63 )
64import Control.Concurrent.Chan (newChan, writeChan, readChan)
65import Control.Concurrent.STM
66 ( atomically
67 )
68import Control.Distributed.Process.Internal.CQueue
69 ( CQueue
70 , enqueue
71 , newCQueue
72 , mkWeakCQueue
73 )
74import qualified Network.Transport as NT
75 ( Transport
76 , EndPoint
77 , newEndPoint
78 , receive
79 , Event(..)
80 , EventErrorCode(..)
81 , TransportError(..)
82 , address
83 , closeEndPoint
84 , ConnectionId
85 , Connection
86 , close
87 , EndPointAddress
88 , Reliability(ReliableOrdered)
89 )
90import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
91import System.Random (randomIO)
92import Control.Distributed.Static (RemoteTable, Closure)
93import qualified Control.Distributed.Static as Static
94 ( unclosure
95 , initRemoteTable
96 )
97import Control.Distributed.Process.Internal.Types
98 ( NodeId(..)
99 , LocalProcessId(..)
100 , ProcessId(..)
101 , LocalNode(..)
102 , Tracer(InactiveTracer)
103 , LocalNodeState(..)
104 , LocalProcess(..)
105 , LocalProcessState(..)
106 , Process(..)
107 , DiedReason(..)
108 , NCMsg(..)
109 , ProcessSignal(..)
110 , localPidCounter
111 , localPidUnique
112 , localProcessWithId
113 , localConnections
114 , forever'
115 , MonitorRef(..)
116 , ProcessMonitorNotification(..)
117 , NodeMonitorNotification(..)
118 , PortMonitorNotification(..)
119 , ProcessExitException(..)
120 , ProcessLinkException(..)
121 , NodeLinkException(..)
122 , PortLinkException(..)
123 , DidUnmonitor(..)
124 , DidUnlinkProcess(..)
125 , DidUnlinkNode(..)
126 , DidUnlinkPort(..)
127 , SpawnRef
128 , DidSpawn(..)
129 , Message
130 , TypedChannel(..)
131 , Identifier(..)
132 , nodeOf
133 , ProcessInfo(..)
134 , ProcessInfoNone(..)
135 , SendPortId(..)
136 , typedChannelWithId
137 , RegisterReply(..)
138 , WhereIsReply(..)
139 , messageToPayload
140 , payloadToMessage
141 , createMessage
142 , runLocalProcess
143 , firstNonReservedProcessId
144 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
145 )
146import Control.Distributed.Process.Internal.Trace
147 ( TraceArg(..)
148 , traceFormat
149 , startTracing
150 , stopTracer
151 )
152import Control.Distributed.Process.Serializable (Serializable)
153import Control.Distributed.Process.Internal.Messaging
154 ( sendBinary
155 , sendMessage
156 , sendPayload
157 , closeImplicitReconnections
158 , impliesDeathOf
159 )
160import Control.Distributed.Process.Internal.Primitives
161 ( register
162 , finally
163 , receiveWait
164 , match
165 , sendChan
166 )
167import Control.Distributed.Process.Internal.Types (SendPort)
168import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
169import Control.Distributed.Process.Internal.WeakTQueue (writeTQueue)
170import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
171 ( mapMaybe
172 , mapDefault
173 )
175--------------------------------------------------------------------------------
176-- Initialization --
177--------------------------------------------------------------------------------
179initRemoteTable :: RemoteTable
180initRemoteTable = BuiltIn.remoteTable Static.initRemoteTable
182-- | Initialize a new local node.
183newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
184newLocalNode transport rtable = do
185 mEndPoint <- NT.newEndPoint transport
186 case mEndPoint of
187 Left ex -> throwIO ex
188 Right endPoint -> do
189 localNode <- createBareLocalNode endPoint rtable
190 startServiceProcesses localNode
191 return localNode
193-- | Create a new local node (without any service processes running)
194createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
195createBareLocalNode endPoint rtable = do
196 unq <- randomIO
197 state <- newMVar LocalNodeState
198 { _localProcesses = Map.empty
199 , _localPidCounter = firstNonReservedProcessId
200 , _localPidUnique = unq
201 , _localConnections = Map.empty
202 }
203 ctrlChan <- newChan
204 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
205 , localEndPoint = endPoint
206 , localState = state
207 , localCtrlChan = ctrlChan
208 , localTracer = InactiveTracer
209 , remoteTable = rtable
210 }
211 tracedNode <- startTracing node
212 void . forkIO $ runNodeController tracedNode
213 void . forkIO $ handleIncomingMessages tracedNode
214 return tracedNode
216-- | Start and register the service processes on a node
217-- (for now, this is only the logger)
218startServiceProcesses :: LocalNode -> IO ()
219startServiceProcesses node = do
220 logger <- forkProcess node loop
221 runProcess node $ register "logger" logger
222 where
223 loop = do
224 receiveWait
225 [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do
226 liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string
227 loop
228 , match $ \((time, string) :: (String, String)) -> do
229 -- this is a 'trace' message from the local node tracer
230 liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
231 loop
232 , match $ \(ch :: SendPort ()) -> -- a shutdown request
233 sendChan ch ()
234 ]
236-- | Force-close a local node
237--
238-- TODO: for now we just close the associated endpoint
239closeLocalNode :: LocalNode -> IO ()
240closeLocalNode node =
241 NT.closeEndPoint (localEndPoint node)
243-- | Run a process on a local node and wait for it to finish
244runProcess :: LocalNode -> Process () -> IO ()
245runProcess node proc = do
246 done <- newEmptyMVar
247 void $ forkProcess node (proc `finally` liftIO (putMVar done ()))
248 takeMVar done
250-- | Spawn a new process on a local node
251forkProcess :: LocalNode -> Process () -> IO ProcessId
252forkProcess node proc = modifyMVar (localState node) startProcess
253 where
254 startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)
255 startProcess st = do
256 let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
257 , lpidUnique = st ^. localPidUnique
258 }
259 let pid = ProcessId { processNodeId = localNodeId node
260 , processLocalId = lpid
261 }
262 pst <- newMVar LocalProcessState { _monitorCounter = 0
263 , _spawnCounter = 0
264 , _channelCounter = 0
265 , _typedChannels = Map.empty
266 }
267 queue <- newCQueue
268 weakQueue <- mkWeakCQueue queue (return ())
269 (_, lproc) <- fixIO $ \ ~(tid, _) -> do
270 let lproc = LocalProcess { processQueue = queue
271 , processWeakQ = weakQueue
272 , processId = pid
273 , processState = pst
274 , processThread = tid
275 , processNode = node
276 }
277 tid' <- forkIO $ do
278 reason <- Exception.catch
279 (runLocalProcess lproc proc >> return DiedNormal)
280 (return . DiedException . (show :: SomeException -> String))
282 -- [Unified: Table 4, rules termination and exiting]
283 modifyMVar_ (localState node) (cleanupProcess pid)
284 writeChan (localCtrlChan node) NCMsg
285 { ctrlMsgSender = ProcessIdentifier pid
286 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
287 }
288 return (tid', lproc)
290 if lpidCounter lpid == maxBound
291 then do
292 newUnique <- randomIO
293 return ( (localProcessWithId lpid ^= Just lproc)
294 . (localPidCounter ^= firstNonReservedProcessId)
295 . (localPidUnique ^= newUnique)
296 $ st
297 , pid
298 )
299 else
300 return ( (localProcessWithId lpid ^= Just lproc)
301 . (localPidCounter ^: (+ 1))
302 $ st
303 , pid
304 )
306 cleanupProcess :: ProcessId -> LocalNodeState -> IO LocalNodeState
307 cleanupProcess pid st = do
308 let pid' = ProcessIdentifier pid
309 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (st ^. localConnections)
310 mapM_ (NT.close . fst) (Map.elems affected)
311 return $ (localProcessWithId (processLocalId pid) ^= Nothing)
312 . (localConnections ^= unaffected)
313 $ st
315--------------------------------------------------------------------------------
316-- Handle incoming messages --
317--------------------------------------------------------------------------------
319type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
321data IncomingTarget =
322 Uninit
323 | ToProc (Weak (CQueue Message))
324 | ToChan TypedChannel
325 | ToNode
327data ConnectionState = ConnectionState {
328 _incoming :: !(Map NT.ConnectionId IncomingConnection)
329 , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
330 }
332initConnectionState :: ConnectionState
333initConnectionState = ConnectionState {
334 _incoming = Map.empty
335 , _incomingFrom = Map.empty
336 }
338incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
339incoming = accessor _incoming (\conns st -> st { _incoming = conns })
341incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
342incomingAt cid = incoming >>> DAC.mapMaybe cid
344incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
345incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
346 where
347 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
349handleIncomingMessages :: LocalNode -> IO ()
350handleIncomingMessages node = go initConnectionState
351 where
352 go :: ConnectionState -> IO ()
353 go !st = do
354 event <- NT.receive endpoint
355 case event of
356 NT.ConnectionOpened cid rel theirAddr ->
357 if rel == NT.ReliableOrdered
358 then go ( (incomingAt cid ^= Just (theirAddr, Uninit))
359 . (incomingFrom theirAddr ^: Set.insert cid)
360 $ st
361 )
362 else invalidRequest cid st
363 NT.Received cid payload ->
364 case st ^. incomingAt cid of
365 Just (_, ToProc weakQueue) -> do
366 mQueue <- deRefWeak weakQueue
367 forM_ mQueue $ \queue -> do
368 -- TODO: if we find that the queue is Nothing, should we remove
369 -- it from the NC state? (and same for channels, below)
370 let msg = payloadToMessage payload
371 enqueue queue msg -- 'enqueue' is strict
372 go st
373 Just (_, ToChan (TypedChannel chan')) -> do
374 mChan <- deRefWeak chan'
375 -- If mChan is Nothing, the process has given up the read end of
376 -- the channel and we simply ignore the incoming message
377 forM_ mChan $ \chan -> atomically $
378 -- We make sure the message is fully decoded when it is enqueued
379 writeTQueue chan $! decode (BSL.fromChunks payload)
380 go st
381 Just (_, ToNode) -> do
382 let ctrlMsg = decode . BSL.fromChunks $ payload
383 writeChan ctrlChan $! ctrlMsg
384 go st
385 Just (src, Uninit) ->
386 case decode (BSL.fromChunks payload) of
387 ProcessIdentifier pid -> do
388 let lpid = processLocalId pid
389 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
390 case mProc of
391 Just proc ->
392 go (incomingAt cid ^= Just (src, ToProc (processWeakQ proc)) $ st)
393 Nothing ->
394 invalidRequest cid st
395 SendPortIdentifier chId -> do
396 let lcid = sendPortLocalId chId
397 lpid = processLocalId (sendPortProcessId chId)
398 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
399 case mProc of
400 Just proc -> do
401 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
402 case mChannel of
403 Just channel ->
404 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
405 Nothing ->
406 invalidRequest cid st
407 Nothing ->
408 invalidRequest cid st
409 NodeIdentifier nid ->
410 if nid == localNodeId node
411 then go (incomingAt cid ^= Just (src, ToNode) $ st)
412 else invalidRequest cid st
413 Nothing ->
414 invalidRequest cid st
415 NT.ConnectionClosed cid ->
416 case st ^. incomingAt cid of
417 Nothing ->
418 invalidRequest cid st
419 Just (src, _) ->
420 go ( (incomingAt cid ^= Nothing)
421 . (incomingFrom src ^: Set.delete cid)
422 $ st
423 )
424 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
425 -- [Unified table 9, rule node_disconnect]
426 let nid = NodeIdentifier $ NodeId theirAddr
427 writeChan ctrlChan NCMsg
428 { ctrlMsgSender = nid
429 , ctrlMsgSignal = Died nid DiedDisconnect
430 }
431 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
432 closeImplicitReconnections node nid
433 go ( (incomingFrom theirAddr ^= Set.empty)
434 . (incoming ^: Map.filterWithKey (const . notLost))
435 $ st
436 )
437 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
438 fatal $ "Cloud Haskell fatal error: end point failed: " ++ str
439 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
440 fatal $ "Cloud Haskell fatal error: transport failed: " ++ str
441 NT.EndPointClosed ->
442 stopTracer (localTracer node) >> return ()
443 NT.ReceivedMulticast _ _ ->
444 -- If we received a multicast message, something went horribly wrong
445 -- and we just give up
446 fatal "Cloud Haskell fatal error: received unexpected multicast"
448 fatal :: String -> IO ()
449 fatal msg = stopTracer (localTracer node) >> fail msg
451 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
452 invalidRequest cid st = do
453 -- TODO: We should treat this as a fatal error on the part of the remote
454 -- node. That is, we should report the remote node as having died, and we
455 -- should close incoming connections (this requires a Transport layer
456 -- extension).
457 traceEventFmtIO node "" [(TraceStr "[network] invalid request: "),
458 (Trace cid)]
459 go ( incomingAt cid ^= Nothing
460 $ st
461 )
463 state = localState node
464 endpoint = localEndPoint node
465 ctrlChan = localCtrlChan node
467--------------------------------------------------------------------------------
468-- Top-level access to the node controller --
469--------------------------------------------------------------------------------
471runNodeController :: LocalNode -> IO ()
472runNodeController =
473 runReaderT (evalStateT (unNC nodeController) initNCState)
475--------------------------------------------------------------------------------
476-- Internal data types --
477--------------------------------------------------------------------------------
479data NCState = NCState
480 { -- Mapping from remote processes to linked local processes
481 _links :: !(Map Identifier (Set ProcessId))
482 -- Mapping from remote processes to monitoring local processes
483 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
484 -- Process registry: names and where they live, mapped to the PIDs
485 , _registeredHere :: !(Map String ProcessId)
486 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
487 }
489newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
490 deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)
492initNCState :: NCState
493initNCState = NCState { _links = Map.empty
494 , _monitors = Map.empty
495 , _registeredHere = Map.empty
496 , _registeredOnNodes = Map.empty
497 }
499-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
500-- type is deliberately not exported so it cannot be caught explicitly.
501data ProcessKillException =
502 ProcessKillException !ProcessId !String
503 deriving (Typeable)
505instance Exception ProcessKillException
506instance Show ProcessKillException where
507 show (ProcessKillException pid reason) =
508 "killed-by=" ++ show pid ++ ",reason=" ++ reason
510--------------------------------------------------------------------------------
511-- Tracing/Debugging --
512--------------------------------------------------------------------------------
514-- [Issue #104]
516traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
517traceNotifyDied node ident reason =
518 case reason of
519 DiedNormal -> return ()
520 _ -> traceNcEventFmt node " "
521 [(TraceStr "[node-controller]"),
522 (Trace ident),
523 (Trace reason)]
525traceNcEventFmt :: LocalNode -> String -> [TraceArg] -> NC ()
526traceNcEventFmt node fmt args =
527 liftIO $ traceEventFmtIO node fmt args
529traceEventFmtIO :: LocalNode
530 -> String
531 -> [TraceArg]
532 -> IO ()
533traceEventFmtIO node fmt args =
534 withLocalTracer node $ \t -> traceFormat t fmt args
536withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()
537withLocalTracer node act = act (localTracer node)
539--------------------------------------------------------------------------------
540-- Core functionality --
541--------------------------------------------------------------------------------
543-- [Unified: Table 7]
544nodeController :: NC ()
545nodeController = do
546 node <- ask
547 forever' $ do
548 msg <- liftIO $ readChan (localCtrlChan node)
550 -- [Unified: Table 7, rule nc_forward]
551 case destNid (ctrlMsgSignal msg) of
552 Just nid' | nid' /= localNodeId node ->
553 liftIO $ sendBinary node
554 (ctrlMsgSender msg)
555 (NodeIdentifier nid')
556 WithImplicitReconnect
557 msg
558 _ ->
559 return ()
561 case msg of
562 NCMsg (ProcessIdentifier from) (Link them) ->
563 ncEffectMonitor from them Nothing
564 NCMsg (ProcessIdentifier from) (Monitor ref) ->
565 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
566 NCMsg (ProcessIdentifier from) (Unlink them) ->
567 ncEffectUnlink from them
568 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
569 ncEffectUnmonitor from ref
570 NCMsg _from (Died ident reason) ->
571 ncEffectDied ident reason
572 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
573 ncEffectSpawn from proc ref
574 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
575 ncEffectRegister from label atnode pid force
576 NCMsg (ProcessIdentifier from) (WhereIs label) ->
577 ncEffectWhereIs from label
578 NCMsg from (NamedSend label msg') ->
579 ncEffectNamedSend from label msg'
580 NCMsg (ProcessIdentifier from) (Kill to reason) ->
581 ncEffectKill from to reason
582 NCMsg (ProcessIdentifier from) (Exit to reason) ->
583 ncEffectExit from to reason
584 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
585 ncEffectGetInfo from pid
586 unexpected ->
587 error $ "nodeController: unexpected message " ++ show unexpected
589-- [Unified: Table 10]
590ncEffectMonitor :: ProcessId -- ^ Who's watching?
591 -> Identifier -- ^ Who's being watched?
592 -> Maybe MonitorRef -- ^ 'Nothing' to link
593 -> NC ()
594ncEffectMonitor from them mRef = do
595 node <- ask
596 shouldLink <-
597 if not (isLocal node them)
598 then return True
599 else isValidLocalIdentifier them
600 case (shouldLink, isLocal node (ProcessIdentifier from)) of
601 (True, _) -> -- [Unified: first rule]
602 case mRef of
603 Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
604 Nothing -> modify' $ linksFor them ^: Set.insert from
605 (False, True) -> -- [Unified: second rule]
606 notifyDied from them DiedUnknownId mRef
607 (False, False) -> -- [Unified: third rule]
608 -- TODO: this is the right sender according to the Unified semantics,
609 -- but perhaps having 'them' as the sender would make more sense
610 -- (see also: notifyDied)
611 liftIO $ sendBinary node
612 (NodeIdentifier $ localNodeId node)
613 (NodeIdentifier $ processNodeId from)
614 WithImplicitReconnect
615 NCMsg
616 { ctrlMsgSender = NodeIdentifier (localNodeId node)
617 , ctrlMsgSignal = Died them DiedUnknownId
618 }
620-- [Unified: Table 11]
621ncEffectUnlink :: ProcessId -> Identifier -> NC ()
622ncEffectUnlink from them = do
623 node <- ask
624 when (isLocal node (ProcessIdentifier from)) $
625 case them of
626 ProcessIdentifier pid ->
627 postAsMessage from $ DidUnlinkProcess pid
628 NodeIdentifier nid ->
629 postAsMessage from $ DidUnlinkNode nid
630 SendPortIdentifier cid ->
631 postAsMessage from $ DidUnlinkPort cid
632 modify' $ linksFor them ^: Set.delete from
634-- [Unified: Table 11]
635ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
636ncEffectUnmonitor from ref = do
637 node <- ask
638 when (isLocal node (ProcessIdentifier from)) $
639 postAsMessage from $ DidUnmonitor ref
640 modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
642-- [Unified: Table 12]
643ncEffectDied :: Identifier -> DiedReason -> NC ()
644ncEffectDied ident reason = do
645 node <- ask
646 traceNotifyDied node ident reason
647 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
648 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
650-- _registry :: !(Map (String,NodeId) ProcessId)
652 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
654 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
655 forM_ uss $ \us ->
656 when (localOnly <= isLocal node (ProcessIdentifier us)) $
657 notifyDied us them reason Nothing
659 forM_ (Map.toList affectedMons) $ \(them, refs) ->
660 forM_ refs $ \(us, ref) ->
661 when (localOnly <= isLocal node (ProcessIdentifier us)) $
662 notifyDied us them reason (Just ref)
664 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
666 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
668 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
669 mapM (\(pid,nidlist) ->
670 case ident `impliesDeathOf` ProcessIdentifier pid of
671 True ->
672 do forM_ nidlist $ \(nid,_) ->
673 when (not $ isLocal node (NodeIdentifier nid))
674 (forwardNameDeath node nid)
675 return Nothing
676 False -> return $ Just (pid,nidlist) )
677 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
678 where
679 forwardNameDeath node nid =
680 liftIO $ sendBinary node
681 (NodeIdentifier $ localNodeId node)
682 (NodeIdentifier $ nid)
683 WithImplicitReconnect
684 NCMsg
685 { ctrlMsgSender = NodeIdentifier (localNodeId node)
686 , ctrlMsgSignal = Died ident reason
687 }
689-- [Unified: Table 13]
690ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
691ncEffectSpawn pid cProc ref = do
692 mProc <- unClosure cProc
693 -- If the closure does not exist, we spawn a process that throws an exception
694 -- This allows the remote node to find out what's happening
695 let proc = case mProc of
696 Left err -> fail $ "Error: Could not resolve closure: " ++ err
697 Right p -> p
698 node <- ask
699 pid' <- liftIO $ forkProcess node proc
700 liftIO $ sendMessage node
701 (NodeIdentifier (localNodeId node))
702 (ProcessIdentifier pid)
703 WithImplicitReconnect
704 (DidSpawn ref pid')
706-- Unified semantics does not explicitly describe how to implement 'register',
707-- but mentions it's "very similar to nsend" (Table 14)
708-- We send a response indicated if the operation is invalid
709ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
710ncEffectRegister from label atnode mPid reregistration = do
711 node <- ask
712 currentVal <- gets (^. registeredHereFor label)
713 isOk <-
714 case mPid of
715 Nothing -> -- unregister request
716 return $ isJust currentVal
717 Just thepid -> -- register request
718 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
719 return $ (isNothing currentVal /= reregistration) &&
720 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
721 if isLocal node (NodeIdentifier atnode)
722 then do when (isOk) $
723 do modify' $ registeredHereFor label ^= mPid
724 updateRemote node currentVal mPid
725 liftIO $ sendMessage node
726 (NodeIdentifier (localNodeId node))
727 (ProcessIdentifier from)
728 WithImplicitReconnect
729 (RegisterReply label isOk)
730 else let operation =
731 case reregistration of
732 True -> flip decList
733 False -> flip incList
734 in case mPid of
735 Nothing -> return ()
736 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
737 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
738 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
739 forward node (processNodeId newval) (Register label atnode (Just newval) False)
740 updateRemote node Nothing (Just newval) =
741 forward node (processNodeId newval) (Register label atnode (Just newval) False)
742 updateRemote node (Just oldval) Nothing =
743 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
744 updateRemote _ _ _ = return ()
745 maybeify f Nothing = unmaybeify $ f []
746 maybeify f (Just x) = unmaybeify $ f x
747 unmaybeify [] = Nothing
748 unmaybeify x = Just x
749 incList [] tag = [(tag,1)]
750 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
751 incList (x:xs) tag = x : incList xs tag
752 decList [] _ = []
753 decList ((atag,1):xs) tag | atag == tag = xs
754 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
755 decList (x:xs) tag = x:decList xs tag
756 forward node to reg =
757 when (not $ isLocal node (NodeIdentifier to)) $
758 liftIO $ sendBinary node
759 (ProcessIdentifier from)
760 (NodeIdentifier to)
761 WithImplicitReconnect
762 NCMsg
763 { ctrlMsgSender = ProcessIdentifier from
764 , ctrlMsgSignal = reg
765 }
768-- Unified semantics does not explicitly describe 'whereis'
769ncEffectWhereIs :: ProcessId -> String -> NC ()
770ncEffectWhereIs from label = do
771 node <- ask
772 mPid <- gets (^. registeredHereFor label)
773 liftIO $ sendMessage node
774 (NodeIdentifier (localNodeId node))
775 (ProcessIdentifier from)
776 WithImplicitReconnect
777 (WhereIsReply label mPid)
779-- [Unified: Table 14]
780ncEffectNamedSend :: Identifier -> String -> Message -> NC ()
781ncEffectNamedSend from label msg = do
782 node <- ask
783 mPid <- gets (^. registeredHereFor label)
784 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
785 forM_ mPid $ \pid ->
786 liftIO $ sendPayload node
787 from
788 (ProcessIdentifier pid)
789 NoImplicitReconnect
790 (messageToPayload msg)
792-- [Issue #69]
793ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
794ncEffectKill from to reason = do
795 node <- ask
796 when (isLocal node (ProcessIdentifier to)) $
797 throwException to $ ProcessKillException from reason
799-- [Issue #69]
800ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
801ncEffectExit from to reason = do
802 node <- ask
803 when (isLocal node (ProcessIdentifier to)) $
804 throwException to $ ProcessExitException from reason
806-- [Issue #89]
807ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
808ncEffectGetInfo from pid =
809 let lpid = processLocalId pid
810 them = (ProcessIdentifier pid)
811 in do
812 node <- ask
813 mProc <- liftIO $
814 withMVar (localState node) $ return . (^. localProcessWithId lpid)
815 case mProc of
816 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
817 from node (ProcessInfoNone DiedUnknownId)
818 Just _ -> do
819 itsLinks <- gets (^. linksFor them)
820 itsMons <- gets (^. monitorsFor them)
821 registered <- gets (^. registeredHere)
823 let reg = registeredNames registered
824 dispatch (isLocal node (ProcessIdentifier from))
825 from
826 node
827 ProcessInfo {
828 infoNode = (processNodeId pid)
829 , infoRegisteredNames = reg
830 -- we cannot populate this field
831 , infoMessageQueueLength = Nothing
832 , infoMonitors = Set.toList itsMons
833 , infoLinks = Set.toList itsLinks
834 }
835 where dispatch :: (Serializable a, Show a)
836 => Bool
837 -> ProcessId
838 -> LocalNode
839 -> a
840 -> NC ()
841 dispatch True dest _ pInfo = postAsMessage dest $ pInfo
842 dispatch False dest node pInfo = do
843 liftIO $ sendMessage node
844 (NodeIdentifier (localNodeId node))
845 (ProcessIdentifier dest)
846 WithImplicitReconnect
847 pInfo
849 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
850 then (k:ks)
851 else ks) []
853--------------------------------------------------------------------------------
854-- Auxiliary --
855--------------------------------------------------------------------------------
857notifyDied :: ProcessId -- ^ Who to notify?
858 -> Identifier -- ^ Who died?
859 -> DiedReason -- ^ How did they die?
860 -> Maybe MonitorRef -- ^ 'Nothing' for linking
861 -> NC ()
862notifyDied dest src reason mRef = do
863 node <- ask
864 case (isLocal node (ProcessIdentifier dest), mRef, src) of
865 (True, Just ref, ProcessIdentifier pid) ->
866 postAsMessage dest $ ProcessMonitorNotification ref pid reason
867 (True, Just ref, NodeIdentifier nid) ->
868 postAsMessage dest $ NodeMonitorNotification ref nid reason
869 (True, Just ref, SendPortIdentifier cid) ->
870 postAsMessage dest $ PortMonitorNotification ref cid reason
871 (True, Nothing, ProcessIdentifier pid) ->
872 throwException dest $ ProcessLinkException pid reason
873 (True, Nothing, NodeIdentifier pid) ->
874 throwException dest $ NodeLinkException pid reason
875 (True, Nothing, SendPortIdentifier pid) ->
876 throwException dest $ PortLinkException pid reason
877 (False, _, _) ->
878 -- The change in sender comes from [Unified: Table 10]
879 liftIO $ sendBinary node
880 (NodeIdentifier $ localNodeId node)
881 (NodeIdentifier $ processNodeId dest)
882 WithImplicitReconnect
883 NCMsg
884 { ctrlMsgSender = NodeIdentifier (localNodeId node)
885 , ctrlMsgSignal = Died src reason
886 }
888-- | [Unified: Table 8]
889destNid :: ProcessSignal -> Maybe NodeId
890destNid (Link ident) = Just $ nodeOf ident
891destNid (Unlink ident) = Just $ nodeOf ident
892destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
893destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
894destNid (Spawn _ _) = Nothing
895destNid (Register _ _ _ _) = Nothing
896destNid (WhereIs _) = Nothing
897destNid (NamedSend _ _) = Nothing
898-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
899-- then when a local process dies the monitoring/linking machinery will take
900-- care of notifying remote nodes
901destNid (Died _ _) = Nothing
902destNid (Kill pid _) = Just $ processNodeId pid
903destNid (Exit pid _) = Just $ processNodeId pid
904destNid (GetInfo pid) = Just $ processNodeId pid
906-- | Check if a process is local to our own node
907isLocal :: LocalNode -> Identifier -> Bool
908isLocal nid ident = nodeOf ident == localNodeId nid
910-- | Lookup a local closure
911unClosure :: Typeable a => Closure a -> NC (Either String a)
912unClosure closure = do
913 rtable <- remoteTable <$> ask
914 return (Static.unclosure rtable closure)
916-- | Check if an identifier refers to a valid local object
917isValidLocalIdentifier :: Identifier -> NC Bool
918isValidLocalIdentifier ident = do
919 node <- ask
920 liftIO . withMVar (localState node) $ \nSt ->
921 case ident of
922 NodeIdentifier nid ->
923 return $ nid == localNodeId node
924 ProcessIdentifier pid -> do
925 let mProc = nSt ^. localProcessWithId (processLocalId pid)
926 return $ isJust mProc
927 SendPortIdentifier cid -> do
928 let pid = sendPortProcessId cid
929 mProc = nSt ^. localProcessWithId (processLocalId pid)
930 case mProc of
931 Nothing -> return False
932 Just proc -> withMVar (processState proc) $ \pSt -> do
933 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
934 return $ isJust mCh
936--------------------------------------------------------------------------------
937-- Messages to local processes --
938--------------------------------------------------------------------------------
940postAsMessage :: Serializable a => ProcessId -> a -> NC ()
941postAsMessage pid = postMessage pid . createMessage
943postMessage :: ProcessId -> Message -> NC ()
944postMessage pid msg = do
945 withLocalProc pid $ \p -> enqueue (processQueue p) msg
947throwException :: Exception e => ProcessId -> e -> NC ()
948throwException pid e = withLocalProc pid $ \p ->
949 throwTo (processThread p) e
951withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()
952withLocalProc pid p = do
953 node <- ask
954 liftIO $ do
955 -- By [Unified: table 6, rule missing_process] messages to dead processes
956 -- can silently be dropped
957 let lpid = processLocalId pid
958 mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
959 forM_ mProc p
961--------------------------------------------------------------------------------
962-- Accessors --
963--------------------------------------------------------------------------------
965links :: Accessor NCState (Map Identifier (Set ProcessId))
966links = accessor _links (\ls st -> st { _links = ls })
968monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
969monitors = accessor _monitors (\ms st -> st { _monitors = ms })
971registeredHere :: Accessor NCState (Map String ProcessId)
972registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
974registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
975registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
977linksFor :: Identifier -> Accessor NCState (Set ProcessId)
978linksFor ident = links >>> DAC.mapDefault Set.empty ident
980monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
981monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
983registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
984registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
986registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
987registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
989-- | @splitNotif ident@ splits a notifications map into those
990-- notifications that should trigger when 'ident' fails and those links that
991-- should not.
992--
993-- There is a hierarchy between identifiers: failure of a node implies failure
994-- of all processes on that node, and failure of a process implies failure of
995-- all typed channels to that process. In other words, if 'ident' refers to a
996-- node, then the /should trigger/ set will include
997--
998-- * the notifications for the node specifically
999-- * the notifications for processes on that node, and
1000-- * the notifications for typed channels to processes on that node.
1002-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1003-- include
1005-- * the notifications for that process specifically and
1006-- * the notifications for typed channels to that process.
1008-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1009splitNotif :: Identifier
1010 -> Map Identifier a
1011 -> (Map Identifier a, Map Identifier a)
1012splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1014--------------------------------------------------------------------------------
1015-- Auxiliary --
1016--------------------------------------------------------------------------------
1018-- | Modify and evaluate the state
1019modify' :: MonadState s m => (s -> s) -> m ()
1020modify' f = StateT.get >>= \s -> StateT.put $! f s