git.iko.soy/streamly
on master help sign in
Remove the parent pointer from EventF
3744363Harendra Kumar
src/Duct/AsyncT.hs+3−3
compare with master
unifiedsplit
@@ -362,14 +362,14 @@ waitForChildren chan pendingRef = do
362-- | Run a transient computation with a default initial state
362-- | Run a transient computation with a default initial state
363runTransient :: forall m a. MonadIO m => AsyncT m a -> m (Maybe a, EventF)
363runTransient :: forall m a. MonadIO m => AsyncT m a -> m (Maybe a, EventF)
364runTransient t = do
364runTransient t = do
365− zombieChan <- liftIO $ atomically newTChan
365+ childChan <- liftIO $ atomically newTChan
366 pendingRef <- liftIO $ newIORef []
366 pendingRef <- liftIO $ newIORef []
367 credit <- liftIO $ newIORef maxBound
367 credit <- liftIO $ newIORef maxBound
368
368
369 r <- runStateT (runAsyncT t) $ initEventF
369 r <- runStateT (runAsyncT t) $ initEventF
370− (empty :: AsyncT m a) zombieChan pendingRef credit
370+ (empty :: AsyncT m a) childChan pendingRef credit
371
371
372− waitForChildren zombieChan pendingRef
372+ waitForChildren childChan pendingRef
373 return r
373 return r
374
374
375-- | Run an 'AsyncT m' computation. Returns the result of the computation or
375-- | Run an 'AsyncT m' computation. Returns the result of the computation or
src/Duct/Event.hs+7−8
@@ -41,9 +41,8 @@ data EventF = forall m a b. EventF
41
41
42 , mfSequence :: Int
42 , mfSequence :: Int
43
43
44− , parent :: Maybe EventF -- We need the channel of the parent, but do we
44+ , parentChannel :: Maybe (TChan ThreadId)
45− -- need the parent itself?
45+ -- ^ Our parent thread's channel to communicate to when we die
46− -- ^ The parent of this thread
47
46
48 -- Thread creation and cleanup handling. When a new thread is created the
47 -- Thread creation and cleanup handling. When a new thread is created the
49 -- parent records it in the 'children' field and 'threadCredit' is
48 -- parent records it in the 'children' field and 'threadCredit' is
@@ -55,8 +54,8 @@ data EventF = forall m a b. EventF
55 -- parent thread waits on the channel until all its children are cleaned
54 -- parent thread waits on the channel until all its children are cleaned
56 -- up.
55 -- up.
57
56
58− , zombieChannel :: TChan ThreadId
57+ , childChannel :: TChan ThreadId
59− -- ^ A channel for the immediate children to communicate to when they die.
58+ -- ^ A channel for the immediate children to communicate to us when they die.
60 -- Each thread has its own dedicated channel for its children
59 -- Each thread has its own dedicated channel for its children
61
60
62 -- We always track the child threads, otherwise the programmer needs to
61 -- We always track the child threads, otherwise the programmer needs to
@@ -90,14 +89,14 @@ initEventF
90 -> IORef [ThreadId]
89 -> IORef [ThreadId]
91 -> IORef Int
90 -> IORef Int
92 -> EventF
91 -> EventF
93−initEventF x zombieChan pending credit =
92+initEventF x childChan pending credit =
94 EventF { event = mempty
93 EventF { event = mempty
95 , xcomp = x
94 , xcomp = x
96 , fcomp = []
95 , fcomp = []
97 , mfData = mempty
96 , mfData = mempty
98 , mfSequence = 0
97 , mfSequence = 0
99− , parent = Nothing
98+ , parentChannel = Nothing
100− , zombieChannel = zombieChan
99+ , childChannel = childChan
101 , pendingThreads = pending
100 , pendingThreads = pending
102 , threadCredit = credit }
101 , threadCredit = credit }
103
102
src/Duct/Threads.hs+7−7
@@ -129,16 +129,16 @@ forkIt current proc = do
129 -- update the new thread before reclaiming zombies so that if it exited
129 -- update the new thread before reclaiming zombies so that if it exited
130 -- already reclaim finds it in the list and does not panic.
130 -- already reclaim finds it in the list and does not panic.
131 liftIO $ modifyIORef (pendingThreads cur) $ (\ts -> tid:ts)
131 liftIO $ modifyIORef (pendingThreads cur) $ (\ts -> tid:ts)
132− tryReclaimZombies (zombieChannel cur) (pendingThreads cur)
132+ tryReclaimZombies (childChannel cur) (pendingThreads cur)
133
133
134 newChildCont cur = do
134 newChildCont cur = do
135 pendingRef <- liftIO $ newIORef []
135 pendingRef <- liftIO $ newIORef []
136 chan <- liftIO $ atomically newTChan
136 chan <- liftIO $ atomically newTChan
137 -- shares the threadCredit of the parent by default
137 -- shares the threadCredit of the parent by default
138 return $ cur
138 return $ cur
139− { parent = Just cur
139+ { parentChannel = Just (childChannel cur)
140 , pendingThreads = pendingRef
140 , pendingThreads = pendingRef
141− , zombieChannel = chan
141+ , childChannel = chan
142 }
142 }
143
143
144 beforeExit cur res = do
144 beforeExit cur res = do
@@ -148,13 +148,13 @@ forkIt current proc = do
148 Right _ -> -- collect the results?
148 Right _ -> -- collect the results?
149 return ()
149 return ()
150
150
151− waitForChildren (zombieChannel cur) (pendingThreads cur)
151+ waitForChildren (childChannel cur) (pendingThreads cur)
152 signalQSemB (threadCredit cur)
152 signalQSemB (threadCredit cur)
153 -- 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
154 -- forked by some parent.
154 -- forked by some parent.
155− let p = fromJust (parent cur)
155+ let p = fromJust (parentChannel cur)
156 tid <- myThreadId
156 tid <- myThreadId
157− liftIO $ atomically $ writeTChan (zombieChannel p) tid
157+ liftIO $ atomically $ writeTChan p tid
158
158
159forkMaybe :: (MonadBaseControl IO m, MonadIO m) => EventF -> (EventF -> m ()) -> m ()
159forkMaybe :: (MonadBaseControl IO m, MonadIO m) => EventF -> (EventF -> m ()) -> m ()
160forkMaybe current toRun = do
160forkMaybe current toRun = do
@@ -169,7 +169,7 @@ forkMaybe current toRun = do
169 -- special handling for those cases. Add those to
169 -- special handling for those cases. Add those to
170 -- unreclaimable list? And always execute them in an
170 -- unreclaimable list? And always execute them in an
171 -- async thread, cannot use sync for those.
171 -- async thread, cannot use sync for those.
172− waitForOneChild (zombieChannel current)
172+ waitForOneChild (childChannel current)
173 (pendingThreads current)
173 (pendingThreads current)
174 forkMaybe current toRun
174 forkMaybe current toRun
175 True -> forkIt current toRun
175 True -> forkIt current toRun