master...9b4a059 · distributed-process

git.iko.soy/distributed-process

7 files changed+346−117

since splitsnapshots

unifiedsplit

distributed-process.cabal+5−1 src/Control/Distributed/Process.hs+26−4 src/Control/Distributed/Process/Internal/Messaging.hs+4−4 src/Control/Distributed/Process/Internal/Primitives.hs+8−61 src/Control/Distributed/Process/Internal/ThreadPool.hs+78 src/Control/Distributed/Process/Internal/Types.hs+58−4 src/Control/Distributed/Process/Node.hs+167−43

distributed-process.cabal+5−1

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

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

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

5656 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

5861 Exposed-modules: Control.Distributed.Process,

5962 Control.Distributed.Process.Closure,

6063 Control.Distributed.Process.Debug,

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

6770 Control.Distributed.Process.Internal.StrictContainerAccessors,

6871 Control.Distributed.Process.Internal.StrictList,

6972 Control.Distributed.Process.Internal.StrictMVar,

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

7074 Control.Distributed.Process.Internal.Types,

7175 Control.Distributed.Process.Internal.WeakTQueue,

7276 Control.Distributed.Process.Management,

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

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

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

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

168168import Control.Monad.Reader (ask)

169+import Control.Concurrent (killThread)

169170import Control.Concurrent.MVar

170171 ( MVar

171172 , newEmptyMVar

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

200201 , RegisterReply(..)

201202 , LocalProcess(processNode)

202203 , Message

204+ , localProcessWithId

203205 )

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

205207import Control.Distributed.Process.Internal.Primitives

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

307309 , reconnectPort

308310 )

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

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

313+ ( processThread

314+ , withValidLocalState

315+ )

310316import Control.Distributed.Process.Internal.Spawn

311317 ( -- Spawning Processes/Channels

312318 spawn

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

316322 , spawnSupervised

317323 , call

318324 )

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+

320330

321331-- INTERNAL NOTES

322332--

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

416426callLocal proc = mask $ \release -> do

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

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

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

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

421− either throw return rs

429+ lproc <- ask

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

443+ either throwIO return rs

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

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

1818import Control.Concurrent (forkIO)

1919import Control.Concurrent.Chan (writeChan)

2020import Control.Exception (mask_)

21−import Control.Monad (unless)

21+import Control.Monad (unless, void)

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

2323import Control.Monad.Reader (ask)

2424import qualified Network.Transport as NT

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

3030 , close

3131 )

3232import Control.Distributed.Process.Internal.Types

