4d7024f · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Initial poc of lazy streaming

4d7024fHarendra Kumar9 years ago

src/Asyncly.hs+6−6

compare with master

unifiedsplit

@@ -129,10 +129,10 @@ module Asyncly

129 , threads

130

131 -- * Fork and Gather

132− , async

132+ --, async

133− , makeAsync

133+ --, makeAsync

134 , each

135− , gather

135+ --, gather

136

137 -- * Compose

138 , before

@@ -143,9 +143,9 @@ module Asyncly

143 , (>>|)

144

145 -- * Record and Play

146− , waitRecord

146+ --, waitRecord

147− , waitRecord_

147+ --, waitRecord_

148− , playRecordings

148+ --, playRecordings

149 )

150where

151

src/Asyncly/AsyncT.hs+42−24

@@ -24,8 +24,8 @@ module Asyncly.AsyncT

24 ( AsyncT (..)

26 , threads

27− , async

27+-- , async

28− , makeAsync

28+-- , makeAsync

29 , each

31 , before

@@ -46,7 +46,7 @@ import Control.Monad.Base (MonadBase (..), liftBaseDefault)

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

47import Control.Monad.State (MonadIO (..), MonadPlus (..),

48 StateT (..), liftM, runStateT,

49− modify)

49+ get, gets, put, modify)

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

51import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),

52 MonadTransControl (..),

@@ -60,7 +60,8 @@ import Control.Monad.Trans.Recorder (MonadRecorder(..))

61import Asyncly.Threads

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

63−newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Maybe a) }

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

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

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

66-- Utilities

67-- Utilities

@@ -76,23 +77,38 @@ dbg _ = return ()

76-- Monad

77-- Monad

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

78------------------------------------------------------------------------------

80+stop :: Monad m => AsyncT m a

81+stop = AsyncT . return $ Stop

82+

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

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

93+parallel :: Monad m

94+ => AsyncT m a

95+ -> AsyncT m a

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

97+parallel ma mb = do

98+ step <- runAction ma

99+ case step of

100+ Stop -> runAsyncT mb

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

102+

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

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

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

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

81− return a = AsyncT . return $ Just a

105+ return a = AsyncT . return $ Yield a stop

82− m >>= f = bind

106+

83−

84− where

85−

86− {-# NOINLINE[1] bind #-}

87− bind = AsyncT $ do

107+ m >>= f = AsyncT $ do

88− modify $ \Context {continuations = fs, ..} ->

108+ step <- runAction m

89− Context {continuations = (unsafeCoerce f) : fs, ..}

90− x <- runAsyncT m

91− modify $ \Context {continuations = (_ : fs), ..} ->

92− Context {continuations = fs, ..}

93− case x of

109+ case step of

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

94− Just y -> runAsyncT $ f y

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

95− Nothing -> return Nothing

112

97------------------------------------------------------------------------------

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

98-- Functor

114-- Functor

@@ -132,10 +148,8 @@ instance MonadAsync m => Applicative (AsyncT m) where

132

148

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

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

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

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

135− empty = AsyncT $ return Nothing

151+ empty = stop

136− (<|>) m1 m2 = AsyncT $ do

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

137− r <- runAsyncTask False (runAsyncT m1)

138− maybe (runAsyncT m2) (return . Just) r

139

153

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

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

141 mzero = empty

155 mzero = empty

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

162-------------------------------------------------------------------------------

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

163

177

164instance MonadTrans AsyncT where

178instance MonadTrans AsyncT where

165− lift mx = AsyncT $ lift mx >>= return . Just

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

166

180

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

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

168 liftBase = liftBaseDefault

182 liftBase = liftBaseDefault

@@ -171,6 +185,7 @@ instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m) where

171-- monad-control

185-- monad-control

172-------------------------------------------------------------------------------

186-------------------------------------------------------------------------------

173

187

