@@ -96,7 +96,7 @@ import qualified Data.ByteString.Lazy as BSL ( ByteString
9696 )
9797import Data.Binary (Binary, decode, encode, put, get, getWord8, putWord8)
9898import Data.Map (Map)
99−import qualified Data.Map as Map (empty, lookup, insert, delete)
99+import qualified Data.Map as Map (empty, lookup, insert, delete, keys)
100100import qualified Data.List as List (delete)
101101import Data.Int (Int32)
102102import Data.Typeable (Typeable)
@@ -106,8 +106,8 @@ import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
106106import Control.Monad.IO.Class (MonadIO, liftIO)
107107import Control.Applicative ((<$>))
108108import Control.Category ((>>>))
109−import Control.Exception (Exception, throwIO, catch)
110−import Control.Concurrent (forkIO)
109+import Control.Exception (Exception, throwIO, catch, throwTo)
110+import Control.Concurrent (forkIO, myThreadId)
111111import Control.Concurrent.MVar ( MVar
112112 , newMVar
113113 , withMVar
@@ -141,12 +141,14 @@ import qualified Network.Transport as NT ( Transport
141141 , newEndPoint
142142 , receive
143143 , Event(..)
144+ , EventErrorCode(..)
145+ , TransportError(..)
144146 , address
145147 , closeEndPoint
146148 )
147149import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
148150import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
149−import qualified Data.Accessor.Container as DAC (mapMaybe)
151+import qualified Data.Accessor.Container as DAC (mapMaybe, mapDefault)
150152import System.Random (randomIO)
151153
152154-- | A local process ID consists of a seed which distinguishes processes from
@@ -173,15 +175,16 @@ data Message = Message
173175
174176-- | Local nodes
175177data LocalNode = LocalNode
176− { localEndPoint :: NT.EndPoint
177− , localState :: MVar LocalNodeState
178+ { localEndPoint :: NT.EndPoint
179+ , localState :: MVar LocalNodeState
178180 }
179181
180182-- | Local node state
181183data LocalNodeState = LocalNodeState
182− { _localProcesses :: Map LocalProcessId LocalProcess
183− , _localPidCounter :: Int32
184− , _localPidUnique :: Int32
184+ { _localProcesses :: Map LocalProcessId LocalProcess
185+ , _localPidCounter :: Int32
186+ , _localPidUnique :: Int32
187+ , _monitorActions :: Map ProcessId [IO ()]
185188 }
186189
187190-- | Processes running on our local node
@@ -194,8 +197,7 @@ data LocalProcess = LocalProcess
194197
195198-- | Local process state
196199data LocalProcessState = LocalProcessState
197− { _monitorActions :: Map ProcessId MonitorAction
198− , _connections :: Map ProcessId NT.Connection
200+ { _connections :: Map ProcessId NT.Connection
199201 }
200202
201203-- The Cloud Haskell 'Process' type
@@ -234,9 +236,22 @@ getSelfPid = processId <$> ask
234236-- monitors (http://www.erlang.org/doc/reference_manual/processes.html#id82613)
235237monitor :: ProcessId -> MonitorAction -> Process ()
236238monitor them action = do
239+ proc <- ask
240+ liftIO $ do
241+ tid <- myThreadId
242+ let err = ProcessMonitorException them SrNoPing
243+ act = case action of
244+ MaMonitor -> enqueue (processQueue proc) $ createMessage err
245+ MaLink -> throwTo tid err
246+ MaLinkError -> fail "monitor: MaLinkError not implemented"
247+ modifyMVar_ (localState . processNode $ proc) $
248+ return . (monitorActionsFor them ^: (act :))
249+
250+{-
237251 ourState <- processState <$> ask
238252 liftIO $ modifyMVar_ ourState $ \st ->
239253 return . (monitorActionFor them ^= Just action) $ st
254+-}
240255
241256-- | Have one process monitor another
242257--
@@ -375,9 +390,10 @@ newLocalNode transport = do
375390 Right endPoint -> do
376391 unq <- randomIO
377392 state <- newMVar LocalNodeState
378− { _localProcesses = Map.empty
379− , _localPidCounter = 0
380− , _localPidUnique = unq
393+ { _localProcesses = Map.empty
394+ , _localPidCounter = 0
395+ , _localPidUnique = unq
396+ , _monitorActions = Map.empty
381397 }
382398 let node = LocalNode { localEndPoint = endPoint
383399 , localState = state
@@ -401,8 +417,7 @@ forkProcess node proc = do
401417 let pid = ProcessId { processAddress = NT.address (localEndPoint node)
402418 , processLocalId = lpid
403419 }
404− pst <- newMVar LocalProcessState { _connections = Map.empty
405− , _monitorActions = Map.empty
420+ pst <- newMVar LocalProcessState { _connections = Map.empty
406421 }
407422 let lproc = LocalProcess { processQueue = queue
408423 , processNode = node
@@ -446,9 +461,14 @@ handleIncomingMessages node = go [] Map.empty
446461 fail "handleIncomingMessages: TODO 2"
447462 NT.ConnectionClosed cid ->
448463 go (List.delete cid halfOpenConns) (Map.delete cid openConns)
464+ NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr _) _) -> do
465+ -- TODO: we could cache this information
466+ pids <- withMVar state $ \st ->
467+ return $ filter ((== theirAddr) . processAddress) . Map.keys
468+ $ st ^. monitorActions
469+ forM_ pids $ remoteProcessFailed node
449470 NT.ErrorEvent _ ->
450− -- fail "handleIncomingMessages: TODO 3"
451− go halfOpenConns openConns
471+ fail "handleIncomingMessages: TODO 4"
452472 NT.EndPointClosed ->
453473 return ()
454474 NT.ReceivedMulticast _ _ ->
@@ -496,16 +516,19 @@ createConnectionTo them = do
496516 return . Just $ conn
497517 Left _err -> do
498518 -- TODO: should probably pass this error to remoteProcessFailed
499− remoteProcessFailed them
519+ node <- processNode <$> ask
520+ liftIO $ remoteProcessFailed node them
500521 return Nothing
501522
502523sendBinary :: ProcessId -> NT.Connection -> [BSS.ByteString] -> Process ()
503524sendBinary them conn payload = do
504525 result <- liftIO $ NT.send conn payload
505526 case result of
506− Right () -> return ()
507− Left _ -> remoteProcessFailed them
508−
527+ Right () ->
528+ return ()
529+ Left _ -> do
530+ node <- processNode <$> ask
531+ liftIO $ remoteProcessFailed node them
509532
510533sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()
511534sendMessage them conn (Message fp enc) =
@@ -529,8 +552,13 @@ payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss
529552 , lpidUnique = NTI.decodeInt32 bs2
530553 }
531554
532−remoteProcessFailed :: ProcessId -> Process ()
533−remoteProcessFailed them = do
555+remoteProcessFailed :: LocalNode -> ProcessId -> IO ()
556+remoteProcessFailed node them = do
557+ liftIO $ modifyMVar_ (localState node) $ \st -> do
558+ sequence_ (st ^. monitorActionsFor them)
559+ return (monitorActionsFor them ^= [] $ st)
560+
561+{-do
534562 proc <- ask
535563 let ourState = processState proc
536564 -- We only execute monitor actions once
@@ -547,6 +575,7 @@ remoteProcessFailed them = do
547575 throwIO err
548576 Nothing ->
549577 return ()
578+ -}
550579
551580createMessage :: Serializable a => a -> Message
552581createMessage a = Message (fingerprint a) (encode a)
@@ -580,17 +609,17 @@ localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter =
580609localPidUnique :: Accessor LocalNodeState Int32
581610localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })
582611
612+monitorActions :: Accessor LocalNodeState (Map ProcessId [IO ()])
613+monitorActions = accessor _monitorActions (\as st -> st { _monitorActions = as })
614+
583615localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)
584616localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid
585617
618+monitorActionsFor :: ProcessId -> Accessor LocalNodeState [IO ()]
619+monitorActionsFor pid = monitorActions >>> DAC.mapDefault [] pid
620+
586621connections :: Accessor LocalProcessState (Map ProcessId NT.Connection)
587622connections = accessor _connections (\conns st -> st { _connections = conns })
588623
589−monitorActions :: Accessor LocalProcessState (Map ProcessId MonitorAction)
590−monitorActions = accessor _monitorActions (\ms st -> st { _monitorActions = ms})
591−
592624connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)
593625connectionTo pid = connections >>> DAC.mapMaybe pid
594−
595−monitorActionFor :: ProcessId -> Accessor LocalProcessState (Maybe MonitorAction)
596−monitorActionFor pid = monitorActions >>> DAC.mapMaybe pid