35e512d · distributed-process

git.iko.soy/distributed-process

on master help sign in

Reorganize the code a little bit

35e512dEdsko de Vries14 years ago

distributed-process/distributed-process.cabal+2−1

compare with master

unifiedsplit

@@ -22,7 +22,8 @@ Library

22 network-transport >= 0.1

23 Exposed-modules: Control.Distributed.Process,

24 Control.Distributed.Process.Serializable,

25− Control.Distributed.Process.Closure

25+ Control.Distributed.Process.Closure,

26+ Control.Distributed.Process.Node

26 Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification

27 Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification

27 ghc-options: -Wall

28 ghc-options: -Wall

28 HS-Source-Dirs: src

29 HS-Source-Dirs: src

distributed-process/src/Control/Distributed/Process.hs+71−300

@@ -16,7 +16,6 @@ module Control.Distributed.Process

16 , send

17 , getSelfPid

18 , RemoteTable

19− , initRemoteTable

20 -- * Matching messages

19 -- * Matching messages

21 , Match

20 , Match

22 , match

21 , match

@@ -35,99 +34,61 @@ module Control.Distributed.Process

35 , DiedReason(..)

34 , DiedReason(..)

36 , DidUnmonitor(..)

35 , DidUnmonitor(..)

37 , DidUnlink(..)

36 , DidUnlink(..)

38− -- * Initialization

37+ -- * Closures

39− , newLocalNode

38+ , Closure

40− , forkProcess

39+ , Static

41− , runProcess

40+ , unClosure

42 -- * Auxiliary API

41 -- * Auxiliary API

43− , closeLocalNode

44 , catch

42 , catch

45 , expectTimeout

43 , expectTimeout

46 , spawnAsync

44 , spawnAsync

47 ) where

45 ) where

49import Prelude hiding (catch)

47import Prelude hiding (catch)

50−import System.IO (fixIO)

51−import qualified Data.ByteString as BSS (ByteString)

52−import qualified Data.ByteString.Lazy as BSL (fromChunks)

53import Data.Binary (decode)

48import Data.Binary (decode)

54−import Data.Map (Map)

49+import Data.Typeable (Typeable)

55−import qualified Data.Map as Map (empty, lookup, insert, delete)

50+import Data.Dynamic (fromDyn)

56−import qualified Data.List as List (delete)

51+import qualified Data.Map as Map ((!))

57−import Data.Set (Set)

58−import qualified Data.Set as Set (empty, insert, delete, member)

59−import Data.Foldable (forM_)

60−import Control.Monad (void)

61−import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

52+import Control.Monad.Reader (ask)

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

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

54+import qualified Control.Monad.Trans.Class as Trans (lift)

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

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

64−import Control.Exception (Exception, throwIO, SomeException)

56+import Control.Exception (Exception, throw)

65import qualified Control.Exception as Exception (catch)

57import qualified Control.Exception as Exception (catch)

66−import Control.Concurrent (forkIO)

58+import Control.Concurrent.MVar (modifyMVar)

67−import Control.Concurrent.MVar ( newMVar

68− , withMVar

69− , modifyMVar

70− , modifyMVar_

71− , newEmptyMVar

72− , putMVar

73− , takeMVar

74− )

75−import Control.Concurrent.Chan (newChan, writeChan)

59+import Control.Concurrent.Chan (writeChan)

76−import Control.Distributed.Process.Internal.CQueue (enqueue, dequeue, newCQueue)

60+import Control.Distributed.Process.Internal.CQueue (dequeue)

77import Control.Distributed.Process.Serializable (Serializable, fingerprint)

61import Control.Distributed.Process.Serializable (Serializable, fingerprint)

78−import qualified Network.Transport as NT ( Transport

79− , EndPoint

80− , Connection

81− , Reliability(ReliableOrdered)

82− , defaultConnectHints

83− , send

84− , connect

85− , close

86− , newEndPoint

87− , receive

88− , Event(..)

89− , EventErrorCode(..)

90− , TransportError(..)

91− , address

92− , closeEndPoint

93− , ConnectionId

94− )

95−import Data.Accessor ((^.), (^=), (^:))

62+import Data.Accessor ((^.), (^:))

96−import System.Random (randomIO)

97−import Control.Distributed.Process.Internal.NodeController (runNodeController)

63+import Control.Distributed.Process.Internal.Types

98−import Control.Distributed.Process.Internal ( RemoteTable

64+ ( RemoteTable

99− , initRemoteTable

100− , NodeId(..)

65+ , NodeId(..)

101− , LocalProcessId(..)

102− , ProcessId(..)

66+ , ProcessId(..)

103− , LocalNode(..)

67+ , LocalNode(..)

104− , LocalNodeState(..)

105− , LocalProcess(..)

68+ , LocalProcess(..)

106− , LocalProcessState(..)

69+ , Process(..)

107− , Process(..)

70+ , runLocalProcess

108− , Closure

71+ , Closure(..)

72+ , Static(..)

109− , Message(..)

73+ , Message(..)

110− , MonitorRef(..)

74+ , MonitorRef(..)

111− , MonitorNotification(..)

75+ , MonitorNotification(..)

112− , LinkException(..)

76+ , LinkException(..)

113− , DidUnmonitor(..)

77+ , DidUnmonitor(..)

114− , DidUnlink(..)

78+ , DidUnlink(..)

115− , DiedReason(..)

79+ , DiedReason(..)

116− , SpawnRef(..)

80+ , SpawnRef(..)

117− , NCMsg(..)

81+ , NCMsg(..)

118− , ProcessSignal(..)

82+ , ProcessSignal(..)

119− , createMessage

120− , messageToPayload

121− , payloadToMessage

122− , payloadToId

123− , idToPayload

124− , localPidCounter

125− , localPidUnique

126− , localProcessWithId

127− , monitorCounter

83+ , monitorCounter

128− , spawnCounter

84+ , spawnCounter

129− , connectionTo

85+ , MessageT

86+ , Closure(..)

87+ )

88+import Control.Distributed.Process.Internal.MessageT

89+ ( sendMessage

90+ , getLocalNode

130− )

91+ )

