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