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