master...ca136cb · distributed-process

git.iko.soy/distributed-process

on ca136cb help sign in

distributed-process.cabal+5−1

since splitsnapshots

unifiedsplit

@@ -54,7 +54,10 @@ Library

54 ghc-prim >= 0.2 && < 0.5,

55 distributed-static >= 0.2 && < 0.4,

56 rank1dynamic >= 0.1 && < 0.4,

57− syb >= 0.3 && < 0.7

57+ syb >= 0.3 && < 0.7,

58+ transformers-base >= 0.3 && < 0.5,

59+ monad-control >= 0.4,

60+ exceptions >= 0.5

58 Exposed-modules: Control.Distributed.Process,

61 Exposed-modules: Control.Distributed.Process,

59 Control.Distributed.Process.Closure,

62 Control.Distributed.Process.Closure,

60 Control.Distributed.Process.Debug,

63 Control.Distributed.Process.Debug,

@@ -67,6 +70,7 @@ Library

67 Control.Distributed.Process.Internal.StrictContainerAccessors,

70 Control.Distributed.Process.Internal.StrictContainerAccessors,

68 Control.Distributed.Process.Internal.StrictList,

71 Control.Distributed.Process.Internal.StrictList,

69 Control.Distributed.Process.Internal.StrictMVar,

72 Control.Distributed.Process.Internal.StrictMVar,

73+ Control.Distributed.Process.Internal.ThreadPool,

70 Control.Distributed.Process.Internal.Types,

74 Control.Distributed.Process.Internal.Types,

71 Control.Distributed.Process.Internal.WeakTQueue,

75 Control.Distributed.Process.Internal.WeakTQueue,

72 Control.Distributed.Process.Management,

76 Control.Distributed.Process.Management,

src/Control/Distributed/Process.hs+26−4

@@ -166,6 +166,7 @@ import Prelude hiding (catch)

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

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

168import Control.Monad.Reader (ask)

169+import Control.Concurrent (killThread)

169import Control.Concurrent.MVar

170import Control.Concurrent.MVar

170 ( MVar

171 ( MVar

171 , newEmptyMVar

172 , newEmptyMVar

@@ -200,6 +201,7 @@ import Control.Distributed.Process.Internal.Types

200 , RegisterReply(..)

201 , RegisterReply(..)

201 , LocalProcess(processNode)

202 , LocalProcess(processNode)

202 , Message

203 , Message

204+ , localProcessWithId

203 )

205 )

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

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

205import Control.Distributed.Process.Internal.Primitives

207import Control.Distributed.Process.Internal.Primitives

@@ -307,6 +309,10 @@ import Control.Distributed.Process.Internal.Primitives

307 , reconnectPort

309 , reconnectPort

308 )

310 )

309import Control.Distributed.Process.Node (forkProcess)

311import Control.Distributed.Process.Node (forkProcess)

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

313+ ( processThread

314+ , withValidLocalState

315+ )

310import Control.Distributed.Process.Internal.Spawn

316import Control.Distributed.Process.Internal.Spawn

311 ( -- Spawning Processes/Channels

317 ( -- Spawning Processes/Channels

312 spawn

318 spawn

@@ -316,7 +322,11 @@ import Control.Distributed.Process.Internal.Spawn

316 , spawnSupervised

322 , spawnSupervised

317 , call

323 , call

318 )

324 )

319−import Control.Exception (SomeException, throw)

325+import Control.Exception (SomeException, throwIO, uninterruptibleMask_)

326+import qualified Control.Exception as Exception (onException)

327+import Control.Monad (forM_)

328+import Data.Accessor ((^.))

329+

320

330

321-- INTERNAL NOTES

331-- INTERNAL NOTES

322--

332--

@@ -416,6 +426,18 @@ callLocal :: Process a -> Process a

416callLocal proc = mask $ \release -> do

426callLocal proc = mask $ \release -> do

417 mv <- liftIO newEmptyMVar :: Process (MVar (Either SomeException a))

427 mv <- liftIO newEmptyMVar :: Process (MVar (Either SomeException a))

418 child <- spawnLocal $ try (release proc) >>= liftIO . putMVar mv

428 child <- spawnLocal $ try (release proc) >>= liftIO . putMVar mv

419− rs <- liftIO (takeMVar mv) `onException`

429+ lproc <- ask

420− (kill child "exception in parent process" >> liftIO (takeMVar mv))

430+ liftIO $ do

431+ rs <- Exception.onException (takeMVar mv) $ uninterruptibleMask_ $

432+ -- Exceptions need to be prevented from interrupting the clean up or

433+ -- the original exception which caused entering the handler could be

434+ -- forgotten. For instance, this could have a problematic effect

435+ -- when the original exception was meant to kill the thread and the

436+ -- second exception doesn't (like the exception thrown by

437+ -- 'System.Timeout.timeout').

438+ do mchildThreadId <- withValidLocalState (processNode lproc) $

439+ \vst -> return $ fmap processThread $

440+ vst ^. localProcessWithId (processLocalId child)

441+ forM_ mchildThreadId killThread

442+ takeMVar mv

421− either throw return rs

443+ either throwIO return rs

src/Control/Distributed/Process/Internal/Messaging.hs+4−4

@@ -18,7 +18,7 @@ import Control.Distributed.Process.Serializable ()

18import Control.Concurrent (forkIO)

19import Control.Concurrent.Chan (writeChan)

20import Control.Exception (mask_)

21−import Control.Monad (unless)

21+import Control.Monad (unless, void)

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

23import Control.Monad.Reader (ask)

24import qualified Network.Transport as NT

