git.iko.soy/distributed-process
on master help sign in
Fix for DP-110. The NC does not block anymore on network operations.

A pool of threads is used for network operations. One thread is used per NodeId. Conflicts: src/Control/Distributed/Process/Internal/Types.hs src/Control/Distributed/Process/Node.hs

commit46228c2aef79e3cbebee0c6062089c6c5f171b13 authorFacundo Domínguez <facundo.dominguez@tweag.io> dateThu 23 Apr 2015, 11:25 (11 years ago) parent53c7fe6
4 files changed+106−7
compare with master
unifiedsplit
distributed-process.cabal+1 src/Control/Distributed/Process/Internal/ThreadPool.hs+78 src/Control/Distributed/Process/Internal/Types.hs+3 src/Control/Distributed/Process/Node.hs+24−7
distributed-process.cabal+1
@@ -69,6 +69,7 @@ Library
6969 Control.Distributed.Process.Internal.StrictContainerAccessors,
7070 Control.Distributed.Process.Internal.StrictList,
7171 Control.Distributed.Process.Internal.StrictMVar,
72+ Control.Distributed.Process.Internal.ThreadPool,
7273 Control.Distributed.Process.Internal.Types,
7374 Control.Distributed.Process.Internal.WeakTQueue,
7475 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
137137 , modifyMVar
138138 , modifyMVar_
139139 )
140+import Control.Distributed.Process.Internal.ThreadPool
140141import Control.Distributed.Process.Internal.WeakTQueue (TQueue)
141142import Control.Distributed.Static (RemoteTable, Closure)
142143import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)
@@ -266,6 +267,8 @@ data LocalNode = LocalNode
266267 -- | Runtime lookup table for supporting closures
267268 -- TODO: this should be part of the CH state, not the local endpoint state
268269 , remoteTable :: !RemoteTable
270+ -- The pool of threads and queues to send messages
271+ , localSendPool :: ThreadPool NodeId ThreadId
269272 }
270273
271274data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect
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