e6fc1c5 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Start work on monitoring

e6fc1c5Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+132−60

compare with master

unifiedsplit

@@ -67,8 +67,12 @@ module Control.Distributed.Process

67 , newLocalNode

68 , forkProcess

69 , runProcess

70+ -- * Auxiliary API

71+ , closeLocalNode

72+ , pcatch

70 ) where

73 ) where

75+import Prelude hiding (catch)

72import qualified Data.ByteString as BSS (ByteString, concat, splitAt)

76import qualified Data.ByteString as BSS (ByteString, concat, splitAt)

73import qualified Data.ByteString.Lazy as BSL ( ByteString

77import qualified Data.ByteString.Lazy as BSL ( ByteString

74 , toChunks

78 , toChunks

@@ -81,12 +85,13 @@ import qualified Data.Map as Map (empty, lookup, insert, delete)

81import qualified Data.List as List (delete)

85import qualified Data.List as List (delete)

82import Data.Int (Int32)

86import Data.Int (Int32)

83import Data.Typeable (Typeable)

87import Data.Typeable (Typeable)

88+import Data.Foldable (forM_)

84import Control.Monad (void, liftM, liftM2)

89import Control.Monad (void, liftM, liftM2)

85import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

90import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

86import Control.Monad.IO.Class (MonadIO, liftIO)

91import Control.Monad.IO.Class (MonadIO, liftIO)

87import Control.Applicative ((<$>))

92import Control.Applicative ((<$>))

88import Control.Category ((>>>))

93import Control.Category ((>>>))

89−import Control.Exception (Exception, throwIO)

94+import Control.Exception (Exception, throwIO, catch)

90import Control.Concurrent (forkIO)

95import Control.Concurrent (forkIO)

91import Control.Concurrent.MVar ( MVar

96import Control.Concurrent.MVar ( MVar

92 , newMVar

97 , newMVar

@@ -122,6 +127,7 @@ import qualified Network.Transport as NT ( Transport

122 , receive

127 , receive

123 , Event(..)

128 , Event(..)

124 , address

129 , address

130+ , closeEndPoint

125 )

131 )

126import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)

132import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)

127import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

133import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

@@ -158,8 +164,7 @@ data LocalNode = LocalNode

158

164

159-- | Local node state

165-- | Local node state

160data LocalNodeState = LocalNodeState

166data LocalNodeState = LocalNodeState

161− { _localConnections :: Map ProcessId NT.Connection

162− , _localProcesses :: Map LocalProcessId LocalProcess

167+ { _localProcesses :: Map LocalProcessId LocalProcess

163 , _localPidCounter :: Int32

168 , _localPidCounter :: Int32

164 , _localPidUnique :: Int32

169 , _localPidUnique :: Int32

165 }

170 }

@@ -169,6 +174,13 @@ data LocalProcess = LocalProcess

169 { processQueue :: CQueue Message

174 { processQueue :: CQueue Message

170 , processNode :: LocalNode

175 , processNode :: LocalNode

171 , processId :: ProcessId

176 , 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

172 }

184 }

173

185

174-- The Cloud Haskell 'Process' type

186-- The Cloud Haskell 'Process' type

@@ -189,25 +201,31 @@ expect = do

189

201

190-- | Send a message

202-- | Send a message

191send :: Serializable a => ProcessId -> a -> Process ()

203send :: Serializable a => ProcessId -> a -> Process ()

192−send pid msg = do

204+send them msg = do

193 -- This requires a lookup on every send. If we want to avoid that we need to

205 -- This requires a lookup on every send. If we want to avoid that we need to

194 -- modify serializable to allow for stateful (IO) deserialization

206 -- modify serializable to allow for stateful (IO) deserialization

195− node <- processNode <$> ask

207+ mConn <- getConnectionTo them

196− liftIO $ do

208+ forM_ mConn $ \conn -> sendMessage them conn (createMessage msg)

197− conn <- connectionTo node pid

198− sendBinary node pid conn ( encodeFingerprint (fingerprint msg)

199− : BSL.toChunks (encode msg)

200− )

201

209

202-- | Our own process ID

210-- | Our own process ID

203getSelfPid :: Process ProcessId

211getSelfPid :: Process ProcessId

204getSelfPid = processId <$> ask

212getSelfPid = processId <$> ask

205

213

206-- | Variation on 'monitorProcess' where the monitor is the current process

214-- | 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)

207monitor :: ProcessId -> MonitorAction -> Process ()

219monitor :: 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

209

224

210-- | Have one process monitor another

225-- | 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)

211monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()

229monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()