@@ -30,7 +30,7 @@ import qualified Network.Transport as NT

30 , close

31 )

32import Control.Distributed.Process.Internal.Types

33− ( LocalNode(localState, localEndPoint, localCtrlChan)

33+ ( LocalNode(localEndPoint, localCtrlChan)

34 , withValidLocalState

35 , modifyValidLocalState

36 , modifyValidLocalState_

@@ -146,9 +146,9 @@ disconnect node from to = mask_ $ do

146 return (vst, return ())

147 Just (conn, _) -> do

148 return ( localConnectionBetween from to ^= Nothing $ vst

149− , NT.close conn

149+ , void $ forkIO $ NT.close conn

150 )

151− forM_ mio forkIO

151+ forM_ mio id

152

153closeImplicitReconnections :: LocalNode -> Identifier -> IO ()

154closeImplicitReconnections node to = mask_ $ do

src/Control/Distributed/Process/Internal/Primitives.hs+8−61

@@ -137,9 +137,8 @@ import System.Timeout (timeout)

137import Control.Monad (when)

138import Control.Monad.Reader (ask)

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

140+import Control.Monad.Catch hiding (Handler, catches)

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

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

141−import Control.Exception (Exception(..), throw, throwIO, SomeException)

142−import qualified Control.Exception as Ex (catch, mask, mask_, try)

143import Control.Distributed.Process.Internal.StrictMVar

142import Control.Distributed.Process.Internal.StrictMVar

144 ( StrictMVar

143 ( StrictMVar

145 , modifyMVar

144 , modifyMVar

@@ -699,7 +698,7 @@ instance Exception ProcessTerminationException

699

698

700-- | Terminate immediately (throws a ProcessTerminationException)

699-- | Terminate immediately (throws a ProcessTerminationException)

701terminate :: Process a

700terminate :: Process a

702−terminate = liftIO $ throwIO ProcessTerminationException

701+terminate = throwM ProcessTerminationException

703

702

704-- [Issue #110]

703-- [Issue #110]

705-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.

704-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.

@@ -709,7 +708,7 @@ die reason = do

709-- could be decoded by a handler passed to 'catchExit', re-thrown or even

708-- could be decoded by a handler passed to 'catchExit', re-thrown or even

710-- passed to another process via the tracing mechanism.

709-- passed to another process via the tracing mechanism.

711 pid <- getSelfPid

710 pid <- getSelfPid

712− liftIO $ throwIO (ProcessExitException pid (createMessage reason))

711+ throwM (ProcessExitException pid (createMessage reason))

713

712

714-- | Forceful request to kill a process. Where 'exit' provides an exception

713-- | Forceful request to kill a process. Where 'exit' provides an exception

715-- that can be caught and handled, 'kill' throws an unexposed exception type

714-- that can be caught and handled, 'kill' throws an unexposed exception type

@@ -747,7 +746,7 @@ catchExit act exitHandler = catch act handleExit

747 handleExit ex@(ProcessExitException from msg) =

746 handleExit ex@(ProcessExitException from msg) =

748 if messageFingerprint msg == fingerprint (undefined :: a)

747 if messageFingerprint msg == fingerprint (undefined :: a)

749 then exitHandler from decoded

748 then exitHandler from decoded

750− else liftIO $ throwIO ex

749+ else throwM ex

751 where

750 where

752 decoded :: a

751 decoded :: a

753 -- Make sure the value is fully decoded so that we don't hang to

752 -- Make sure the value is fully decoded so that we don't hang to

@@ -771,7 +770,7 @@ catchesExit act handlers = catch act ((flip handleExit) handlers)

771 handleExit :: ProcessExitException

770 handleExit :: ProcessExitException

772 -> [(ProcessId -> Message -> Process (Maybe b))]

771 -> [(ProcessId -> Message -> Process (Maybe b))]

773 -> Process b

772 -> Process b

774− handleExit ex [] = liftIO $ throwIO ex

773+ handleExit ex [] = throwM ex

775 handleExit ex@(ProcessExitException from msg) (h:hs) = do

774 handleExit ex@(ProcessExitException from msg) (h:hs) = do

776 r <- h from msg

775 r <- h from msg

777 case r of

776 case r of

@@ -934,71 +933,19 @@ unmonitor ref = do

934-- Exception handling --

933-- Exception handling --

935--------------------------------------------------------------------------------

934--------------------------------------------------------------------------------

936

935

937−-- | Lift 'Control.Exception.catch'

938−catch :: Exception e => Process a -> (e -> Process a) -> Process a

939−catch p h = do

940− lproc <- ask

941− liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)

942−

943−-- | Lift 'Control.Exception.try'

944−try :: Exception e => Process a -> Process (Either e a)

945−try p = do

946− lproc <- ask

947− liftIO $ Ex.try (runLocalProcess lproc p)

948−

949−-- | Lift 'Control.Exception.mask'

950−mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b

951−mask p = do

952− lproc <- ask

953− liftIO $ Ex.mask $ \restore ->

954− runLocalProcess lproc (p (liftRestore restore))

955− where

956− liftRestore :: (forall a. IO a -> IO a)

957− -> (forall a. Process a -> Process a)

958− liftRestore restoreIO = \p2 -> do

959− ourLocalProc <- ask

960− liftIO $ restoreIO $ runLocalProcess ourLocalProc p2

961−

962−-- | Lift 'Control.Exception.mask_'

963−mask_ :: Process a -> Process a

964−mask_ p = do

965− lproc <- ask

966− liftIO $ Ex.mask_ $ runLocalProcess lproc p

967−

968−-- | Lift 'Control.Exception.onException'

969−onException :: Process a -> Process b -> Process a

970−onException p what = p `catch` \e -> do _ <- what

971− liftIO $ throwIO (e :: SomeException)

972−

973−-- | Lift 'Control.Exception.bracket'

974−bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c

975−bracket before after thing =

976− mask $ \restore -> do

977− a <- before

978− r <- restore (thing a) `onException` after a

979− _ <- after a

980− return r

981−

982−-- | Lift 'Control.Exception.bracket_'

983−bracket_ :: Process a -> Process b -> Process c -> Process c

984−bracket_ before after thing = bracket before (const after) (const thing)

985−

986−-- | Lift 'Control.Exception.finally'

987−finally :: Process a -> Process b -> Process a

988−finally a sequel = bracket_ (return ()) sequel a

989−

990-- | You need this when using 'catches'

936-- | You need this when using 'catches'

991data Handler a = forall e . Exception e => Handler (e -> Process a)

937data Handler a = forall e . Exception e => Handler (e -> Process a)

992

938

993instance Functor Handler where

939instance Functor Handler where

994 fmap f (Handler h) = Handler (fmap f . h)

940 fmap f (Handler h) = Handler (fmap f . h)

995

941

942+

996-- | Lift 'Control.Exception.catches'

943-- | Lift 'Control.Exception.catches'

997catches :: Process a -> [Handler a] -> Process a

944catches :: Process a -> [Handler a] -> Process a

998catches proc handlers = proc `catch` catchesHandler handlers

945catches proc handlers = proc `catch` catchesHandler handlers

999

946

1000catchesHandler :: [Handler a] -> SomeException -> Process a

947catchesHandler :: [Handler a] -> SomeException -> Process a

1001−catchesHandler handlers e = foldr tryHandler (throw e) handlers

948+catchesHandler handlers e = foldr tryHandler (throwM e) handlers

1002 where tryHandler (Handler handler) res

949 where tryHandler (Handler handler) res

1003 = case fromException e of

950 = case fromException e of

1004 Just e' -> handler e'

951 Just e' -> handler e'

@@ -1138,7 +1085,7 @@ unregister label = do

1138handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()

1085handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()

1139handleRegistrationReply label ok owner =

1086handleRegistrationReply label ok owner =

1140 when (not ok) $

1087 when (not ok) $

1141− liftIO $ throwIO $ ProcessRegistrationException label owner

1088+ throwM $ ProcessRegistrationException label owner

1142

1089

1143-- | Remove a process from a remote registry (asynchronous).

1090-- | Remove a process from a remote registry (asynchronous).

1144--

1091--

src/Control/Distributed/Process/Internal/ThreadPool.hsnew file+78

@@ -0,0 +1,78 @@

1+-- | An implementation of a pool of threads

2+--

3+{-# LANGUAGE RecursiveDo #-}

4+module Control.Distributed.Process.Internal.ThreadPool

5+ ( newThreadPool

6+ , submitTask

7+ , lookupWorker

8+ , ThreadPool

9+ ) where

10+

11+import Control.Exception

12+import Control.Monad

13+import Data.IORef

14+import qualified Data.Map as Map

15+

16+

17+-- | A pool of worker threads that execute tasks.

18+--

19+-- Each worker thread is named with a key @k@. Tasks are submitted to a

20+-- specific worker using its key. While the worker is busy the tasks are queued.

21+-- When there are no more queued tasks the worker ceases to exist.

22+--

23+-- The next time a task is submitted the worker will be respawned.

24+--

25+newtype ThreadPool k w = ThreadPool (IORef (Map.Map k (Maybe (IO ()), w)))

26+

27+-- Each worker has an entry in the map with a closure that contains all

28+-- queued actions fot it.

29+--

30+-- No entry in the map is kept for defunct workers.

31+

32+-- | Creates a pool with no workers.

33+newThreadPool :: IO (ThreadPool k w)

34+newThreadPool = fmap ThreadPool $ newIORef Map.empty

35+

36+-- | @submitTask pool fork k task@ submits a task for the worker @k@.

37+--

38+-- If worker @k@ is busy, then the task is queued until the worker is available.

39+--

40+-- If worker @k@ does not exist, then the given @fork@ operation is used to

41+-- spawn the worker. @fork@ returns whatever information is deemed useful for

42+-- later retrieval via 'lookupWorker'.

43+--

44+submitTask :: Ord k

45+ => ThreadPool k w

46+ -> (IO () -> IO w)

47+ -> k -> IO () -> IO ()

48+submitTask (ThreadPool mapRef) fork k task = mdo

49+ m' <- join $ atomicModifyIORef mapRef $ \m ->

50+ case Map.lookup k m of

51+ -- There is no worker for this key, create one.

52+ Nothing -> ( m'

53+ , do w <- fork $ flip onException terminateWorker $ do

54+ task

55+ continue

56+ return $ Map.insert k (Nothing, w) m

57+ )

58+ -- Queue an action for the existing worker.

59+ Just (mp, w) ->

60+ (m', return $ Map.insert k (Just $ maybe task (>> task) mp, w) m)

61+ return ()

62+ where

63+ continue = join $ atomicModifyIORef mapRef $ \m ->

64+ case Map.lookup k m of

65+ -- Execute the next batch of queued actions.

66+ Just (Just p, w) -> (Map.insert k (Nothing, w) m, p >> continue)

67+ -- There are no more queued actions. Terminate the worker.

68+ Just (Nothing, w) -> (Map.delete k m, return ())

69+ -- The worker key was removed already (?)

70+ Nothing -> (m, return ())

71+ -- Remove the worker key regardless of whether there are more queued

72+ -- actions.

73+ terminateWorker = atomicModifyIORef mapRef $ \m -> (Map.delete k m, ())

74+

75+-- | Looks up a worker with the given key.

76+lookupWorker :: Ord k => ThreadPool k w -> k -> IO (Maybe w)

77+lookupWorker (ThreadPool mapRef) k =

78+ atomicModifyIORef mapRef $ \m -> (m, fmap snd $ Map.lookup k m)

src/Control/Distributed/Process/Internal/Types.hs+48−4

@@ -1,9 +1,12 @@

1+{-# LANGUAGE CPP #-}

1{-# LANGUAGE DeriveDataTypeable #-}

2{-# LANGUAGE DeriveDataTypeable #-}

3+{-# LANGUAGE DeriveGeneric #-}

2{-# LANGUAGE ExistentialQuantification #-}

4{-# LANGUAGE ExistentialQuantification #-}

3{-# LANGUAGE GeneralizedNewtypeDeriving #-}

5{-# LANGUAGE GeneralizedNewtypeDeriving #-}

4{-# LANGUAGE GADTs #-}

6{-# LANGUAGE GADTs #-}

5−{-# LANGUAGE DeriveGeneric #-}

7+{-# LANGUAGE MultiParamTypeClasses #-}

8+{-# LANGUAGE TypeFamilies #-}

6−{-# LANGUAGE CPP #-}

9+{-# LANGUAGE Rank2Types #-}

8-- | Types used throughout the Cloud Haskell framework

11-- | Types used throughout the Cloud Haskell framework

9--

12--

@@ -112,11 +115,14 @@ import Control.Concurrent (ThreadId)

112import Control.Concurrent.Chan (Chan)

115import Control.Concurrent.Chan (Chan)

113import Control.Concurrent.STM (STM)

116import Control.Concurrent.STM (STM)

114import Control.Concurrent.STM.TChan (TChan)

117import Control.Concurrent.STM.TChan (TChan)

118+import Control.Monad.Catch (MonadThrow(..), MonadCatch(..), MonadMask(..))

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

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

116import Control.Applicative (Applicative, Alternative, (<$>), (<*>))

120import Control.Applicative (Applicative, Alternative, (<$>), (<*>))

117import Control.Monad.Fix (MonadFix)

121import Control.Monad.Fix (MonadFix)

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

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

119−import Control.Monad.IO.Class (MonadIO)

123+import Control.Monad.IO.Class (MonadIO(..))

124+import Control.Monad.Base (MonadBase(..))

125+import Control.Monad.Trans.Control (MonadBaseControl(..))

120import Control.Distributed.Process.Serializable

126import Control.Distributed.Process.Serializable

121 ( Fingerprint

127 ( Fingerprint

122 , Serializable

128 , Serializable

@@ -133,6 +139,7 @@ import Control.Distributed.Process.Internal.StrictMVar

133 , modifyMVar

139 , modifyMVar

134 , modifyMVar_

140 , modifyMVar_

135 )

141 )

142+import Control.Distributed.Process.Internal.ThreadPool

136import Control.Distributed.Process.Internal.WeakTQueue (TQueue)

143import Control.Distributed.Process.Internal.WeakTQueue (TQueue)

137import Control.Distributed.Static (RemoteTable, Closure)

144import Control.Distributed.Static (RemoteTable, Closure)

138import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)

145import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)

@@ -262,6 +269,8 @@ data LocalNode = LocalNode

262 -- | Runtime lookup table for supporting closures

269 -- | Runtime lookup table for supporting closures

263 -- TODO: this should be part of the CH state, not the local endpoint state

270 -- TODO: this should be part of the CH state, not the local endpoint state

264 , remoteTable :: !RemoteTable

271 , remoteTable :: !RemoteTable

272+ -- The pool of threads and queues to send messages

273+ , localSendPool :: ThreadPool NodeId ThreadId

265 }

274 }

266

275

267data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

276data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

@@ -345,7 +354,42 @@ data LocalProcessState = LocalProcessState

345newtype Process a = Process {

354newtype Process a = Process {

346 unProcess :: ReaderT LocalProcess IO a

355 unProcess :: ReaderT LocalProcess IO a

347 }

356 }

357+ deriving ( Applicative, Functor, Monad, MonadFix, MonadIO

348− deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative, MonadFix)

358+ , MonadReader LocalProcess, Typeable

359+ )

360+

361+instance MonadBase IO Process where

362+ liftBase = liftIO

363+instance MonadThrow Process where

364+ throwM = liftIO . throwIO

365+instance MonadCatch Process where

366+ catch p h = do

367+ lproc <- ask

368+ liftIO $ catch (runLocalProcess lproc p) (runLocalProcess lproc . h)

369+instance MonadMask Process where

370+ mask p = do

371+ lproc <- ask

372+ liftIO $ mask $ \restore ->

373+ runLocalProcess lproc (p (liftRestore restore))

374+ where

375+ liftRestore :: (forall a. IO a -> IO a)

376+ -> (forall a. Process a -> Process a)

377+ liftRestore restoreIO = \p2 -> do

378+ ourLocalProc <- ask

379+ liftIO $ restoreIO $ runLocalProcess ourLocalProc p2

380+

381+

382+

383+newtype StMProcess a = StMProcess { unStMProcess :: StM (ReaderT LocalProcess IO) a }

384+

385+

386+instance MonadBaseControl IO Process where

387+ type StM Process a = StMProcess a

388+

389+ liftBaseWith f = Process $ liftBaseWith $ \run ->

390+ f $ fmap StMProcess . run . unProcess

391+

392+ restoreM = Process . restoreM . unStMProcess

349

393

350--------------------------------------------------------------------------------

394--------------------------------------------------------------------------------

351-- Typed channels --

395-- Typed channels --

src/Control/Distributed/Process/Node.hs+167−43

@@ -5,6 +5,7 @@

5{-# LANGUAGE BangPatterns #-}

6{-# LANGUAGE GeneralizedNewtypeDeriving #-}

7{-# LANGUAGE RecordWildCards #-}

8+{-# LANGUAGE MagicHash #-}

9-- | Local nodes

10-- | Local nodes

10--

11--

@@ -28,6 +29,7 @@ import System.IO (fixIO, hPutStrLn, stderr)

28import System.Mem.Weak (Weak, deRefWeak)

29import System.Mem.Weak (Weak, deRefWeak)

29import qualified Data.ByteString.Lazy as BSL (fromChunks)

30import qualified Data.ByteString.Lazy as BSL (fromChunks)

30import Data.Binary (decode)

31import Data.Binary (decode)

32+import Data.Function (fix)

31import Data.Map (Map)

33import Data.Map (Map)

32import qualified Data.Map as Map

34import qualified Data.Map as Map

33 ( empty

35 ( empty

@@ -65,14 +67,24 @@ import Control.Exception

65 , Exception

67 , Exception

66 , throwTo

68 , throwTo

67 , uninterruptibleMask_

69 , uninterruptibleMask_

70+ , getMaskingState

71+ , MaskingState(..)

68 )

72 )

69import qualified Control.Exception as Exception

73import qualified Control.Exception as Exception

70 ( Handler(..)

74 ( Handler(..)

71 , catch

75 , catch

72 , catches

76 , catches

73 , finally

77 , finally

78+ , bracket

79+ , mask_

80+ )

81+import Control.Concurrent

82+ ( forkIO

83+ , forkIOWithUnmask

84+ , killThread

85+ , myThreadId

86+ , ThreadId

74 )

87 )

75−import Control.Concurrent (forkIO, forkIOWithUnmask, killThread, myThreadId)

76import Control.Distributed.Process.Internal.StrictMVar

88import Control.Distributed.Process.Internal.StrictMVar

77 ( newMVar

89 ( newMVar

78 , withMVar

90 , withMVar

@@ -83,8 +95,18 @@ import Control.Distributed.Process.Internal.StrictMVar

83 , putMVar

95 , putMVar

84 , takeMVar

96 , takeMVar

85 )

97 )

98+import Control.Distributed.Process.Internal.ThreadPool

99+ ( newThreadPool

100+ , submitTask

101+ , lookupWorker

102+ , ThreadPool

103+ )

86import Control.Concurrent.Chan (newChan, writeChan, readChan)

104import Control.Concurrent.Chan (newChan, writeChan, readChan)

87−import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)

105+import qualified Control.Concurrent.MVar as MVar

106+ ( newEmptyMVar

107+ , takeMVar

108+ , putMVar

109+ )

88import Control.Concurrent.STM

110import Control.Concurrent.STM

89 ( atomically

111 ( atomically

90 )

112 )

@@ -218,8 +240,19 @@ import qualified Control.Distributed.Process.Internal.StrictContainerAccessors a

218 ( mapMaybe

240 ( mapMaybe

219 , mapDefault

241 , mapDefault

220 )

242 )

243+import GHC.IO (IO(..), unsafeUnmask)

244+import GHC.Base ( maskAsyncExceptions# )

245+

246+import System.Timeout

221import Unsafe.Coerce

247import Unsafe.Coerce

222

248

249+-- Remove these when this is fixed:

250+-- https://ghc.haskell.org/trac/ghc/ticket/10149

251+block :: IO a -> IO a

252+block (IO io) = IO $ maskAsyncExceptions# io

253+unblock :: IO a -> IO a

254+unblock = unsafeUnmask

255+

223--------------------------------------------------------------------------------

256--------------------------------------------------------------------------------

224-- Initialization --

257-- Initialization --

225--------------------------------------------------------------------------------

258--------------------------------------------------------------------------------

@@ -249,12 +282,14 @@ createBareLocalNode endPoint rtable = do

249 , _localConnections = Map.empty

282 , _localConnections = Map.empty

250 }

283 }

251 ctrlChan <- newChan

284 ctrlChan <- newChan

285+ sendPool <- newThreadPool

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

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

253 , localEndPoint = endPoint

287 , localEndPoint = endPoint

254 , localState = state

288 , localState = state

255 , localCtrlChan = ctrlChan

289 , localCtrlChan = ctrlChan

256 , localEventBus = MxEventBusInitialising

290 , localEventBus = MxEventBusInitialising

257 , remoteTable = rtable

291 , remoteTable = rtable

292+ , localSendPool = sendPool

258 }

293 }

259 tracedNode <- startMxAgent node

294 tracedNode <- startMxAgent node

260

295

@@ -361,11 +396,14 @@ runProcess node proc = do

361

396

362-- | Spawn a new process on a local node

397-- | Spawn a new process on a local node

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

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

364−forkProcess node proc =

399+forkProcess node proc = do

400+ ms <- getMaskingState

365− modifyMVarMasked (localState node) startProcess

401+ modifyMVarMasked (localState node) (startProcess ms)

366 where

402 where

403+ startProcess :: MaskingState

404+ -> LocalNodeState

367− startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)

405+ -> IO (LocalNodeState, ProcessId)

368− startProcess (LocalNodeValid vst) = do

406+ startProcess ms (LocalNodeValid vst) = do

369 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter

407 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter

370 , lpidUnique = vst ^. localPidUnique

408 , lpidUnique = vst ^. localPidUnique

371 }

409 }

@@ -387,7 +425,13 @@ forkProcess node proc =

387 , processThread = tid

425 , processThread = tid

388 , processNode = node

426 , processNode = node

389 }

427 }

428+ -- Rewrite this code when this is fixed:

429+ -- https://ghc.haskell.org/trac/ghc/ticket/10149

430+ let unmask = case ms of

431+ Unmasked -> unblock

432+ MaskedInterruptible -> block

433+ MaskedUninterruptible -> id

390− tid' <- uninterruptibleMask_ $ forkIOWithUnmask $ \unmask -> do

434+ tid' <- uninterruptibleMask_ $ forkIO $ do

391 reason <- Exception.catches

435 reason <- Exception.catches

392 (unmask $ runLocalProcess lproc proc >> return DiedNormal)

436 (unmask $ runLocalProcess lproc proc >> return DiedNormal)

393 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do

437 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do

@@ -433,7 +477,7 @@ forkProcess node proc =

433 $ vst

477 $ vst

434 , pid

478 , pid

435 )

479 )

436− startProcess LocalNodeClosed =

480+ startProcess _ LocalNodeClosed =

437 throwIO $ NodeClosedException $ localNodeId node

481 throwIO $ NodeClosedException $ localNodeId node

438

482

439 cleanupProcess :: ProcessId

483 cleanupProcess :: ProcessId

@@ -491,24 +535,78 @@ incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr

491 where

535 where

492 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })

536 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })

493

537

538+-- | And 'EventMonitor' monitors the handling of events reporting when an event

539+-- is not handled promptly. It monitors events which are handled in sequence.

540+data EventMonitor a = EventMonitor

541+ { -- | Notifies that an event was received. Takes the event as argument.

542+ eventMonitorGotEvent :: a -> IO ()

543+ -- | Notifies that the last received event was handled.

544+ --

545+ -- Might be called even when there is no event, in which case it will have

546+ -- no effect.

547+ , eventMonitorHandledEvent :: IO ()

548+ -- | Stops the event monitor.

549+ , eventMonitorStop :: IO ()

550+ }

551+

552+-- | `eventMon <- monitorThread t action`

553+--

554+-- Creates an event monitor that executes `action` when an event hasn't been

555+-- handled withing `t` microseconds. `action` takes as argument the event that

556+-- wasn't handled timely.

557+--

558+monitorThread :: Int -> (a -> IO ()) -> IO (EventMonitor a)

559+monitorThread tmicrosecs action = do

560+ mv <- MVar.newEmptyMVar

561+ tid <- forkIO $ forever' $ do

562+ -- loop until some event is reported

563+ ev <- fix $ \loop -> MVar.takeMVar mv >>= maybe loop return

564+ -- wait until the event is handled

565+ Exception.mask_ (timeout tmicrosecs $ MVar.takeMVar mv) >>=

566+ maybe (do action ev -- report the unhandled event

567+ -- wait for the slow event handler

568+ void $ MVar.takeMVar mv

569+ )

570+ (const $ return ())

571+ return EventMonitor

572+ { eventMonitorGotEvent = MVar.putMVar mv . Just

573+ , eventMonitorHandledEvent = MVar.putMVar mv Nothing

574+ , eventMonitorStop = killThread tid

575+ }

576+

494handleIncomingMessages :: LocalNode -> IO ()

577handleIncomingMessages :: LocalNode -> IO ()

495−handleIncomingMessages node = go initConnectionState

578+handleIncomingMessages node =

579+ Exception.bracket

580+ (monitorThread 1000000 $ \ev ->

581+ hPutStrLn stderr $

582+ "handleIncomingMessages blocked" ++ show (localNodeId node) ++

583+ " on " ++ show ev

584+ )

585+ eventMonitorStop $ \eventMon ->

586+ (go eventMon initConnectionState

496− `Exception.catch` \(NodeClosedException _) -> return ()

587+ `Exception.catch` \(NodeClosedException _) -> return ()

588+ ) `Exception.catch` \e -> do

589+ hPutStrLn stderr $

590+ "handleIncomingMessages failed: " ++

591+ show (localNodeId node, e :: SomeException)

592+ throwIO e

497 where

593 where

498− go :: ConnectionState -> IO ()

594+ go :: EventMonitor NT.Event -> ConnectionState -> IO ()

499− go !st = do

595+ go eventMon !st = do

596+ eventMonitorHandledEvent eventMon

500 event <- NT.receive endpoint

597 event <- NT.receive endpoint

598+ eventMonitorGotEvent eventMon event

501 case event of

599 case event of

502 NT.ConnectionOpened cid rel theirAddr ->

600 NT.ConnectionOpened cid rel theirAddr ->

503 if rel == NT.ReliableOrdered

601 if rel == NT.ReliableOrdered

504 then

602 then

505 trace node (MxConnected cid theirAddr)

603 trace node (MxConnected cid theirAddr)

506− >> go (

604+ >> go eventMon (

507 (incomingAt cid ^= Just (theirAddr, Uninit))

605 (incomingAt cid ^= Just (theirAddr, Uninit))

508 . (incomingFrom theirAddr ^: Set.insert cid)

606 . (incomingFrom theirAddr ^: Set.insert cid)

509 $ st

607 $ st

510 )

608 )

511− else invalidRequest cid st

609+ else invalidRequest eventMon cid st

512 NT.Received cid payload ->

610 NT.Received cid payload ->

513 case st ^. incomingAt cid of

611 case st ^. incomingAt cid of

514 Just (_, ToProc pid weakQueue) -> do

612 Just (_, ToProc pid weakQueue) -> do

@@ -519,7 +617,7 @@ handleIncomingMessages node = go initConnectionState

519 let msg = payloadToMessage payload

617 let msg = payloadToMessage payload

520 enqueue queue msg -- 'enqueue' is strict

618 enqueue queue msg -- 'enqueue' is strict

521 trace node (MxReceived pid msg)

619 trace node (MxReceived pid msg)

522− go st

620+ go eventMon st

523 Just (_, ToChan (TypedChannel chan')) -> do

621 Just (_, ToChan (TypedChannel chan')) -> do

524 mChan <- deRefWeak chan'

622 mChan <- deRefWeak chan'

525 -- If mChan is Nothing, the process has given up the read end of

623 -- If mChan is Nothing, the process has given up the read end of

@@ -527,11 +625,11 @@ handleIncomingMessages node = go initConnectionState

527 forM_ mChan $ \chan -> atomically $

625 forM_ mChan $ \chan -> atomically $

528 -- We make sure the message is fully decoded when it is enqueued

626 -- We make sure the message is fully decoded when it is enqueued

529 writeTQueue chan $! decode (BSL.fromChunks payload)

627 writeTQueue chan $! decode (BSL.fromChunks payload)

530− go st

628+ go eventMon st

531 Just (_, ToNode) -> do

629 Just (_, ToNode) -> do

532 let ctrlMsg = decode . BSL.fromChunks $ payload

630 let ctrlMsg = decode . BSL.fromChunks $ payload

533 writeChan ctrlChan $! ctrlMsg

631 writeChan ctrlChan $! ctrlMsg

534− go st

632+ go eventMon st

535 Just (src, Uninit) ->

633 Just (src, Uninit) ->

536 case decode (BSL.fromChunks payload) of

634 case decode (BSL.fromChunks payload) of

537 ProcessIdentifier pid -> do

635 ProcessIdentifier pid -> do

@@ -539,9 +637,9 @@ handleIncomingMessages node = go initConnectionState

539 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)

637 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)

540 case mProc of

638 case mProc of

541 Just proc ->

639 Just proc ->

542− go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)

640+ go eventMon (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)

543 Nothing ->

641 Nothing ->

544− invalidRequest cid st

642+ invalidRequest eventMon cid st

545 SendPortIdentifier chId -> do

643 SendPortIdentifier chId -> do

546 let lcid = sendPortLocalId chId

644 let lcid = sendPortLocalId chId

547 lpid = processLocalId (sendPortProcessId chId)

645 lpid = processLocalId (sendPortProcessId chId)

@@ -551,24 +649,25 @@ handleIncomingMessages node = go initConnectionState

551 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

649 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

552 case mChannel of

650 case mChannel of

553 Just channel ->

651 Just channel ->

652+ go eventMon

554− go (incomingAt cid ^= Just (src, ToChan channel) $ st)

653+ (incomingAt cid ^= Just (src, ToChan channel) $ st)

555 Nothing ->

654 Nothing ->

556− invalidRequest cid st

655+ invalidRequest eventMon cid st

557 Nothing ->

656 Nothing ->

558− invalidRequest cid st

657+ invalidRequest eventMon cid st

559 NodeIdentifier nid ->

658 NodeIdentifier nid ->

560 if nid == localNodeId node

659 if nid == localNodeId node

561− then go (incomingAt cid ^= Just (src, ToNode) $ st)

660+ then go eventMon (incomingAt cid ^= Just (src, ToNode) $ st)

562− else invalidRequest cid st

661+ else invalidRequest eventMon cid st

563 Nothing ->

662 Nothing ->

564− invalidRequest cid st

663+ invalidRequest eventMon cid st

565 NT.ConnectionClosed cid ->

664 NT.ConnectionClosed cid ->

566 case st ^. incomingAt cid of

665 case st ^. incomingAt cid of

567 Nothing ->

666 Nothing ->

568− invalidRequest cid st

667+ invalidRequest eventMon cid st

569 Just (src, _) -> do

668 Just (src, _) -> do

570 trace node (MxDisconnected cid src)

669 trace node (MxDisconnected cid src)

571− go ( (incomingAt cid ^= Nothing)

670+ go eventMon ( (incomingAt cid ^= Nothing)

572 . (incomingFrom src ^: Set.delete cid)

671 . (incomingFrom src ^: Set.delete cid)

573 $ st

672 $ st

574 )

673 )

@@ -580,8 +679,10 @@ handleIncomingMessages node = go initConnectionState

580 , ctrlMsgSignal = Died nid DiedDisconnect

679 , ctrlMsgSignal = Died nid DiedDisconnect

581 }

680 }

582 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))

681 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))

682+ liftIO $ lookupWorker (localSendPool node) (NodeId theirAddr)

683+ >>= maybe (return ()) killThread

583 closeImplicitReconnections node nid

684 closeImplicitReconnections node nid

584− go ( (incomingFrom theirAddr ^= Set.empty)

685+ go eventMon ( (incomingFrom theirAddr ^= Set.empty)

585 . (incoming ^: Map.filterWithKey (const . notLost))

686 . (incoming ^: Map.filterWithKey (const . notLost))

586 $ st

687 $ st

587 )

688 )

@@ -596,15 +697,17 @@ handleIncomingMessages node = go initConnectionState

596 -- and we just give up

697 -- and we just give up

597 fail "Cloud Haskell fatal error: received unexpected multicast"

698 fail "Cloud Haskell fatal error: received unexpected multicast"

598

699

700+ invalidRequest :: EventMonitor NT.Event

701+ -> NT.ConnectionId

599− invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()

702+ -> ConnectionState -> IO ()

600− invalidRequest cid st = do

703+ invalidRequest eventMon cid st = do

601 -- TODO: We should treat this as a fatal error on the part of the remote

704 -- TODO: We should treat this as a fatal error on the part of the remote

602 -- node. That is, we should report the remote node as having died, and we

705 -- node. That is, we should report the remote node as having died, and we

603 -- should close incoming connections (this requires a Transport layer

706 -- should close incoming connections (this requires a Transport layer

604 -- extension).

707 -- extension).

605 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),