188+{-

174instance MonadTransControl AsyncT where

189instance MonadTransControl AsyncT where

175 type StT AsyncT a = (Maybe a, Context)

190 type StT AsyncT a = (Maybe a, Context)

176 liftWith f = AsyncT $ StateT $ \s ->

191 liftWith f = AsyncT $ StateT $ \s ->

@@ -186,6 +201,7 @@ instance (MonadBaseControl b m, MonadAsync m) => MonadBaseControl b (AsyncT m) w

186 restoreM = defaultRestoreM

201 restoreM = defaultRestoreM

187 {-# INLINABLE liftBaseWith #-}

202 {-# INLINABLE liftBaseWith #-}

188 {-# INLINABLE restoreM #-}

203 {-# INLINABLE restoreM #-}

204+-}

189

205

190------------------------------------------------------------------------------

206------------------------------------------------------------------------------

191-- Standard transformer instances

207-- Standard transformer instances

@@ -271,11 +287,13 @@ infixr 1 >>|

271-- is not useful and should not be used. Even in an 'Alternative' composition

287-- is not useful and should not be used. Even in an 'Alternative' composition

272-- 'async' is not useful as the last action as the last action always runs in

288-- 'async' is not useful as the last action as the last action always runs in

273-- the current thread.

289-- the current thread.

290+{-

274async :: MonadAsync m => AsyncT m a -> AsyncT m a

291async :: MonadAsync m => AsyncT m a -> AsyncT m a

275async action = AsyncT $ runAsyncTask True (runAsyncT action)

292async action = AsyncT $ runAsyncTask True (runAsyncT action)

276

293

277makeAsync :: MonadAsync m => ((a -> m ()) -> m ()) -> AsyncT m a

294makeAsync :: MonadAsync m => ((a -> m ()) -> m ()) -> AsyncT m a

278makeAsync = AsyncT . makeCont

295makeAsync = AsyncT . makeCont

296+-}

279

297

280-- scatter

298-- scatter

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

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

src/Asyncly/RunAsync.hs+57

@@ -13,15 +13,18 @@

13module Asyncly.RunAsync

14 ( wait

15 , wait_

16+ {-

16 , gather

17 , gather

18 , waitRecord_

19 , waitRecord_

19 , waitRecord

20 , waitRecord

20 , playRecordings

21 , playRecordings

22+ -}

21 )

23 )

22where

24where

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

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

27+import Control.Monad (liftM)

25import Control.Monad.Catch (MonadCatch, throwM, try)

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

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

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

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

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

@@ -34,6 +37,58 @@ import Control.Monad.Trans.Recorder (MonadRecorder(..), RecorderT,

34import Asyncly.Threads

37import Asyncly.Threads

35import Asyncly.AsyncT

38import Asyncly.AsyncT

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

41+getContext f lref = do

42+ childChan <- atomically newTChan

43+ pendingRef <- newIORef []

44+ credit <- newIORef maxBound

45+ return $ initContext childChan pendingRef credit f lref

46+

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

48+-- results.

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

50+wait_ m = do

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

52+ _ <- runStateT (run m) ctx

53+ return ()

54+

55+ where

56+

57+ run m = runAsyncT m >>= \x ->

58+ case x of

59+ Yield _ r -> run r

60+ -- XXX pull from children here

61+ _ -> return ()

62+

63+{-

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

65+wait_ m = do

66+ ctx <- liftIO $ getContext (return :: a -> AsyncT IO a) Nothing

67+ _ <- runStateT (run m) ctx

68+ return ()

69+

70+ where

71+

72+ run m = runAsyncT m >>= \x ->

73+ case x of

74+ Yield _ r -> run r

75+ -- XXX pull from children here

76+ _ -> return ()

77+-}

78+

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

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

81+

82+ where

83+

84+ run ma ctx = do

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

86+ case a of

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

88+ -- XXX pull from children here

89+ Stop -> return []

90+

91+{-

37------------------------------------------------------------------------------

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

38-- Running the monad

93-- Running the monad

39------------------------------------------------------------------------------

94------------------------------------------------------------------------------

@@ -133,3 +188,5 @@ waitRecord_ m = do

133 runRecorderT blank (waitAsync (const (return Nothing)) (Just lref) m)

188 runRecorderT blank (waitAsync (const (return Nothing)) (Just lref) m)

134 logs <- liftIO $ readIORef lref

189 logs <- liftIO $ readIORef lref

135 return logs

190 return logs

191+

192+-}

src/Asyncly/Threads.hs+13−6

@@ -15,16 +15,18 @@ module Asyncly.Threads

15 ( -- ChildEvent(..)

16 MonadAsync

17 , Context(..)

18+ , Step (..)

19+ , AsyncT (..)

18 , initContext

20 , initContext

19− , runAsyncTask

21+ --, runAsyncTask

20− , makeCont

22+ --, makeCont

21 , threadCtl

23 , threadCtl

22 -- , Location(..)

24 -- , Location(..)

23 -- , getLocation

25 -- , getLocation

24 -- , setLocation

26 -- , setLocation

26 -- , waitForChildren

28 -- , waitForChildren

27− , handleResult

29+ --, handleResult

28 )

30 )

29where

31where

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

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

58type 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+

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

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

59-- Parent child thread communication types

64-- Parent child thread communication types

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

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

@@ -128,7 +133,7 @@ initContext

128 :: TChan ChildEvent

133 :: TChan ChildEvent

129 -> IORef [ThreadId]

134 -> IORef [ThreadId]

130 -> IORef Int

135 -> IORef Int

131− -> (a -> StateT Context m (Maybe a))

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

132 -> Maybe (IORef [Recording])

137 -> Maybe (IORef [Recording])

133 -> Context

138 -> Context

134initContext childChan pending credit finalizer lref =

139initContext childChan pending credit finalizer lref =

@@ -155,6 +160,7 @@ getLocation = gets location

155setLocation :: Monad m => Location -> StateT Context m ()

160setLocation :: Monad m => Location -> StateT Context m ()

156setLocation loc = modify $ \Context { .. } -> Context { location = loc, .. }

161setLocation loc = modify $ \Context { .. } -> Context { location = loc, .. }

157

162

163+{-

158------------------------------------------------------------------------------

164------------------------------------------------------------------------------

159-- Pick up from where we left in the previous thread

165-- Pick up from where we left in the previous thread

160------------------------------------------------------------------------------

166------------------------------------------------------------------------------

@@ -431,6 +437,7 @@ makeCont cbsetter = do

431 return ()

437 return ()

432 spawningParentDone

438 spawningParentDone

433

439

440+-}

434------------------------------------------------------------------------------

441------------------------------------------------------------------------------

435-- Controlling thread quota

442-- Controlling thread quota

436------------------------------------------------------------------------------

443------------------------------------------------------------------------------

@@ -441,8 +448,8 @@ makeCont cbsetter = do

441-- 'async'.

448-- 'async'.

442threadCtl :: MonadAsync m

449threadCtl :: MonadAsync m

443 => Int

450 => Int

444− -> StateT Context m (Maybe a)

451+ -> StateT Context m a

445− -> StateT Context m (Maybe a)

452+ -> StateT Context m a

446threadCtl n action = do

453threadCtl n action = do

447 oldCr <- gets threadCredit

454 oldCr <- gets threadCredit

448 newCr <- liftIO $ newIORef n

455 newCr <- liftIO $ newIORef n