212monitorProcess = undefined

230monitorProcess = undefined

213

231

@@ -215,6 +233,24 @@ monitorProcess = undefined

215linkProcess :: ProcessId -> Process ()

233linkProcess :: ProcessId -> Process ()

216linkProcess = undefined

234linkProcess = undefined

217

235

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+

218--------------------------------------------------------------------------------

254--------------------------------------------------------------------------------

219-- Monitoring --

255-- Monitoring --

220-- --

256-- --

@@ -315,8 +351,7 @@ newLocalNode transport = do

315 Right endPoint -> do

351 Right endPoint -> do

316 unq <- randomIO

352 unq <- randomIO

317 state <- newMVar LocalNodeState

353 state <- newMVar LocalNodeState

318− { _localConnections = Map.empty

319− , _localProcesses = Map.empty

354+ { _localProcesses = Map.empty

320 , _localPidCounter = 0

355 , _localPidCounter = 0

321 , _localPidUnique = unq

356 , _localPidUnique = unq

322 }

357 }

@@ -335,25 +370,29 @@ runProcess node proc = do

335forkProcess :: LocalNode -> Process () -> IO ProcessId

370forkProcess :: LocalNode -> Process () -> IO ProcessId

336forkProcess node proc = do

371forkProcess node proc = do

337 queue <- newCQueue

372 queue <- newCQueue

338− state <- modifyMVar (localState node) $ \st -> do

373+ lproc <- modifyMVar (localState node) $ \st -> do

339 let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter

374 let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter

340 , lpidUnique = st ^. localPidUnique

375 , lpidUnique = st ^. localPidUnique

341 }

376 }

342 let pid = ProcessId { processAddress = NT.address (localEndPoint node)

377 let pid = ProcessId { processAddress = NT.address (localEndPoint node)

343 , processLocalId = lpid

378 , processLocalId = lpid

344 }

379 }

380+ pst <- newMVar LocalProcessState { _connections = Map.empty

381+ , _monitorActions = Map.empty

382+ }

345− let state = LocalProcess { processQueue = queue

383+ let lproc = LocalProcess { processQueue = queue

346 , processNode = node

384 , processNode = node

347 , processId = pid

385 , processId = pid

386+ , processState = pst

348 }

387 }

349 -- TODO: if the counter overflows we should pick a new unique

388 -- TODO: if the counter overflows we should pick a new unique

350− return ( (localProcessWithId lpid ^= Just state)

389+ return ( (localProcessWithId lpid ^= Just lproc)

351 . (localPidCounter ^: (+ 1))

390 . (localPidCounter ^: (+ 1))

352 $ st

391 $ st

353− , state

392+ , lproc

354 )

393 )

355− void . forkIO $ runReaderT (unProcess proc) state

394+ void . forkIO $ runLocalProcess lproc proc

356− return (processId state)

395+ return (processId lproc)

357

396

358handleIncomingMessages :: LocalNode -> IO ()

397handleIncomingMessages :: LocalNode -> IO ()

359handleIncomingMessages node = go [] Map.empty

398handleIncomingMessages node = go [] Map.empty

@@ -397,27 +436,27 @@ handleIncomingMessages node = go [] Map.empty

397-- Auxiliary functions --

436-- Auxiliary functions --

398--------------------------------------------------------------------------------

437--------------------------------------------------------------------------------

399

438

400−connectionTo :: LocalNode -> ProcessId -> IO NT.Connection

439+getConnectionTo :: ProcessId -> Process (Maybe NT.Connection)

401−connectionTo node pid = do

440+getConnectionTo pid = do

441+ ourState <- processState <$> ask

402− mConn <- withMVar (localState node) $ return . (^. localConnectionTo pid)

442+ mConn <- liftIO $ withMVar ourState $ return . (^. connectionTo pid)

403 case mConn of

443 case mConn of

404− Just conn -> return conn

444+ Just conn -> return . Just $ conn

405− Nothing -> createConnectionTo node pid

445+ Nothing -> createConnectionTo pid

406−

446+

407−createConnectionTo :: LocalNode -> ProcessId -> IO NT.Connection

447+createConnectionTo :: ProcessId -> Process (Maybe NT.Connection)

408−createConnectionTo node pid = do

448+createConnectionTo pid = do

449+ proc <- ask

409− mConn <- NT.connect (localEndPoint node)

450+ mConn <- liftIO $ NT.connect (localEndPoint . processNode $ proc)

410− (processAddress pid)

451+ (processAddress pid)

411− NT.ReliableOrdered

452+ NT.ReliableOrdered

