df9febd · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Use threadId tracking instead of EventF for children

df9febdHarendra Kumar9 years ago

src/Duct/AsyncT.hs+4−6

compare with master

unifiedsplit

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

39import Data.Dynamic (Typeable)

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

41 writeIORef)

42+import Data.List (delete)

42import Data.Maybe (isJust, isNothing)

43import Data.Maybe (isJust, isNothing)

43import Unsafe.Coerce (unsafeCoerce)

44import Unsafe.Coerce (unsafeCoerce)

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

339-- Thread management

340-- Thread management

340------------------------------------------------------------------------------

341------------------------------------------------------------------------------

341

342

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

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

343drainChildren chan pending =

344drainChildren chan pending =

344 if length pending == 0

345 if length pending == 0

345 then return ()

346 then return ()

346 else do

347 else do

347 tid <- liftIO $ atomically $ readTChan chan

348 tid <- liftIO $ atomically $ readTChan chan

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

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

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

350+ drainChildren chan $ delete tid pending

350

351

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

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

352waitForChildren chan pendingRef = do

353waitForChildren chan pendingRef = do

353 pending <- liftIO $ readIORef pendingRef

354 pending <- liftIO $ readIORef pendingRef

354− tid <- liftIO myThreadId

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

356 drainChildren chan pending

355 drainChildren chan pending

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

358 liftIO $ writeIORef pendingRef []

356 liftIO $ writeIORef pendingRef []

359

357

360------------------------------------------------------------------------------

358------------------------------------------------------------------------------

src/Duct/Event.hs+16−12

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

26type SData = ()

28+-- XXX rename to Context

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

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

29data EventF = forall m a b. EventF

30data 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?

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

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

33 , xcomp :: m a

35 , xcomp :: m a

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

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

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

40 , mfSequence :: Int

42 , mfSequence :: Int

41− , threadId :: ThreadId

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

43− , parent :: Maybe EventF

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

46+ -- need the parent itself?

44 -- ^ The parent of this thread

47 -- ^ The parent of this thread

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

66 --

69 --

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

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

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

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

70− , pendingThreads :: IORef [EventF]

73+ , pendingThreads :: IORef [ThreadId]

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

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

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

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

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

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

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

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

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

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

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

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

79 --

83 --

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

85 :: m a

89 :: m a

86 -> ThreadId

90 -> ThreadId

87 -> TChan ThreadId

91 -> TChan ThreadId

88− -> IORef [EventF]

92+ -> IORef [ThreadId]

89 -> IORef Int

93 -> IORef Int

90 -> EventF

94 -> EventF

91initEventF x th zombieChan pending credit =

95initEventF x th zombieChan pending credit =

src/Duct/Threads.hs+15−14

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

23import Data.IORef

24 (IORef, atomicModifyIORef, readIORef, writeIORef, modifyIORef,

25 newIORef)

26+import Data.List (delete)

26import Data.Maybe (fromJust)

27import Data.Maybe (fromJust)

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

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

28import Control.Monad.State

29import Control.Monad.State

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

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

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

49------------------------------------------------------------------------------

50------------------------------------------------------------------------------

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

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

52tryReclaimZombies chan pendingRef = do

53tryReclaimZombies chan pendingRef = do

53 pending <- liftIO $ readIORef pendingRef

54 pending <- liftIO $ readIORef pendingRef

54− if length pending == 0

55+ case pending of

55− then return ()

56+ [] -> return ()

56− else do

57+ _ -> do

57 mtid <- liftIO $ atomically $ tryReadTChan chan

58 mtid <- liftIO $ atomically $ tryReadTChan chan

58 case mtid of

59 case mtid of

59 Nothing -> return ()

60 Nothing -> return ()

60 Just tid -> do

61 Just tid -> do

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

62+ liftIO $ writeIORef pendingRef $ delete tid pending

62 tryReclaimZombies chan pendingRef

63 tryReclaimZombies chan pendingRef

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

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

65waitForOneChild chan pendingRef = do

66waitForOneChild chan pendingRef = do

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

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

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

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

68 tid <- liftIO $ atomically $ readTChan chan

69 tid <- liftIO $ atomically $ readTChan chan

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

70+ liftIO $ modifyIORef pendingRef $ delete tid

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

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

72killChildren :: EventF -> IO ()

73killChildren :: EventF -> IO ()

73killChildren ctx = do

74killChildren ctx = do

74 ths <- readIORef (pendingThreads ctx)

75 ths <- readIORef (pendingThreads ctx)

75− mapM_ (killThread . threadId) ths

76+ mapM_ killThread ths

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

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

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

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

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

120 $ \res -> do

121 $ \res -> do

121 thId <- myThreadId

122 thId <- myThreadId

122 beforeExit child{threadId = thId} res

123 beforeExit child{threadId = thId} res

123− updatePendingThreads current child{threadId = tid}

124+ updatePendingThreads current tid

124

125

125 where

126 where

126

127

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

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

128− updatePendingThreads cur child = do

129+ updatePendingThreads cur tid = do

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

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

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

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

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

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

132 tryReclaimZombies (zombieChannel cur) (pendingThreads cur)

133 tryReclaimZombies (zombieChannel cur) (pendingThreads cur)

133

134

134 newChildCont cur = do

135 newChildCont cur = do