9b2ca30 · streamly · git.iko.soy

git.iko.soy/streamly

Add the "for" parallel API and change "each" to serial

commit9b2ca30312a4c9bb83af1c424b5f5ebd5d08600a authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 17 Aug 2017, 04:42 (9 years ago) parent63f965a

4 files changed+20−12

compare with master

unifiedsplit

benchmark/Main.hs+9−7 src/Asyncly.hs+1 src/Asyncly/AsyncT.hs+1−4 src/Asyncly/RunAsync.hs+9−1

benchmark/Main.hs+9−7

@@ -18,7 +18,7 @@ import qualified Conduit.Simple as S

1818

1919-- Transient

2020import Control.Monad (guard)

21−import Control.Applicative (Alternative)

21+import Control.Applicative (Alternative(..))

2222import Data.IORef (IORef, newIORef, writeIORef)

2323import Data.Atomics (atomicModifyIORefCAS)

2424import System.IO.Unsafe (unsafePerformIO)

@@ -102,12 +102,14 @@ asyncly_basic :: IO Int

102102asyncly_basic = do

103103 writeIORef count 0

104104 xs <- A.toList $ do

105− A.each [1..1000000 :: Int]

106− >>= afilter even

107− >>= amap (+1)

108− >>= adrop 1000

109− >>= amap (+1)

110− >>= afilter (\x -> x `mod` 2 == 0)

105+ -- A.each [1..1000000 :: Int]

106+ (A.for [1..1000000 :: Int] $ \x ->

107+ return x

108+ >>= afilter even

109+ >>= amap (+1)

110+ >>= adrop 1000

111+ >>= amap (+1)

112+ >>= afilter (\x -> x `mod` 2 == 0))

111113 assert (Prelude.length xs == 499000) $

112114 return (Prelude.length xs)

113115

src/Asyncly.hs+1

@@ -127,6 +127,7 @@ module Asyncly

127127 , runAsyncly

128128 , toList

129129 , each

130+ , for

130131

131132{-

132133 -- * Fork and Gather

src/Asyncly/AsyncT.hs+1−4

@@ -187,9 +187,6 @@ push ctx m = go (Just ctx) m

187187 channelYield pchan a

188188 channelDone pchan

189189

190− -- XXX If other threads are running, we should yield to the channel only

191− -- when the channel becomes empty so that we do not start more threads

192− -- unnecessarily.

193190 {-# INLINE continue #-}

194191 continue a c mx = do

195192 channelYield pchan a

@@ -288,7 +285,7 @@ fork m1 m2 = AsyncT $ \ctx stp yld -> do

288285 Just c -> (runAsyncT (pushFork c m1 m2)) ctx stp yld

289286

290287instance MonadAsync m => Alternative (AsyncT m) where

291− empty = AsyncT $ \_ stp _ -> stp

288+ empty = mempty

292289

293290 m1 <|> m2 = AsyncT $ \ctx stp yld -> do

294291 doAsync <- liftIO $ canFork

src/Asyncly/RunAsync.hs+9−1

@@ -17,6 +17,7 @@ module Asyncly.RunAsync

1717 , runAsyncly

1818 , toList

1919 , each

20+ , for

2021 , runAsynclyRecorded

2122-- , toListRecorded

2223 , playRecordings

@@ -30,6 +31,7 @@ import Control.Monad.Catch (MonadThrow)

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

3132import Control.Monad.Trans.Class (MonadTrans (lift))

3233import Control.Monad.State (StateT(..), runStateT)

34+import Data.Monoid ((<>))

3335import Data.IORef (IORef, newIORef, readIORef)

3436

3537import Control.Monad.Trans.Recorder (MonadRecorder(..), RecorderT,

@@ -104,9 +106,15 @@ toList m = run Nothing m

104106 {-# INLINE run #-}

105107 run ctx mx = (runAsyncT mx) ctx stop yield

106108

109+-- | Run a given function concurrently on the list and collect the results.

110+{-# INLINABLE for #-}

111+for :: MonadAsync m => [a] -> (a -> AsyncT m b) -> AsynclyT m b

112+for xs f = foldr (<|>) empty $ map f xs

113+

114+-- XXX rename to fromList?

107115{-# INLINABLE each #-}

108116each :: MonadAsync m => [a] -> AsynclyT m a

109−each xs = foldr (<|>) empty $ map return xs

117+each xs = foldr (<>) empty $ map return xs

110118

111119------------------------------------------------------------------------------

112120-- Logging