412− NT.defaultConnectHints

453+ NT.defaultConnectHints

413 case mConn of

454 case mConn of

414 Right conn -> do

455 Right conn -> do

415− mConn' <- modifyMVar (localState node) $ \st ->

456+ mConn' <- liftIO $ modifyMVar (processState proc) $ \st ->

416− case st ^. localConnectionTo pid of

457+ case st ^. connectionTo pid of

417 Just conn' -> return (st, Just conn')

458 Just conn' -> return (st, Just conn')

418− Nothing -> return ( localConnectionTo pid ^= Just conn $ st

459+ Nothing -> return (connectionTo pid ^= Just conn $ st, Nothing)

419− , Nothing

420− )

421 case mConn' of

460 case mConn' of

422 Just conn' -> do

461 Just conn' -> do

423 -- Somebody else already created a connection while we weren't looking

462 -- Somebody else already created a connection while we weren't looking

@@ -425,27 +464,30 @@ createConnectionTo node pid = do

425 -- connection because that would limit concurrency too much, and

464 -- connection because that would limit concurrency too much, and

426 -- since Network.Transport supports lgihtweight connections creating

465 -- since Network.Transport supports lgihtweight connections creating

427 -- an unnecessary connection now and then is cheap anyway)

466 -- an unnecessary connection now and then is cheap anyway)

428− NT.close conn

467+ liftIO $ NT.close conn

429− return conn'

468+ return . Just $ conn'

430 Nothing -> do

469 Nothing -> do

431− sendBinary node pid conn $ lpidToPayload (processLocalId pid)

470+ sendBinary pid conn $ lpidToPayload (processLocalId pid)

432− return conn

471+ return . Just $ conn

433− Left err ->

472+ Left _err -> do

434− throwIO err

473+ -- TODO: should probably pass this error to remoteProcessFailed

435−

474+ remoteProcessFailed pid

436−sendBinary :: LocalNode

475+ return Nothing

437− -> ProcessId

476+

438− -> NT.Connection

439− -> [BSS.ByteString]

477+sendBinary :: ProcessId -> NT.Connection -> [BSS.ByteString] -> Process ()

440− -> IO ()

441−sendBinary node pid conn msg = do

478+sendBinary pid conn payload = do

442− result <- NT.send conn msg

479+ result <- liftIO $ NT.send conn payload

443 case result of

480 case result of

444 Right () -> return ()

481 Right () -> return ()

445 Left err -> do

482 Left err -> do

483+ ourState <- processState <$> ask

446− modifyMVar_ (localState node) $

484+ liftIO $ modifyMVar_ ourState $

447− return . (localConnectionTo pid ^= Nothing)

485+ return . (connectionTo pid ^= Nothing)

448− throwIO err

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

449

491

450payloadToMessage :: [BSS.ByteString] -> Message

492payloadToMessage :: [BSS.ByteString] -> Message

451payloadToMessage payload = Message fp msg

493payloadToMessage payload = Message fp msg

@@ -464,7 +506,31 @@ payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss

464 in LocalProcessId { lpidCounter = NTI.decodeInt32 bs1

506 in LocalProcessId { lpidCounter = NTI.decodeInt32 bs1

465 , lpidUnique = NTI.decodeInt32 bs2

507 , lpidUnique = NTI.decodeInt32 bs2

466 }

508 }

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+

468--------------------------------------------------------------------------------

534--------------------------------------------------------------------------------

469-- Binary instances --

535-- Binary instances --

470--------------------------------------------------------------------------------

536--------------------------------------------------------------------------------

@@ -481,9 +547,6 @@ instance Binary ProcessId where

481-- Accessors --

547-- Accessors --

482--------------------------------------------------------------------------------

548--------------------------------------------------------------------------------

483

549

484−localConnections :: Accessor LocalNodeState (Map ProcessId NT.Connection)

485−localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })

486−

487localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess)

550localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess)

488localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })

551localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })

489

552

@@ -493,8 +556,17 @@ localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter =

493localPidUnique :: Accessor LocalNodeState Int32

556localPidUnique :: Accessor LocalNodeState Int32

494localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })

557localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })

495

558

496−localConnectionTo :: ProcessId -> Accessor LocalNodeState (Maybe NT.Connection)

497−localConnectionTo pid = localConnections >>> DAC.mapMaybe pid

498−

499localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)

559localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)

500localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid

560localProcessWithId 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

distributed-process/tests/TestCH.hs+61−5

@@ -1,5 +1,6 @@

1module Main where

3+import Prelude hiding (catch)

