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