c1f0a86 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Continuation based implementation

c1f0a86Harendra Kumar9 years ago

asyncly.cabal+1−1

compare with master

unifiedsplit

@@ -26,7 +26,7 @@ library

26 exposed-modules: Asyncly

27 , Asyncly.Threads

28 , Asyncly.AsyncT

29− , Asyncly.RunAsync

29+ -- , Asyncly.RunAsync

31 default-language: Haskell2010

32 ghc-options: -Wall -fwarn-identities -fwarn-incomplete-record-updates

src/Asyncly.hs+8−6

@@ -120,17 +120,18 @@

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

121

122module Asyncly

123− ( AsynclyT

123+ ( -- AsynclyT

124− , MonadAsync

124+ MonadAsync

125 , AsyncT

126

127 -- * Run

128− , wait

128+-- , wait

129− , wait_

129+-- , wait_

130− , threads

130+-- , threads

131 , waitA_

132 , eachA

133

134+{-

134 -- * Fork and Gather

135 -- * Fork and Gather

135 --, async

136 --, async

136 --, makeAsync

137 --, makeAsync

@@ -144,6 +145,7 @@ module Asyncly

144 , (>>*)

145 , (>>*)

145 , afterfirst

146 , afterfirst

146 , (>>|)

147 , (>>|)

148+ -}

147

149

148 -- * Record and Play

150 -- * Record and Play

149 --, waitRecord

151 --, waitRecord

@@ -154,4 +156,4 @@ where

154

156

155import Asyncly.Threads

157import Asyncly.Threads

156import Asyncly.AsyncT

158import Asyncly.AsyncT

157−import Asyncly.RunAsync

159+-- import Asyncly.RunAsync

src/Asyncly/AsyncT.hs+46−41

@@ -22,19 +22,21 @@

22--

23module Asyncly.AsyncT

24 ( AsyncT (..)

25− , Step (..)

25+ -- , Step (..)

26− , stop

26+ -- , stop

27 , waitA_

28 , eachA

29-- , async

30-- , makeAsync

32+{-

32 , before

33 , before

33 , (*>>)

34 , (*>>)

34 , thereafter

35 , thereafter

35 , (>>*)

36 , (>>*)

36 , afterfirst

37 , afterfirst

37 , (>>|)

38 , (>>|)

39+ -}

39 -- internal

41 -- internal

40 , dbg

42 , dbg

@@ -42,6 +44,7 @@ module Asyncly.AsyncT

42where

44where

44import Control.Applicative (Alternative (..))

46import Control.Applicative (Alternative (..))

47+import Data.Maybe (maybe)

45import Control.Monad (ap, mzero, when)

48import Control.Monad (ap, mzero, when)

46import Control.Monad.Base (MonadBase (..), liftBaseDefault)

49import Control.Monad.Base (MonadBase (..), liftBaseDefault)

47import Control.Monad.Catch (MonadThrow, throwM)

50import Control.Monad.Catch (MonadThrow, throwM)

@@ -61,8 +64,15 @@ import Control.Monad.Trans.Recorder (MonadRecorder(..))

62import Asyncly.Threads (MonadAsync, Context)

65import Asyncly.Threads (MonadAsync, Context)

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

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

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

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

69+-- stop value is needed for Alternative instance

70+newtype AsyncT m a = AsyncT {

71+ runAsyncT :: forall r.

72+ m r

73+ -> (a -> Maybe (AsyncT m a) -> m r)

74+ -> m r

75+ }

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

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

68-- Utilities

78-- Utilities

@@ -78,36 +88,33 @@ dbg _ = return ()

78-- Monad

88-- Monad

79------------------------------------------------------------------------------

89------------------------------------------------------------------------------

81−stop :: Monad m => AsyncT m a

91+-- XXX need to wait for the async threads in case of a stop is returned due to

82−stop = AsyncT . return $ Stop

92+-- async thread spawning.

83−

84−-- TBD real parallel

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

93{-# SPECIALIZE parallel

90 :: AsyncT IO a

94 :: AsyncT IO a

91 -> AsyncT IO a

95 -> AsyncT IO a

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

96+ -> AsyncT IO a #-}

97+{-# INLINE parallel #-}

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

98+parallel :: Monad m => AsyncT m a -> AsyncT m a -> AsyncT m a

94−parallel ma mb = do

99+parallel (AsyncT m1) m2 = AsyncT $ \stp yld ->

95− runAsyncT ma >>= \step ->

100+ m1 ((runAsyncT m2) stp yld) (\a ax ->

96− case step of

101+ case ax of

97− Stop -> runAsyncT mb

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

102+ Just mx -> yld a (Just (parallel mx m2))

103+ Nothing -> yld a (Just m2)

104+ )

105

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

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

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

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

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

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

103− return a = AsyncT . return $ Yield a stop

109+ return a = AsyncT $ \_ yld -> yld a Nothing

104

110

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

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

106− m >>= f = AsyncT $ do

112+ AsyncT m >>= f = AsyncT $ \stp yld ->

107− runAsyncT m >>= \step ->

113+ m stp (\a ma ->

108− case step of

114+ case ma of

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

115+ Just mx -> (runAsyncT (parallel (f a) (mx >>= f))) stp yld

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

116+ Nothing -> (runAsyncT (f a)) stp yld

117+ )

111

118

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

119------------------------------------------------------------------------------

113-- Functor

120-- Functor

@@ -146,15 +153,16 @@ instance MonadAsync m => Applicative (AsyncT m) where

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

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

147

154

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

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

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

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

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

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

151− empty = stop

158+ empty = AsyncT $ \stp _ -> stp

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

159+ (<|>) m1 m2 = parallel m1 m2

153

160

154instance MonadAsync m => MonadPlus (AsyncT m) where

161instance MonadAsync m => MonadPlus (AsyncT m) where

155 mzero = empty

162 mzero = empty

156 mplus = (<|>)

163 mplus = (<|>)

157

164

165+{-

158instance (Monoid a, MonadAsync m) => Monoid (AsyncT m a) where

166instance (Monoid a, MonadAsync m) => Monoid (AsyncT m a) where

159 mappend x y = mappend <$> x <*> y

167 mappend x y = mappend <$> x <*> y

160 mempty = return mempty

168 mempty = return mempty

@@ -171,12 +179,13 @@ instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where

171 abs f = f >>= return . abs

179 abs f = f >>= return . abs

172 signum f = f >>= return . signum

180 signum f = f >>= return . signum

173

181

182+-}

174-------------------------------------------------------------------------------

183-------------------------------------------------------------------------------

175-- AsyncT transformer

184-- AsyncT transformer

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

185-------------------------------------------------------------------------------

177

186

178instance MonadTrans AsyncT where

187instance MonadTrans AsyncT where

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

188+ lift mx = AsyncT $ \_ yld -> mx >>= (\a -> (yld a Nothing))

180

189

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

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

182 liftBase = liftBaseDefault

191 liftBase = liftBaseDefault

@@ -226,6 +235,7 @@ instance (MonadAsync m, MonadRecorder m) => MonadRecorder (AsyncT m) where

226-- Special compositions

235-- Special compositions

227------------------------------------------------------------------------------

236------------------------------------------------------------------------------

228

237

238+{-

229-- | Run @m a@ before running @m b@. The result of @m a@ is discarded.

239-- | Run @m a@ before running @m b@. The result of @m a@ is discarded.

230before :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b

240before :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b

231before ma mb = AsyncT $ do

241before ma mb = AsyncT $ do

@@ -264,6 +274,7 @@ infixr 1 >>|

264-- | Same as 'afterfirst'.

274-- | Same as 'afterfirst'.

265(>>|) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a

275(>>|) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a

266(>>|) = afterfirst

276(>>|) = afterfirst

277+-}

267

278

268------------------------------------------------------------------------------

279------------------------------------------------------------------------------

269-- Async primitives

280-- Async primitives

@@ -295,17 +306,9 @@ makeAsync :: MonadAsync m => ((a -> m ()) -> m ()) -> AsyncT m a

295makeAsync = AsyncT . makeCont

306makeAsync = AsyncT . makeCont

296-}

307-}

297

308

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

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

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

310waitA_ :: (MonadAsync m) => AsyncT m a -> m ()

300−waitA_ (AsyncT m) = run m

311+waitA_ (AsyncT m) = m (return ()) (\_ r -> maybe (return ()) waitA_ r)

301−

302− where

303−

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

305− case x of

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

307− -- XXX pull from children here

308− _ -> return ()

309

312

310{-

313{-

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

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

@@ -321,6 +324,8 @@ waitA (AsyncT m) = run m

321-}

324-}

322

325

323-- scatter

326-- scatter

327+-- XXX unthreaded case can be more efficient by yielding continuously rather

328+-- than folding a yield followed by Nothing in the parallel case.

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

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

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

330eachA :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)

326 => [a] -> AsyncT m a

331 => [a] -> AsyncT m a