0a69aec · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Refactor, simplify, add comments

0a69aecHarendra Kumar9 years ago

src/Duct/AsyncT.hs+35−30

compare with master

unifiedsplit

@@ -12,11 +12,9 @@

12module Duct.AsyncT

13 ( AsyncT (..)

14 , waitAsync

15− , compose

16 , (<**)

15 , (<**)

17 , onNothing

16 , onNothing

18 , RemoteStatus (..)

17 , RemoteStatus (..)

19− , back

20 , waitForChildren

18 , waitForChildren

21 , dbg

19 , dbg

22 )

20 )

@@ -44,13 +42,10 @@ import Data.Maybe (isJust, isNothing)

44import Unsafe.Coerce (unsafeCoerce)

42import Unsafe.Coerce (unsafeCoerce)

46import Duct.Event

44import Duct.Event

47−import Duct.Log

49-- import Debug.Trace

46-- import Debug.Trace

51−type StateM m = StateT EventF m

52−

53−newtype AsyncT m a = AsyncT { runAsyncT :: StateM m (Maybe a) }

48+newtype AsyncT m a = AsyncT { runAsyncT :: StateT EventF m (Maybe a) }

55------------------------------------------------------------------------------

50------------------------------------------------------------------------------

56-- Remote data types

51-- Remote data types

@@ -89,6 +84,7 @@ instance Monad (AsyncT m) => Functor (AsyncT m) where

89-- Applicative

84-- Applicative

90------------------------------------------------------------------------------

85------------------------------------------------------------------------------

87+{-

92-- | Warning: Radically untyped stuff. handle with care

88-- | Warning: Radically untyped stuff. handle with care

93getContinuations :: Monad m => StateM m [a -> AsyncT m b]

89getContinuations :: Monad m => StateM m [a -> AsyncT m b]

94getContinuations = do

90getContinuations = do

@@ -109,6 +105,8 @@ restoreStack :: MonadState EventF m => t -> m ()

109restoreStack fs = modify $ \EventF {..} ->

105restoreStack fs = modify $ \EventF {..} ->

110 EventF { event = Nothing, fcomp = (unsafeCoerce fs), .. }

106 EventF { event = Nothing, fcomp = (unsafeCoerce fs), .. }

111

107

108+-}

109+

112instance Monad m => Applicative (AsyncT m) where

110instance Monad m => Applicative (AsyncT m) where

113 pure a = AsyncT . return $ Just a

111 pure a = AsyncT . return $ Just a

114 m1 <*> m2 = do { x1 <- m1; x2 <- m2; return (x1 x2) }

112 m1 <*> m2 = do { x1 <- m1; x2 <- m2; return (x1 x2) }

@@ -145,24 +143,29 @@ tailsafe (_:xs) = xs

145

143

146instance Monad m => Monad (AsyncT m) where

144instance Monad m => Monad (AsyncT m) where

147 return = pure

145 return = pure

148− m >>= f = AsyncT $ do

149− -- save m in xcomp and add f to the head of fcomp

150− -- _ <- setEventCont m f

151− modify $ \EventF { fcomp = fs, .. } ->

152− EventF { xcomp = unsafeCoerce m, fcomp = unsafeCoerce f : fs, .. }

153

146

154− -- run m under this modified environment

147+ m >>= f = AsyncT $ do

148+ -- Save the 'm' and 'f', in case we migrate to a new thread before this

149+ -- bind operation completes, we run the operation manually in the new

150+ -- thread.

151+ modify $ \EventF { fstack = fs, .. } ->

152+ EventF { currentm = unsafeCoerce m

153+ , fstack = unsafeCoerce f : fs

154+ , .. }

155+

156+ -- Inner bind operations in 'm' add their 'f' to fstack. If we migrate to

157+ -- a new thread somewhere in the middle we unwind the fstack and run these

158+ -- functions manually after migration.

155 mres <- runAsyncT m

159 mres <- runAsyncT m

156− -- remove the head of fcomp apply it to mres and that becomes the new xcomp.

160+

157− -- is head of fcomp always the same as f?

158 let res = case mres of

161 let res = case mres of

159 Nothing -> empty

162 Nothing -> empty

160 Just x -> f x

163 Just x -> f x

161

164

162− -- XXX why would we have more than one continuation? Can we have just one

163− -- instead of a list and remove it after every bind?

164− modify $ \EventF { fcomp = fs, .. } ->

165+ modify $ \EventF { fstack = fs, .. } ->

165− EventF { xcomp = unsafeCoerce res, fcomp = tailsafe fs, .. }

166+ EventF { currentm = unsafeCoerce res

167+ , fstack = tailsafe fs

168+ , .. }

166

169

167 runAsyncT res