131

132-- INTERNAL NOTES

93-- INTERNAL NOTES

133--

94--

@@ -198,11 +159,9 @@ expect = receiveWait [match return]

198

159

199-- | Send a message

160-- | Send a message

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

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

201−send them msg = do

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

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

203− -- modify serializable to allow for stateful (IO) deserialization

163+-- modify serializable to allow for stateful (IO) deserialization

204− mConn <- getConnectionTo them

164+send them msg = lift $ sendMessage (Left them) msg

205− forM_ mConn $ \conn -> sendMessage them conn (createMessage msg)

206

165

207-- | Our own process ID

166-- | Our own process ID

208getSelfPid :: Process ProcessId

167getSelfPid :: Process ProcessId

@@ -268,20 +227,26 @@ receiveTimeout t ms = do

268 Just proc -> Just <$> proc

227 Just proc -> Just <$> proc

269

228

270--------------------------------------------------------------------------------

229--------------------------------------------------------------------------------

271−-- Auxiliary API --

230+-- Closures --

272--------------------------------------------------------------------------------

231--------------------------------------------------------------------------------

273

232

274−-- | Force-close a local node

233+-- | Deserialize a closure

275−--

234+unClosure :: Typeable a => Closure a -> Process a

276−-- TODO: for now we just close the associated endpoint

235+unClosure (Closure (Static label) env) = do

277−closeLocalNode :: LocalNode -> IO ()

236+ dec <- lookupStatic label

278−closeLocalNode node =

237+ return (dec env)

279− NT.closeEndPoint (localEndPoint node)

238+

239+--------------------------------------------------------------------------------

240+-- Auxiliary API --

241+--------------------------------------------------------------------------------

280

242

281-- | Catch exceptions within a process

243-- | Catch exceptions within a process

282catch :: Exception e => Process a -> (e -> Process a) -> Process a

244catch :: Exception e => Process a -> (e -> Process a) -> Process a

283catch p h = do

245catch p h = do

284− run <- runLocalProcess <$> ask

246+ node <- lift getLocalNode

247+ lproc <- ask

248+ let run :: Process a -> IO a

249+ run proc = runLocalProcess node proc lproc

285 liftIO $ Exception.catch (run p) (run . h)

250 liftIO $ Exception.catch (run p) (run . h)

286

251

287-- | Like 'expect' but with a timeout

252-- | Like 'expect' but with a timeout

@@ -297,210 +262,10 @@ spawnAsync _nid proc = do

297 postCtrlMsg (Spawn proc spawnRef)

262 postCtrlMsg (Spawn proc spawnRef)

298 return spawnRef

263 return spawnRef

299

264

300−--------------------------------------------------------------------------------

301−-- Initialization --

302−--------------------------------------------------------------------------------

303−

304−-- | Initialize a new local node.

305−--

306−-- Note that proper Cloud Haskell initialization and configuration is still

307−-- to do.

308−newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode

309−newLocalNode transport rtable = do

310− mEndPoint <- NT.newEndPoint transport

311− case mEndPoint of

312− Left ex -> throwIO ex

313− Right endPoint -> do

314− unq <- randomIO

315− state <- newMVar LocalNodeState

316− { _localProcesses = Map.empty

317− , _localPidCounter = 0

318− , _localPidUnique = unq

319− }

320− ctrlChan <- newChan

321− let node = LocalNode { localNodeId = NodeId $ NT.address endPoint

322− , localEndPoint = endPoint

323− , localState = state

324− , localCtrlChan = ctrlChan

325− , remoteTable = rtable

326− }

327− void . forkIO $ runNodeController node

328− void . forkIO $ handleIncomingMessages node

329− return node

330−

331−-- | Run a process on a local node and wait for it to finish

332−runProcess :: LocalNode -> Process () -> IO ()

333−runProcess node proc = do

334− done <- newEmptyMVar

335− void $ forkProcess node (proc >> liftIO (putMVar done ()))

336− takeMVar done

337−

338−-- | Spawn a new process on a local node

339−forkProcess :: LocalNode -> Process () -> IO ProcessId

340−forkProcess node proc = modifyMVar (localState node) $ \st -> do

341− let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter

342− , lpidUnique = st ^. localPidUnique

343− }

344− let pid = ProcessId { processNodeId = localNodeId node

345− , processLocalId = lpid

346− }

347− pst <- newMVar LocalProcessState { _connections = Map.empty

348− , _monitorCounter = 0

349− , _spawnCounter = 0

350− }

351− queue <- newCQueue

352− (_, lproc) <- fixIO $ \ ~(tid, _) -> do

353− let lproc = LocalProcess { processQueue = queue

354− , processNode = node

355− , processId = pid

356− , processState = pst

357− , processThread = tid

358− }

359− tid' <- forkIO $ do

360− reason <- Exception.catch

361− (runLocalProcess lproc proc >> return DiedNormal)

362− (return . DiedException . (show :: SomeException -> String))

363− -- [Unified: Table 4, rules termination and exiting]

364− modifyMVar_ (localState node) $

365− return . (localProcessWithId lpid ^= Nothing)

366− writeChan (localCtrlChan node) NCMsg

367− { ctrlMsgSender = Left pid

368− , ctrlMsgSignal = Died (Left pid) reason

369− }

