@@ -67,8 +67,12 @@ module Control.Distributed.Process
6767 , newLocalNode
6868 , forkProcess
6969 , runProcess
70+ -- * Auxiliary API
71+ , closeLocalNode
72+ , pcatch
7073 ) where
7174
75+import Prelude hiding (catch)
7276import qualified Data.ByteString as BSS (ByteString, concat, splitAt)
7377import qualified Data.ByteString.Lazy as BSL ( ByteString
7478 , toChunks
@@ -81,12 +85,13 @@ import qualified Data.Map as Map (empty, lookup, insert, delete)
8185import qualified Data.List as List (delete)
8286import Data.Int (Int32)
8387import Data.Typeable (Typeable)
88+import Data.Foldable (forM_)
8489import Control.Monad (void, liftM, liftM2)
8590import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
8691import Control.Monad.IO.Class (MonadIO, liftIO)
8792import Control.Applicative ((<$>))
8893import Control.Category ((>>>))
89−import Control.Exception (Exception, throwIO)
94+import Control.Exception (Exception, throwIO, catch)
9095import Control.Concurrent (forkIO)
9196import Control.Concurrent.MVar ( MVar
9297 , newMVar
@@ -122,6 +127,7 @@ import qualified Network.Transport as NT ( Transport
122127 , receive
123128 , Event(..)
124129 , address
130+ , closeEndPoint
125131 )
126132import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
127133import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
@@ -158,8 +164,7 @@ data LocalNode = LocalNode
158164
159165-- | Local node state
160166data LocalNodeState = LocalNodeState
161− { _localConnections :: Map ProcessId NT.Connection
162− , _localProcesses :: Map LocalProcessId LocalProcess
167+ { _localProcesses :: Map LocalProcessId LocalProcess
163168 , _localPidCounter :: Int32
164169 , _localPidUnique :: Int32
165170 }
@@ -169,6 +174,13 @@ data LocalProcess = LocalProcess
169174 { processQueue :: CQueue Message
170175 , processNode :: LocalNode
171176 , processId :: ProcessId
177+ , processState :: MVar LocalProcessState
178+ }
179+
180+-- | Local process state
181+data LocalProcessState = LocalProcessState
182+ { _monitorActions :: Map ProcessId MonitorAction
183+ , _connections :: Map ProcessId NT.Connection
172184 }
173185
174186-- The Cloud Haskell 'Process' type
@@ -189,25 +201,31 @@ expect = do
189201
190202-- | Send a message
191203send :: Serializable a => ProcessId -> a -> Process ()
192−send pid msg = do
204+send them msg = do
193205 -- This requires a lookup on every send. If we want to avoid that we need to
194206 -- modify serializable to allow for stateful (IO) deserialization
195− node <- processNode <$> ask
196− liftIO $ do
197− conn <- connectionTo node pid
198− sendBinary node pid conn ( encodeFingerprint (fingerprint msg)
199− : BSL.toChunks (encode msg)
200− )
207+ mConn <- getConnectionTo them
208+ forM_ mConn $ \conn -> sendMessage them conn (createMessage msg)
201209
202210-- | Our own process ID
203211getSelfPid :: Process ProcessId
204212getSelfPid = processId <$> ask
205213
206214-- | Variation on 'monitorProcess' where the monitor is the current process
215+--
216+-- TODO: we need to decide what to do with multiple calls to monitor with
217+-- the same argument. The Erlang manual is very specific: it creates multiple
218+-- monitors (http://www.erlang.org/doc/reference_manual/processes.html#id82613)
207219monitor :: ProcessId -> MonitorAction -> Process ()
208−monitor _them = undefined
220+monitor them action = do
221+ ourState <- processState <$> ask
222+ liftIO $ modifyMVar_ ourState $ \st ->
223+ return . (monitorActionFor them ^= Just action) $ st
209224
210225-- | Have one process monitor another
226+--
227+-- TODO: this is a weird function to expose at the API level ('monitor' is more
228+-- natural and more Erlang-like)
211229monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()
212230monitorProcess = undefined
213231
@@ -215,6 +233,24 @@ monitorProcess = undefined
215233linkProcess :: ProcessId -> Process ()
216234linkProcess = undefined
217235
236+--------------------------------------------------------------------------------
237+-- Auxiliary API --
238+--------------------------------------------------------------------------------
239+
240+-- Force-close a local node
241+--
242+-- TODO: for now we just close the associated endpoint
243+closeLocalNode :: LocalNode -> IO ()
244+closeLocalNode node =
245+ NT.closeEndPoint (localEndPoint node)
246+
247+-- Catch exceptions within a process
248+-- TODO: should this be called simply 'catch'?
249+pcatch :: Exception e => Process a -> (e -> Process a) -> Process a
250+pcatch p h = do
251+ run <- runLocalProcess <$> ask
252+ liftIO $ catch (run p) (run . h)
253+
218254--------------------------------------------------------------------------------
219255-- Monitoring --
220256-- --
@@ -315,8 +351,7 @@ newLocalNode transport = do
315351 Right endPoint -> do
316352 unq <- randomIO
317353 state <- newMVar LocalNodeState
318− { _localConnections = Map.empty
319− , _localProcesses = Map.empty
354+ { _localProcesses = Map.empty
320355 , _localPidCounter = 0
321356 , _localPidUnique = unq
322357 }
@@ -335,25 +370,29 @@ runProcess node proc = do
335370forkProcess :: LocalNode -> Process () -> IO ProcessId
336371forkProcess node proc = do
337372 queue <- newCQueue
338− state <- modifyMVar (localState node) $ \st -> do
373+ lproc <- modifyMVar (localState node) $ \st -> do
339374 let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
340375 , lpidUnique = st ^. localPidUnique
341376 }
342377 let pid = ProcessId { processAddress = NT.address (localEndPoint node)
343378 , processLocalId = lpid
344379 }
345− let state = LocalProcess { processQueue = queue
380+ pst <- newMVar LocalProcessState { _connections = Map.empty
381+ , _monitorActions = Map.empty
382+ }
383+ let lproc = LocalProcess { processQueue = queue
346384 , processNode = node
347385 , processId = pid
386+ , processState = pst
348387 }
349388 -- TODO: if the counter overflows we should pick a new unique
350− return ( (localProcessWithId lpid ^= Just state)
389+ return ( (localProcessWithId lpid ^= Just lproc)
351390 . (localPidCounter ^: (+ 1))
352391 $ st
353− , state
392+ , lproc
354393 )
355− void . forkIO $ runReaderT (unProcess proc) state
356− return (processId state)
394+ void . forkIO $ runLocalProcess lproc proc
395+ return (processId lproc)
357396
358397handleIncomingMessages :: LocalNode -> IO ()
359398handleIncomingMessages node = go [] Map.empty
@@ -397,27 +436,27 @@ handleIncomingMessages node = go [] Map.empty
397436-- Auxiliary functions --
398437--------------------------------------------------------------------------------
399438
400−connectionTo :: LocalNode -> ProcessId -> IO NT.Connection
401−connectionTo node pid = do
402− mConn <- withMVar (localState node) $ return . (^. localConnectionTo pid)
439+getConnectionTo :: ProcessId -> Process (Maybe NT.Connection)
440+getConnectionTo pid = do
441+ ourState <- processState <$> ask
442+ mConn <- liftIO $ withMVar ourState $ return . (^. connectionTo pid)
403443 case mConn of
404− Just conn -> return conn
405− Nothing -> createConnectionTo node pid
406−
407−createConnectionTo :: LocalNode -> ProcessId -> IO NT.Connection
408−createConnectionTo node pid = do
409− mConn <- NT.connect (localEndPoint node)
410− (processAddress pid)
411− NT.ReliableOrdered
412− NT.defaultConnectHints
444+ Just conn -> return . Just $ conn
445+ Nothing -> createConnectionTo pid
446+
447+createConnectionTo :: ProcessId -> Process (Maybe NT.Connection)
448+createConnectionTo pid = do
449+ proc <- ask
450+ mConn <- liftIO $ NT.connect (localEndPoint . processNode $ proc)
451+ (processAddress pid)
452+ NT.ReliableOrdered
453+ NT.defaultConnectHints
413454 case mConn of
414455 Right conn -> do
415− mConn' <- modifyMVar (localState node) $ \st ->
416− case st ^. localConnectionTo pid of
456+ mConn' <- liftIO $ modifyMVar (processState proc) $ \st ->
457+ case st ^. connectionTo pid of
417458 Just conn' -> return (st, Just conn')
418− Nothing -> return ( localConnectionTo pid ^= Just conn $ st
419− , Nothing
420− )
459+ Nothing -> return (connectionTo pid ^= Just conn $ st, Nothing)
421460 case mConn' of
422461 Just conn' -> do
423462 -- Somebody else already created a connection while we weren't looking
@@ -425,27 +464,30 @@ createConnectionTo node pid = do
425464 -- connection because that would limit concurrency too much, and
426465 -- since Network.Transport supports lgihtweight connections creating
427466 -- an unnecessary connection now and then is cheap anyway)
428− NT.close conn
429− return conn'
467+ liftIO $ NT.close conn
468+ return . Just $ conn'
430469 Nothing -> do
431− sendBinary node pid conn $ lpidToPayload (processLocalId pid)
432− return conn
433− Left err ->
434− throwIO err
435−
436−sendBinary :: LocalNode
437− -> ProcessId
438− -> NT.Connection
439− -> [BSS.ByteString]
440− -> IO ()
441−sendBinary node pid conn msg = do
442− result <- NT.send conn msg
470+ sendBinary pid conn $ lpidToPayload (processLocalId pid)
471+ return . Just $ conn
472+ Left _err -> do
473+ -- TODO: should probably pass this error to remoteProcessFailed
474+ remoteProcessFailed pid
475+ return Nothing
476+
477+sendBinary :: ProcessId -> NT.Connection -> [BSS.ByteString] -> Process ()
478+sendBinary pid conn payload = do
479+ result <- liftIO $ NT.send conn payload
443480 case result of
444481 Right () -> return ()
445482 Left err -> do
446− modifyMVar_ (localState node) $
447− return . (localConnectionTo pid ^= Nothing)
448− throwIO err
483+ ourState <- processState <$> ask
484+ liftIO $ modifyMVar_ ourState $
485+ return . (connectionTo pid ^= Nothing)
486+ liftIO $ throwIO err
487+
488+sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()
489+sendMessage pid conn (Message fp enc) =
490+ sendBinary pid conn $ encodeFingerprint fp : BSL.toChunks enc
449491
450492payloadToMessage :: [BSS.ByteString] -> Message
451493payloadToMessage payload = Message fp msg
@@ -464,7 +506,31 @@ payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss
464506 in LocalProcessId { lpidCounter = NTI.decodeInt32 bs1
465507 , lpidUnique = NTI.decodeInt32 bs2
466508 }
467−
509+
510+remoteProcessFailed :: ProcessId -> Process ()
511+remoteProcessFailed them = do
512+ proc <- ask
513+ let ourState = processState proc
514+ action <- liftIO $ withMVar ourState $ return . (^. monitorActionFor them)
515+ let err = ProcessMonitorException them SrNoPing
516+ liftIO $ case action of
517+ Just MaMonitor ->
518+ -- TODO: can/should we avoid this encoding/decoding?
519+ enqueue (processQueue proc) $ createMessage err
520+ Just MaLink ->
521+ throwIO err
522+ Just MaLinkError ->
523+ throwIO err
524+ Nothing ->
525+ return ()
526+
527+createMessage :: Serializable a => a -> Message
528+createMessage a = Message (fingerprint a) (encode a)
529+
530+-- This is most definitely NOT exported
531+runLocalProcess :: LocalProcess -> Process a -> IO a
532+runLocalProcess lproc proc = runReaderT (unProcess proc) lproc
533+
468534--------------------------------------------------------------------------------
469535-- Binary instances --
470536--------------------------------------------------------------------------------
@@ -481,9 +547,6 @@ instance Binary ProcessId where
481547-- Accessors --
482548--------------------------------------------------------------------------------
483549
484−localConnections :: Accessor LocalNodeState (Map ProcessId NT.Connection)
485−localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })
486−
487550localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess)
488551localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })
489552
@@ -493,8 +556,17 @@ localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter =
493556localPidUnique :: Accessor LocalNodeState Int32
494557localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })
495558
496−localConnectionTo :: ProcessId -> Accessor LocalNodeState (Maybe NT.Connection)
497−localConnectionTo pid = localConnections >>> DAC.mapMaybe pid
498−
499559localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)
500560localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid
561+
562+connections :: Accessor LocalProcessState (Map ProcessId NT.Connection)
563+connections = accessor _connections (\conns st -> st { _connections = conns })
564+
565+monitorActions :: Accessor LocalProcessState (Map ProcessId MonitorAction)
566+monitorActions = accessor _monitorActions (\ms st -> st { _monitorActions = ms})
567+
568+connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)
569+connectionTo pid = connections >>> DAC.mapMaybe pid
570+
571+monitorActionFor :: ProcessId -> Accessor LocalProcessState (Maybe MonitorAction)
572+monitorActionFor pid = monitorActions >>> DAC.mapMaybe pid