170 runAsyncT res

168

171

@@ -178,15 +181,11 @@ instance (Monoid a, Monad (AsyncT m)) => Monoid (AsyncT m a) where

178-- Backtracking

181-- Backtracking

179------------------------------------------------------------------------------

182------------------------------------------------------------------------------

180

183

184+{-

181-- | Run the closure (the 'x' in 'x >>= f') of the current bind operation.

185-- | Run the closure (the 'x' in 'x >>= f') of the current bind operation.

182runClosure :: EventF -> StateM m (Maybe a)

186runClosure :: EventF -> StateM m (Maybe a)

183runClosure EventF { xcomp = x } = runAsyncT (unsafeCoerce x)

187runClosure EventF { xcomp = x } = runAsyncT (unsafeCoerce x)

184

188

185−-- | Compose a list of continuations.

186−compose :: MonadIO m => [a -> AsyncT m a] -> (a -> AsyncT m b)

187−compose [] = const empty

188−compose (f:fs) = \x -> f x >>= compose fs

189−

190-- | Run the continuation (the 'f' in 'x >>= f') of the current bind operation with the current state.

189-- | Run the continuation (the 'f' in 'x >>= f') of the current bind operation with the current state.

191runContinuation :: MonadIO m => EventF -> a -> StateM m (Maybe b)

190runContinuation :: MonadIO m => EventF -> a -> StateM m (Maybe b)

192runContinuation EventF { fcomp = fs } =

191runContinuation EventF { fcomp = fs } =

@@ -226,18 +225,14 @@ back reason = AsyncT $ do

226 Nothing -> runContinuation first x

225 Nothing -> runContinuation first x

227 justreason -> goBackt $ Backtrack justreason bs

226 justreason -> goBackt $ Backtrack justreason bs

228

227

228+-}

229+

229------------------------------------------------------------------------------

230------------------------------------------------------------------------------

230-- MonadIO

231-- MonadIO

231------------------------------------------------------------------------------

232------------------------------------------------------------------------------

232

233

233instance MonadIO m => MonadIO (AsyncT m) where

234instance MonadIO m => MonadIO (AsyncT m) where

234− liftIO mx = do

235− ex <- liftIO' $ (mx >>= return . Right) `catch`

236− (\(e :: SomeException) -> return $ Left e)

237− case ex of

238− Left e -> back e

239− Right x -> return x

240− where liftIO' x = AsyncT $ liftIO x >>= return . Just

235+ liftIO mx = AsyncT $ liftIO mx >>= return . Just

241

236

242-------------------------------------------------------------------------------

237-------------------------------------------------------------------------------

243-- AsyncT transformer

238-- AsyncT transformer

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

403 abs f = f >>= return . abs

398 abs f = f >>= return . abs

404 signum f = f >>= return . signum

399 signum f = f >>= return . signum

405

400

401+{-

406class AdditionalOperators m where

402class AdditionalOperators m where

407

403

408 -- | Run @m a@ discarding its result before running @m b@.

404 -- | Run @m a@ discarding its result before running @m b@.

@@ -469,3 +465,12 @@ infixr 1 <***, <**, **>

469 runAsyncT mb

465 runAsyncT mb

470 return $ Just x

466 return $ Just x

471

467

468+-}

469+

470+(<**) :: Monad m => AsyncT m a -> AsyncT m b -> AsyncT m a

471+(<**) ma mb = AsyncT $ do

472+ a <- runAsyncT ma

473+ runAsyncT mb

474+ return a

475+

476+infixr 1 <**

src/Duct/Event.hs+9−4

@@ -32,13 +32,18 @@ data EventF = forall m a b. EventF

32 -- away with this?

33 -- ^ event data to use in a continuation in a new thread

35− , xcomp :: m a

35+ -- the 'm' and 'f' in an 'm >>= f' operation of the monad

36− , fcomp :: [b -> m b]

36+ -- In nested binds we store the current m only, but the whole stack of fs

37+ , currentm :: m a

38+ , fstack :: [b -> m b]

37 -- ^ List of continuations

39 -- ^ List of continuations

39 , mfData :: M.Map TypeRep SData

41 , mfData :: M.Map TypeRep SData

40 -- ^ State data accessed with get or put operations

42 -- ^ State data accessed with get or put operations

44+ -- XXX All of the following can be removed

45+ -- Even the remote status can be removed by the same logic

46+

42 , mfSequence :: Int

47 , mfSequence :: Int

44 , parentChannel :: Maybe (TChan ThreadId)

49 , parentChannel :: Maybe (TChan ThreadId)

@@ -91,8 +96,8 @@ initEventF

91 -> EventF

96 -> EventF

92initEventF x childChan pending credit =

