git.iko.soy/streamly
on master help sign in
Gain some perf by moving code out of fast path
62bca27Harendra Kumar
benchmark/Main.hs+1−2
compare with master
unifiedsplit
@@ -104,8 +104,7 @@ asyncly_basic :: IO Int
104asyncly_basic = do
104asyncly_basic = do
105 writeIORef count 0
105 writeIORef count 0
106 xs <- A.toList $ do
106 xs <- A.toList $ do
107− -- A.each [1..1000000 :: Int]
108− (A.for [1..1000000 :: Int] $ \x ->
107+ (A.forEachWith (A.<|) [1..1000000 :: Int] $ \x ->
109 return x
108 return x
110 >>= afilter even
109 >>= afilter even
111 >>= amap (+1)
110 >>= amap (+1)
src/Asyncly/AsyncT.hs+12−9
@@ -265,6 +265,17 @@ handleException e ctx tid = do
265 liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
265 liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
266 throwM e
266 throwM e
267
267
268+{-# NOINLINE sendWorkerWait #-}
269+sendWorkerWait :: MonadAsync m => Context m a -> m ()
270+sendWorkerWait ctx = do
271+ liftIO $ threadDelay 4
272+ let workQ = workQueue ctx
273+ outQ = outputQueue ctx
274+ workQEmpty <- liftIO $ atomically $ isEmptyTBQueue workQ
275+ outQEmpty <- liftIO $ atomically $ isEmptyTBQueue outQ
276+ when (not workQEmpty && outQEmpty) $ pushWorker ctx
277+ void $ liftIO $ atomically $ peekTBQueue (outputQueue ctx)
278+
268-- We re-raise any exceptions received from the child threads, that way
279-- We re-raise any exceptions received from the child threads, that way
269-- exceptions get propagated to the top level computation and can be handled
280-- exceptions get propagated to the top level computation and can be handled
270-- there.
281-- there.
@@ -279,15 +290,7 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
279
290
280 res <- liftIO $ atomically $ tryReadTBQueue (outputQueue ctx)
291 res <- liftIO $ atomically $ tryReadTBQueue (outputQueue ctx)
281 case res of
292 case res of
282− Nothing -> do
283− liftIO $ threadDelay 4
284− let workQ = workQueue ctx
285− outQ = outputQueue ctx
286− workQEmpty <- liftIO $ atomically $ isEmptyTBQueue workQ
287− outQEmpty <- liftIO $ atomically $ isEmptyTBQueue outQ
288− when (not workQEmpty && outQEmpty) $ pushWorker ctx
289− void $ liftIO $ atomically $ peekTBQueue (outputQueue ctx)
290− continue
293+ Nothing -> sendWorkerWait ctx >> continue
291 Just ev ->
294 Just ev ->
292 case ev of
295 case ev of
293 ChildYield a -> yield a
296 ChildYield a -> yield a