git.iko.soy/distributed-process
on f575a46 help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1214 lines 48 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE DeriveDataTypeable #-}
3{-# LANGUAGE ScopedTypeVariables #-}
4{-# LANGUAGE RankNTypes #-}
5{-# LANGUAGE BangPatterns #-}
6{-# LANGUAGE GeneralizedNewtypeDeriving #-}
7{-# LANGUAGE RecordWildCards #-}
9-- | Local nodes
10--
11module Control.Distributed.Process.Node
12 ( LocalNode
13 , newLocalNode
14 , closeLocalNode
15 , forkProcess
16 , runProcess
17 , initRemoteTable
18 , localNodeId
19 ) where
21-- TODO: Calls to 'sendBinary' and co (by the NC) may stall the node controller.
23#if ! MIN_VERSION_base(4,6,0)
24import Prelude hiding (catch)
25#endif
27import System.IO (fixIO, hPutStrLn, stderr)
28import System.Mem.Weak (Weak, deRefWeak)
29import qualified Data.ByteString.Lazy as BSL (fromChunks)
30import Data.Binary (decode)
31import Data.Map (Map)
32import qualified Data.Map as Map
33 ( empty
34 , toList
35 , fromList
36 , filter
37 , partitionWithKey
38 , elems
39 , size
40 , filterWithKey
41 , foldlWithKey
42 )
43import Data.Set (Set)
44import qualified Data.Set as Set
45 ( empty
46 , insert
47 , delete
48 , member
49 , toList
50 )
51import Data.Foldable (forM_)
52import Data.Maybe (isJust, fromJust, isNothing, catMaybes)
53import Data.Typeable (Typeable)
54import Control.Category ((>>>))
55import Control.Applicative (Applicative, (<$>))
56import Control.Monad (void, when)
57import Control.Monad.IO.Class (MonadIO, liftIO)
58import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)
59import qualified Control.Monad.State.Strict as StateT (get, put)
60import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
61import Control.Exception
62 ( throwIO
63 , AsyncException(ThreadKilled)
64 , SomeException
65 , Exception
66 , throwTo
67 , uninterruptibleMask_
68 )
69import qualified Control.Exception as Exception (Handler(..), catches, finally)
70import Control.Concurrent (forkIO, forkIOWithUnmask, killThread, myThreadId)
71import Control.Distributed.Process.Internal.StrictMVar
72 ( newMVar
73 , withMVar
74 , modifyMVarMasked
75 , modifyMVar_
76 , newEmptyMVar
77 , putMVar
78 , takeMVar
79 )
80import Control.Concurrent.Chan (newChan, writeChan, readChan)
81import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
82import Control.Concurrent.STM
83 ( atomically
84 )
85import Control.Distributed.Process.Internal.CQueue
86 ( CQueue
87 , enqueue
88 , newCQueue
89 , mkWeakCQueue
90 , queueSize
91 )
92import qualified Network.Transport as NT
93 ( Transport
94 , EndPoint
95 , newEndPoint
96 , receive
97 , Event(..)
98 , EventErrorCode(..)
99 , TransportError(..)
100 , address
101 , closeEndPoint
102 , ConnectionId
103 , close
104 , EndPointAddress
105 , Reliability(ReliableOrdered)
106 )
107import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
108import System.Random (randomIO)
109import Control.Distributed.Static (RemoteTable, Closure)
110import qualified Control.Distributed.Static as Static
111 ( unclosure
112 , initRemoteTable
113 )
114import Control.Distributed.Process.Internal.Types
115 ( NodeId(..)
116 , LocalProcessId(..)
117 , ProcessId(..)
118 , LocalNode(..)
119 , MxEventBus(..)
120 , LocalNodeState(..)
121 , ValidLocalNodeState(..)
122 , withValidLocalState
123 , modifyValidLocalState_
124 , LocalProcess(..)
125 , LocalProcessState(..)
126 , Process(..)
127 , DiedReason(..)
128 , NCMsg(..)
129 , ProcessSignal(..)
130 , localPidCounter
131 , localPidUnique
132 , localProcessWithId
133 , localProcesses
134 , localConnections
135 , forever'
136 , MonitorRef(..)
137 , ProcessMonitorNotification(..)
138 , NodeMonitorNotification(..)
139 , PortMonitorNotification(..)
140 , ProcessExitException(..)
141 , ProcessLinkException(..)
142 , NodeLinkException(..)
143 , PortLinkException(..)
144 , DidUnmonitor(..)
145 , DidUnlinkProcess(..)
146 , DidUnlinkNode(..)
147 , DidUnlinkPort(..)
148 , SpawnRef
149 , DidSpawn(..)
150 , Message(..)
151 , TypedChannel(..)
152 , Identifier(..)
153 , nodeOf
154 , ProcessInfo(..)
155 , ProcessInfoNone(..)
156 , NodeStats(..)
157 , SendPortId(..)
158 , typedChannelWithId
159 , RegisterReply(..)
160 , WhereIsReply(..)
161 , payloadToMessage
162 , createUnencodedMessage
163 , runLocalProcess
164 , firstNonReservedProcessId
165 , ImplicitReconnect(WithImplicitReconnect)
166 )
167import Control.Distributed.Process.Management.Internal.Agent
168 ( mxAgentController
169 )
170import qualified Control.Distributed.Process.Management.Internal.Table as Table
171 ( mxTableCoordinator
172 , startTableCoordinator
173 )
174import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
175 ( remoteTable
176 )
177import Control.Distributed.Process.Management.Internal.Trace.Tracer
178 ( defaultTracer
179 )
180import Control.Distributed.Process.Management.Internal.Trace.Types
181 ( TraceArg(..)
182 , traceEvent
183 , traceLogFmt
184 , enableTrace
185 )
186import Control.Distributed.Process.Management.Internal.Types
187 ( MxEvent(..)
188 )
189import Control.Distributed.Process.Serializable (Serializable)
190import Control.Distributed.Process.Internal.Messaging
191 ( sendBinary
192 , sendMessage
193 , closeImplicitReconnections
194 , impliesDeathOf
195 )
196import Control.Distributed.Process.Internal.Primitives
197 ( register
198 , finally
199 , receiveWait
200 , match
201 , sendChan
202 , catch
203 , unwrapMessage
204 )
205import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))
206import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
207import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
208import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
209 ( mapMaybe
210 , mapDefault
211 )
212import Unsafe.Coerce
214--------------------------------------------------------------------------------
215-- Initialization --
216--------------------------------------------------------------------------------
218initRemoteTable :: RemoteTable
219initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable
221-- | Initialize a new local node.
222newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
223newLocalNode transport rtable = do
224 mEndPoint <- NT.newEndPoint transport
225 case mEndPoint of
226 Left ex -> throwIO ex
227 Right endPoint -> do
228 localNode <- createBareLocalNode endPoint rtable
229 startServiceProcesses localNode
230 return localNode
232-- | Create a new local node (without any service processes running)
233createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
234createBareLocalNode endPoint rtable = do
235 unq <- randomIO
236 state <- newMVar $ LocalNodeValid $ ValidLocalNodeState
237 { _localProcesses = Map.empty
238 , _localPidCounter = firstNonReservedProcessId
239 , _localPidUnique = unq
240 , _localConnections = Map.empty
241 }
242 ctrlChan <- newChan
243 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
244 , localEndPoint = endPoint
245 , localState = state
246 , localCtrlChan = ctrlChan
247 , localEventBus = MxEventBusInitialising
248 , remoteTable = rtable
249 }
250 tracedNode <- startMxAgent node
252 -- Once the NC terminates, the endpoint isn't much use,
253 void $ forkIO $ Exception.finally (runNodeController tracedNode)
254 (NT.closeEndPoint (localEndPoint node))
256 -- whilst a closed/failing endpoint will terminate the NC
257 void $ forkIO $ Exception.finally (handleIncomingMessages tracedNode)
258 (stopNC node)
260 return tracedNode
261 where
262 stopNC node =
263 writeChan (localCtrlChan node) NCMsg
264 { ctrlMsgSender = NodeIdentifier (localNodeId node)
265 , ctrlMsgSignal = SigShutdown
266 }
268startMxAgent :: LocalNode -> IO LocalNode
269startMxAgent node = do
270 -- see note [tracer/forkProcess races]
271 let fork = forkProcess node
272 mv <- MVar.newEmptyMVar
273 pid <- fork $ mxAgentController fork mv
274 (tracer', wqRef, mxNew') <- MVar.takeMVar mv
275 return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }
277startDefaultTracer :: LocalNode -> IO ()
278startDefaultTracer node' = do
279 let t = localEventBus node'
280 case t of
281 MxEventBus _ (Tracer pid _) _ _ -> do
282 runProcess node' $ register "trace.controller" pid
283 pid' <- forkProcess node' defaultTracer
284 enableTrace (localEventBus node') pid'
285 runProcess node' $ register "tracer.initial" pid'
286 _ -> return ()
288-- TODO: we need a better mechanism for defining and registering services
290-- | Start and register the service processes on a node
291startServiceProcesses :: LocalNode -> IO ()
292startServiceProcesses node = do
293 -- tracing /spawns/ relies on the tracer being enabled, but we start
294 -- the default tracer first, even though it might @nsend@ to the logger
295 -- before /that/ process has started - this is a totally harmless race
296 -- however, so we deliberably ignore it
297 startDefaultTracer node
298 tableCoordinatorPid <- fork $ Table.startTableCoordinator fork
299 runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid
300 logger <- forkProcess node loop
301 runProcess node $ do
302 register "logger" logger
303 -- The trace.logger is used for tracing to the console to avoid feedback
304 -- loops during tracing if the user reregisters the "logger" with a custom
305 -- process which uses 'send' or other primitives which are traced.
306 register "trace.logger" logger
307 where
308 fork = forkProcess node
310 loop = do
311 receiveWait
312 [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do
313 liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string
314 loop
315 , match $ \((time, string) :: (String, String)) -> do
316 -- this is a 'trace' message from the local node tracer
317 liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
318 loop
319 , match $ \(ch :: SendPort ()) -> -- a shutdown request
320 sendChan ch ()
321 ]
323-- | Force-close a local node, killing all processes on that node.
324closeLocalNode :: LocalNode -> IO ()
325closeLocalNode node = do
326 modifyMVar_ (localState node) $ \st -> case st of
327 LocalNodeValid vst -> do
328 forM_ (vst ^. localProcesses) $ \lproc ->
329 -- Semantics of 'throwTo' guarantee that target thread will get
330 -- delivered an exception. Therefore, target thread will be killed
331 -- eventually and that's as good as we can do. No need to wait for
332 -- thread to actually finish dying.
333 killThread (processThread lproc)
334 return LocalNodeClosed
335 LocalNodeClosed -> return LocalNodeClosed
336 -- This call will have the effect of shutting down the NC as well (see
337 -- 'createBareLocalNode').
338 NT.closeEndPoint (localEndPoint node)
340-- | Run a process on a local node and wait for it to finish
341runProcess :: LocalNode -> Process () -> IO ()
342runProcess node proc = do
343 done <- newEmptyMVar
344 tid <- myThreadId
345 void $ forkProcess node $ do
346 catch (proc `finally` liftIO (putMVar done ()))
347 (\(ex :: SomeException) -> liftIO $ throwTo tid ex)
348 takeMVar done
350-- | Spawn a new process on a local node
351forkProcess :: LocalNode -> Process () -> IO ProcessId
352forkProcess node proc =
353 modifyMVarMasked (localState node) startProcess
354 where
355 startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)
356 startProcess (LocalNodeValid vst) = do
357 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
358 , lpidUnique = vst ^. localPidUnique
359 }
360 let pid = ProcessId { processNodeId = localNodeId node
361 , processLocalId = lpid
362 }
363 pst <- newMVar LocalProcessState { _monitorCounter = 0
364 , _spawnCounter = 0
365 , _channelCounter = 0
366 , _typedChannels = Map.empty
367 }
368 queue <- newCQueue
369 weakQueue <- mkWeakCQueue queue (return ())
370 (_, lproc) <- fixIO $ \ ~(tid, _) -> do
371 let lproc = LocalProcess { processQueue = queue
372 , processWeakQ = weakQueue
373 , processId = pid
374 , processState = pst
375 , processThread = tid
376 , processNode = node
377 }
378 tid' <- uninterruptibleMask_ $ forkIOWithUnmask $ \unmask -> do
379 reason <- Exception.catches
380 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
381 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
382 mMsg <- unwrapMessage msg :: IO (Maybe String)
383 case mMsg of
384 Nothing -> return $ DiedException $ show ex
385 Just m -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))
386 , (Exception.Handler
387 (return . DiedException . (show :: SomeException -> String)))]
389 -- [Unified: Table 4, rules termination and exiting]
390 modifyValidLocalState_ node (cleanupProcess pid)
391 writeChan (localCtrlChan node) NCMsg
392 { ctrlMsgSender = ProcessIdentifier pid
393 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
394 }
395 return (tid', lproc)
397 -- see note [tracer/forkProcess races]
398 trace node (MxSpawned pid)
400 if lpidCounter lpid == maxBound
401 then do
402 -- TODO: this doesn't look right at all - how do we know
403 -- that newUnique represents a process id that is available!?
404 newUnique <- randomIO
405 return ( LocalNodeValid
406 $ (localProcessWithId lpid ^= Just lproc)
407 . (localPidCounter ^= firstNonReservedProcessId)
408 . (localPidUnique ^= newUnique)
409 $ vst
410 , pid
411 )
412 else
413 return ( LocalNodeValid
414 $ (localProcessWithId lpid ^= Just lproc)
415 . (localPidCounter ^: (+ 1))
416 $ vst
417 , pid
418 )
419 startProcess LocalNodeClosed = throwIO $ userError $ "Node closed " ++ show (localNodeId node)
421 cleanupProcess :: ProcessId -> ValidLocalNodeState -> IO ValidLocalNodeState
422 cleanupProcess pid vst = do
423 let pid' = ProcessIdentifier pid
424 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
425 mapM_ (NT.close . fst) (Map.elems affected)
426 return $ (localProcessWithId (processLocalId pid) ^= Nothing)
427 . (localConnections ^= unaffected)
428 $ vst
430-- note [tracer/forkProcess races]
431--
432-- Our startTracing function uses forkProcess to start the trace controller
433-- process, and of course forkProcess attempts to call traceEvent once the
434-- process has started. This is harmless, as the localEventBus is not updated
435-- until /after/ the initial forkProcess completes, so the first call to
436-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
437--
439--------------------------------------------------------------------------------
440-- Handle incoming messages --
441--------------------------------------------------------------------------------
443type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
445data IncomingTarget =
446 Uninit
447 | ToProc ProcessId (Weak (CQueue Message))
448 | ToChan TypedChannel
449 | ToNode
451data ConnectionState = ConnectionState {
452 _incoming :: !(Map NT.ConnectionId IncomingConnection)
453 , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
454 }
456initConnectionState :: ConnectionState
457initConnectionState = ConnectionState {
458 _incoming = Map.empty
459 , _incomingFrom = Map.empty
460 }
462incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
463incoming = accessor _incoming (\conns st -> st { _incoming = conns })
465incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
466incomingAt cid = incoming >>> DAC.mapMaybe cid
468incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
469incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
470 where
471 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
473handleIncomingMessages :: LocalNode -> IO ()
474handleIncomingMessages node = go initConnectionState
475 where
476 go :: ConnectionState -> IO ()
477 go !st = do
478 event <- NT.receive endpoint
479 case event of
480 NT.ConnectionOpened cid rel theirAddr ->
481 if rel == NT.ReliableOrdered
482 then
483 trace node (MxConnected cid theirAddr)
484 >> go (
485 (incomingAt cid ^= Just (theirAddr, Uninit))
486 . (incomingFrom theirAddr ^: Set.insert cid)
487 $ st
488 )
489 else invalidRequest cid st
490 NT.Received cid payload ->
491 case st ^. incomingAt cid of
492 Just (_, ToProc pid weakQueue) -> do
493 mQueue <- deRefWeak weakQueue
494 forM_ mQueue $ \queue -> do
495 -- TODO: if we find that the queue is Nothing, should we remove
496 -- it from the NC state? (and same for channels, below)
497 let msg = payloadToMessage payload
498 enqueue queue msg -- 'enqueue' is strict
499 trace node (MxReceived pid msg)
500 go st
501 Just (_, ToChan (TypedChannel chan')) -> do
502 mChan <- deRefWeak chan'
503 -- If mChan is Nothing, the process has given up the read end of
504 -- the channel and we simply ignore the incoming message
505 forM_ mChan $ \chan -> atomically $
506 -- We make sure the message is fully decoded when it is enqueued
507 writeTQueue chan $! decode (BSL.fromChunks payload)
508 go st
509 Just (_, ToNode) -> do
510 let ctrlMsg = decode . BSL.fromChunks $ payload
511 writeChan ctrlChan $! ctrlMsg
512 go st
513 Just (src, Uninit) ->
514 case decode (BSL.fromChunks payload) of
515 ProcessIdentifier pid -> do
516 let lpid = processLocalId pid
517 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
518 case mProc of
519 Just proc ->
520 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
521 Nothing ->
522 invalidRequest cid st
523 SendPortIdentifier chId -> do
524 let lcid = sendPortLocalId chId
525 lpid = processLocalId (sendPortProcessId chId)
526 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
527 case mProc of
528 Just proc -> do
529 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
530 case mChannel of
531 Just channel ->
532 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
533 Nothing ->
534 invalidRequest cid st
535 Nothing ->
536 invalidRequest cid st
537 NodeIdentifier nid ->
538 if nid == localNodeId node
539 then go (incomingAt cid ^= Just (src, ToNode) $ st)
540 else invalidRequest cid st
541 Nothing ->
542 invalidRequest cid st
543 NT.ConnectionClosed cid ->
544 case st ^. incomingAt cid of
545 Nothing ->
546 invalidRequest cid st
547 Just (src, _) -> do
548 trace node (MxDisconnected cid src)
549 go ( (incomingAt cid ^= Nothing)
550 . (incomingFrom src ^: Set.delete cid)
551 $ st
552 )
553 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
554 -- [Unified table 9, rule node_disconnect]
555 let nid = NodeIdentifier $ NodeId theirAddr
556 writeChan ctrlChan NCMsg
557 { ctrlMsgSender = nid
558 , ctrlMsgSignal = Died nid DiedDisconnect
559 }
560 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
561 closeImplicitReconnections node nid
562 go ( (incomingFrom theirAddr ^= Set.empty)
563 . (incoming ^: Map.filterWithKey (const . notLost))
564 $ st
565 )
566 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
567 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
568 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
569 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
570 NT.EndPointClosed ->
571 return ()
572 NT.ReceivedMulticast _ _ ->
573 -- If we received a multicast message, something went horribly wrong
574 -- and we just give up
575 fail "Cloud Haskell fatal error: received unexpected multicast"
577 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
578 invalidRequest cid st = do
579 -- TODO: We should treat this as a fatal error on the part of the remote
580 -- node. That is, we should report the remote node as having died, and we
581 -- should close incoming connections (this requires a Transport layer
582 -- extension).
583 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
584 (Trace cid)]
585 go ( incomingAt cid ^= Nothing
586 $ st
587 )
589 state = localState node
590 endpoint = localEndPoint node
591 ctrlChan = localCtrlChan node
593--------------------------------------------------------------------------------
594-- Top-level access to the node controller --
595--------------------------------------------------------------------------------
597runNodeController :: LocalNode -> IO ()
598runNodeController =
599 runReaderT (evalStateT (unNC nodeController) initNCState)
601--------------------------------------------------------------------------------
602-- Internal data types --
603--------------------------------------------------------------------------------
605data NCState = NCState
606 { -- Mapping from remote processes to linked local processes
607 _links :: !(Map Identifier (Set ProcessId))
608 -- Mapping from remote processes to monitoring local processes
609 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
610 -- Process registry: names and where they live, mapped to the PIDs
611 , _registeredHere :: !(Map String ProcessId)
612 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
613 }
615newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
616 deriving ( Applicative
617 , Functor
618 , Monad
619 , MonadIO
620 , MonadState NCState
621 , MonadReader LocalNode
622 )
624initNCState :: NCState
625initNCState = NCState { _links = Map.empty
626 , _monitors = Map.empty
627 , _registeredHere = Map.empty
628 , _registeredOnNodes = Map.empty
629 }
631-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
632-- type is deliberately not exported so it cannot be caught explicitly.
633data ProcessKillException =
634 ProcessKillException !ProcessId !String
635 deriving (Typeable)
637instance Exception ProcessKillException
638instance Show ProcessKillException where
639 show (ProcessKillException pid reason) =
640 "killed-by=" ++ show pid ++ ",reason=" ++ reason
642--------------------------------------------------------------------------------
643-- Tracing/Debugging --
644--------------------------------------------------------------------------------
646-- [Issue #104 / DP-13]
648traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
649traceNotifyDied node ident reason =
650 -- TODO: sendPortDied notifications
651 liftIO $ withLocalTracer node $ \t ->
652 case ident of
653 (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
654 (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
655 _ -> return ()
657traceEventFmtIO :: LocalNode
658 -> String
659 -> [TraceArg]
660 -> IO ()
661traceEventFmtIO node fmt args =
662 withLocalTracer node $ \t -> traceLogFmt t fmt args
664trace :: LocalNode -> MxEvent -> IO ()
665trace node ev = withLocalTracer node $ \t -> traceEvent t ev
667withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
668withLocalTracer node act = act (localEventBus node)
670--------------------------------------------------------------------------------
671-- Core functionality --
672--------------------------------------------------------------------------------
674-- [Unified: Table 7]
675nodeController :: NC ()
676nodeController = do
677 node <- ask
678 forever' $ do
679 msg <- liftIO $ readChan (localCtrlChan node)
681 -- [Unified: Table 7, rule nc_forward]
682 case destNid (ctrlMsgSignal msg) of
683 Just nid' | nid' /= localNodeId node ->
684 liftIO $ sendBinary node
685 (ctrlMsgSender msg)
686 (NodeIdentifier nid')
687 WithImplicitReconnect
688 msg
689 _ ->
690 return ()
692 case msg of
693 NCMsg (ProcessIdentifier from) (Link them) ->
694 ncEffectMonitor from them Nothing
695 NCMsg (ProcessIdentifier from) (Monitor ref) ->
696 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
697 NCMsg (ProcessIdentifier from) (Unlink them) ->
698 ncEffectUnlink from them
699 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
700 ncEffectUnmonitor from ref
701 NCMsg _from (Died ident reason) ->
702 ncEffectDied ident reason
703 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
704 ncEffectSpawn from proc ref
705 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
706 ncEffectRegister from label atnode pid force
707 NCMsg (ProcessIdentifier from) (WhereIs label) ->
708 ncEffectWhereIs from label
709 NCMsg _ (NamedSend label msg') ->
710 ncEffectNamedSend label msg'
711 NCMsg _ (UnreliableSend lpid msg') ->
712 ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
713 NCMsg _ (LocalSend to msg') ->
714 ncEffectLocalSend node to msg'
715 NCMsg _ (LocalPortSend to msg') ->
716 ncEffectLocalPortSend to msg'
717 NCMsg (ProcessIdentifier from) (Kill to reason) ->
718 ncEffectKill from to reason
719 NCMsg (ProcessIdentifier from) (Exit to reason) ->
720 ncEffectExit from to reason
721 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
722 ncEffectGetInfo from pid
723 NCMsg _ SigShutdown ->
724 liftIO $ do
725 NT.closeEndPoint (localEndPoint node)
726 `Exception.finally` throwIO ThreadKilled
727 -- ThreadKilled seems to make more sense than fail/error here
728 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
729 ncEffectGetNodeStats from nid
730 unexpected ->
731 error $ "nodeController: unexpected message " ++ show unexpected
733-- [Unified: Table 10]
734ncEffectMonitor :: ProcessId -- ^ Who's watching?
735 -> Identifier -- ^ Who's being watched?
736 -> Maybe MonitorRef -- ^ 'Nothing' to link
737 -> NC ()
738ncEffectMonitor from them mRef = do
739 node <- ask
740 shouldLink <-
741 if not (isLocal node them)
742 then return True
743 else isValidLocalIdentifier them
744 case (shouldLink, isLocal node (ProcessIdentifier from)) of
745 (True, _) -> -- [Unified: first rule]
746 case mRef of
747 Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
748 Nothing -> modify' $ linksFor them ^: Set.insert from
749 (False, True) -> -- [Unified: second rule]
750 notifyDied from them DiedUnknownId mRef
751 (False, False) -> -- [Unified: third rule]
752 -- TODO: this is the right sender according to the Unified semantics,
753 -- but perhaps having 'them' as the sender would make more sense
754 -- (see also: notifyDied)
755 liftIO $ sendBinary node
756 (NodeIdentifier $ localNodeId node)
757 (NodeIdentifier $ processNodeId from)
758 WithImplicitReconnect
759 NCMsg
760 { ctrlMsgSender = NodeIdentifier (localNodeId node)
761 , ctrlMsgSignal = Died them DiedUnknownId
762 }
764-- [Unified: Table 11]
765ncEffectUnlink :: ProcessId -> Identifier -> NC ()
766ncEffectUnlink from them = do
767 node <- ask
768 when (isLocal node (ProcessIdentifier from)) $
769 case them of
770 ProcessIdentifier pid ->
771 postAsMessage from $ DidUnlinkProcess pid
772 NodeIdentifier nid ->
773 postAsMessage from $ DidUnlinkNode nid
774 SendPortIdentifier cid ->
775 postAsMessage from $ DidUnlinkPort cid
776 modify' $ linksFor them ^: Set.delete from
778-- [Unified: Table 11]
779ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
780ncEffectUnmonitor from ref = do
781 node <- ask
782 when (isLocal node (ProcessIdentifier from)) $
783 postAsMessage from $ DidUnmonitor ref
784 modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
786-- [Unified: Table 12]
787ncEffectDied :: Identifier -> DiedReason -> NC ()
788ncEffectDied ident reason = do
789 node <- ask
790 traceNotifyDied node ident reason
791 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
792 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
794-- _registry :: !(Map (String,NodeId) ProcessId)
796 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
798 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
799 forM_ uss $ \us ->
800 when (localOnly <= isLocal node (ProcessIdentifier us)) $
801 notifyDied us them reason Nothing
803 forM_ (Map.toList affectedMons) $ \(them, refs) ->
804 forM_ refs $ \(us, ref) ->
805 when (localOnly <= isLocal node (ProcessIdentifier us)) $
806 notifyDied us them reason (Just ref)
808 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
810 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
812 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
813 mapM (\(pid,nidlist) ->
814 case ident `impliesDeathOf` ProcessIdentifier pid of
815 True ->
816 do forM_ nidlist $ \(nid,_) ->
817 when (not $ isLocal node (NodeIdentifier nid))
818 (forwardNameDeath node nid)
819 return Nothing
820 False -> return $ Just (pid,nidlist) )
821 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
822 where
823 forwardNameDeath node nid =
824 liftIO $ sendBinary node
825 (NodeIdentifier $ localNodeId node)
826 (NodeIdentifier $ nid)
827 WithImplicitReconnect
828 NCMsg
829 { ctrlMsgSender = NodeIdentifier (localNodeId node)
830 , ctrlMsgSignal = Died ident reason
831 }
833-- [Unified: Table 13]
834ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
835ncEffectSpawn pid cProc ref = do
836 mProc <- unClosure cProc
837 -- If the closure does not exist, we spawn a process that throws an exception
838 -- This allows the remote node to find out what's happening
839 -- TODO:
840 let proc = case mProc of
841 Left err -> fail $ "Error: Could not resolve closure: " ++ err
842 Right p -> p
843 node <- ask
844 pid' <- liftIO $ forkProcess node proc
845 liftIO $ sendMessage node
846 (NodeIdentifier (localNodeId node))
847 (ProcessIdentifier pid)
848 WithImplicitReconnect
849 (DidSpawn ref pid')
851-- Unified semantics does not explicitly describe how to implement 'register',
852-- but mentions it's "very similar to nsend" (Table 14)
853-- We send a response indicated if the operation is invalid
854ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
855ncEffectRegister from label atnode mPid reregistration = do
856 node <- ask
857 currentVal <- gets (^. registeredHereFor label)
858 isOk <-
859 case mPid of
860 Nothing -> -- unregister request
861 return $ isJust currentVal
862 Just thepid -> -- register request
863 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
864 return $ (isNothing currentVal /= reregistration) &&
865 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
866 if isLocal node (NodeIdentifier atnode)
867 then do when (isOk) $
868 do modify' $ registeredHereFor label ^= mPid
869 updateRemote node currentVal mPid
870 case mPid of
871 (Just p) -> liftIO $ trace node (MxRegistered p label)
872 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
873 liftIO $ sendMessage node
874 (NodeIdentifier (localNodeId node))
875 (ProcessIdentifier from)
876 WithImplicitReconnect
877 (RegisterReply label isOk)
878 else let operation =
879 case reregistration of
880 True -> flip decList
881 False -> flip incList
882 in case mPid of
883 Nothing -> return ()
884 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
885 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
886 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
887 forward node (processNodeId newval) (Register label atnode (Just newval) False)
888 updateRemote node Nothing (Just newval) =
889 forward node (processNodeId newval) (Register label atnode (Just newval) False)
890 updateRemote node (Just oldval) Nothing =
891 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
892 updateRemote _ _ _ = return ()
893 maybeify f Nothing = unmaybeify $ f []
894 maybeify f (Just x) = unmaybeify $ f x
896 unmaybeify [] = Nothing
897 unmaybeify x = Just x
898 incList [] tag = [(tag,1)]
899 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
900 incList (x:xs) tag = x : incList xs tag
901 decList [] _ = []
902 decList ((atag,1):xs) tag | atag == tag = xs
903 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
904 decList (x:xs) tag = x:decList xs tag
905 forward node to reg =
906 when (not $ isLocal node (NodeIdentifier to)) $
907 liftIO $ sendBinary node
908 (ProcessIdentifier from)
909 (NodeIdentifier to)
910 WithImplicitReconnect
911 NCMsg
912 { ctrlMsgSender = ProcessIdentifier from
913 , ctrlMsgSignal = reg
914 }
917-- Unified semantics does not explicitly describe 'whereis'
918ncEffectWhereIs :: ProcessId -> String -> NC ()
919ncEffectWhereIs from label = do
920 node <- ask
921 mPid <- gets (^. registeredHereFor label)
922 liftIO $ sendMessage node
923 (NodeIdentifier (localNodeId node))
924 (ProcessIdentifier from)
925 WithImplicitReconnect
926 (WhereIsReply label mPid)
928-- [Unified: Table 14]
929ncEffectNamedSend :: String -> Message -> NC ()
930ncEffectNamedSend label msg = do
931 mPid <- gets (^. registeredHereFor label)
932 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
933 forM_ mPid $ \pid -> postMessage pid msg
935-- [Issue #DP-20]
936ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
937ncEffectLocalSend node to msg =
938 liftIO $ withLocalProc node to $ \p -> do
939 enqueue (processQueue p) msg
940 trace node (MxReceived to msg)
942-- [Issue #DP-20]
943ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
944ncEffectLocalPortSend from msg = do
945 node <- ask
946 let pid = sendPortProcessId from
947 cid = sendPortLocalId from
948 liftIO $ withLocalProc node pid $ \proc -> do
949 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
950 case mChan of
951 -- in the unlikely event we know nothing about this channel id,
952 -- there's little to be done - perhaps some logging/tracing though...
953 Nothing -> return ()
954 Just (TypedChannel chan') -> do
955 -- If ch is Nothing, the process has given up the read end of
956 -- the channel and we simply ignore the incoming message - this
957 ch <- deRefWeak chan'
958 forM_ ch $ \chan -> deliverChan msg chan
959 where deliverChan :: forall a . Message -> TQueue a -> IO ()
960 deliverChan (UnencodedMessage _ raw) chan' =
961 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
962 deliverChan (EncodedMessage _ _) _ =
963 -- this will not happen unless someone screws with Primitives.hs
964 error "invalid local channel delivery"
966-- [Issue #69]
967ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
968ncEffectKill from to reason = do
969 node <- ask
970 when (isLocal node (ProcessIdentifier to)) $
971 throwException to $ ProcessKillException from reason
973-- [Issue #69]
974ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
975ncEffectExit from to reason = do
976 node <- ask
977 when (isLocal node (ProcessIdentifier to)) $
978 throwException to $ ProcessExitException from reason
980-- [Issue #89]
981ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
982ncEffectGetInfo from pid =
983 let lpid = processLocalId pid
984 them = (ProcessIdentifier pid)
985 in do
986 node <- ask
987 mProc <- liftIO $ withValidLocalState node
988 $ return . (^. localProcessWithId lpid)
989 case mProc of
990 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
991 from node (ProcessInfoNone DiedUnknownId)
992 Just proc -> do
993 itsLinks <- gets (^. linksFor them)
994 itsMons <- gets (^. monitorsFor them)
995 registered <- gets (^. registeredHere)
996 size <- liftIO $ queueSize $ processQueue $ proc
998 let reg = registeredNames registered
999 dispatch (isLocal node (ProcessIdentifier from))
1000 from
1001 node
1002 ProcessInfo {
1003 infoNode = (processNodeId pid)
1004 , infoRegisteredNames = reg
1005 , infoMessageQueueLength = size
1006 , infoMonitors = Set.toList itsMons
1007 , infoLinks = Set.toList itsLinks
1009 where dispatch :: (Serializable a, Show a)
1010 => Bool
1011 -> ProcessId
1012 -> LocalNode
1013 -> a
1014 -> NC ()
1015 dispatch True dest _ pInfo = postAsMessage dest $ pInfo
1016 dispatch False dest node pInfo = do
1017 liftIO $ sendMessage node
1018 (NodeIdentifier (localNodeId node))
1019 (ProcessIdentifier dest)
1020 WithImplicitReconnect
1021 pInfo
1023 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1024 then (k:ks)
1025 else ks) []
1027ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1028ncEffectGetNodeStats from _nid = do
1029 node <- ask
1030 ncState <- StateT.get
1031 nodeState <- liftIO $ withValidLocalState node return
1032 let stats =
1033 NodeStats {
1034 nodeStatsNode = localNodeId node
1035 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1036 , nodeStatsMonitors = Map.size $ ncState ^. monitors
1037 , nodeStatsLinks = Map.size $ ncState ^. links
1038 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1040 postAsMessage from stats
1042--------------------------------------------------------------------------------
1043-- Auxiliary --
1044--------------------------------------------------------------------------------
1046notifyDied :: ProcessId -- ^ Who to notify?
1047 -> Identifier -- ^ Who died?
1048 -> DiedReason -- ^ How did they die?
1049 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1050 -> NC ()
1051notifyDied dest src reason mRef = do
1052 node <- ask
1053 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1054 (True, Just ref, ProcessIdentifier pid) ->
1055 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1056 (True, Just ref, NodeIdentifier nid) ->
1057 postAsMessage dest $ NodeMonitorNotification ref nid reason
1058 (True, Just ref, SendPortIdentifier cid) ->
1059 postAsMessage dest $ PortMonitorNotification ref cid reason
1060 (True, Nothing, ProcessIdentifier pid) ->
1061 throwException dest $ ProcessLinkException pid reason
1062 (True, Nothing, NodeIdentifier pid) ->
1063 throwException dest $ NodeLinkException pid reason
1064 (True, Nothing, SendPortIdentifier pid) ->
1065 throwException dest $ PortLinkException pid reason
1066 (False, _, _) ->
1067 -- The change in sender comes from [Unified: Table 10]
1068 liftIO $ sendBinary node
1069 (NodeIdentifier $ localNodeId node)
1070 (NodeIdentifier $ processNodeId dest)
1071 WithImplicitReconnect
1072 NCMsg
1073 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1074 , ctrlMsgSignal = Died src reason
1077-- | [Unified: Table 8]
1078destNid :: ProcessSignal -> Maybe NodeId
1079destNid (Link ident) = Just $ nodeOf ident
1080destNid (Unlink ident) = Just $ nodeOf ident
1081destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1082destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1083destNid (Spawn _ _) = Nothing
1084destNid (Register _ _ _ _) = Nothing
1085destNid (WhereIs _) = Nothing
1086destNid (NamedSend _ _) = Nothing
1087destNid (UnreliableSend _ _) = Nothing
1088-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1089-- then when a local process dies the monitoring/linking machinery will take
1090-- care of notifying remote nodes
1091destNid (Died _ _) = Nothing
1092destNid (Kill pid _) = Just $ processNodeId pid
1093destNid (Exit pid _) = Just $ processNodeId pid
1094destNid (GetInfo pid) = Just $ processNodeId pid
1095destNid (GetNodeStats nid) = Just nid
1096destNid (LocalSend pid _) = Just $ processNodeId pid
1097destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1098destNid (SigShutdown) = Nothing
1100-- | Check if a process is local to our own node
1101isLocal :: LocalNode -> Identifier -> Bool
1102isLocal nid ident = nodeOf ident == localNodeId nid
1104-- | Lookup a local closure
1105unClosure :: Typeable a => Closure a -> NC (Either String a)
1106unClosure closure = do
1107 rtable <- remoteTable <$> ask
1108 return (Static.unclosure rtable closure)
1110-- | Check if an identifier refers to a valid local object
1111isValidLocalIdentifier :: Identifier -> NC Bool
1112isValidLocalIdentifier ident = do
1113 node <- ask
1114 liftIO . withValidLocalState node $ \nSt ->
1115 case ident of
1116 NodeIdentifier nid ->
1117 return $ nid == localNodeId node
1118 ProcessIdentifier pid -> do
1119 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1120 return $ isJust mProc
1121 SendPortIdentifier cid -> do
1122 let pid = sendPortProcessId cid
1123 mProc = nSt ^. localProcessWithId (processLocalId pid)
1124 case mProc of
1125 Nothing -> return False
1126 Just proc -> withMVar (processState proc) $ \pSt -> do
1127 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1128 return $ isJust mCh
1130--------------------------------------------------------------------------------
1131-- Messages to local processes --
1132--------------------------------------------------------------------------------
1134postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1135postAsMessage pid = postMessage pid . createUnencodedMessage
1137postMessage :: ProcessId -> Message -> NC ()
1138postMessage pid msg = do
1139 node <- ask
1140 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1142throwException :: Exception e => ProcessId -> e -> NC ()
1143throwException pid e = do
1144 node <- ask
1145 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1147withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1148withLocalProc node pid p =
1149 -- By [Unified: table 6, rule missing_process] messages to dead processes
1150 -- can silently be dropped
1151 let lpid = processLocalId pid in do
1152 withValidLocalState node $ \vst ->
1153 forM_ (vst ^. localProcessWithId lpid) p
1155--------------------------------------------------------------------------------
1156-- Accessors --
1157--------------------------------------------------------------------------------
1159links :: Accessor NCState (Map Identifier (Set ProcessId))
1160links = accessor _links (\ls st -> st { _links = ls })
1162monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1163monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1165registeredHere :: Accessor NCState (Map String ProcessId)
1166registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1168registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1169registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1171linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1172linksFor ident = links >>> DAC.mapDefault Set.empty ident
1174monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1175monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1177registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1178registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1180registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1181registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1183-- | @splitNotif ident@ splits a notifications map into those
1184-- notifications that should trigger when 'ident' fails and those links that
1185-- should not.
1187-- There is a hierarchy between identifiers: failure of a node implies failure
1188-- of all processes on that node, and failure of a process implies failure of
1189-- all typed channels to that process. In other words, if 'ident' refers to a
1190-- node, then the /should trigger/ set will include
1192-- * the notifications for the node specifically
1193-- * the notifications for processes on that node, and
1194-- * the notifications for typed channels to processes on that node.
1196-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1197-- include
1199-- * the notifications for that process specifically and
1200-- * the notifications for typed channels to that process.
1202-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1203splitNotif :: Identifier
1204 -> Map Identifier a
1205 -> (Map Identifier a, Map Identifier a)
1206splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1208--------------------------------------------------------------------------------
1209-- Auxiliary --
1210--------------------------------------------------------------------------------
1212-- | Modify and evaluate the state
1213modify' :: MonadState s m => (s -> s) -> m ()
1214modify' f = StateT.get >>= \s -> StateT.put $! f s