git.iko.soy/distributed-process
on 9b4a059 help sign in
660d554
swap
13 ahead, 209 behind
distributed-process.cabal+5−1
since splitsnapshots
unifiedsplit
@@ -54,7 +54,10 @@ Library
54 ghc-prim >= 0.2 && < 0.5,
54 ghc-prim >= 0.2 && < 0.5,
55 distributed-static >= 0.2 && < 0.4,
55 distributed-static >= 0.2 && < 0.4,
56 rank1dynamic >= 0.1 && < 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)
166import Control.Monad.IO.Class (liftIO)
167import Control.Applicative ((<$>))
167import Control.Applicative ((<$>))
168import Control.Monad.Reader (ask)
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)
18import Control.Concurrent (forkIO)
19import Control.Concurrent.Chan (writeChan)
19import Control.Concurrent.Chan (writeChan)
20import Control.Exception (mask_)
20import Control.Exception (mask_)
21−import Control.Monad (unless)
21+import Control.Monad (unless, void)
22import Control.Monad.IO.Class (liftIO)
22import Control.Monad.IO.Class (liftIO)
23import Control.Monad.Reader (ask)
23import Control.Monad.Reader (ask)
24import qualified Network.Transport as NT
24import qualified Network.Transport as NT
@@ -30,7 +30,7 @@ import qualified Network.Transport as NT
30 , close
30 , close
31 )
31 )
32import Control.Distributed.Process.Internal.Types
32import Control.Distributed.Process.Internal.Types
33− ( LocalNode(localState, localEndPoint, localCtrlChan)
33+ ( LocalNode(localEndPoint, localCtrlChan)
34 , withValidLocalState
34 , withValidLocalState
35 , modifyValidLocalState
35 , modifyValidLocalState
36 , modifyValidLocalState_
36 , modifyValidLocalState_
@@ -146,9 +146,9 @@ disconnect node from to = mask_ $ do
146 return (vst, return ())
146 return (vst, return ())
147 Just (conn, _) -> do
147 Just (conn, _) -> do
148 return ( localConnectionBetween from to ^= Nothing $ vst
148 return ( localConnectionBetween from to ^= Nothing $ vst
149− , NT.close conn
149+ , void $ forkIO $ NT.close conn
150 )
150 )
151− forM_ mio forkIO
151+ forM_ mio id
152
152
153closeImplicitReconnections :: LocalNode -> Identifier -> IO ()
153closeImplicitReconnections :: LocalNode -> Identifier -> IO ()
154closeImplicitReconnections node to = mask_ $ do
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)
137import Control.Monad (when)
138import Control.Monad.Reader (ask)
138import Control.Monad.Reader (ask)
139import Control.Monad.IO.Class (liftIO)
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+58−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 #-}
7
10
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,52 @@ 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+ 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
349
403
350--------------------------------------------------------------------------------
404--------------------------------------------------------------------------------
351-- Typed channels --
405-- Typed channels --
src/Control/Distributed/Process/Node.hs+167−43
@@ -5,6 +5,7 @@
5{-# LANGUAGE BangPatterns #-}
5{-# LANGUAGE BangPatterns #-}
6{-# LANGUAGE GeneralizedNewtypeDeriving #-}
6{-# LANGUAGE GeneralizedNewtypeDeriving #-}
7{-# LANGUAGE RecordWildCards #-}
7{-# LANGUAGE RecordWildCards #-}
8+{-# LANGUAGE MagicHash #-}
8
9
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