master...e648b03 · distributed-process

git.iko.soy/distributed-process

6 files changed+197−20

since splitsnapshots

unifiedsplit

distributed-process.cabal+4−1 src/Control/Distributed/Process.hs+62−4 src/Control/Distributed/Process/Internal/Messaging.hs+4−4 src/Control/Distributed/Process/Internal/ThreadPool.hs+78 src/Control/Distributed/Process/Internal/Types.hs+25−4 src/Control/Distributed/Process/Node.hs+24−7

distributed-process.cabal+4−1

@@ -54,7 +54,9 @@ 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.3 && < 0.4

5860 Exposed-modules: Control.Distributed.Process,

5961 Control.Distributed.Process.Closure,

6062 Control.Distributed.Process.Debug,

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

6769 Control.Distributed.Process.Internal.StrictContainerAccessors,

6870 Control.Distributed.Process.Internal.StrictList,

6971 Control.Distributed.Process.Internal.StrictMVar,

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

7073 Control.Distributed.Process.Internal.Types,

7174 Control.Distributed.Process.Internal.WeakTQueue,

7275 Control.Distributed.Process.Management,

src/Control/Distributed/Process.hs+62−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 (forkIO)

169170import Control.Concurrent.MVar

170171 ( MVar

171172 , newEmptyMVar

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

199200 , WhereIsReply(..)

200201 , RegisterReply(..)

201202 , LocalProcess(processNode)

203+ , runLocalProcess

202204 , Message

203205 )

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

@@ -307,6 +309,12 @@ 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+ , localState

315+ , LocalNodeState(..)

316+ , withValidLocalState

317+ )

310318import Control.Distributed.Process.Internal.Spawn

311319 ( -- Spawning Processes/Channels

312320 spawn

@@ -316,7 +324,10 @@ import Control.Distributed.Process.Internal.Spawn

316324 , spawnSupervised

317325 , call

318326 )

319−import Control.Exception (SomeException, throw)

327+import Control.Distributed.Process.Internal.StrictMVar (readMVar)

328+import Control.Exception (SomeException, throw, throwTo, throwIO)

329+import Control.Monad (forM_)

330+

320331

321332-- INTERNAL NOTES

322333--

@@ -415,7 +426,54 @@ spawnChannelLocal proc = do

415426callLocal :: Process a -> Process a

416427callLocal proc = mask $ \release -> do

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

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

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

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

429+ (spInit, rpInit) <- newChan -- TODO: Remove when spawnLocal inherits the

430+ -- masking state.

431+ child <- spawnLocal $ mask_ $

432+ try (sendChan spInit () >> release proc) >>= liftIO . putMVar mv

433+ rs <- liftIO (takeMVar mv)

434+ `onException`

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

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

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

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

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

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

441+ --

442+ -- Using 'uninterruptibleMask_' would be problematic as there is no

443+ -- way to be sure the node isn't closing when the cleanup executes.

444+ -- Successfully running the cleanup is necessary to ensure

445+ -- termination. If exceptions are masked uninterruptibly,

446+ -- 'receiveChan' might block indefinitely and 'kill' might have no

447+ -- effect.

448+ --

449+ do (exs0, _) <- collectExceptions [] $

450+ whenNotClosed $ receiveChan rpInit

451+ (exs1, _) <- collectExceptions exs0 $

452+ whenNotClosed $ kill child "callLocal was interrupted"

453+ (exs, mr) <- collectExceptions exs1 $

454+ whenNotClosed $ liftIO (takeMVar mv)

455+ -- Forward exceptions asynchronously.

456+ lproc <- ask

457+ liftIO $ forkIO $ forM_ (reverse exs) $

458+ throwTo (processThread lproc)

459+ here <- getSelfNode

460+ liftIO $

461+ maybe (throwIO $ userError $ "Node closed " ++ show here)

462+ return mr

421463 either throw return rs

464+ where

465+ -- Retries a given action until it succeeds.

466+ collectExceptions :: [SomeException]

467+ -> Process a

468+ -> Process ([SomeException], a)

469+ collectExceptions exs p =

470+ try p >>= either (\e -> collectExceptions (e : exs) p) (return . (,) exs)

471+