708 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),

606 (Trace cid)]

709 (Trace cid)]

607− go ( incomingAt cid ^= Nothing

710+ go eventMon ( incomingAt cid ^= Nothing

608 $ st

711 $ st

609 )

712 )

610

713

@@ -616,9 +719,21 @@ handleIncomingMessages node = go initConnectionState

616--------------------------------------------------------------------------------

719--------------------------------------------------------------------------------

617

720

618runNodeController :: LocalNode -> IO ()

721runNodeController :: LocalNode -> IO ()

619−runNodeController node =

722+runNodeController node = do

723+ Exception.bracket

724+ (monitorThread 1000000 $ \ev ->

725+ hPutStrLn stderr $

726+ "nodeController blocked" ++ show (localNodeId node) ++

727+ " on " ++ show ev

728+ )

729+ eventMonitorStop $ \eventMon ->

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

730+ (runReaderT (evalStateT (unNC (nodeController eventMon)) initNCState) node

621− `Exception.catch` \(NodeClosedException _) -> return ()

731+ `Exception.catch` \(NodeClosedException _) -> return ()

732+ ) `Exception.catch` \e -> do

733+ hPutStrLn stderr $

734+ "nodeController failed: " ++

735+ show (localNodeId node, e :: SomeException)

736+ throwIO e