370− return (tid', lproc)

371−

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

373− return ( (localProcessWithId lpid ^= Just lproc)

374− . (localPidCounter ^: (+ 1))

375− $ st

376− , pid

377− )

378−

379−handleIncomingMessages :: LocalNode -> IO ()

380−handleIncomingMessages node = go [] Map.empty Set.empty

381− where

382− go :: [NT.ConnectionId] -- ^ Connections whose purpose we don't yet know

383− -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes

384− -> Set NT.ConnectionId -- ^ Connections to our controller

385− -> IO ()

386− go uninitConns procConns ctrlConns = do

387− event <- NT.receive endpoint

388− case event of

389− NT.ConnectionOpened cid _rel _theirAddr ->

390− go (cid : uninitConns) procConns ctrlConns

391− NT.Received cid payload ->

392− case Map.lookup cid procConns of

393− Just proc -> do

394− let msg = payloadToMessage payload

395− enqueue (processQueue proc) msg

396− go uninitConns procConns ctrlConns

397− Nothing | cid `Set.member` ctrlConns -> do

398− writeChan ctrlChan (decode . BSL.fromChunks $ payload)

399− go uninitConns procConns ctrlConns

400− Nothing | cid `elem` uninitConns ->

401− case payloadToId payload of

402− Just lpid -> do

403− mProc <- withMVar state $ return . (^. localProcessWithId lpid)

404− case mProc of

405− Just proc ->

406− go (List.delete cid uninitConns)

407− (Map.insert cid proc procConns)

408− ctrlConns

409− Nothing ->

410− fail "handleIncomingMessages: TODO 1"

411− Nothing ->

412− go (List.delete cid uninitConns)

413− procConns

414− (Set.insert cid ctrlConns)

415− _ ->

416− fail "handleIncomingMessages: TODO 2"

417− NT.ConnectionClosed cid ->

418− go (List.delete cid uninitConns)

419− (Map.delete cid procConns)

420− (Set.delete cid ctrlConns)

421− NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) _) _) -> do

422− -- [Unified table 9, rule node_disconnect]

423− let nid = Right $ NodeId theirAddr

424− writeChan ctrlChan NCMsg

425− { ctrlMsgSender = nid

426− , ctrlMsgSignal = Died nid DiedDisconnect

427− }

428− NT.ErrorEvent _ ->

429− fail "handleIncomingMessages: TODO 4"

430− NT.EndPointClosed ->

431− return ()

432− NT.ReceivedMulticast _ _ ->

433− fail "Unexpected multicast"

434−

435− state = localState node

436− endpoint = localEndPoint node

437− ctrlChan = localCtrlChan node

438−

439--------------------------------------------------------------------------------

265--------------------------------------------------------------------------------

440-- Auxiliary functions --

266-- Auxiliary functions --

441--------------------------------------------------------------------------------

267--------------------------------------------------------------------------------

442

268

443−getConnectionTo :: ProcessId -> Process (Maybe NT.Connection)

444−getConnectionTo them = do

445− ourState <- processState <$> ask

446− mConn <- liftIO $ withMVar ourState $ return . (^. connectionTo them)

447− case mConn of

448− Just conn -> return . Just $ conn

449− Nothing -> createConnectionTo them

450−

451−createConnectionTo :: ProcessId -> Process (Maybe NT.Connection)

452−createConnectionTo them = do

453− proc <- ask

454− mConn <- liftIO $ NT.connect (localEndPoint . processNode $ proc)

455− (nodeAddress . processNodeId $ them)

456− NT.ReliableOrdered

457− NT.defaultConnectHints

458− case mConn of

459− Right conn -> do

460− mConn' <- liftIO $ modifyMVar (processState proc) $ \st ->

461− case st ^. connectionTo them of

462− Just conn' -> return (st, Just conn')

463− Nothing -> return (connectionTo them ^= Just conn $ st, Nothing)

464− case mConn' of

465− Just conn' -> do

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

467− -- (We don't want to keep localConnections locked while creating the

468− -- connection because that would limit concurrency too much, and

469− -- since Network.Transport supports lgihtweight connections creating

470− -- an unnecessary connection now and then is cheap anyway)

471− liftIO $ NT.close conn

472− return . Just $ conn'

473− Nothing -> do

474− sendBinary them conn . idToPayload . Just . processLocalId $ them

475− return . Just $ conn

476− Left _err -> do

477− -- TODO: should probably pass this error to remoteProcessFailed

478− node <- processNode <$> ask

479− liftIO $ remoteProcessFailed node them

480− return Nothing

481−

482−sendBinary :: ProcessId -> NT.Connection -> [BSS.ByteString] -> Process ()

483−sendBinary them conn payload = do

484− result <- liftIO $ NT.send conn payload

485− case result of

486− Right () ->

487− return ()

488− Left _ -> do

489− node <- processNode <$> ask

490− liftIO $ remoteProcessFailed node them

491−

492−sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()

493−sendMessage them conn = sendBinary them conn . messageToPayload

494−

495−remoteProcessFailed :: LocalNode -> ProcessId -> IO ()

496−remoteProcessFailed node them = do

497− -- [Unified: Table 9 rule node_disconnect]

498− let nid = Right (processNodeId them)

499− writeChan (localCtrlChan node) NCMsg

500− { ctrlMsgSender = nid

501− , ctrlMsgSignal = Died nid DiedDisconnect

502− }

503−

504getMonitorRefFor :: ProcessId -> Process MonitorRef

269getMonitorRefFor :: ProcessId -> Process MonitorRef

505getMonitorRefFor pid = do

270getMonitorRefFor pid = do

506 proc <- ask

271 proc <- ask

@@ -519,15 +284,21 @@ getSpawnRef = do

519 , SpawnRef counter

284 , SpawnRef counter

520 )

285 )

521

286

522−-- This is most definitely NOT exported

523−runLocalProcess :: LocalProcess -> Process a -> IO a

524−runLocalProcess lproc proc = runReaderT (unProcess proc) lproc

525−

526-- | Post a control message on the local node controller

287-- | Post a control message on the local node controller

527postCtrlMsg :: ProcessSignal -> Process ()

288postCtrlMsg :: ProcessSignal -> Process ()

528postCtrlMsg signal = do

289postCtrlMsg signal = do

529 us <- getSelfPid

290 us <- getSelfPid

530− ctrlChan <- localCtrlChan . processNode <$> ask

291+ ctrlChan <- localCtrlChan <$> lift getLocalNode

531 liftIO $ writeChan ctrlChan NCMsg { ctrlMsgSender = Left us

292 liftIO $ writeChan ctrlChan NCMsg { ctrlMsgSender = Left us

532 , ctrlMsgSignal = signal

293 , ctrlMsgSignal = signal

533 }

294 }

295+