472+ -- Runs the given closure if the node is not closed.

473+ whenNotClosed :: Process a -> Process (Maybe a)

474+ whenNotClosed p = do

475+ lproc <- ask

476+ st <- liftIO $ readMVar (localState $ processNode lproc)

477+ case st of

478+ LocalNodeValid _ -> fmap Just p

479+ LocalNodeClosed -> return Nothing

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/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+25−4

@@ -1,9 +1,11 @@

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

810-- | Types used throughout the Cloud Haskell framework

911--

@@ -116,7 +118,9 @@ import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

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

117119import Control.Monad.Fix (MonadFix)

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

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

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

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

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

120124import Control.Distributed.Process.Serializable

121125 ( Fingerprint

122126 , Serializable

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

133137 , modifyMVar

134138 , modifyMVar_

135139 )

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

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

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

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

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

262267 -- | Runtime lookup table for supporting closures

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

264269 , remoteTable :: !RemoteTable

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

271+ , localSendPool :: ThreadPool NodeId ThreadId

265272 }

266273

267274data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

@@ -345,7 +352,21 @@ data LocalProcessState = LocalProcessState

345352newtype Process a = Process {

346353 unProcess :: ReaderT LocalProcess IO a

347354 }

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

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

356+ , MonadReader LocalProcess, Typeable

357+ )

358+

359+instance MonadBase IO Process where

360+ liftBase = liftIO

361+

362+instance MonadBaseControl IO Process where

363+ newtype StM Process a =

364+ StMProcess { unStMProcess :: StM (ReaderT LocalProcess IO) a }

365+

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

367+ f $ fmap StMProcess . run . unProcess

368+

369+ restoreM = Process . restoreM . unStMProcess

349370

350371--------------------------------------------------------------------------------

351372-- Typed channels --

src/Control/Distributed/Process/Node.hs+24−7

@@ -83,6 +83,12 @@ import Control.Distributed.Process.Internal.StrictMVar

8383 , putMVar

8484 , takeMVar

8585 )

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

87+ ( newThreadPool

88+ , submitTask

89+ , lookupWorker

90+ , ThreadPool

91+ )

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

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

8894import Control.Concurrent.STM

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

249255 , _localConnections = Map.empty

250256 }

251257 ctrlChan <- newChan

258+ sendPool <- newThreadPool

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

253260 , localEndPoint = endPoint

254261 , localState = state

255262 , localCtrlChan = ctrlChan

256263 , localEventBus = MxEventBusInitialising

257264 , remoteTable = rtable

265+ , localSendPool = sendPool

258266 }

259267 tracedNode <- startMxAgent node

260268

@@ -580,6 +588,8 @@ handleIncomingMessages node = go initConnectionState

580588 , ctrlMsgSignal = Died nid DiedDisconnect

581589 }

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

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

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

583593 closeImplicitReconnections node nid

584594 go ( (incomingFrom theirAddr ^= Set.empty)

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

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

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

635645 }

636646

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

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

649+ where

650+ fork io =

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

652+

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

638654 deriving ( Applicative

639655 , Functor

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

644660 )

645661

646662initNCState :: NCState

647−initNCState = NCState { _links = Map.empty

648− , _monitors = Map.empty

649− , _registeredHere = Map.empty

650− , _registeredOnNodes = Map.empty

651− }

663+initNCState = NCState

664+ { _links = Map.empty

665+ , _monitors = Map.empty

666+ , _registeredHere = Map.empty

667+ , _registeredOnNodes = Map.empty

668+ }

652669

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

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

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

666683 node <- ask

667684 if processNodeId pid == localNodeId node

668685 then ncEffectLocalSend node pid msg

669− else liftIO $ sendBinary node

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

670687 (NodeIdentifier $ localNodeId node)

671688 (NodeIdentifier $ processNodeId pid)

672689 WithImplicitReconnect

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

679696 node <- ask

680697 liftIO $ if to == localNodeId node

681698 then writeChan (localCtrlChan node) $! msg

682− else sendBinary node

699+ else submitSendPool node to $ sendBinary node

683700 (NodeIdentifier $ localNodeId node)

684701 (NodeIdentifier to)

685702 WithImplicitReconnect