7e8209e · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Separate the State layer from AsyncT layer

7e8209eHarendra Kumar9 years ago

src/Asyncly.hs+4−1

compare with master

unifiedsplit

@@ -120,13 +120,16 @@

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

121

122module Asyncly

123− ( AsyncT

123+ ( AsynclyT

124 , MonadAsync

125+ , AsyncT

125

126

126 -- * Run

127 -- * Run

127 , wait

128 , wait

128 , wait_

129 , wait_

129 , threads

130 , threads

131+ , waitA_

132+ , eachA

130

133

131 -- * Fork and Gather

134 -- * Fork and Gather

132 --, async

135 --, async

src/Asyncly/AsyncT.hs+56−41

@@ -22,11 +22,12 @@

22--

23module Asyncly.AsyncT

24 ( AsyncT (..)

25−

25+ , Step (..)

26− , threads

26+ , stop

27+ , waitA_

28+ , eachA

27-- , async

29-- , async

28-- , makeAsync

30-- , makeAsync

29− , each

31 , before

32 , before

32 , (*>>)

33 , (*>>)

@@ -58,10 +59,10 @@ import Unsafe.Coerce (unsafeCoerce)

58import Control.Monad.Trans.Recorder (MonadRecorder(..))

59import Control.Monad.Trans.Recorder (MonadRecorder(..))

59--import Debug.Trace (traceM)

60--import Debug.Trace (traceM)

61−import Asyncly.Threads

62+import Asyncly.Threads (MonadAsync, Context)

63−--data Step a m = Stop | Yield a (AsyncT m a)

64+data Step a m = Stop | Yield a (AsyncT m a)

64−--newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Step a m) }

65+newtype AsyncT m a = AsyncT { runAsyncT :: m (Step a m) }

66------------------------------------------------------------------------------

67------------------------------------------------------------------------------

67-- Utilities

68-- Utilities

@@ -80,35 +81,33 @@ dbg _ = return ()

80stop :: Monad m => AsyncT m a

81stop :: Monad m => AsyncT m a

81stop = AsyncT . return $ Stop

82stop = AsyncT . return $ Stop

83−{-# SPECIALIZE runAction :: AsyncT IO a -> StateT Context IO (Step a IO) #-}

84−runAction :: Monad m => AsyncT m a -> StateT Context m (Step a m)

85−runAction m = do

86− ctx <- get

87− (step, ctx') <- lift $ runStateT (runAsyncT m) ctx

88− put ctx'

89− return step

90−

91-- TBD real parallel

84-- TBD real parallel

92−{-# SPECIALIZE parallel :: AsyncT IO a -> AsyncT IO a -> StateT Context IO (Step a IO) #-}

85+{-# SPECIALIZE parallel

86+ :: AsyncT (StateT Context IO) a

87+ -> AsyncT (StateT Context IO) a

88+ -> StateT Context IO (Step a (StateT Context IO)) #-}

89+{-# SPECIALIZE parallel

90+ :: AsyncT IO a

91+ -> AsyncT IO a

92+ -> IO (Step a IO) #-}

93−parallel :: Monad m

93+parallel :: Monad m => AsyncT m a -> AsyncT m a -> m (Step a m)

94− => AsyncT m a

95− -> AsyncT m a

96− -> StateT Context m (Step a m)

97parallel ma mb = do

94parallel ma mb = do

98− step <- runAction ma

95+ runAsyncT ma >>= \step ->

99− case step of

96+ case step of

100− Stop -> runAsyncT mb

97+ Stop -> runAsyncT mb

101− Yield a mx -> return (Yield a (AsyncT (parallel mx mb)))

98+ Yield a mx -> return (Yield a (AsyncT (parallel mx mb)))

102

103instance MonadAsync m => Monad (AsyncT m) where

100instance MonadAsync m => Monad (AsyncT m) where

101+ {-# SPECIALIZE instance Monad (AsyncT (StateT Context IO)) #-}

104 {-# SPECIALIZE instance Monad (AsyncT IO) #-}

102 {-# SPECIALIZE instance Monad (AsyncT IO) #-}

105 return a = AsyncT . return $ Yield a stop

103 return a = AsyncT . return $ Yield a stop

106

104

105+ -- {-# INLINE (>>=) #-}

107 m >>= f = AsyncT $ do

106 m >>= f = AsyncT $ do

108− step <- runAction m

107+ runAsyncT m >>= \step ->

109− case step of

108+ case step of

110− Stop -> return Stop -- XXX need to wait for the child threads

109+ Stop -> return Stop -- XXX need to wait for the child threads

111− Yield a ma -> runAsyncT $ (f a) <|> (ma >>= f)

110+ Yield a ma -> runAsyncT $ (f a) <|> (ma >>= f)

112

111

113------------------------------------------------------------------------------

112------------------------------------------------------------------------------

114-- Functor

113-- Functor

@@ -147,6 +146,7 @@ instance MonadAsync m => Applicative (AsyncT m) where

147-- they can go away rather than waiting indefinitely.

146-- they can go away rather than waiting indefinitely.

148

147

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

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

149+ {-# SPECIALIZE instance Alternative (AsyncT (StateT Context IO)) #-}

150 {-# SPECIALIZE instance Alternative (AsyncT IO) #-}

151 empty = stop

152 (<|>) m1 m2 = AsyncT $ parallel m1 m2

@@ -176,7 +176,7 @@ instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where

176-------------------------------------------------------------------------------

177

178instance MonadTrans AsyncT where

179− lift mx = AsyncT $ lift mx >>= return . (\a -> (Yield a stop))

179+ lift mx = AsyncT $ mx >>= return . (\a -> (Yield a stop))

180

181instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m) where

182 liftBase = liftBaseDefault

@@ -295,18 +295,33 @@ makeAsync :: MonadAsync m => ((a -> m ()) -> m ()) -> AsyncT m a

295makeAsync = AsyncT . makeCont

296-}

297

298−-- scatter

299−{-# SPECIALIZE each :: [a] -> AsyncT IO a #-}

298+{-# SPECIALIZE waitA_ :: AsyncT IO a -> IO () #-}

300−each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)

299+waitA_ :: (MonadAsync m) => AsyncT m a -> m ()

301− => [a] -> AsyncT m a

300+waitA_ (AsyncT m) = run m

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

303

301

304−------------------------------------------------------------------------------

302+ where

305−-- Controlling thread quota

303+

306−------------------------------------------------------------------------------

304+ run m = m >>= \x ->

305+ case x of

306+ Yield _ r -> run (runAsyncT r)

307+ -- XXX pull from children here

308+ _ -> return ()

307

309

308−-- | Runs a computation under a given thread limit. A limit of 0 means all new

310+{-

309−-- tasks start synchronously in the current thread unless overridden by

311+waitA :: forall m a. (MonadAsync m, MonadCatch m) => AsyncT m a -> m [a]

310−-- 'async'.

312+waitA (AsyncT m) = run m

311−threads :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a

313+

312−threads n action = AsyncT $ threadCtl n (runAsyncT action)

314+ where

315+

316+ run ma = do

317+ ma >>= \a -> case a of

318+ Yield x mb -> liftM (x :) (run mb ctx')

319+ -- XXX pull from children here

320+ Stop -> return []

321+-}

322+

323+-- scatter

324+{-# SPECIALIZE eachA :: [a] -> AsyncT IO a #-}

325+eachA :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)

326+ => [a] -> AsyncT m a

327+eachA xs = foldr (<|>) empty $ map return xs

src/Asyncly/RunAsync.hs+61−10

@@ -1,5 +1,8 @@

1−{-# LANGUAGE FlexibleContexts #-}

1+{-# LANGUAGE FlexibleContexts #-}

2−{-# LANGUAGE ScopedTypeVariables #-}

2+{-# LANGUAGE ScopedTypeVariables #-}

3+{-# LANGUAGE StandaloneDeriving #-}

4+{-# LANGUAGE GeneralizedNewtypeDeriving #-}

5+{-# LANGUAGE UndecidableInstances #-}

4-- |

7-- |

5-- Module : Asyncly.Threads

8-- Module : Asyncly.Threads

@@ -13,6 +16,9 @@

13module Asyncly.RunAsync

16module Asyncly.RunAsync

14 ( wait

17 ( wait

15 , wait_

18 , wait_

19+ , AsynclyT

20+ , threads

21+ , each

16 {-

22 {-

17 , gather

23 , gather

@@ -23,10 +29,13 @@ module Asyncly.RunAsync

23 )

29 )

24where

30where

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

26import Control.Concurrent.STM (atomically, newTChan)

33import Control.Concurrent.STM (atomically, newTChan)

27import Control.Monad (liftM)

34import Control.Monad (liftM)

28−import Control.Monad.Catch (MonadCatch, throwM, try)

35+import Control.Monad.Catch (MonadCatch, MonadThrow, throwM, try)

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

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

37+import Control.Monad.Trans.Class (MonadTrans (lift))

38+import Control.Monad.Trans.Control (MonadBaseControl)

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

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

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

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

32import Data.Atomics (atomicModifyIORefCAS)

41import Data.Atomics (atomicModifyIORefCAS)

@@ -34,21 +43,34 @@ import Data.IORef (IORef, newIORef, readIORef)

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

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

36 Recording, blank, runRecorderT)

45 Recording, blank, runRecorderT)

46+import GHC.Prim (Any)

47+

37import Asyncly.Threads

48import Asyncly.Threads

38import Asyncly.AsyncT

49import Asyncly.AsyncT

51+newtype AsynclyT m a = AsynclyT { runAsynclyT :: AsyncT (StateT Context m) a }

52+

53+deriving instance MonadAsync m => Functor (AsynclyT m)

54+deriving instance MonadAsync m => Applicative (AsynclyT m)

55+deriving instance MonadAsync m => Alternative (AsynclyT m)

56+deriving instance MonadAsync m => Monad (AsynclyT m)

57+deriving instance MonadAsync m => MonadIO (AsynclyT m)

58+instance MonadTrans (AsynclyT) where

59+ lift mx = AsynclyT $ AsyncT $ lift mx >>= return . (\a -> (Yield a stop))

60+

40−getContext :: Monad m => (a -> AsyncT m a) -> Maybe (IORef [Recording]) -> IO Context

61+getContext :: Maybe (IORef [Recording]) -> IO Context

41−getContext f lref = do

62+getContext lref = do

42 childChan <- atomically newTChan

63 childChan <- atomically newTChan

43 pendingRef <- newIORef []

64 pendingRef <- newIORef []

44 credit <- newIORef maxBound

65 credit <- newIORef maxBound

45− return $ initContext childChan pendingRef credit f lref

66+ return $ initContext childChan pendingRef credit lref

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

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

48-- results.

69-- results.

70+{-# SPECIALIZE wait_ :: AsynclyT IO a -> IO () #-}

49−wait_ :: forall m a. (MonadAsync m, MonadCatch m) => AsyncT m a -> m ()

71+wait_ :: forall m a. (MonadAsync m, MonadCatch m) => AsynclyT m a -> m ()

50−wait_ m = do

72+wait_ (AsynclyT m) = do

51− ctx <- liftIO $ getContext (return :: a -> AsyncT m a) Nothing

73+ ctx <- liftIO $ getContext Nothing

52 _ <- runStateT (run m) ctx

74 _ <- runStateT (run m) ctx

53 return ()

75 return ()

@@ -60,6 +82,35 @@ wait_ m = do

60 -- XXX pull from children here

82 -- XXX pull from children here

61 _ -> return ()

83 _ -> return ()

85+wait :: forall m a. (MonadAsync m, MonadCatch m) => AsynclyT m a -> m [a]

86+wait (AsynclyT m) = liftIO (getContext Nothing) >>= run m

87+

88+ where

89+

90+ run ma ctx = do

91+ (a, ctx') <- runStateT (runAsyncT ma) ctx

92+ case a of

93+ Yield x mb -> liftM (x :) (run mb ctx')

94+ -- XXX pull from children here

95+ Stop -> return []

96+

97+

98+-- scatter

99+{-# SPECIALIZE each :: [a] -> AsynclyT IO a #-}

100+each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)

101+ => [a] -> AsynclyT m a

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

103+

104+------------------------------------------------------------------------------

105+-- Controlling thread quota

106+------------------------------------------------------------------------------

107+

108+-- | Runs a computation under a given thread limit. A limit of 0 means all new

109+-- tasks start synchronously in the current thread unless overridden by

110+-- 'async'.

111+threads :: MonadAsync m => Int -> AsynclyT m a -> AsynclyT m a

112+threads n action = AsynclyT $ AsyncT $ threadCtl n (runAsyncT $ runAsynclyT action)

113+

63{-

114{-

64wait_ :: AsyncT IO a -> IO ()

115wait_ :: AsyncT IO a -> IO ()

65wait_ m = do

116wait_ m = do

@@ -74,7 +125,6 @@ wait_ m = do

74 Yield _ r -> run r

125 Yield _ r -> run r

75 -- XXX pull from children here

126 -- XXX pull from children here

76 _ -> return ()

127 _ -> return ()

77−-}

128

79wait :: forall m a. (MonadAsync m, MonadCatch m) => AsyncT m a -> m [a]

129wait :: forall m a. (MonadAsync m, MonadCatch m) => AsyncT m a -> m [a]

80wait m = liftIO (getContext (return :: a -> AsyncT m a) Nothing) >>= run m

130wait m = liftIO (getContext (return :: a -> AsyncT m a) Nothing) >>= run m

@@ -87,6 +137,7 @@ wait m = liftIO (getContext (return :: a -> AsyncT m a) Nothing) >>= run m

87 Yield x mb -> liftM (x :) (run mb ctx')

137 Yield x mb -> liftM (x :) (run mb ctx')

88 -- XXX pull from children here

138 -- XXX pull from children here

89 Stop -> return []

139 Stop -> return []

140+-}

141

91{-

142{-

92------------------------------------------------------------------------------

143------------------------------------------------------------------------------

src/Asyncly/Threads.hs+3−13

@@ -15,8 +15,6 @@ module Asyncly.Threads

15 ( -- ChildEvent(..)

16 MonadAsync

17 , Context(..)

18− , Step (..)

19− , AsyncT (..)

20 , initContext

18 , initContext

21 --, runAsyncTask

19 --, runAsyncTask

22 --, makeCont

20 --, makeCont

@@ -57,9 +55,6 @@ import Control.Monad.Trans.Recorder (Paused (..), Recording)

58type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

56type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

60−data Step a m = Stop | Yield a (AsyncT m a)

61−newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Step a m) }

62−

63------------------------------------------------------------------------------

58------------------------------------------------------------------------------

64-- Parent child thread communication types

59-- Parent child thread communication types

65------------------------------------------------------------------------------

60------------------------------------------------------------------------------

@@ -76,11 +71,8 @@ data Context = Context

76 -- Execution state

71 -- Execution state

77 ---------------------------------------------------------------------------

72 ---------------------------------------------------------------------------

79− -- a -> AsyncT m b

80− continuations :: [Any]

81−

82 -- When we suspend we save the logs in this IORef and exit.

74 -- When we suspend we save the logs in this IORef and exit.

83− , logsRef :: Maybe (IORef [Recording])

75+ logsRef :: Maybe (IORef [Recording])

84 -- XXX this functionality can be in a separate layer?

76 -- XXX this functionality can be in a separate layer?

85 , location :: Location

77 , location :: Location

@@ -133,12 +125,10 @@ initContext

133 :: TChan ChildEvent

125 :: TChan ChildEvent

134 -> IORef [ThreadId]

126 -> IORef [ThreadId]

135 -> IORef Int

127 -> IORef Int

136− -> (a -> AsyncT m b)

137 -> Maybe (IORef [Recording])

128 -> Maybe (IORef [Recording])

138 -> Context

129 -> Context

139−initContext childChan pending credit finalizer lref =

130+initContext childChan pending credit lref =

140− Context { continuations = [unsafeCoerce finalizer]

141− , logsRef = lref

131+ Context { logsRef = lref

142 , location = Worker

132 , location = Worker

143 , childChannel = childChan

133 , childChannel = childChan

144 , pendingThreads = pending

134 , pendingThreads = pending

test/Main.hs+2

@@ -25,6 +25,7 @@ main = hspec $ do

25 wait (do x <- return 1; y <- return 2; return (x + y)) `shouldReturn` ([3] :: [Int])

26 it "Alternative composition of sync tasks" $

27 ((wait $ threads 0 $ (return 0 <|> return 1)) >>= return . sort) `shouldReturn` ([0, 1] :: [Int])

28+ {-

28 it "Alternative composition of async and sync tasks" $

29 it "Alternative composition of async and sync tasks" $

29 ((wait (threads 0 ((async (return 0) <|> return 1)))) >>= return . sort)

30 ((wait (threads 0 ((async (return 0) <|> return 1)))) >>= return . sort)

30 `shouldReturn` ([0,1] :: [Int])

31 `shouldReturn` ([0,1] :: [Int])

@@ -141,3 +142,4 @@ generalExample = do

141 return (x11 + y11 + z11)

142 return (x11 + y11 + z11)

142 return (x1 + y1 + z1)

143 return (x1 + y1 + z1)

143 return (x + y + z)

144 return (x + y + z)

145+ -}