296+lookupStatic :: Typeable a => String -> Process a

297+lookupStatic label = do

298+ rtable <- remoteTable <$> lift getLocalNode

299+ return $ fromDyn (rtable Map.! label) (throw typeError)

300+ where

301+ typeError = userError "lookupStatic type error"

302+

303+lift :: MessageT IO a -> Process a

304+lift = Process . Trans.lift

distributed-process/src/Control/Distributed/Process/Closure.hs+34−59

@@ -39,53 +39,45 @@

39--

40-- See Section 6, /Faking It/, of /Towards Haskell in the Cloud/ for more info.

41{-# LANGUAGE TemplateHaskell #-}

42−module Control.Distributed.Process.Closure

42+module Control.Distributed.Process.Closure (remotable, mkClosure) where

43− ( unClosure

44− , remotable

45− , mkClosure

46− ) where

48import Prelude hiding (lookup)

44import Prelude hiding (lookup)

49import Data.ByteString.Lazy (ByteString)

45import Data.ByteString.Lazy (ByteString)

50import Data.Binary (encode, decode)

46import Data.Binary (encode, decode)

51−import qualified Data.Map as Map (insert, (!))

47+import qualified Data.Map as Map (insert)

52−import Data.Dynamic (toDyn, fromDyn)

48+import Data.Dynamic (toDyn)

53−import Data.Typeable (Typeable)

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

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

55−import Control.Monad.Reader (ask)

50+import Language.Haskell.TH

56−import Control.Exception (throw)

57−import Language.Haskell.TH ( -- Q monad and operations

51+ ( -- Q monad and operations

58− Q

52+ Q

59− , reify

53+ , reify

60− -- Names

54+ -- Names

61− , Name

55+ , Name

62− , mkName

56+ , mkName

63− , nameBase

57+ , nameBase

64− -- Algebraic data types

58+ -- Algebraic data types

65− , Dec

59+ , Dec

66− , Exp

60+ , Exp

67− , Type(AppT, ArrowT)

61+ , Type(AppT, ArrowT)

68− , Info(VarI)

62+ , Info(VarI)

69− -- Lifted constructors

63+ -- Lifted constructors

70− -- .. Literals

64+ -- .. Literals

71− , stringL

65+ , stringL

72− -- .. Patterns

66+ -- .. Patterns

73− , normalB

67+ , normalB

74− , clause

68+ , clause

75− -- .. Expressions

69+ -- .. Expressions

76− , varE

70+ , varE

77− , litE

71+ , litE

78− -- .. Top-level declarations

72+ -- .. Top-level declarations

79− , funD

73+ , funD

80− , sigD

74+ , sigD

81− )

75+ )

82−import Control.Distributed.Process (Process)

83−import Control.Distributed.Process.Internal ( RemoteTable

76+import Control.Distributed.Process.Internal.Types

84− , LocalProcess(processNode)

77+ ( RemoteTable

85− , LocalNode(remoteTable)

86− , Closure(..)

78+ , Closure(..)

87− , Static(..)

79+ , Static(..)

88− )

80+ )

90--------------------------------------------------------------------------------

82--------------------------------------------------------------------------------

91-- Top-level API --

83-- Top-level API --

@@ -99,27 +91,10 @@ remotable ns = do

99 rtable <- createMetaData inserts

91 rtable <- createMetaData inserts

100 return $ concat closures ++ concat decoders ++ rtable

92 return $ concat closures ++ concat decoders ++ rtable

101

102−-- | Deserialize a closure

103−unClosure :: Typeable a => Closure a -> Process a

104−unClosure (Closure (Static label) env) = do

105− dec <- lookupStatic label

106− return (dec env)

107−

108-- | Create a closure

94-- | Create a closure

109mkClosure :: Name -> Q Exp

95mkClosure :: Name -> Q Exp

110mkClosure = varE . closureName

96mkClosure = varE . closureName

111

112−--------------------------------------------------------------------------------

113−-- Internal (Cloud Haskell state) --

114−--------------------------------------------------------------------------------

115−

116−lookupStatic :: Typeable a => String -> Process a

117−lookupStatic label = do

118− rtable <- remoteTable . processNode <$> ask

119− return $ fromDyn (rtable Map.! label) (throw typeError)

120− where

121− typeError = userError "lookupStatic type error"

122−

123--------------------------------------------------------------------------------

98--------------------------------------------------------------------------------

124-- Internal (Template Haskell) --

99-- Internal (Template Haskell) --

125--------------------------------------------------------------------------------

100--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Internal/NodeController.hs+72−147

@@ -7,56 +7,51 @@ import Data.Map (Map)

7import qualified Data.Map as Map (empty, toList)

8import Data.Set (Set)

9import qualified Data.Set as Set (empty, insert, delete)

10−import qualified Data.ByteString as BSS (ByteString)

11−import qualified Data.ByteString.Lazy as BSL (toChunks)

12−import Data.Binary (encode)

13import Data.Foldable (forM_)

10import Data.Foldable (forM_)

14import Data.Maybe (isJust)

11import Data.Maybe (isJust)

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

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

16−import qualified Data.Accessor.Container as DAC (mapMaybe, mapDefault)

13+import qualified Data.Accessor.Container as DAC (mapDefault)

17import Control.Category ((>>>))

14import Control.Category ((>>>))

18−import Control.Monad (when, unless, forever)

15+import Control.Monad (when, forever)

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

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

20import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)

17import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)

21−import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

18+import qualified Control.Monad.Trans.Class as Trans (lift)

22import Control.Concurrent.MVar (withMVar)

19import Control.Concurrent.MVar (withMVar)

23−import Control.Concurrent.Chan (readChan, writeChan)

20+import Control.Concurrent.Chan (readChan)

24import Control.Exception (Exception, throwTo)

21import Control.Exception (Exception, throwTo)

25−import qualified Network.Transport as NT ( EndPoint

26− , Connection

27− , Reliability(ReliableOrdered)

28− , defaultConnectHints

29− , send

30− , connect

31− )

32import Control.Distributed.Process.Serializable (Serializable)

