master...caf93de · distributed-process

git.iko.soy/distributed-process

on caf93de help sign in

distributed-process.cabal+4−1

since splitsnapshots

unifiedsplit

@@ -54,7 +54,9 @@ Library

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

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

56 rank1dynamic >= 0.1 && < 0.4,

57− syb >= 0.3 && < 0.7

57+ syb >= 0.3 && < 0.7,

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

59+ monad-control >= 0.3 && < 0.4

58 Exposed-modules: Control.Distributed.Process,

60 Exposed-modules: Control.Distributed.Process,

59 Control.Distributed.Process.Closure,

61 Control.Distributed.Process.Closure,

60 Control.Distributed.Process.Debug,

62 Control.Distributed.Process.Debug,

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

67 Control.Distributed.Process.Internal.StrictContainerAccessors,

69 Control.Distributed.Process.Internal.StrictContainerAccessors,

68 Control.Distributed.Process.Internal.StrictList,

70 Control.Distributed.Process.Internal.StrictList,

69 Control.Distributed.Process.Internal.StrictMVar,

71 Control.Distributed.Process.Internal.StrictMVar,

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

70 Control.Distributed.Process.Internal.Types,

73 Control.Distributed.Process.Internal.Types,

71 Control.Distributed.Process.Internal.WeakTQueue,

74 Control.Distributed.Process.Internal.WeakTQueue,

72 Control.Distributed.Process.Management,

75 Control.Distributed.Process.Management,

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

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

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

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

168import Control.Monad.Reader (ask)

169+import Control.Concurrent (killThread)

169import Control.Concurrent.MVar

170import Control.Concurrent.MVar

170 ( MVar

171 ( MVar

171 , newEmptyMVar

172 , newEmptyMVar

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

200 , RegisterReply(..)

201 , RegisterReply(..)

201 , LocalProcess(processNode)

202 , LocalProcess(processNode)

202 , Message

203 , Message

204+ , localProcessWithId

203 )

205 )

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

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

205import Control.Distributed.Process.Internal.Primitives

207import Control.Distributed.Process.Internal.Primitives

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

307 , reconnectPort

309 , reconnectPort

308 )

310 )

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

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

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

313+ ( processThread

314+ , withValidLocalState

315+ )

310import Control.Distributed.Process.Internal.Spawn

316import Control.Distributed.Process.Internal.Spawn

311 ( -- Spawning Processes/Channels

317 ( -- Spawning Processes/Channels

312 spawn

318 spawn

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

316 , spawnSupervised

322 , spawnSupervised

317 , call

323 , call

318 )

324 )

319−import Control.Exception (SomeException, throw)

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

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

327+import Control.Monad (forM_)

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

329+

320

330

321-- INTERNAL NOTES

331-- INTERNAL NOTES

322--

332--

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

416callLocal proc = mask $ \release -> do

426callLocal proc = mask $ \release -> do

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

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

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

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

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

429+ lproc <- ask

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

430+ liftIO $ do

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

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

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

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

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

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

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

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

439+ \vst -> return $ fmap processThread $

440+ vst ^. localProcessWithId (processLocalId child)

441+ forM_ mchildThreadId killThread

442+ takeMVar mv

421− either throw return rs

443+ either throwIO return rs

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

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

18import Control.Concurrent (forkIO)

19import Control.Concurrent.Chan (writeChan)

20import Control.Exception (mask_)

21−import Control.Monad (unless)

21+import Control.Monad (unless, void)

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

23import Control.Monad.Reader (ask)

24import qualified Network.Transport as NT

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

30 , close

31 )

32import Control.Distributed.Process.Internal.Types