3import Data.Binary (Binary)

4import Data.Binary (Binary)

4import Data.Typeable (Typeable)

5import Data.Typeable (Typeable)

5import Control.Concurrent (forkIO)

6import Control.Concurrent (forkIO)

@@ -8,8 +9,10 @@ import Control.Concurrent.MVar ( newEmptyMVar

8 , takeMVar

9 , takeMVar

9 , readMVar

10 , readMVar

10 )

11 )

12+import Control.Exception (catch)

11import Control.Monad (replicateM_)

13import Control.Monad (replicateM_)

12import Control.Distributed.Process

14import Control.Distributed.Process

15+import qualified Network.Transport as NT (Transport)

13import Network.Transport.TCP (createTransport, defaultTCPParameters)

16import Network.Transport.TCP (createTransport, defaultTCPParameters)

15newtype Ping = Ping ProcessId

18newtype Ping = Ping ProcessId

@@ -18,6 +21,7 @@ newtype Ping = Ping ProcessId

18newtype Pong = Pong ProcessId

21newtype Pong = Pong ProcessId

19 deriving (Typeable, Binary, Show)

22 deriving (Typeable, Binary, Show)

24+-- | The ping server from the paper

21ping :: Process ()

25ping :: Process ()

22ping = do

26ping = do

23 Pong partner <- expect

27 Pong partner <- expect

@@ -25,21 +29,20 @@ ping = do

25 send partner (Ping self)

29 send partner (Ping self)

26 ping

30 ping

32+-- | Basic ping test

28−testPing :: IO ()

33+testPing :: NT.Transport -> IO ()

29−testPing = do

34+testPing transport = do

30 serverAddr <- newEmptyMVar

35 serverAddr <- newEmptyMVar

31 clientDone <- newEmptyMVar

36 clientDone <- newEmptyMVar

33 -- Server

38 -- Server

34 forkIO $ do

39 forkIO $ do

35− Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

36 localNode <- newLocalNode transport

40 localNode <- newLocalNode transport

37 addr <- forkProcess localNode ping

41 addr <- forkProcess localNode ping

38 putMVar serverAddr addr

42 putMVar serverAddr addr

40 -- Client

44 -- Client

41 forkIO $ do

45 forkIO $ do

42− Right transport <- createTransport "127.0.0.1" "8081" defaultTCPParameters

43 localNode <- newLocalNode transport

46 localNode <- newLocalNode transport

44 pingServer <- readMVar serverAddr

47 pingServer <- readMVar serverAddr

@@ -58,5 +61,58 @@ testPing = do

59 takeMVar clientDone

62 takeMVar clientDone

64+-- | Set up monitoring, send first message to an unreachable node

65+testMonitor1 :: NT.Transport -> IO ()

66+testMonitor1 transport = do

67+ deadProcess <- newEmptyMVar

68+ done <- newEmptyMVar

69+

70+ forkIO $ do

71+ localNode <- newLocalNode transport

72+ addr <- forkProcess localNode $ return ()

73+ closeLocalNode localNode

74+ putMVar deadProcess addr

75+

76+ forkIO $ do

77+ localNode <- newLocalNode transport

78+ theirAddr <- readMVar deadProcess

79+ runProcess localNode $ do

80+ monitor theirAddr MaMonitor

81+ send theirAddr "Hi"

82+ ProcessMonitorException pid SrNoPing <- expect

83+ True <- return $ pid == theirAddr

84+ return ()

85+ putMVar done ()

86+

87+ takeMVar done

88+

89+-- Like 'testMonitor1', but throw an exception instead

90+testMonitor2 :: NT.Transport -> IO ()

91+testMonitor2 transport = do

92+ deadProcess <- newEmptyMVar

93+ done <- newEmptyMVar

94+

95+ forkIO $ do

96+ localNode <- newLocalNode transport

97+ addr <- forkProcess localNode $ return ()

98+ closeLocalNode localNode

99+ putMVar deadProcess addr

100+

101+ forkIO $ do

102+ localNode <- newLocalNode transport

103+ theirAddr <- readMVar deadProcess

104+ runProcess localNode $ do

105+ monitor theirAddr MaLink

106+ pcatch (send theirAddr "Hi") $ \(ProcessMonitorException pid SrNoPing) -> do

107+ True <- return $ pid == theirAddr

108+ return ()

109+ putMVar done ()

110+

111+ takeMVar done

112+

61main :: IO ()

113main :: IO ()

114+main = do

115+ Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

62−main = testPing

116+ -- testPing transport

117+ testMonitor1 transport

118+ testMonitor2 transport