22import Control.Distributed.Process.Serializable (Serializable)

33import Control.Distributed.Process.Internal.CQueue (enqueue)

23import Control.Distributed.Process.Internal.CQueue (enqueue)

34−import Control.Distributed.Process.Internal ( NodeId(..)

24+import Control.Distributed.Process.Internal.Types

25+ ( NodeId(..)

35− , LocalProcessId(..)

26+ , LocalProcessId(..)

36− , ProcessId(..)

27+ , ProcessId(..)

37− , LocalNode(..)

28+ , LocalNode(..)

38− , LocalProcess(..)

29+ , LocalProcess(..)

39− , MonitorRef(..)

30+ , MonitorRef(..)

40− , MonitorNotification(..)

31+ , MonitorNotification(..)

41− , LinkException(..)

32+ , LinkException(..)

42− , DidUnmonitor(..)

33+ , DidUnmonitor(..)

43− , DidUnlink(..)

34+ , DidUnlink(..)

44− , DiedReason(..)

35+ , DiedReason(..)

45− , Identifier

46− , NCMsg(..)

36+ , NCMsg(..)

47− , ProcessSignal(..)

37+ , ProcessSignal(..)

48− , createMessage

38+ , createMessage

49− , idToPayload

50− , localProcessWithId

39+ , localProcessWithId

40+ , MessageT

41+ , runMessageT

42+ )

43+import Control.Distributed.Process.Internal.MessageT

44+ ( sendBinary

45+ , getLocalNode

51− )

46+ )

53--------------------------------------------------------------------------------

48--------------------------------------------------------------------------------

54-- Top-level access to the node controller --

49-- Top-level access to the node controller --

55--------------------------------------------------------------------------------

50--------------------------------------------------------------------------------

57runNodeController :: LocalNode -> IO ()

52runNodeController :: LocalNode -> IO ()

58−runNodeController node =

53+runNodeController node =

59− evalStateT (runReaderT (unNC nodeController) node) initNCState

54+ runMessageT node (evalStateT (unNC nodeController) initNCState)

61--------------------------------------------------------------------------------

56--------------------------------------------------------------------------------

62-- Internal data types --

57-- Internal data types --

@@ -64,16 +59,21 @@ runNodeController node =

65data NCState = NCState

60data NCState = NCState

66 { -- Mapping from remote processes to linked local processes

61 { -- Mapping from remote processes to linked local processes

67− _links :: Map NodeId (Map LocalProcessId (Set ProcessId))

62+ _links :: Map NodeId (Map LocalProcessId (Set ProcessId))

68 -- Mapping from remote processes to monitoring local processes

63 -- Mapping from remote processes to monitoring local processes

69− , _monitors :: Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef)))

64+ , _monitors :: Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef)))

70− -- The node controller maintains its own set of connections

71− -- TODO: still not convinced that this is correct

72− , _connections :: Map Identifier NT.Connection

73 }

65 }

75−newtype NC a = NC { unNC :: ReaderT LocalNode (StateT NCState IO) a }

67+newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }

76− deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)

68+ deriving (Functor, Monad, MonadIO, MonadState NCState)

69+

70+lift :: MessageT IO a -> NC a

71+lift = NC . Trans.lift

72+

73+initNCState :: NCState

74+initNCState = NCState { _links = Map.empty

75+ , _monitors = Map.empty

76+ }

78--------------------------------------------------------------------------------

79-- Core functionality --

@@ -82,13 +82,13 @@ newtype NC a = NC { unNC :: ReaderT LocalNode (StateT NCState IO) a }

82-- [Unified: Table 7]

83nodeController :: NC ()

84nodeController = do

85− node <- ask

85+ node <- lift getLocalNode

86 forever $ do

87 msg <- liftIO $ readChan (localCtrlChan node)

89 -- [Unified: Table 7, rule nc_forward]

90 case destNid (ctrlMsgSignal msg) of

91− Just nid' | nid' /= localNodeId node -> sendCtrlMsg nid' msg