33− ( LocalNode(localState, localEndPoint, localCtrlChan)

33+ ( LocalNode(localEndPoint, localCtrlChan)

34 , withValidLocalState

35 , modifyValidLocalState

36 , modifyValidLocalState_

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

146 return (vst, return ())

147 Just (conn, _) -> do

148 return ( localConnectionBetween from to ^= Nothing $ vst

149− , NT.close conn

149+ , void $ forkIO $ NT.close conn

150 )

151− forM_ mio forkIO

151+ forM_ mio id

152

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

154closeImplicitReconnections node to = mask_ $ do

src/Control/Distributed/Process/Internal/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 #-}

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

6−{-# LANGUAGE CPP #-}

8+{-# LANGUAGE TypeFamilies #-}

8-- | Types used throughout the Cloud Haskell framework

10-- | Types used throughout the Cloud Haskell framework

9--

11--

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

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

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

117import Control.Monad.Fix (MonadFix)

119import Control.Monad.Fix (MonadFix)

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

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

120import Control.Distributed.Process.Serializable

124import Control.Distributed.Process.Serializable

121 ( Fingerprint

125 ( Fingerprint

122 , Serializable

126 , Serializable

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

133 , modifyMVar

137 , modifyMVar

134 , modifyMVar_

138 , modifyMVar_

135 )

139 )

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

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

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

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

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

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

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

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

262 -- | Runtime lookup table for supporting closures

267 -- | Runtime lookup table for supporting closures

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

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

264 , remoteTable :: !RemoteTable

269 , remoteTable :: !RemoteTable

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

271+ , localSendPool :: ThreadPool NodeId ThreadId

265 }

272 }

266

273

267data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

274data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

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

345newtype Process a = Process {

352newtype Process a = Process {

346 unProcess :: ReaderT LocalProcess IO a

353 unProcess :: ReaderT LocalProcess IO a

347 }

354 }

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

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

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

349

370

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

371--------------------------------------------------------------------------------

351-- Typed channels --

372-- Typed channels --

src/Control/Distributed/Process/Node.hs+52−13

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

5{-# LANGUAGE BangPatterns #-}

6{-# LANGUAGE GeneralizedNewtypeDeriving #-}

7{-# LANGUAGE RecordWildCards #-}

8+{-# LANGUAGE MagicHash #-}

9-- | Local nodes

10-- | Local nodes

10--

11--

@@ -65,6 +66,8 @@ import Control.Exception

65 , Exception

66 , Exception

66 , throwTo

67 , throwTo

67 , uninterruptibleMask_

68 , uninterruptibleMask_

69+ , getMaskingState

70+ , MaskingState(..)

68 )

71 )

69import qualified Control.Exception as Exception

72import qualified Control.Exception as Exception

70 ( Handler(..)

73 ( Handler(..)

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

83 , putMVar

86 , putMVar

84 , takeMVar

87 , takeMVar

85 )

88 )

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

90+ ( newThreadPool

91+ , submitTask

92+ , lookupWorker

93+ , ThreadPool

94+ )

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

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

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

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

88import Control.Concurrent.STM

97import Control.Concurrent.STM

@@ -218,8 +227,18 @@ import qualified Control.Distributed.Process.Internal.StrictContainerAccessors a

218 ( mapMaybe

227 ( mapMaybe

219 , mapDefault

228 , mapDefault

220 )

229 )

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

231+import GHC.Base ( maskAsyncExceptions# )

232+

221import Unsafe.Coerce

233import Unsafe.Coerce

222

234

235+-- Remove these when this is fixed:

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

237+block :: IO a -> IO a

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

239+unblock :: IO a -> IO a

240+unblock = unsafeUnmask

241+

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

242--------------------------------------------------------------------------------

224-- Initialization --

243-- Initialization --

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

244--------------------------------------------------------------------------------

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

249 , _localConnections = Map.empty

268 , _localConnections = Map.empty

250 }

269 }

251 ctrlChan <- newChan

270 ctrlChan <- newChan

271+ sendPool <- newThreadPool

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

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

253 , localEndPoint = endPoint

273 , localEndPoint = endPoint

254 , localState = state

274 , localState = state

255 , localCtrlChan = ctrlChan

275 , localCtrlChan = ctrlChan

256 , localEventBus = MxEventBusInitialising

276 , localEventBus = MxEventBusInitialising

257 , remoteTable = rtable

277 , remoteTable = rtable

278+ , localSendPool = sendPool

258 }

279 }

259 tracedNode <- startMxAgent node

280 tracedNode <- startMxAgent node

260

281

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

361

382

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

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

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

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

364−forkProcess node proc =

385+forkProcess node proc = do

386+ ms <- getMaskingState

365− modifyMVarMasked (localState node) startProcess

387+ modifyMVarMasked (localState node) (startProcess ms)

366 where

388 where

389+ startProcess :: MaskingState

390+ -> LocalNodeState

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

391+ -> IO (LocalNodeState, ProcessId)

368− startProcess (LocalNodeValid vst) = do

392+ startProcess ms (LocalNodeValid vst) = do

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

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

370 , lpidUnique = vst ^. localPidUnique

394 , lpidUnique = vst ^. localPidUnique

371 }

