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