91+ Just nid' | nid' /= localNodeId node -> lift $ sendBinary (Right nid') msg

92 _ -> return ()

94 case msg of

@@ -106,15 +106,8 @@ nodeController = do

106 ncEffectProcessDied pid reason

107 NCMsg _from (Spawn _ _) ->

108 fail "spawn not implemented"

109− -- Error cases

109+ unexpected ->

110− NCMsg (Right _) (Link _) ->

110+ error $ "Unexpected message " ++ show unexpected

111− error "Link message from a node?"

112− NCMsg (Right _) (Monitor _ _) ->

113− error "Monitor message from a node?"

114− NCMsg (Right _) (Unlink _) ->

115− error "Unlink message from a node?"

116− NCMsg (Right _) (Unmonitor _) ->

117− error "Unmonitor message from a node?"

118

111

119-- [Unified: Table 10]

112-- [Unified: Table 10]

120ncEffectMonitor :: ProcessId -- ^ Who's watching?

113ncEffectMonitor :: ProcessId -- ^ Who's watching?

@@ -122,7 +115,7 @@ ncEffectMonitor :: ProcessId -- ^ Who's watching?

122 -> Maybe MonitorRef -- ^ 'Nothing' to link

115 -> Maybe MonitorRef -- ^ 'Nothing' to link

123 -> NC ()

116 -> NC ()

124ncEffectMonitor from them mRef = do

117ncEffectMonitor from them mRef = do

125− node <- ask

118+ node <- lift getLocalNode

126 shouldLink <-

119 shouldLink <-

127 if not (isLocal node them)

120 if not (isLocal node them)

128 then return True

121 then return True

@@ -136,7 +129,7 @@ ncEffectMonitor from them mRef = do

136 (False, True) -> -- [Unified: second rule]

129 (False, True) -> -- [Unified: second rule]

137 notifyDied from them DiedNoProc mRef

130 notifyDied from them DiedNoProc mRef

138 (False, False) -> -- [Unified: third rule]

131 (False, False) -> -- [Unified: third rule]

139− sendCtrlMsg (processNodeId from) NCMsg

132+ lift $ sendBinary (Right $ processNodeId from) NCMsg

140 { ctrlMsgSender = Right (localNodeId node)

133 { ctrlMsgSender = Right (localNodeId node)

141 , ctrlMsgSignal = Died (Left them) DiedNoProc

134 , ctrlMsgSignal = Died (Left them) DiedNoProc

142 }

135 }

@@ -144,21 +137,21 @@ ncEffectMonitor from them mRef = do

144-- [Unified: Table 11]

137-- [Unified: Table 11]

145ncEffectUnlink :: ProcessId -> ProcessId -> NC ()

138ncEffectUnlink :: ProcessId -> ProcessId -> NC ()

146ncEffectUnlink from them = do

139ncEffectUnlink from them = do

147− node <- ask

140+ node <- lift getLocalNode

148 when (isLocal node from) $ postMessage from $ DidUnlink them

141 when (isLocal node from) $ postMessage from $ DidUnlink them

149 modify $ linksForProcess them ^: Set.delete from

142 modify $ linksForProcess them ^: Set.delete from

150

143

151-- [Unified: Table 11]

144-- [Unified: Table 11]

152ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()

145ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()

153ncEffectUnmonitor from ref = do

146ncEffectUnmonitor from ref = do

154− node <- ask

147+ node <- lift getLocalNode

155 when (isLocal node from) $ postMessage from $ DidUnmonitor ref

148 when (isLocal node from) $ postMessage from $ DidUnmonitor ref

156 modify $ monitorsFor (monitorRefPid ref) from ^: Set.delete ref

149 modify $ monitorsFor (monitorRefPid ref) from ^: Set.delete ref

157

150

158-- [Unified: Table 12, bottom rule]

151-- [Unified: Table 12, bottom rule]

159ncEffectNodeDied :: NodeId -> DiedReason -> NC ()

152ncEffectNodeDied :: NodeId -> DiedReason -> NC ()

160ncEffectNodeDied nid reason = do

153ncEffectNodeDied nid reason = do

161− node <- ask

154+ node <- lift getLocalNode

162 lnks <- gets (^. linksForNode nid)

155 lnks <- gets (^. linksForNode nid)

163 mons <- gets (^. monitorsForNode nid)

156 mons <- gets (^. monitorsForNode nid)

164

157

@@ -195,20 +188,16 @@ ncEffectProcessDied pid reason = do

195 . (monitorsForProcess pid ^= Map.empty)

188 . (monitorsForProcess pid ^= Map.empty)

196

189

197--------------------------------------------------------------------------------

190--------------------------------------------------------------------------------

198−-- Connections --

191+-- Auxiliary --

199--------------------------------------------------------------------------------

192--------------------------------------------------------------------------------

200

193

201−-- | Check if a process is local to our own node

202−isLocal :: LocalNode -> ProcessId -> Bool

203−isLocal nid pid = processNodeId pid == localNodeId nid

204−

205notifyDied :: ProcessId -- ^ Who to notify?

194notifyDied :: ProcessId -- ^ Who to notify?

206 -> ProcessId -- ^ Who died?

195 -> ProcessId -- ^ Who died?

207 -> DiedReason -- ^ How did they die?

196 -> DiedReason -- ^ How did they die?

208 -> Maybe MonitorRef -- ^ 'Nothing' for linking

197 -> Maybe MonitorRef -- ^ 'Nothing' for linking

209 -> NC ()

198 -> NC ()

210notifyDied dest src reason mRef = do

199notifyDied dest src reason mRef = do

211− node <- ask

200+ node <- lift getLocalNode

212 case (isLocal node dest, mRef) of

201 case (isLocal node dest, mRef) of

213 (True, Just ref) ->

202 (True, Just ref) ->

214 postMessage dest $ MonitorNotification ref src reason

203 postMessage dest $ MonitorNotification ref src reason

@@ -217,11 +206,29 @@ notifyDied dest src reason mRef = do

217 (False, _) ->

206 (False, _) ->

218 -- TODO: why the change in sender? How does that affect 'reconnect' semantics?

207 -- TODO: why the change in sender? How does that affect 'reconnect' semantics?

219 -- (see [Unified: Table 10]

208 -- (see [Unified: Table 10]

220− sendCtrlMsg (processNodeId dest) NCMsg

209+ lift $ sendBinary (Right $ processNodeId dest) NCMsg

221 { ctrlMsgSender = Right (localNodeId node)

210 { ctrlMsgSender = Right (localNodeId node)

222 , ctrlMsgSignal = Died (Left src) reason

211 , ctrlMsgSignal = Died (Left src) reason

223 }

212 }

224

213

214+-- | [Unified: Table 8]

215+destNid :: ProcessSignal -> Maybe NodeId

216+destNid (Link pid) = Just . processNodeId $ pid

217+destNid (Unlink pid) = Just . processNodeId $ pid

218+destNid (Monitor pid _) = Just . processNodeId $ pid

219+destNid (Unmonitor ref) = Just . processNodeId . monitorRefPid $ ref

220+destNid (Died (Right _) _) = Nothing

221+destNid (Died (Left _pid) _) = fail "destNid: TODO"

222+destNid (Spawn _ _) = Nothing

223+

224+-- | Check if a process is local to our own node

225+isLocal :: LocalNode -> ProcessId -> Bool

226+isLocal nid pid = processNodeId pid == localNodeId nid

227+

228+--------------------------------------------------------------------------------

229+-- Messages to local processes --

230+--------------------------------------------------------------------------------

231+

225postMessage :: Serializable a => ProcessId -> a -> NC ()

232postMessage :: Serializable a => ProcessId -> a -> NC ()

226postMessage pid msg = withLocalProc pid $ \p ->

233postMessage pid msg = withLocalProc pid $ \p ->

227 enqueue (processQueue p) (createMessage msg)

234 enqueue (processQueue p) (createMessage msg)

@@ -232,7 +239,7 @@ throwException pid e = withLocalProc pid $ \p ->

232

239

233withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()

240withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()

234withLocalProc pid p = do

241withLocalProc pid p = do

235− node <- ask

242+ node <- lift getLocalNode

236 liftIO $ do

243 liftIO $ do

237 -- By [Unified: table 6, rule missing_process] messages to dead processes

244 -- By [Unified: table 6, rule missing_process] messages to dead processes

238 -- can silently be dropped

245 -- can silently be dropped

@@ -240,82 +247,6 @@ withLocalProc pid p = do

240 mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)

247 mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)

241 forM_ mProc p

248 forM_ mProc p

242

249

243−sendTo :: Identifier -> [BSS.ByteString] -> NC ()

244−sendTo them payload = do

245− mConn <- connTo them

246− didSend <- case mConn of

247− Just conn -> do

248− didSend <- liftIO $ NT.send conn payload

249− case didSend of

250− Left _ -> return False

251− Right () -> return True

252− Nothing ->

253− return False

254− unless didSend $ do

255− -- [Unified: Table 9, rule node_disconnect]

256− node <- ask

257− liftIO . writeChan (localCtrlChan node) $ NCMsg

258− { ctrlMsgSender = them

259− , ctrlMsgSignal = Died them DiedDisconnect

260− }

261−

262−sendCtrlMsg :: NodeId -> NCMsg -> NC ()

263−sendCtrlMsg dest = sendTo (Right dest) . BSL.toChunks . encode

264−

265−connTo :: Identifier -> NC (Maybe NT.Connection)

266−connTo them = do

267− mConn <- gets (^. connectionTo them)

268− case mConn of

269− Just conn -> return (Just conn)

270− Nothing -> createConnTo them

271−

272−createConnTo :: Identifier -> NC (Maybe NT.Connection)

273−createConnTo them = do

274− node <- ask

275− mConn <- liftIO $ NT.connect (localEndPoint node)

276− addr

277− NT.ReliableOrdered

278− NT.defaultConnectHints

279− case mConn of

280− Right conn -> do

281− didSend <- liftIO $ NT.send conn firstMsg

282− case didSend of

283− Left _ ->

284− return Nothing

285− Right () -> do

286− modify $ connectionTo them ^= Just conn

287− return $ Just conn

288− Left _ ->

289− return Nothing

290− where

291− (addr, firstMsg) = case them of

292− Left pid -> ( nodeAddress (processNodeId pid)

293− , idToPayload (Just $ processLocalId pid)

294− )

295− Right nid -> ( nodeAddress nid

296− , idToPayload Nothing

297− )

298−

299−--------------------------------------------------------------------------------

300−-- Auxiliary --

301−--------------------------------------------------------------------------------

302−

303−destNid :: ProcessSignal -> Maybe NodeId

304−destNid (Link pid) = Just . processNodeId $ pid

305−destNid (Unlink pid) = Just . processNodeId $ pid

306−destNid (Monitor pid _) = Just . processNodeId $ pid

307−destNid (Unmonitor ref) = Just . processNodeId . monitorRefPid $ ref

308−destNid (Died (Right _) _) = Nothing

309−destNid (Died (Left _pid) _) = fail "destNid: TODO"

310−destNid (Spawn _ _) = Nothing

311−

312−initNCState :: NCState

313−initNCState = NCState

314− { _links = Map.empty

315− , _monitors = Map.empty

316− , _connections = Map.empty

317− }

318−

319--------------------------------------------------------------------------------

250--------------------------------------------------------------------------------

320-- Accessors --

251-- Accessors --

321--------------------------------------------------------------------------------

252--------------------------------------------------------------------------------

@@ -326,12 +257,6 @@ links = accessor _links (\lnks st -> st { _links = lnks })

326monitors :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))

