4e26aa5 · streamly · git.iko.soy

git.iko.soy/streamly

Remove stale code, rename the type to AsyncT

commit4e26aa558bcce9e9de9d58fd0638b99b0ad23372 authorHarendra Kumar <harendra.kumar@gmail.com> dateMon 28 Aug 2017, 08:05 (9 years ago) parentb9603cc

6 files changed+25−57

compare with master

unifiedsplit

benchmark/Main.hs+3−3 examples/echo-loop.hs+3−3 examples/loop.hs+4−4 examples/map-reduce.hs+1−1 src/Asyncly.hs+1−1 src/Asyncly/RunAsync.hs+13−45

benchmark/Main.hs+3−3

@@ -89,13 +89,13 @@ transient_basic = T.keep' $ T.threads 0 $ do

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

9090 T.exit (Prelude.length xs)

9191

92−amap :: (Int -> Int) -> Int -> A.AsynclyT IO Int

92+amap :: (Int -> Int) -> Int -> A.AsyncT IO Int

9393amap = Main.map

9494

95−afilter :: (Int -> Bool) -> Int -> A.AsynclyT IO Int

95+afilter :: (Int -> Bool) -> Int -> A.AsyncT IO Int

9696afilter = Main.filter

9797

98−adrop :: Int -> Int -> A.AsynclyT IO Int

98+adrop :: Int -> Int -> A.AsyncT IO Int

9999adrop = Main.drop

100100

101101asyncly_basic :: IO Int

examples/echo-loop.hs+3−3

@@ -1,14 +1,14 @@

11import Control.Monad.IO.Class (liftIO)

22import Data.Monoid ((<>))

4−import Asyncly (AsynclyT, runAsyncly)

4+import Asyncly (AsyncT, runAsyncly)

6−input :: AsynclyT IO String

6+input :: AsyncT IO String

77input = do

88 string <- liftIO getLine

99 return string <> input

1010

11−output :: AsynclyT IO String -> AsynclyT IO ()

11+output :: AsyncT IO String -> AsyncT IO ()

1212output strings = do

1313 s <- strings

1414 liftIO $ putStrLn s

examples/loop.hs+4−4

@@ -37,7 +37,7 @@ main = do

3737

3838 -- Generates a value and then loops. Can be used to generate an infinite

3939 -- stream. Interleaves the generator and the consumer.

40− loopTail :: Int -> AsynclyT IO Int

40+ loopTail :: Int -> AsyncT IO Int

4141 loopTail x = do

4242 liftIO $ putStrLn "LoopTail..."

4343 return x <> (if x < 3 then loopTail (x + 1) else empty)

@@ -45,7 +45,7 @@ main = do

4545 -- Loops and then generates a value. The consumer can run only after the

4646 -- loop has finished. An infinite generator will not let the consumer run

4747 -- at all.

48− loopHead :: Int -> AsynclyT IO Int

48+ loopHead :: Int -> AsyncT IO Int

4949 loopHead x = do

5050 liftIO $ putStrLn "LoopHead..."

5151 (if x < 3 then loopHead (x + 1) else empty) <> return x

@@ -59,12 +59,12 @@ main = do

5959 -- then the action on the right is also spawned concurrently. In other

6060 -- words, both actions may run concurrently based on the need.

6161

62− loopTailA :: Int -> AsynclyT IO Int

62+ loopTailA :: Int -> AsyncT IO Int

6363 loopTailA x = do

6464 liftIO $ putStrLn "LoopTailA..."

6565 return x <|> (if x < 3 then loopTailA (x + 1) else empty)

6666

67− loopHeadA :: Int -> AsynclyT IO Int

67+ loopHeadA :: Int -> AsyncT IO Int

6868 loopHeadA x = do

6969 liftIO $ putStrLn "LoopHeadA..."

7070 (if x < 3 then loopHeadA (x + 1) else empty) <|> return x

examples/map-reduce.hs+1−1

@@ -2,5 +2,5 @@ import Data.List (sum)

22import Asyncly

44main = do

5− xs <- toList $ for [1..100] $ \x -> return (x * x) :: AsynclyT IO Int

5+ xs <- toList $ for [1..100] $ \x -> return (x * x) :: AsyncT IO Int

66 print . sum $ xs

src/Asyncly.hs+1−1

@@ -120,7 +120,7 @@

120120-- of the state machine. It is an immutable state machine!

121121

122122module Asyncly

