df9febd · streamly · git.iko.soy

git.iko.soy/streamly

Use threadId tracking instead of EventF for children

commitdf9febd5bdf201e1cb9553b1ad3638383d45c805 authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 8 Jun 2017, 00:34 (9 years ago) parent1bcf579

3 files changed+35−32

compare with master

unifiedsplit

src/Duct/AsyncT.hs+4−6 src/Duct/Event.hs+16−12 src/Duct/Threads.hs+15−14

src/Duct/AsyncT.hs+4−6

@@ -39,6 +39,7 @@ import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),

3939import Data.Dynamic (Typeable)

4040import Data.IORef (IORef, newIORef, readIORef,

4141 writeIORef)

42+import Data.List (delete)

4243import Data.Maybe (isJust, isNothing)

4344import Unsafe.Coerce (unsafeCoerce)

4445

@@ -339,22 +340,19 @@ instance (MonadBaseControl b m, MonadIO m) => MonadBaseControl b (AsyncT m) wher

339340-- Thread management

340341------------------------------------------------------------------------------

341342

342−drainChildren :: MonadIO m => TChan ThreadId -> [EventF] -> m ()

343+drainChildren :: MonadIO m => TChan ThreadId -> [ThreadId] -> m ()

343344drainChildren chan pending =

344345 if length pending == 0

345346 then return ()

346347 else do

347348 tid <- liftIO $ atomically $ readTChan chan

348349 dbg $ "Reaping child: " ++ show tid

349− drainChildren chan $ filter (\x -> threadId x /= tid) pending

350+ drainChildren chan $ delete tid pending

350351

351−waitForChildren :: MonadIO m => TChan ThreadId -> IORef [EventF] -> m ()

352+waitForChildren :: MonadIO m => TChan ThreadId -> IORef [ThreadId] -> m ()

352353waitForChildren chan pendingRef = do

353354 pending <- liftIO $ readIORef pendingRef

354− tid <- liftIO myThreadId

355− dbg $ "Waiting for children: " ++ show tid

356355 drainChildren chan pending

357− dbg $ "Done: " ++ show tid

358356 liftIO $ writeIORef pendingRef []

359357

360358------------------------------------------------------------------------------

src/Duct/Event.hs+16−12

@@ -25,9 +25,11 @@ import Unsafe.Coerce (unsafeCoerce)

2525

2626type SData = ()

2727

28+-- XXX rename to Context

2829-- | EventF describes the context of a TransientIO computation.

2930data EventF = forall m a b. EventF