97initEventF x childChan pending credit =

93 EventF { event = mempty

98 EventF { event = mempty

94− , xcomp = x

99+ , currentm = x

95− , fcomp = []

100+ , fstack = []

96 , mfData = mempty

101 , mfData = mempty

97 , mfSequence = 0

102 , mfSequence = 0

98 , parentChannel = Nothing

103 , parentChannel = Nothing

src/Duct/Threads.hs+19−11

@@ -12,7 +12,7 @@ module Duct.Threads

12 )

13where

15−import Control.Applicative ((<|>))

15+import Control.Applicative ((<|>), empty)

16import Control.Concurrent (ThreadId, forkIO, killThread,

17 myThreadId, threadDelay)

18import Control.Concurrent.STM (TChan, atomically, newTChan,

@@ -38,14 +38,22 @@ import Duct.AsyncT

38import Duct.Event

40------------------------------------------------------------------------------

41−-- Utilities

41+-- Pick up from where we left in the previous thread

42------------------------------------------------------------------------------

44−-- | Run the closure and the continuation using the state data of the calling thread

44+-- | Continue execution of the closure that we were executing when we migrated

45+-- to a new thread. Run the stack of pending functions in fstack. The types

46+-- don't match. We just coerce the types here, we know that they actually match.

47+

45−runCont :: (MonadIO m, Monad (AsyncT m)) => EventF -> StateT EventF m (Maybe a)

48+continue :: (MonadIO m, Monad (AsyncT m)) => EventF -> StateT EventF m (Maybe a)

46−runCont EventF { xcomp = x, fcomp = fs } = runAsyncT $ do

49+continue EventF { currentm = x, fstack = fs } =

47− r <- unsafeCoerce x

50+ runAsyncT $ unsafeCoerce x >>= runfStack (unsafeCoerce fs)

48− compose (unsafeCoerce fs) r

51+

52+ where

53+

54+ runfStack :: Monad m => [a -> AsyncT m a] -> a -> AsyncT m b

55+ runfStack [] _ = empty

56+ runfStack (f:fs) x = f x >>= runfStack fs

50------------------------------------------------------------------------------

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

51-- Thread Management (creation, reaping and killing)

59-- Thread Management (creation, reaping and killing)

@@ -210,7 +218,7 @@ genAsyncEvents parentc rec = forkMaybe parentc loop

210 -- pass on the io result and then run the continuation

218 -- pass on the io result and then run the continuation

211 runContWith dat cont = do

219 runContWith dat cont = do

212 let cont' = cont { event = Just $ unsafeCoerce dat }

220 let cont' = cont { event = Just $ unsafeCoerce dat }

213− _ <- runStateT (runCont cont') cont'

221+ _ <- runStateT (continue cont') cont'

214 return ()

222 return ()

215

223

216-- | Run an IO action one or more times to generate a stream of tasks. The IO

224-- | Run an IO action one or more times to generate a stream of tasks. The IO

@@ -248,7 +256,7 @@ waitEvents io = do

248 mr <- parallel (SMore <$> io)

256 mr <- parallel (SMore <$> io)

249 case mr of

257 case mr of

250 SMore x -> return x

258 SMore x -> return x

251− SError e -> back e

259+ -- SError e -> back e

252

260

253-- | Run an IO computation asynchronously and generate a single task carrying

261-- | Run an IO computation asynchronously and generate a single task carrying

254-- the result of the computation when it completes. See 'parallel' for notes on

262-- the result of the computation when it completes. See 'parallel' for notes on

@@ -258,7 +266,7 @@ async io = do

258 mr <- parallel (SLast <$> io)

266 mr <- parallel (SLast <$> io)

259 case mr of

267 case mr of

260 SLast x -> return x

268 SLast x -> return x

261− SError e -> back e

269+ -- SError e -> back e

262

270

263-- | Force an async computation to run synchronously. It can be useful in an

271-- | Force an async computation to run synchronously. It can be useful in an

264-- 'Alternative' composition to run the alternative only after finishing a

272-- 'Alternative' composition to run the alternative only after finishing a

@@ -310,7 +318,7 @@ react setHandler iob = AsyncT $ do

310 case event cont of

318 case event cont of

311 Nothing -> do

319 Nothing -> do

312 lift $ setHandler $ \dat ->do

320 lift $ setHandler $ \dat ->do

313− runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat}

321+ runStateT (continue cont) cont{event= Just $ unsafeCoerce dat}

314 liftIO iob

322 liftIO iob

315 was <- getData `onNothing` return NoRemote

323 was <- getData `onNothing` return NoRemote

316 when (was /= WasRemote) $ setData WasParallel

324 when (was /= WasRemote) $ setData WasParallel