git.iko.soy/streamly
on master help sign in
Perf - do not save pending work in the context
36ecdf8Harendra Kumar
src/Asyncly/AsyncT.hs+16−20
compare with master
unifiedsplit
@@ -82,10 +82,9 @@ data ChildEvent a =
82-- State threaded around the monad for thread management
82-- State threaded around the monad for thread management
83------------------------------------------------------------------------------
83------------------------------------------------------------------------------
84
84
85−data Context m a =
85+data Context a =
86 Context { childChannel :: TChan (ChildEvent a)
86 Context { childChannel :: TChan (ChildEvent a)
87 , pullSide :: Bool
87 , pullSide :: Bool
88− , pendingWork :: IORef (AsyncT m a)
89 }
88 }
90
89
91-- The 'Maybe (AsyncT m a)' is redundant as we can use 'stop' value for the
90-- The 'Maybe (AsyncT m a)' is redundant as we can use 'stop' value for the
@@ -99,9 +98,9 @@ data Context m a =
99newtype AsyncT m a =
98newtype AsyncT m a =
100 AsyncT {
99 AsyncT {
101 runAsyncT :: forall r.
100 runAsyncT :: forall r.
102− Maybe (Context m a) -- state
101+ Maybe (Context a) -- state
103 -> m r -- stop
102 -> m r -- stop
104− -> (a -> Maybe (Context m a) -> Maybe (AsyncT m a) -> m r) -- yield
103+ -> (a -> Maybe (Context a) -> Maybe (AsyncT m a) -> m r) -- yield
105 -> m r
104 -> m r
106 }
105 }
107
106
@@ -178,7 +177,7 @@ doFork action preExit =
178 return ()
177 return ()
179
178
180{-# NOINLINE push #-}
179{-# NOINLINE push #-}
181−push :: MonadIO m => Context m a -> AsyncT m a -> m ()
180+push :: MonadIO m => Context a -> AsyncT m a -> m ()
182push context action = run (Just context) action
181push context action = run (Just context) action
183
182
184 where
183 where
@@ -211,7 +210,7 @@ handlePushException pchan e = do
211
210
212{-# NOINLINE pushSideDispatch #-}
211{-# NOINLINE pushSideDispatch #-}
213pushSideDispatch :: MonadAsync m
212pushSideDispatch :: MonadAsync m
214− => Context m a -> AsyncT m a -> AsyncT m a -> AsyncT m a
213+ => Context a -> AsyncT m a -> AsyncT m a -> AsyncT m a
215pushSideDispatch ctx m1 m2 = AsyncT $ \_ stp yld -> do
214pushSideDispatch ctx m1 m2 = AsyncT $ \_ stp yld -> do
216 let chan = childChannel ctx
215 let chan = childChannel ctx
217 _ <- doFork (push ctx m1) (handlePushException chan)
216 _ <- doFork (push ctx m1) (handlePushException chan)
@@ -223,15 +222,14 @@ pushSideDispatch ctx m1 m2 = AsyncT $ \_ stp yld -> do
223-- there.
222-- there.
224{-# NOINLINE pullDispatch #-}
223{-# NOINLINE pullDispatch #-}
225pullDispatch :: (MonadIO m, MonadThrow m)
224pullDispatch :: (MonadIO m, MonadThrow m)
226− => Context m a -> Bool -> AsyncT m a
225+ => Context a -> AsyncT m a -> Bool -> AsyncT m a
227−pullDispatch ctx dispatch = AsyncT $ \_ stp yld -> do
226+pullDispatch ctx m dispatch = AsyncT $ \_ stp yld -> do
228 let chan = childChannel ctx
227 let chan = childChannel ctx
229 eHandler e = handleException e stp
228 eHandler e = handleException e stp
230 evHandler ev = handleEvent ev stp yld
229 evHandler ev = handleEvent ev stp yld
231 maybeEvHandler ev = maybe (continue stp yld) evHandler ev
230 maybeEvHandler ev = maybe (continue stp yld) evHandler ev
232
231
233 if dispatch then do
232 if dispatch then do
234− m <- liftIO $ readIORef (pendingWork ctx)
235 (runAsyncT m) (Just ctx) stp yld
233 (runAsyncT m) (Just ctx) stp yld
236 else do
234 else do
237 res <- liftIO $ try $ atomically $ tryReadTChan chan
235 res <- liftIO $ try $ atomically $ tryReadTChan chan
@@ -245,13 +243,13 @@ pullDispatch ctx dispatch = AsyncT $ \_ stp yld -> do
245 Nothing -> throwM e
243 Nothing -> throwM e
246
244
247 {-# INLINE continue #-}
245 {-# INLINE continue #-}
248− continue stp yld = (runAsyncT (pullDispatch ctx True)) (Just ctx) stp yld
246+ continue stp yld = (runAsyncT (pullDispatch ctx m True)) (Just ctx) stp yld
249
247
250 {-# INLINE handleEvent #-}
248 {-# INLINE handleEvent #-}
251 handleEvent ev stp yld =
249 handleEvent ev stp yld =
252 case ev of
250 case ev of
253− ChildYield a -> yld a Nothing (Just (pullDispatch ctx False))
251+ ChildYield a -> yld a Nothing (Just (pullDispatch ctx m False))
254− ChildDone _ a -> yld a Nothing (Just (pullDispatch ctx True))
252+ ChildDone _ a -> yld a Nothing (Just (pullDispatch ctx m True))
255 -- XXX kill threads
253 -- XXX kill threads
256 ChildStop _ e -> maybe (continue stp yld) throwM e
254 ChildStop _ e -> maybe (continue stp yld) throwM e
257
255
@@ -259,13 +257,12 @@ pullDispatch ctx dispatch = AsyncT $ \_ stp yld -> do
259-- m2 in the parent thread. Any exceptions are also pushed to the channel.
257-- m2 in the parent thread. Any exceptions are also pushed to the channel.
260{-# INLINE pullSideDispatch #-}
258{-# INLINE pullSideDispatch #-}
261pullSideDispatch :: MonadAsync m
259pullSideDispatch :: MonadAsync m
262− => Context m a -> AsyncT m a -> AsyncT m a -> AsyncT m a
260+ => Context a -> AsyncT m a -> AsyncT m a -> AsyncT m a
263pullSideDispatch ctx m1 m2 = AsyncT $ \_ stp yld -> do
261pullSideDispatch ctx m1 m2 = AsyncT $ \_ stp yld -> do
264 let chan = childChannel ctx
262 let chan = childChannel ctx
265 _ <- doFork (push (ctx {pullSide = False}) m1) (handlePushException chan)
263 _ <- doFork (push (ctx {pullSide = False}) m1) (handlePushException chan)
266 -- liftIO $ threadDelay 0
264 -- liftIO $ threadDelay 0
267− liftIO $ writeIORef (pendingWork ctx) m2
268− (runAsyncT (pullDispatch ctx False)) Nothing stp yld
265+ (runAsyncT (pullDispatch ctx m2 False)) Nothing stp yld
269
266
270{-# NOINLINE pullDrain #-}
267{-# NOINLINE pullDrain #-}
271pullDrain :: (MonadIO m, MonadThrow m) => TChan (ChildEvent a) -> AsyncT m a
268pullDrain :: (MonadIO m, MonadThrow m) => TChan (ChildEvent a) -> AsyncT m a
@@ -300,17 +297,16 @@ pullDrain chan = AsyncT $ \_ stp yld -> do
300{-# NOINLINE pullFork #-}
297{-# NOINLINE pullFork #-}
301pullFork :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
298pullFork :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
302pullFork m1 m2 = AsyncT $ \_ stp yld -> do
299pullFork m1 m2 = AsyncT $ \_ stp yld -> do
303− ctx <- liftIO $ newContext (m1 <|> m2)
300+ ctx <- liftIO $ newContext
301+ let m = pullDispatch ctx ((m1 <|> m2) <> pullDrain (childChannel ctx)) True
304− (runAsyncT (pullDispatch ctx True)) Nothing stp yld
302+ (runAsyncT m) Nothing stp yld
305
303
306 where
304 where
307
305
308− newContext m = do
306+ newContext = do
309 channel <- atomically newTChan
307 channel <- atomically newTChan
310− work <- liftIO $ newIORef (m <> pullDrain channel)
311 return $ Context { childChannel = channel
308 return $ Context { childChannel = channel
312 , pullSide = True
309 , pullSide = True
313− , pendingWork = work
314 }
310 }
315
311
316-- Concurrency rate control. Our objective is to create more threads on
312-- Concurrency rate control. Our objective is to create more threads on