257monitors :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))

327monitors = accessor _monitors (\mons st -> st { _monitors = mons })

258monitors = accessor _monitors (\mons st -> st { _monitors = mons })

328

259

329−connections :: Accessor NCState (Map Identifier NT.Connection)

330−connections = accessor _connections (\conns st -> st { _connections = conns })

331−

332−connectionTo :: Identifier -> Accessor NCState (Maybe NT.Connection)

333−connectionTo them = connections >>> DAC.mapMaybe them

334−

335linksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))

260linksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))

336linksForNode nid = links >>> DAC.mapDefault Map.empty nid

261linksForNode nid = links >>> DAC.mapDefault Map.empty nid

337

262

distributed-process/src/Control/Distributed/Process/Internal.hs → distributed-process/src/Control/Distributed/Process/Internal/Types.hsrenamed+38−13

@@ -1,4 +1,6 @@

1+-- | We collect all types used internally in a single module to avoid using

2+-- mutually recursive modules

1−module Control.Distributed.Process.Internal

3+module Control.Distributed.Process.Internal.Types

2 ( -- * Global CH state

4 ( -- * Global CH state

3 RemoteTable

5 RemoteTable

4 , initRemoteTable

6 , initRemoteTable

@@ -13,6 +15,7 @@ module Control.Distributed.Process.Internal

13 , LocalProcess(..)

15 , LocalProcess(..)

14 , LocalProcessState(..)

16 , LocalProcessState(..)

15 , Process(..)

17 , Process(..)

18+ , runLocalProcess

16 -- * Closures

19 -- * Closures

17 , Static(..)

20 , Static(..)

18 , Closure(..)

21 , Closure(..)

@@ -40,10 +43,12 @@ module Control.Distributed.Process.Internal

40 , localPidCounter

43 , localPidCounter

41 , localPidUnique

44 , localPidUnique

42 , localProcessWithId

45 , localProcessWithId

43− , connections

44 , monitorCounter

46 , monitorCounter

45 , spawnCounter

47 , spawnCounter

46− , connectionTo

48+ -- * MessageT monad

49+ , MessageT(..)

50+ , MessageState(..)

51+ , runMessageT

47 ) where

52 ) where

49import Control.Concurrent (ThreadId)

54import Control.Concurrent (ThreadId)

@@ -65,8 +70,9 @@ import qualified Data.ByteString as BSS (ByteString, concat, splitAt)

65import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

70import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

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

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

67import Control.Applicative ((<$>), (<*>))

