master...56e5305 · distributed-process

git.iko.soy/distributed-process

on 56e5305 help sign in

distributed-process.cabal+1

since splitsnapshots

unifiedsplit

@@ -68,6 +68,7 @@ Library

68 Control.Distributed.Process.Internal.StrictContainerAccessors,

69 Control.Distributed.Process.Internal.StrictList,

70 Control.Distributed.Process.Internal.StrictMVar,

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

71 Control.Distributed.Process.Internal.Types,

72 Control.Distributed.Process.Internal.Types,

72 Control.Distributed.Process.Internal.WeakTQueue,

73 Control.Distributed.Process.Internal.WeakTQueue,

73 Control.Distributed.Process.Management,

74 Control.Distributed.Process.Management,

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

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

137 , modifyMVar

138 , modifyMVar_

139 )

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

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

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

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

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

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

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

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

267 -- | Runtime lookup table for supporting closures

268 -- | Runtime lookup table for supporting closures

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

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

269 , remoteTable :: !RemoteTable

270 , remoteTable :: !RemoteTable

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

272+ , localSendPool :: ThreadPool NodeId ThreadId

270 }

273 }

271

274

272data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

275data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

src/Control/Distributed/Process/Node.hs+18−2

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

91 , putMVar

92 , takeMVar

93 )

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

95+ ( newThreadPool

96+ , submitTask

97+ , lookupWorker

98+ , ThreadPool

99+ )

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

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

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

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

96import Control.Concurrent.STM

102import Control.Concurrent.STM

@@ -264,12 +270,14 @@ createBareLocalNode endPoint rtable = do

264 , _localConnections = Map.empty

270 , _localConnections = Map.empty

265 }

271 }

266 ctrlChan <- newChan

272 ctrlChan <- newChan

273+ sendPool <- newThreadPool

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

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

268 , localEndPoint = endPoint

275 , localEndPoint = endPoint

269 , localState = state

276 , localState = state

270 , localCtrlChan = ctrlChan

277 , localCtrlChan = ctrlChan

271 , localEventBus = MxEventBusInitialising

278 , localEventBus = MxEventBusInitialising

272 , remoteTable = rtable

279 , remoteTable = rtable

280+ , localSendPool = sendPool

273 }

281 }

274 tracedNode <- startMxAgent node

282 tracedNode <- startMxAgent node

275

283

@@ -613,6 +621,8 @@ handleIncomingMessages node = go initConnectionState

613 , ctrlMsgSignal = Died nid DiedDisconnect

621 , ctrlMsgSignal = Died nid DiedDisconnect

614 }

622 }

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

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

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

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

616 closeImplicitReconnections node nid

626 closeImplicitReconnections node nid

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

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

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

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

@@ -669,6 +679,12 @@ data NCState = NCState

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

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

670 }

680 }

671

681

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

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

684+ where

685+ fork io =

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

687+

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

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

673 deriving ( Applicative

689 deriving ( Applicative

674 , Functor

690 , Functor

@@ -704,7 +720,7 @@ ncSendToProcessAndTrace shouldTrace pid msg = do

704 node <- ask

720 node <- ask

705 if processNodeId pid == localNodeId node

721 if processNodeId pid == localNodeId node

706 then ncEffectLocalSendAndTrace shouldTrace node pid msg

722 then ncEffectLocalSendAndTrace shouldTrace node pid msg

707− else liftIO $ sendBinary node

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

708 (NodeIdentifier $ localNodeId node)

724 (NodeIdentifier $ localNodeId node)

709 (NodeIdentifier $ processNodeId pid)

725 (NodeIdentifier $ processNodeId pid)

710 WithImplicitReconnect

726 WithImplicitReconnect

@@ -717,7 +733,7 @@ ncSendToNode to msg = do

717 node <- ask

733 node <- ask

718 liftIO $ if to == localNodeId node

734 liftIO $ if to == localNodeId node

719 then writeChan (localCtrlChan node) $! msg

735 then writeChan (localCtrlChan node) $! msg

720− else sendBinary node

736+ else submitSendPool node to $ sendBinary node

721 (NodeIdentifier $ localNodeId node)

737 (NodeIdentifier $ localNodeId node)

722 (NodeIdentifier to)

738 (NodeIdentifier to)

723 WithImplicitReconnect

739 WithImplicitReconnect