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