72import Control.Applicative ((<$>), (<*>))

68−import Control.Monad.Reader (MonadReader(..), ReaderT)

73+import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

69import Control.Monad.IO.Class (MonadIO)

74import Control.Monad.IO.Class (MonadIO)

75+import Control.Monad.State (MonadState, StateT, evalStateT)

70import Control.Distributed.Process.Serializable ( Fingerprint

76import Control.Distributed.Process.Serializable ( Fingerprint

71 , Serializable

77 , Serializable

72 , encodeFingerprint

78 , encodeFingerprint

@@ -146,7 +152,6 @@ data LocalNodeState = LocalNodeState

146-- | Processes running on our local node

152-- | Processes running on our local node

147data LocalProcess = LocalProcess

153data LocalProcess = LocalProcess

148 { processQueue :: CQueue Message

154 { processQueue :: CQueue Message

149− , processNode :: LocalNode

150 , processId :: ProcessId

155 , processId :: ProcessId

151 , processState :: MVar LocalProcessState

156 , processState :: MVar LocalProcessState

152 , processThread :: ThreadId

157 , processThread :: ThreadId

@@ -154,15 +159,20 @@ data LocalProcess = LocalProcess

154

159

155-- | Local process state

160-- | Local process state

156data LocalProcessState = LocalProcessState

161data LocalProcessState = LocalProcessState

157− { _connections :: Map ProcessId NT.Connection

158− , _monitorCounter :: Int32

162+ { _monitorCounter :: Int32

159 , _spawnCounter :: Int32

163 , _spawnCounter :: Int32

160 }

164 }

161

165

162-- | The Cloud Haskell 'Process' type

166-- | The Cloud Haskell 'Process' type

167+newtype Process a = Process {

163−newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }

168+ unProcess :: ReaderT LocalProcess (MessageT IO) a

169+ }

164 deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable)

170 deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable)

165

171

172+-- | Deconstructor for 'Process' (not exported to the public API)

173+runLocalProcess :: LocalNode -> Process a -> LocalProcess -> IO a

174+runLocalProcess node proc = runMessageT node . runReaderT (unProcess proc)

175+

166--------------------------------------------------------------------------------

176--------------------------------------------------------------------------------

167-- Closures --

177-- Closures --

168--------------------------------------------------------------------------------

178--------------------------------------------------------------------------------

@@ -377,16 +387,31 @@ localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq

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

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

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

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

379

389

380−connections :: Accessor LocalProcessState (Map ProcessId NT.Connection)

381−connections = accessor _connections (\conns st -> st { _connections = conns })

382−

383monitorCounter :: Accessor LocalProcessState Int32

390monitorCounter :: Accessor LocalProcessState Int32

384monitorCounter = accessor _monitorCounter (\cnt st -> st { _monitorCounter = cnt })

391monitorCounter = accessor _monitorCounter (\cnt st -> st { _monitorCounter = cnt })

385

392

386spawnCounter :: Accessor LocalProcessState Int32

393spawnCounter :: Accessor LocalProcessState Int32

387spawnCounter = accessor _spawnCounter (\cnt st -> st { _spawnCounter = cnt })

394spawnCounter = accessor _spawnCounter (\cnt st -> st { _spawnCounter = cnt })

388

395

389−connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)

396+--------------------------------------------------------------------------------

390−connectionTo pid = connections >>> DAC.mapMaybe pid

397+-- MessageT monad --

398+--------------------------------------------------------------------------------

399+

400+newtype MessageT m a = MessageT { unMessageT :: StateT MessageState m a }

401+ deriving (Functor, Monad, MonadIO, MonadState MessageState)

402+

403+data MessageState = MessageState {

404+ messageLocalNode :: LocalNode

405+ , _messageConnections :: Map Identifier NT.Connection

406+ }

391

407

408+runMessageT :: Monad m => LocalNode -> MessageT m a -> m a

409+runMessageT localNode m =

410+ evalStateT (unMessageT m) $ initMessageState localNode

411+

412+initMessageState :: LocalNode -> MessageState

413+initMessageState localNode = MessageState {

414+ messageLocalNode = localNode

415+ , _messageConnections = Map.empty

416+ }

392

417

distributed-process/tests/TestCH.hs+9−7

@@ -5,12 +5,13 @@ import Data.Binary (Binary(..))

5import Data.Typeable (Typeable)

6import Data.Foldable (forM_)

7import Control.Concurrent (forkIO, threadDelay)

8−import Control.Concurrent.MVar ( MVar

8+import Control.Concurrent.MVar

9+ ( MVar

9− , newEmptyMVar

10+ , newEmptyMVar

10− , putMVar

11+ , putMVar

11− , takeMVar

12+ , takeMVar

12− , readMVar

13+ , readMVar

13− )

14+ )

14import Control.Monad (replicateM_)

15import Control.Monad (replicateM_)

15import Control.Monad.IO.Class (liftIO)

16import Control.Monad.IO.Class (liftIO)

16import Control.Exception (throwIO)

17import Control.Exception (throwIO)

@@ -18,7 +19,8 @@ import Control.Applicative ((<$>), (<*>))

18import qualified Network.Transport as NT (Transport, closeEndPoint)

19import qualified Network.Transport as NT (Transport, closeEndPoint)

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

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

20import Control.Distributed.Process

21import Control.Distributed.Process

21−import Control.Distributed.Process.Internal (LocalNode(localEndPoint))

22+import Control.Distributed.Process.Internal.Types (LocalNode(localEndPoint))

23+import Control.Distributed.Process.Node

22import TestAuxiliary

24import TestAuxiliary

24newtype Ping = Ping ProcessId

26newtype Ping = Ping ProcessId

distributed-process/tests/TestClosure.hs+1

@@ -7,6 +7,7 @@ import Network.Transport (Transport)

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

8import Control.Distributed.Process

9import Control.Distributed.Process.Closure

10+import Control.Distributed.Process.Node

10import TestAuxiliary

11import TestAuxiliary

12addInt :: Int -> Int -> Int

13addInt :: Int -> Int -> Int