622

737

623--------------------------------------------------------------------------------

738--------------------------------------------------------------------------------

624-- Internal data types --

739-- Internal data types --

@@ -634,6 +749,12 @@ data NCState = NCState

634 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])

749 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])

635 }

750 }

636

751

752+submitSendPool :: LocalNode -> NodeId -> IO () -> IO ()

753+submitSendPool node nid task = submitTask (localSendPool node) fork nid task

754+ where

755+ fork io =

756+ forkIO $ io `Exception.catch` \(NodeClosedException _) -> return ()

757+

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

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

638 deriving ( Applicative

759 deriving ( Applicative

639 , Functor

760 , Functor

@@ -644,11 +765,12 @@ newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }

644 )

765 )

645

766

646initNCState :: NCState

767initNCState :: NCState

647−initNCState = NCState { _links = Map.empty

768+initNCState = NCState

769+ { _links = Map.empty

648− , _monitors = Map.empty

770+ , _monitors = Map.empty

649− , _registeredHere = Map.empty

771+ , _registeredHere = Map.empty

650− , _registeredOnNodes = Map.empty

772+ , _registeredOnNodes = Map.empty

651− }

773+ }

652

774

653-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This

775-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This

654-- type is deliberately not exported so it cannot be caught explicitly.

776-- type is deliberately not exported so it cannot be caught explicitly.