33− ( LocalNode(localState, localEndPoint, localCtrlChan)

33+ ( LocalNode(localEndPoint, localCtrlChan)

3434 , withValidLocalState

3535 , modifyValidLocalState

3636 , modifyValidLocalState_

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

146146 return (vst, return ())

147147 Just (conn, _) -> do

148148 return ( localConnectionBetween from to ^= Nothing $ vst

149− , NT.close conn

149+ , void $ forkIO $ NT.close conn

150150 )

151− forM_ mio forkIO

151+ forM_ mio id

152152

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

154154closeImplicitReconnections node to = mask_ $ do

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

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

137137import Control.Monad (when)

138138import Control.Monad.Reader (ask)

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

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

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

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

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

143142import Control.Distributed.Process.Internal.StrictMVar

144143 ( StrictMVar

145144 , modifyMVar

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

699698

700699-- | Terminate immediately (throws a ProcessTerminationException)

701700terminate :: Process a

702−terminate = liftIO $ throwIO ProcessTerminationException

701+terminate = throwM ProcessTerminationException

703702

704703-- [Issue #110]

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

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

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

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

711710 pid <- getSelfPid

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

711+ throwM (ProcessExitException pid (createMessage reason))

713712

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

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

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

747746 handleExit ex@(ProcessExitException from msg) =

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

749748 then exitHandler from decoded

750− else liftIO $ throwIO ex

749+ else throwM ex

751750 where

752751 decoded :: a

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

771770 handleExit :: ProcessExitException

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

773772 -> Process b

774− handleExit ex [] = liftIO $ throwIO ex

773+ handleExit ex [] = throwM ex

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

776775 r <- h from msg

777776 case r of

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

934933-- Exception handling --

935934--------------------------------------------------------------------------------

936935

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−

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

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

992938

993939instance Functor Handler where

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

995941

942+

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

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

998945catches proc handlers = proc `catch` catchesHandler handlers

999946

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

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

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

1002949 where tryHandler (Handler handler) res

1003950 = case fromException e of

1004951 Just e' -> handler e'

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

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

11391086handleRegistrationReply label ok owner =

11401087 when (not ok) $

1141− liftIO $ throwIO $ ProcessRegistrationException label owner

1088+ throwM $ ProcessRegistrationException label owner

11421089

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

11441091--

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+58−4

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

1+{-# LANGUAGE CPP #-}

12{-# LANGUAGE DeriveDataTypeable #-}

3+{-# LANGUAGE DeriveGeneric #-}

24{-# LANGUAGE ExistentialQuantification #-}

35{-# LANGUAGE GeneralizedNewtypeDeriving #-}

46{-# LANGUAGE GADTs #-}

5−{-# LANGUAGE DeriveGeneric #-}

6−{-# LANGUAGE CPP #-}

7+{-# LANGUAGE MultiParamTypeClasses #-}

8+{-# LANGUAGE TypeFamilies #-}

9+{-# LANGUAGE Rank2Types #-}

710

811-- | Types used throughout the Cloud Haskell framework

912--

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

112115import Control.Concurrent.Chan (Chan)

113116import Control.Concurrent.STM (STM)

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

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

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

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

117121import Control.Monad.Fix (MonadFix)

118122import 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(..))

120126import Control.Distributed.Process.Serializable

121127 ( Fingerprint

122128 , Serializable

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

133139 , modifyMVar

134140 , modifyMVar_

135141 )

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

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

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

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

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

262269 -- | Runtime lookup table for supporting closures

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

264271 , remoteTable :: !RemoteTable

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

273+ , localSendPool :: ThreadPool NodeId ThreadId

265274 }

266275

267276data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

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

345354newtype Process a = Process {

346355 unProcess :: ReaderT LocalProcess IO a

347356 }

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

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

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+ uninterruptibleMask p = do

381+ lproc <- ask

382+ liftIO $ uninterruptibleMask $ \restore ->

383+ runLocalProcess lproc (p (liftRestore restore))

384+ where

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

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

387+ liftRestore restoreIO = \p2 -> do

388+ ourLocalProc <- ask

389+ liftIO $ restoreIO $ runLocalProcess ourLocalProc p2

390+

391+

392+

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

394+

395+

396+instance MonadBaseControl IO Process where

397+ type StM Process a = StMProcess a

398+

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

400+ f $ fmap StMProcess . run . unProcess

401+

402+ restoreM = Process . restoreM . unStMProcess

349403

350404--------------------------------------------------------------------------------

351405-- Typed channels --

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

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

55{-# LANGUAGE BangPatterns #-}

66{-# LANGUAGE GeneralizedNewtypeDeriving #-}

77{-# LANGUAGE RecordWildCards #-}

8+{-# LANGUAGE MagicHash #-}

910-- | Local nodes

1011--

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

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

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

3031import Data.Binary (decode)

32+import Data.Function (fix)

3133import Data.Map (Map)

3234import qualified Data.Map as Map

3335 ( empty

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

6567 , Exception

6668 , throwTo

6769 , uninterruptibleMask_

70+ , getMaskingState

71+ , MaskingState(..)

6872 )

6973import qualified Control.Exception as Exception

7074 ( Handler(..)

7175 , catch

7276 , catches

7377 , finally

78+ , bracket

79+ , mask_

80+ )

81+import Control.Concurrent

82+ ( forkIO

83+ , forkIOWithUnmask

84+ , killThread

85+ , myThreadId

86+ , ThreadId

7487 )

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

7688import Control.Distributed.Process.Internal.StrictMVar

7789 ( newMVar

7890 , withMVar

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

8395 , putMVar

8496 , takeMVar

8597 )

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

99+ ( newThreadPool

100+ , submitTask

101+ , lookupWorker

102+ , ThreadPool

103+ )

86104import 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+ )

88110import Control.Concurrent.STM

89111 ( atomically

90112 )

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

218240 ( mapMaybe

219241 , mapDefault

220242 )

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

244+import GHC.Base ( maskAsyncExceptions# )

245+

246+import System.Timeout

221247import Unsafe.Coerce

222248

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+

223256--------------------------------------------------------------------------------

224257-- Initialization --

225258--------------------------------------------------------------------------------

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

249282 , _localConnections = Map.empty

250283 }

251284 ctrlChan <- newChan

285+ sendPool <- newThreadPool

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

253287 , localEndPoint = endPoint

254288 , localState = state

255289 , localCtrlChan = ctrlChan

256290 , localEventBus = MxEventBusInitialising

257291 , remoteTable = rtable

292+ , localSendPool = sendPool

258293 }

259294 tracedNode <- startMxAgent node

260295

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

361396

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

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

364−forkProcess node proc =

365− modifyMVarMasked (localState node) startProcess

399+forkProcess node proc = do

400+ ms <- getMaskingState

401+ modifyMVarMasked (localState node) (startProcess ms)

366402 where

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

368− startProcess (LocalNodeValid vst) = do

403+ startProcess :: MaskingState

404+ -> LocalNodeState

405+ -> IO (LocalNodeState, ProcessId)

406+ startProcess ms (LocalNodeValid vst) = do

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

370408 , lpidUnique = vst ^. localPidUnique

371409 }

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

387425 , processThread = tid

388426 , processNode = node

389427 }

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

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

434+ tid' <- uninterruptibleMask_ $ forkIO $ do

391435 reason <- Exception.catches

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

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

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

433477 $ vst

434478 , pid

435479 )

436− startProcess LocalNodeClosed =

480+ startProcess _ LocalNodeClosed =

437481 throwIO $ NodeClosedException $ localNodeId node

438482

439483 cleanupProcess :: ProcessId

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

491535 where

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

493537

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+

494577handleIncomingMessages :: LocalNode -> IO ()

495−handleIncomingMessages node = go initConnectionState

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

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

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

497593 where

498− go :: ConnectionState -> IO ()

499− go !st = do

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

595+ go eventMon !st = do

596+ eventMonitorHandledEvent eventMon

500597 event <- NT.receive endpoint

598+ eventMonitorGotEvent eventMon event

501599 case event of

502600 NT.ConnectionOpened cid rel theirAddr ->

503601 if rel == NT.ReliableOrdered

504602 then

505603 trace node (MxConnected cid theirAddr)

506− >> go (

604+ >> go eventMon (

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

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

509607 $ st

510608 )

511− else invalidRequest cid st

609+ else invalidRequest eventMon cid st

512610 NT.Received cid payload ->

513611 case st ^. incomingAt cid of

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

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

519617 let msg = payloadToMessage payload

520618 enqueue queue msg -- 'enqueue' is strict

521619 trace node (MxReceived pid msg)

522− go st

620+ go eventMon st

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

524622 mChan <- deRefWeak chan'

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

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

527625 forM_ mChan $ \chan -> atomically $

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

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

530− go st

628+ go eventMon st

531629 Just (_, ToNode) -> do

532630 let ctrlMsg = decode . BSL.fromChunks $ payload

533631 writeChan ctrlChan $! ctrlMsg

534− go st

632+ go eventMon st

535633 Just (src, Uninit) ->

536634 case decode (BSL.fromChunks payload) of

537635 ProcessIdentifier pid -> do

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

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

540638 case mProc of

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

543641 Nothing ->

544− invalidRequest cid st

642+ invalidRequest eventMon cid st

545643 SendPortIdentifier chId -> do

546644 let lcid = sendPortLocalId chId

547645 lpid = processLocalId (sendPortProcessId chId)

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

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

552650 case mChannel of

553651 Just channel ->

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

652+ go eventMon

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

555654 Nothing ->

556− invalidRequest cid st

655+ invalidRequest eventMon cid st

557656 Nothing ->

558− invalidRequest cid st

657+ invalidRequest eventMon cid st

559658 NodeIdentifier nid ->

560659 if nid == localNodeId node

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

562− else invalidRequest cid st

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

661+ else invalidRequest eventMon cid st

563662 Nothing ->

564− invalidRequest cid st

663+ invalidRequest eventMon cid st

565664 NT.ConnectionClosed cid ->

566665 case st ^. incomingAt cid of

567666 Nothing ->

568− invalidRequest cid st

667+ invalidRequest eventMon cid st

569668 Just (src, _) -> do

570669 trace node (MxDisconnected cid src)

571− go ( (incomingAt cid ^= Nothing)

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

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

573672 $ st

574673 )

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

580679 , ctrlMsgSignal = Died nid DiedDisconnect

581680 }

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

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

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

583684 closeImplicitReconnections node nid

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

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

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

586687 $ st

587688 )

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

596697 -- and we just give up

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

598699

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

600− invalidRequest cid st = do

700+ invalidRequest :: EventMonitor NT.Event

701+ -> NT.ConnectionId

702+ -> ConnectionState -> IO ()

703+ invalidRequest eventMon cid st = do

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

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

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

604707 -- extension).

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

606709 (Trace cid)]

607− go ( incomingAt cid ^= Nothing

710+ go eventMon ( incomingAt cid ^= Nothing

608711 $ st

609712 )

610713

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

616719--------------------------------------------------------------------------------

617720

618721runNodeController :: LocalNode -> IO ()

619−runNodeController node =

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

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

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

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

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

622737

623738--------------------------------------------------------------------------------

624739-- Internal data types --

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

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

635750 }

636751

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+

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

638759 deriving ( Applicative

639760 , Functor

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

644765 )

645766

646767initNCState :: NCState

647−initNCState = NCState { _links = Map.empty

648− , _monitors = Map.empty

649− , _registeredHere = Map.empty

650− , _registeredOnNodes = Map.empty

651− }

768+initNCState = NCState

769+ { _links = Map.empty

770+ , _monitors = Map.empty

771+ , _registeredHere = Map.empty

772+ , _registeredOnNodes = Map.empty

773+ }

652774

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

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

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

666788 node <- ask

667789 if processNodeId pid == localNodeId node

668790 then ncEffectLocalSend node pid msg

669− else liftIO $ sendBinary node

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

670792 (NodeIdentifier $ localNodeId node)

671793 (NodeIdentifier $ processNodeId pid)

672794 WithImplicitReconnect

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

679801 node <- ask

680802 liftIO $ if to == localNodeId node

681803 then writeChan (localCtrlChan node) $! msg

682− else sendBinary node

804+ else submitSendPool node to $ sendBinary node

683805 (NodeIdentifier $ localNodeId node)

684806 (NodeIdentifier to)

685807 WithImplicitReconnect

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

718840--------------------------------------------------------------------------------

719841

720842-- [Unified: Table 7]

721−nodeController :: NC ()

722−nodeController = do

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

844+nodeController eventMon = do

723845 node <- ask

724846 forever' $ do

847+ liftIO $ eventMonitorHandledEvent eventMon

725848 msg <- liftIO $ readChan (localCtrlChan node)

849+ liftIO $ eventMonitorGotEvent eventMon msg

726850

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

728852 case destNid (ctrlMsgSignal msg) of