395 }

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

387 , processThread = tid

411 , processThread = tid

388 , processNode = node

412 , processNode = node

389 }

413 }

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

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

416+ let unmask = case ms of

417+ Unmasked -> unblock

418+ MaskedInterruptible -> block

419+ MaskedUninterruptible -> id

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

420+ tid' <- uninterruptibleMask_ $ forkIO $ do

391 reason <- Exception.catches

421 reason <- Exception.catches

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

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

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

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

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

433 $ vst

463 $ vst

434 , pid

464 , pid

435 )

465 )

436− startProcess LocalNodeClosed =

466+ startProcess _ LocalNodeClosed =

437 throwIO $ NodeClosedException $ localNodeId node

467 throwIO $ NodeClosedException $ localNodeId node

438

468

439 cleanupProcess :: ProcessId

469 cleanupProcess :: ProcessId

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

580 , ctrlMsgSignal = Died nid DiedDisconnect

610 , ctrlMsgSignal = Died nid DiedDisconnect

581 }

611 }

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

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

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

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

583 closeImplicitReconnections node nid

615 closeImplicitReconnections node nid

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

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

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

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

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

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

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

635 }

667 }

636

668

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

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

671+ where

672+ fork io =

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

674+

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

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

638 deriving ( Applicative

676 deriving ( Applicative

639 , Functor

677 , Functor

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

644 )

682 )

645

683

646initNCState :: NCState

684initNCState :: NCState

647−initNCState = NCState { _links = Map.empty

685+initNCState = NCState

686+ { _links = Map.empty

648− , _monitors = Map.empty

687+ , _monitors = Map.empty

649− , _registeredHere = Map.empty

688+ , _registeredHere = Map.empty

650− , _registeredOnNodes = Map.empty

689+ , _registeredOnNodes = Map.empty

651− }

690+ }

652

691

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

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

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

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

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

666 node <- ask

705 node <- ask

667 if processNodeId pid == localNodeId node

706 if processNodeId pid == localNodeId node

668 then ncEffectLocalSend node pid msg

707 then ncEffectLocalSend node pid msg

669− else liftIO $ sendBinary node

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

670 (NodeIdentifier $ localNodeId node)

709 (NodeIdentifier $ localNodeId node)

671 (NodeIdentifier $ processNodeId pid)

710 (NodeIdentifier $ processNodeId pid)

672 WithImplicitReconnect

711 WithImplicitReconnect

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

679 node <- ask

718 node <- ask

680 liftIO $ if to == localNodeId node

719 liftIO $ if to == localNodeId node

681 then writeChan (localCtrlChan node) $! msg

720 then writeChan (localCtrlChan node) $! msg

682− else sendBinary node

721+ else submitSendPool node to $ sendBinary node

683 (NodeIdentifier $ localNodeId node)

722 (NodeIdentifier $ localNodeId node)

684 (NodeIdentifier to)

723 (NodeIdentifier to)

685 WithImplicitReconnect

724 WithImplicitReconnect