@@ -666,7 +788,7 @@ ncSendToProcess pid msg = do

666 node <- ask

788 node <- ask

667 if processNodeId pid == localNodeId node

789 if processNodeId pid == localNodeId node

668 then ncEffectLocalSend node pid msg

790 then ncEffectLocalSend node pid msg

669− else liftIO $ sendBinary node

791+ else liftIO $ submitSendPool node (processNodeId pid) $ sendBinary node

670 (NodeIdentifier $ localNodeId node)

792 (NodeIdentifier $ localNodeId node)

671 (NodeIdentifier $ processNodeId pid)

793 (NodeIdentifier $ processNodeId pid)

672 WithImplicitReconnect

794 WithImplicitReconnect

@@ -679,7 +801,7 @@ ncSendToNode to msg = do

679 node <- ask

801 node <- ask

680 liftIO $ if to == localNodeId node

802 liftIO $ if to == localNodeId node

681 then writeChan (localCtrlChan node) $! msg

803 then writeChan (localCtrlChan node) $! msg

682− else sendBinary node

804+ else submitSendPool node to $ sendBinary node

683 (NodeIdentifier $ localNodeId node)

805 (NodeIdentifier $ localNodeId node)

684 (NodeIdentifier to)

806 (NodeIdentifier to)

685 WithImplicitReconnect

807 WithImplicitReconnect

@@ -718,11 +840,13 @@ withLocalTracer node act = act (localEventBus node)

718--------------------------------------------------------------------------------

840--------------------------------------------------------------------------------

719

841

720-- [Unified: Table 7]

842-- [Unified: Table 7]

721−nodeController :: NC ()

843+nodeController :: EventMonitor NCMsg -> NC ()

722−nodeController = do

844+nodeController eventMon = do

723 node <- ask

845 node <- ask

724 forever' $ do

846 forever' $ do

847+ liftIO $ eventMonitorHandledEvent eventMon

725 msg <- liftIO $ readChan (localCtrlChan node)

848 msg <- liftIO $ readChan (localCtrlChan node)

849+ liftIO $ eventMonitorGotEvent eventMon msg

726

850

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

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

728 case destNid (ctrlMsgSignal msg) of

852 case destNid (ctrlMsgSignal msg) of