123− ( AsynclyT

123+ ( AsyncT

124124 , MonadAsync

125125

126126 -- * Run

src/Asyncly/RunAsync.hs+13−45

@@ -4,7 +4,7 @@

44{-# LANGUAGE GeneralizedNewtypeDeriving #-}

66-- |

7−-- Module : Asyncly.Threads

7+-- Module : Asyncly.RunAsync

99--

1010-- License : MIT-style

@@ -13,8 +13,7 @@

1313-- Portability : GHC

1414--

1515module Asyncly.RunAsync

16− ( AsynclyT

17− , runAsyncly

16+ ( runAsyncly

1817 , toList

1918 , each

2019 , for

@@ -37,46 +36,15 @@ import Control.Monad.Trans.Recorder (MonadRecorder(..), RecorderT,

3736 Recording, blank, runRecorderT)

3837import Asyncly.AsyncT

3938

40−-- This transformer runs AsyncT under a state to manage the threads.

41−-- Separating the state from the pure ListT transformer is cleaner but it

42−-- results in 2x performance degradation. At some point if that performance is

43−-- really needed we can combine the two.

44−

45−{-

46−newtype AsynclyT m a = AsynclyT { runAsynclyT :: AsyncT (StateT Context m) a }

47−

48−deriving instance Monad m => Functor (AsynclyT m)

49−deriving instance Monad m => Applicative (AsynclyT m)

50−deriving instance Monad m => Alternative (AsynclyT m)

51−deriving instance Monad m => Monad (AsynclyT m)

52−deriving instance MonadIO m => MonadIO (AsynclyT m)

53−deriving instance MonadThrow m => MonadThrow (AsynclyT m)

54−instance MonadTrans (AsynclyT) where

55− lift mx = AsynclyT $ AsyncT $ \c _ k -> lift mx >>= (\a -> (k a c Nothing))

56−

57−-- XXX orphan instance, use a newtype instead?

58−instance (Monad m, MonadRecorder m)

59− => MonadRecorder (StateT Context m) where

60− getJournal = lift getJournal

61− putJournal = lift . putJournal

62− play = lift . play

63−

64−deriving instance (Monad m, MonadRecorder m)

65− => MonadRecorder (AsynclyT m)

66−

67−-}

68−

69−type AsynclyT = AsyncT

70−

7139------------------------------------------------------------------------------

7240-- Running the monad

7341------------------------------------------------------------------------------

7442

75−-- | Run an 'AsynclyT m' computation, wait for it to finish and discard the

43+-- | Run an 'AsyncT m' computation, wait for it to finish and discard the

7644-- results.

7745{-# INLINABLE runAsynclyLogged #-}

7846runAsynclyLogged :: MonadAsync m

79− => Maybe (IORef [Recording]) -> AsynclyT m a -> m ()

47+ => Maybe (IORef [Recording]) -> AsyncT m a -> m ()

8048runAsynclyLogged lref m = run Nothing m

8149

8250 where

@@ -89,13 +57,13 @@ runAsynclyLogged lref m = run Nothing m

8957

9058 run ct mx = (runAsyncT mx) ct stop yield

9159

92−runAsyncly :: MonadAsync m => AsynclyT m a -> m ()

60+runAsyncly :: MonadAsync m => AsyncT m a -> m ()

9361runAsyncly m = runAsynclyLogged Nothing m

9462

95−-- | Run an 'AsynclyT m' computation and collect the results generated by each

63+-- | Run an 'AsyncT m' computation and collect the results generated by each

9664-- thread of the computation in a list.

9765{-# INLINABLE toList #-}

98−toList :: MonadAsync m => AsynclyT m a -> m [a]

66+toList :: MonadAsync m => AsyncT m a -> m [a]

9967toList m = run Nothing m

10068

10169 where

@@ -111,12 +79,12 @@ toList m = run Nothing m

11179

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

11381{-# INLINABLE for #-}

114−for :: MonadAsync m => [a] -> (a -> AsyncT m b) -> AsynclyT m b

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

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

11684

11785-- XXX rename to fromList?

11886{-# INLINABLE each #-}

119−each :: MonadAsync m => [a] -> AsynclyT m a

87+each :: MonadAsync m => [a] -> AsyncT m a

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

12189

12290------------------------------------------------------------------------------

@@ -125,21 +93,21 @@ each xs = foldr (<>) empty $ map return xs

12593

12694-- | Compose a computation using previously captured logs

12795playRecording :: (MonadAsync m, MonadRecorder m)

128− => AsynclyT m a -> Recording -> AsynclyT m a

96+ => AsyncT m a -> Recording -> AsyncT m a

12997playRecording m recording = play recording >> m

13098

13199-- | Resume an 'AsyncT' computation using previously recorded logs. The

132100-- recording consists of a list of journals one for each thread in the

133101-- computation.

134102playRecordings :: (MonadAsync m, MonadRecorder m)

135− => AsynclyT m a -> [Recording] -> AsynclyT m a

103+ => AsyncT m a -> [Recording] -> AsyncT m a

136104playRecordings m logs = each logs >>= playRecording m

137105

138106{-

139107-- | Run an 'AsyncT' computation with recording enabled, wait for it to finish

140108-- returning results for completed threads and recordings for paused threads.

141109toListRecorded :: (MonadAsync m, MonadCatch m)

142− => AsynclyT m a -> m ([a], [Recording])

110+ => AsyncT m a -> m ([a], [Recording])

143111toListRecorded m = do

144112 resultsRef <- liftIO $ newIORef []

145113 lref <- liftIO $ newIORef []

@@ -152,7 +120,7 @@ toListRecorded m = do

152120-- | Run an 'AsyncT' computation with recording enabled, wait for it to finish

153121-- and discard the results and return the recordings for paused threads, if

154122-- any.

155−runAsynclyRecorded :: MonadAsync m => AsynclyT (RecorderT m) a -> m [Recording]

123+runAsynclyRecorded :: MonadAsync m => AsyncT (RecorderT m) a -> m [Recording]

156124runAsynclyRecorded m = do

157125 lref <- liftIO $ newIORef []

158126 runRecorderT blank (runAsynclyLogged (Just lref) m)