45e8204 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Remove threadId from EventF

45e8204Harendra Kumar9 years ago

src/Duct/AsyncT.hs+2−3

compare with master

unifiedsplit

@@ -23,7 +23,7 @@ module Duct.AsyncT

23where

25import Control.Applicative (Alternative (..))

26−import Control.Concurrent (ThreadId, myThreadId)

26+import Control.Concurrent (ThreadId)

27import Control.Concurrent.STM (TChan, atomically, newTChan,

28 readTChan)

29import Control.Exception (SomeException, catch)

@@ -365,10 +365,9 @@ runTransient t = do

365 zombieChan <- liftIO $ atomically newTChan

366 pendingRef <- liftIO $ newIORef []

367 credit <- liftIO $ newIORef maxBound

368− th <- liftIO $ myThreadId

369

368

370 r <- runStateT (runAsyncT t) $ initEventF

369 r <- runStateT (runAsyncT t) $ initEventF

371− (empty :: AsyncT m a) th zombieChan pendingRef credit

370+ (empty :: AsyncT m a) zombieChan pendingRef credit

372

371

373 waitForChildren zombieChan pendingRef

372 waitForChildren zombieChan pendingRef

374 return r

373 return r

src/Duct/Event.hs+1−4

@@ -40,7 +40,6 @@ data EventF = forall m a b. EventF

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

42 , mfSequence :: Int

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

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

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

46 -- need the parent itself?

45 -- need the parent itself?

@@ -87,18 +86,16 @@ data EventF = forall m a b. EventF

88initEventF

87initEventF

89 :: m a

88 :: m a

90− -> ThreadId

91 -> TChan ThreadId

89 -> TChan ThreadId

92 -> IORef [ThreadId]

90 -> IORef [ThreadId]

93 -> IORef Int

91 -> IORef Int

94 -> EventF

92 -> EventF

95−initEventF x th zombieChan pending credit =

93+initEventF x zombieChan pending credit =

96 EventF { event = mempty

94 EventF { event = mempty

97 , xcomp = x

95 , xcomp = x

98 , fcomp = []

96 , fcomp = []

99 , mfData = mempty

97 , mfData = mempty

100 , mfSequence = 0

98 , mfSequence = 0

101− , threadId = th

102 , parent = Nothing

99 , parent = Nothing

103 , zombieChannel = zombieChan

100 , zombieChannel = zombieChan

104 , pendingThreads = pending

101 , pendingThreads = pending

src/Duct/Threads.hs+4−13

@@ -119,10 +119,7 @@ forkFinally1 action and_then =

119forkIt :: (MonadBaseControl IO m, MonadIO m) => EventF -> (EventF -> m t) -> m ()

120forkIt current proc = do

121 child <- newChildCont current

122− tid <- forkFinally1 (runInChild proc child)

122+ tid <- forkFinally1 (proc child) (beforeExit child)

123− $ \res -> do

124− thId <- myThreadId

125− beforeExit child{threadId = thId} res

126 updatePendingThreads current tid

123 updatePendingThreads current tid

127

124

128 where

125 where

@@ -144,11 +141,6 @@ forkIt current proc = do

144 , zombieChannel = chan

141 , zombieChannel = chan

145 }

142 }

146

143

147− runInChild m child = do

148− th <- liftIO $ myThreadId

149− dbg $ "Forked: child " ++ show th

150− m child{threadId = th}

151−

152 beforeExit cur res = do

144 beforeExit cur res = do

153 case res of

145 case res of

154 Left _exc -> liftIO $ killChildren cur

146 Left _exc -> liftIO $ killChildren cur

@@ -161,7 +153,8 @@ forkIt current proc = do

161 -- We are guaranteed to have a parent because we have been explicitly

153 -- We are guaranteed to have a parent because we have been explicitly

162 -- forked by some parent.

154 -- forked by some parent.

163 let p = fromJust (parent cur)

155 let p = fromJust (parent cur)

156+ tid <- myThreadId

164− liftIO $ atomically $ writeTChan (zombieChannel p) (threadId cur)

157+ liftIO $ atomically $ writeTChan (zombieChannel p) tid

165

158

166forkMaybe :: (MonadBaseControl IO m, MonadIO m) => EventF -> (EventF -> m ()) -> m ()

159forkMaybe :: (MonadBaseControl IO m, MonadIO m) => EventF -> (EventF -> m ()) -> m ()

167forkMaybe current toRun = do

160forkMaybe current toRun = do

@@ -179,9 +172,7 @@ forkMaybe current toRun = do

179 waitForOneChild (zombieChannel current)

172 waitForOneChild (zombieChannel current)

180 (pendingThreads current)

173 (pendingThreads current)

181 forkMaybe current toRun

174 forkMaybe current toRun

182− True -> do

183− dbg $ "Forking: parent " ++ show (threadId current)

184− forkIt current toRun

175+ True -> forkIt current toRun

185

176

186-- | 'StreamData' represents a task in a task stream being generated.

177-- | 'StreamData' represents a task in a task stream being generated.

187data StreamData a =

178data StreamData a =