30− { event :: Maybe SData -- untyped

31+ { event :: Maybe SData -- untyped, XXX rename to mailbox - can we do

32+ -- away with this?

3133 -- ^ event data to use in a continuation in a new thread

3234

3335 , xcomp :: m a

@@ -38,9 +40,10 @@ data EventF = forall m a b. EventF

3840 -- ^ State data accessed with get or put operations

3941

4042 , mfSequence :: Int

41− , threadId :: ThreadId

43+ , threadId :: ThreadId -- Do we need this?

4244

43− , parent :: Maybe EventF

45+ , parent :: Maybe EventF -- We need the channel of the parent, but do we

46+ -- need the parent itself?

4447 -- ^ The parent of this thread

4548

4649 -- Thread creation and cleanup handling. When a new thread is created the

@@ -57,23 +60,24 @@ data EventF = forall m a b. EventF

5760 -- ^ A channel for the immediate children to communicate to when they die.

5861 -- Each thread has its own dedicated channel for its children

5962

60− -- We always track the child threds otherwise the programmer needs to worry

61− -- about if some threads may remain hanging due because they are stuck in

62− -- an infinite loop or a forever blocked IO. Also, it is not clear whether

63− -- there is a significant gain by disabling tracking. Instead of using the

64− -- threadIds we can also keep a count instead, that will allow us to wait

65− -- for all children to drain but we cannot kill them.

63+ -- We always track the child threads, otherwise the programmer needs to

64+ -- worry about if some threads may remain hanging due because they are

65+ -- stuck in an infinite loop or a forever blocked IO. Also, it is not clear

66+ -- whether there is a significant gain by disabling tracking. Instead of

67+ -- using the threadIds we can also keep a count instead, that will allow us

68+ -- to wait for all children to drain but we cannot kill them.

6669 --

6770 -- We need these as IORefs since we need any changes to these to be

6871 -- reflected back in the calling code that waits for the children.

6972

70− , pendingThreads :: IORef [EventF]

73+ , pendingThreads :: IORef [ThreadId]

7174 -- ^ Active immediate child threads of this thread, modified only by this

7275 -- thread, therefore atomic modification is not needed.

7376

7477 -- By default this limit is shared by the current tree of threads. But at

7578 -- any point the 'threads' primitive can be used to start a new limit for

76− -- the subtree under that primitive.

79+ -- the computation under that primitive.

80+ --

7781 -- Shared across many threads, therefore atomic access is required.

7882 -- This is incremented by the child thread itelf before it exits.

7983 --

@@ -85,7 +89,7 @@ initEventF

8589 :: m a

8690 -> ThreadId

8791 -> TChan ThreadId

88− -> IORef [EventF]

92+ -> IORef [ThreadId]

8993 -> IORef Int

9094 -> EventF

9195initEventF x th zombieChan pending credit =

src/Duct/Threads.hs+15−14

@@ -23,6 +23,7 @@ import Data.Dynamic (Typeable)

2323import Data.IORef

2424 (IORef, atomicModifyIORef, readIORef, writeIORef, modifyIORef,

2525 newIORef)

26+import Data.List (delete)

2627import Data.Maybe (fromJust)

2728import Control.Monad.IO.Class (MonadIO(..))

2829import Control.Monad.State

@@ -48,31 +49,31 @@ runCont EventF { xcomp = x, fcomp = fs } = runAsyncT $ do

4849-- Thread Management (creation, reaping and killing)

4950------------------------------------------------------------------------------

5051

51−tryReclaimZombies :: MonadIO m => TChan ThreadId -> IORef [EventF] -> m ()

52+tryReclaimZombies :: MonadIO m => TChan ThreadId -> IORef [ThreadId] -> m ()

5253tryReclaimZombies chan pendingRef = do

5354 pending <- liftIO $ readIORef pendingRef

54− if length pending == 0

55− then return ()

56− else do

55+ case pending of

56+ [] -> return ()

57+ _ -> do

5758 mtid <- liftIO $ atomically $ tryReadTChan chan

5859 case mtid of

5960 Nothing -> return ()

6061 Just tid -> do

61− liftIO $ writeIORef pendingRef $ filter (\x -> threadId x /= tid) pending

62+ liftIO $ writeIORef pendingRef $ delete tid pending

6263 tryReclaimZombies chan pendingRef

6364

64−waitForOneChild :: MonadIO m => TChan ThreadId -> IORef [EventF] -> m ()

65+waitForOneChild :: MonadIO m => TChan ThreadId -> IORef [ThreadId] -> m ()

6566waitForOneChild chan pendingRef = do

6667 -- XXX assert pending must have at least one element

6768 -- assert that the tid is found in our list

6869 tid <- liftIO $ atomically $ readTChan chan

69− liftIO $ modifyIORef pendingRef $ filter (\x -> threadId x /= tid)

70+ liftIO $ modifyIORef pendingRef $ delete tid

7071

7172-- | kill all the child threads associated with the continuation context

7273killChildren :: EventF -> IO ()

7374killChildren ctx = do

7475 ths <- readIORef (pendingThreads ctx)

75− mapM_ (killThread . threadId) ths

76+ mapM_ killThread ths

7677

7778-- XXX this is not a real semaphore as it does not really block on wait,

7879-- instead it returns whether the value is zero or non-zero.

@@ -120,15 +121,15 @@ forkIt current proc = do

120121 $ \res -> do

121122 thId <- myThreadId

122123 beforeExit child{threadId = thId} res

123− updatePendingThreads current child{threadId = tid}

124+ updatePendingThreads current tid

124125

125126 where

126127

127− updatePendingThreads :: MonadIO m => EventF -> EventF -> m ()

128− updatePendingThreads cur child = do

129− -- update the new thread first so that if it exited already reclaim

130− -- finds it in the list and does not panic.

131− liftIO $ modifyIORef (pendingThreads cur) $ (\ts -> child:ts)

128+ updatePendingThreads :: MonadIO m => EventF -> ThreadId -> m ()

129+ updatePendingThreads cur tid = do

130+ -- update the new thread before reclaiming zombies so that if it exited

131+ -- already reclaim finds it in the list and does not panic.

132+ liftIO $ modifyIORef (pendingThreads cur) $ (\ts -> tid:ts)

132133 tryReclaimZombies (zombieChannel cur) (pendingThreads cur)

133134

134135 newChildCont cur = do