1e4950d · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Simplify continuation saving and retrieval

1e4950dHarendra Kumar9 years ago

asyncly.cabal−1

compare with master

unifiedsplit

@@ -24,7 +24,6 @@ cabal-version: >= 1.10

24library

25 hs-source-dirs: src

26 exposed-modules: Asyncly

27− , Asyncly.Context

28 , Asyncly.AsyncT

27 , Asyncly.AsyncT

29 , Asyncly.Threads

28 , Asyncly.Threads

src/Asyncly.hs+1−2

@@ -31,6 +31,5 @@ module Asyncly

31 )

32where

34−import Asyncly.AsyncT (AsyncT)

34+import Asyncly.AsyncT (AsyncT, Log, Loggable)

35−import Asyncly.Context (Log, Loggable)

36import Asyncly.Threads

35import Asyncly.Threads

src/Asyncly/AsyncT.hs+155−4

@@ -1,3 +1,4 @@

1+{-# LANGUAGE ConstraintKinds #-}

1{-# LANGUAGE ExistentialQuantification #-}

2{-# LANGUAGE ExistentialQuantification #-}

2{-# LANGUAGE FlexibleContexts #-}

3{-# LANGUAGE FlexibleContexts #-}

3{-# LANGUAGE FlexibleInstances #-}

4{-# LANGUAGE FlexibleInstances #-}

@@ -11,6 +12,16 @@

12module Asyncly.AsyncT

13module Asyncly.AsyncT

13 ( AsyncT (..)

14 ( AsyncT (..)

15+ , ChildEvent(..)

16+ , Loggable

17+ , Log (..)

18+ , CtxLog (..)

19+ , LogEntry (..)

20+ , Context(..)

21+ , initContext

22+ , Location(..)

23+ , getLocation

24+ , setLocation

14 , (<**)

25 , (<**)

15 , (**>)

26 , (**>)

16 , dbg

27 , dbg

@@ -18,22 +29,156 @@ module Asyncly.AsyncT

18where

29where

20import Control.Applicative (Alternative (..))

31import Control.Applicative (Alternative (..))

32+import Control.Concurrent (ThreadId)

33+import Control.Concurrent.STM (TChan)

34+import Control.Exception (SomeException)

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

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

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

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

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

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

24− StateT (..), liftM, runStateT)

38+ StateT (..), liftM, runStateT,

39+ get, gets, modify)

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

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

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

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

27 MonadTransControl (..),

42 MonadTransControl (..),

28 defaultLiftBaseWith,

43 defaultLiftBaseWith,

29 defaultRestoreM)

44 defaultRestoreM)

30−

45+import Data.IORef (IORef)

31−import Asyncly.Context

46+import GHC.Prim (Any)

47+import Unsafe.Coerce (unsafeCoerce)

33--import Debug.Trace (traceM)

49--import Debug.Trace (traceM)

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

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

53+------------------------------------------------------------------------------

54+-- Log types

55+------------------------------------------------------------------------------

56+

57+-- Logging is done using the 'logged' combinator. It remembers the last value

58+-- returned by a given computation and replays it the next time the computation

59+-- is resumed. However this could be problematic if we do not annotate all

60+-- impure computations. The program can take a different path due to a

61+-- non-logged computation returning a different value. In that case we may

62+-- replay a wrong value. To detect this we can use a unique id for each logging

63+-- site and abort if the id does not match on replay.

64+

65+-- | Constraint type synonym for a value that can be logged.

66+type Loggable a = (Show a, Read a)

67+

68+-- XXX remove the Maybe as we never log Nothing

69+data LogEntry =

70+ Executing -- we are inside this computation, not yet done

71+ | Result (Maybe String) -- computation done, we have the result to replay

72+ deriving (Read, Show)

73+

74+data Log = Log [LogEntry] deriving Show

75+

76+-- log entries and replay entries

77+data CtxLog = CtxLog [LogEntry] [LogEntry] deriving (Read, Show)

78+

79+------------------------------------------------------------------------------

80+-- Parent child thread communication types

81+------------------------------------------------------------------------------

82+

83+data ChildEvent = ChildDone ThreadId (Maybe SomeException)

84+

85+------------------------------------------------------------------------------

86+-- State threaded around the monad

87+------------------------------------------------------------------------------

88+

89+data Context = Context

90+ {

91+ ---------------------------------------------------------------------------

92+ -- Execution state

93+ ---------------------------------------------------------------------------

94+

95+ -- a -> AsyncT m b

96+ continuation :: Any

97+ -- Logging is done by the 'logged' primitive in this journal only if logsRef

98+ -- exists.

99+ , journal :: CtxLog

100+

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

102+ , logsRef :: Maybe (IORef [Log])

103+ , location :: Location

104+

105+ ---------------------------------------------------------------------------

106+ -- Thread creation, communication and cleanup

107+ ---------------------------------------------------------------------------

108+ -- When a new thread is created the parent records it in the

109+ -- 'pendingThreads' field and 'threadCredit' is decremented. When a child

110+ -- thread is done it sends a done event to the parent on an unbounded

111+ -- channel and goes away. Before starting a new thread the parent always

112+ -- processes the unbounded channel to clear up the pending zombies. This

113+ -- strategy ensures that the zombie queue will never grow more than the

114+ -- number of running threads. Before exiting, the parent thread waits on

115+ -- the channel until all its children are cleaned up.

116+ ---------------------------------------------------------------------------

117+

118+ -- XXX setup a cleanup computation to run rather than passing all these

119+ -- params.

120+

121+ , childChannel :: TChan ChildEvent

122+ -- ^ A channel for the immediate children to communicate to us when they

123+ -- die. Each thread has its own dedicated channel for its children

124+

125+ -- We always track the child threads, otherwise the programmer needs to

126+ -- worry about if some threads may remain hanging because they are stuck in

127+ -- an infinite loop or a forever blocked IO. Also, it is not clear whether

128+ -- there is a significant gain by disabling tracking. Instead of using

129+ -- threadIds we can also keep a count instead, that will allow us to wait

130+ -- for all children to drain but we cannot kill them.

131+ --

132+ -- We need these as IORefs since we need any changes to these to be

133+ -- reflected back in the calling code that waits for the children.

134+

135+ , pendingThreads :: IORef [ThreadId]

136+ -- ^ Active immediate child threads spawned by this thread, modified only

137+ -- by this thread, therefore atomic modification is not needed.

138+

139+ -- By default this limit is shared by the current tree of threads. But at

140+ -- any point the 'threads' primitive can be used to start a new limit for

141+ -- the computation under that primitive.

142+ --

143+ -- Shared across many threads, therefore atomic access is required.

144+ -- This is incremented by the child thread itelf before it exits.

145+ --

146+ , threadCredit :: IORef Int

147+ -- ^ How many more threads are allowed to be created?

148+ }

149+

150+initContext

151+ :: TChan ChildEvent

152+ -> IORef [ThreadId]

153+ -> IORef Int

154+ -> (a -> AsyncT m a)

155+ -> Maybe (IORef [Log])

156+ -> Context

157+initContext childChan pending credit finalizer lref =

158+ Context { continuation = unsafeCoerce finalizer

159+ , journal = CtxLog [] []

160+ , logsRef = lref

161+ , location = Worker

162+ , childChannel = unsafeCoerce childChan

163+ , pendingThreads = pending

164+ , threadCredit = credit }

165+

166+------------------------------------------------------------------------------

167+-- Where is the computation running?

168+------------------------------------------------------------------------------

169+

170+-- RemoteNode is used in the Alternative instance. We stop the computation when

171+-- the location is RemoteNode, we do not execute any alternatives.

172+--

173+data Location = Worker | WaitingParent | RemoteNode

174+ deriving (Eq, Show)

175+

176+getLocation :: Monad m => StateT Context m Location

177+getLocation = gets location

178+

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

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

181+

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

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

38-- Utilities

183-- Utilities

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

184------------------------------------------------------------------------------

@@ -81,7 +226,13 @@ instance Monad m => Alternative (AsyncT m) where

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

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

82 return = pure

227 return = pure

83 m >>= f = AsyncT $ do

228 m >>= f = AsyncT $ do

84− saveContext m f >> runAsyncT m >>= restoreContext >>= runAsyncT

229+ modify $ \Context { continuation = g, .. } ->

230+ let k x = f x >>= unsafeCoerce g

231+ in Context { continuation = unsafeCoerce k, .. }

232+

233+ x <- runAsyncT m

234+ Context { continuation = g } <- get

235+ runAsyncT $ maybe mzero (unsafeCoerce g) x

236

86instance Monad m => MonadPlus (AsyncT m) where

237instance Monad m => MonadPlus (AsyncT m) where

87 mzero = empty

238 mzero = empty

src/Asyncly/Context.hsdeleted−256

@@ -1,256 +0,0 @@

1−{-# LANGUAGE ConstraintKinds #-}

2−{-# LANGUAGE ExistentialQuantification #-}

3−{-# LANGUAGE FlexibleContexts #-}

4−{-# LANGUAGE RecordWildCards #-}

5−

6−module Asyncly.Context

7− ( ChildEvent(..)

8− , Loggable

9− , Log (..)

10− , CtxLog (..)

11− , LogEntry (..)

12− , Context(..)

13− , initContext

14− , saveContext

15− , restoreContext

16− , composeContext

17− , setContextMailBox

18− , takeContextMailBox

19− , Location(..)

20− , getLocation

21− , setLocation

22− )

23−where

24−

25−import Control.Concurrent (ThreadId)

26−import Control.Concurrent.STM (TChan)

27−import Control.Exception (SomeException)

28−import Control.Monad.State (StateT, MonadPlus(..))

29−import qualified Control.Monad.Trans.State.Lazy as Lazy (get, gets, modify, put)

30−import Data.Dynamic (Typeable)

31−import Data.IORef (IORef)

32−import Unsafe.Coerce (unsafeCoerce)

33−import GHC.Prim (Any)

34−

35−------------------------------------------------------------------------------

36−-- Log types

37−------------------------------------------------------------------------------

38−

39−-- Logging is done using the 'logged' combinator. It remembers the last value

40−-- returned by a given computation and replays it the next time the computation

41−-- is resumed. However this could be problematic if we do not annotate all

42−-- impure computations. The program can take a different path due to a

43−-- non-logged computation returning a different value. In that case we may

44−-- replay a wrong value. To detect this we can use a unique id for each logging

45−-- site and abort if the id does not match on replay.

46−

47−-- | Constraint type synonym for a value that can be logged.

48−type Loggable a = (Show a, Read a)

49−

50−-- XXX remove the Maybe as we never log Nothing

51−data LogEntry =

52− Executing -- we are inside this computation, not yet done

53− | Result (Maybe String) -- computation done, we have the result to replay

54− deriving (Read, Show)

55−

56−data Log = Log [LogEntry] deriving Show

57−

58−-- log entries and replay entries

59−data CtxLog = CtxLog [LogEntry] [LogEntry] deriving (Read, Show)

60−

61−------------------------------------------------------------------------------

62−-- Parent child thread communication types

63−------------------------------------------------------------------------------

64−

65−data ChildEvent a = ChildDone ThreadId (Maybe SomeException)

66−

67−------------------------------------------------------------------------------

68−-- State threaded around the monad

69−------------------------------------------------------------------------------

70−

71−-- | Describes the context of a computation.

72−data Context = Context

73− {

74− ---------------------------------------------------------------------------

75− -- Execution state

76− ---------------------------------------------------------------------------

77−

78− mailBox :: Maybe Any -- untyped,

79− -- ^ mailBox to send event data to the child thread

80−

81− , currentm :: Any -- untyped, the real type is: m a

82− -- ^ The 'm' in an 'm >>= f' operation of the monad. This tells us where we

83− -- are in the monadic computation at this point.

84−

85− , fstack :: [Any] -- untyped, the real type is: [a -> m b]

86− -- ^ The head of the list stores the 'f' corresponding to 'currentm' in an

87− -- 'm >>= f' operation of the monad. This is stack because when a

88− -- computation has nested binds we need to store the next 'f' for each

89− -- layer of nesting.

90−

91− -- Logging is done by the 'logged' primitive in this journal only if logsRef

92− -- exists.

93− , journal :: CtxLog

94−

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

96− , logsRef :: Maybe (IORef [Log])

97− , location :: Location

98−

99− ---------------------------------------------------------------------------

100− -- User state

101− ---------------------------------------------------------------------------

102−

103− -- , mfData :: M.Map TypeRep Any -- untyped, type coerced

104− -- ^ State data accessed with get or put operations

105−

106− ---------------------------------------------------------------------------

107− -- Thread creation, communication and cleanup

108− ---------------------------------------------------------------------------

109− -- When a new thread is created the parent records it in the

110− -- 'pendingThreads' field and 'threadCredit' is decremented. When a child

111− -- thread is done it sends a done event to the parent on an unbounded

112− -- channel and goes away. Before starting a new thread the parent always

113− -- processes the unbounded channel to clear up the pending zombies. This

114− -- strategy ensures that the zombie queue will never grow more than the

115− -- number of running threads. Before exiting, the parent thread waits on

116− -- the channel until all its children are cleaned up.

117− ---------------------------------------------------------------------------

118−

119− -- XXX setup a cleanup computation to run rather than passing all these

120− -- params.

121−

122− , childChannel :: TChan (ChildEvent Any)

123− -- ^ A channel for the immediate children to communicate to us when they

124− -- die. Each thread has its own dedicated channel for its children

125−

126− -- We always track the child threads, otherwise the programmer needs to

127− -- worry about if some threads may remain hanging because they are stuck in

128− -- an infinite loop or a forever blocked IO. Also, it is not clear whether

129− -- there is a significant gain by disabling tracking. Instead of using

130− -- threadIds we can also keep a count instead, that will allow us to wait

131− -- for all children to drain but we cannot kill them.

132− --

133− -- We need these as IORefs since we need any changes to these to be

134− -- reflected back in the calling code that waits for the children.

135−

136− , pendingThreads :: IORef [ThreadId]

137− -- ^ Active immediate child threads spawned by this thread, modified only

138− -- by this thread, therefore atomic modification is not needed.

139−

140− -- By default this limit is shared by the current tree of threads. But at

141− -- any point the 'threads' primitive can be used to start a new limit for

142− -- the computation under that primitive.

143− --

144− -- Shared across many threads, therefore atomic access is required.

145− -- This is incremented by the child thread itelf before it exits.

146− --

147− , threadCredit :: IORef Int

148− -- ^ How many more threads are allowed to be created?

149− } deriving Typeable

150−

151−initContext

152− :: m a

153− -> TChan (ChildEvent a)

154− -> IORef [ThreadId]

155− -> IORef Int

156− -> (b -> m a)

157− -> Maybe (IORef [Log])

158− -> Context

159−initContext x childChan pending credit finalizer lref =

160− Context { mailBox = Nothing

161− , currentm = unsafeCoerce x

162− , fstack = [unsafeCoerce finalizer]

163− , journal = CtxLog [] []

164− , logsRef = lref

165− , location = Worker

166− -- , mfData = mempty

167− , childChannel = unsafeCoerce childChan

168− , pendingThreads = pending

169− , threadCredit = credit }

170−

171−------------------------------------------------------------------------------

172−-- Where is the computation running?

173−------------------------------------------------------------------------------

174−

175−-- RemoteNode is used in the Alternative instance. We stop the computation when

176−-- the location is RemoteNode, we do not execute any alternatives.

177−--

178−-- WaitingParent is used in logging. When a parent forks a child and waits for

179−-- it, it is marked as WaitingParent and we log a "Wait" entry in the log.

180−--

181−data Location = Worker | WaitingParent | RemoteNode

182− deriving (Typeable, Eq, Show)

183−

184−getLocation :: Monad m => StateT Context m Location

185−getLocation = Lazy.gets location

186−

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

188−setLocation loc = Lazy.modify $ \Context { .. } -> Context { location = loc, .. }

189−

190−------------------------------------------------------------------------------

191−-- Saving and restoring the context of a computation. We perform the monadic

192−-- computation one step at a time storing the context of where we are in the

193−-- computation. At any point the remaining computation can be fully recovered

194−-- from this context.

195−------------------------------------------------------------------------------

196−

197−-- Save the 'm' and 'f', in case we migrate to a new thread before the current

198−-- bind operation completes, we run the operation manually in the new thread

199−-- using this context.

200−saveContext :: Monad m => f a -> (a -> f b) -> StateT Context m ()

201−saveContext m f =

202− Lazy.modify $ \Context { fstack = fs, .. } ->

203− Context { currentm = unsafeCoerce m

204− , fstack = unsafeCoerce f : fs

205− , .. }

206−

207−-- pop the top function from the continuation stack, create the next closure,

208−-- set it as the current closure and return it.

209−restoreContext :: (MonadPlus m1, Monad m) => Maybe a -> StateT Context m (m1 b)

210−restoreContext x = do

211− -- XXX fstack must be non-empty when this is called.

212− ctx@Context { fstack = f:fs } <- Lazy.get

213− let mnext = maybe mzero (unsafeCoerce f) x

214− Lazy.put ctx { currentm = unsafeCoerce mnext, fstack = fs }

215− return mnext

216−

217−-- | We can retrieve a context at any point and resume that context at some

218−-- later point, upon resumption we start executing again from the same point

219−-- where the context was retrieved from.

220−--

221−-- This function composes the stored context to recover the computation.

222−--

223−composeContext :: Monad m => Context -> m a

224−composeContext Context { currentm = m

225− , fstack = fs

226− } =

227− unsafeCoerce m >>= composefStack fs

228−

229− where

230−

231− -- Compose the stack of pending functions in fstack using the bind

232− -- operation of the monad. The types don't match. We just coerce the types

233− -- here, we know that they actually match.

234−

235− -- The last computation is always 'mzero' to stop the whole computation.

236− -- Before returning we save the result of the computation in the context.

237−

238− -- composefStack :: Monad m => [a -> AsyncT m a] -> a -> AsyncT m b

239− composefStack [] _ = error "Bug: this should never be reached"

240− composefStack (f:ff) x = (unsafeCoerce f) x >>= composefStack ff

241−

242−------------------------------------------------------------------------------

243−-- Mailbox

244−------------------------------------------------------------------------------

245−

246−setContextMailBox :: Context -> a -> Context

247−setContextMailBox ctx mbdata = ctx { mailBox = Just $ unsafeCoerce mbdata }

248−

249−takeContextMailBox :: Monad m => StateT Context m (Either Context a)

250−takeContextMailBox = do

251− ctx <- Lazy.get

252− case mailBox ctx of

253− Nothing -> return $ Left ctx

254− Just x -> do

255− Lazy.put ctx { mailBox = Nothing }

256− return $ Right (unsafeCoerce x)

src/Asyncly/Threads.hs+38−55

@@ -42,9 +42,9 @@ import Data.IORef (IORef, atomicModifyIORef,

42 writeIORef)

43import Data.List (delete)

44import Data.Maybe (isJust)

45+import Unsafe.Coerce (unsafeCoerce)

46import Asyncly.AsyncT

47import Asyncly.AsyncT

47−import Asyncly.Context

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

@@ -79,9 +79,9 @@ type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

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

80-- to a new thread.

82−runContext :: Monad m => Context -> StateT Context m ()

82+runContext :: Monad m => Context -> AsyncT m a -> StateT Context m ()

83−runContext ctx = do

83+runContext ctx action = do

84− let s = runAsyncT (composeContext ctx)

84+ let s = runAsyncT $ action >>= unsafeCoerce (continuation ctx)

85 _ <- lift $ runStateT s ctx

86 return ()

@@ -93,7 +93,7 @@ runContext ctx = do

93-- pchan. We can use bounded channels to throttle the creation of threads based

94-- on consumption rate.

95processOneEvent :: MonadIO m

96− => ChildEvent a

96+ => ChildEvent

97 -> [ThreadId]

98 -> m ([ThreadId], Maybe SomeException)

99processOneEvent (ChildDone tid e) pending = do

@@ -101,7 +101,7 @@ processOneEvent (ChildDone tid e) pending = do

101 return (delete tid pending, Nothing)

102

103drainChildren :: MonadIO m

104− => TChan (ChildEvent a)

104+ => TChan ChildEvent

105 -> [ThreadId]

106 -> m ([ThreadId], Maybe SomeException)

107drainChildren cchan pending =

@@ -180,23 +180,27 @@ signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ())

180

181forkFinally1 :: (MonadIO m, MonadBaseControl IO m)

182 => Context

183+ -> AsyncT m a

183 -> (Either SomeException () -> IO ())

184 -> (Either SomeException () -> IO ())

184 -> StateT Context m ThreadId

185 -> StateT Context m ThreadId

185−forkFinally1 ctx preExit =

186+forkFinally1 ctx action preExit =

186 EL.mask $ \restore ->

187 EL.mask $ \restore ->

187 liftBaseWith $ \runInIO -> forkIO $ do

188 liftBaseWith $ \runInIO -> forkIO $ do

188− _ <- runInIO $ EL.try (restore (runContext ctx))

189+ _ <- runInIO $ EL.try (restore (runContext ctx action))

189 >>= liftIO . preExit

190 >>= liftIO . preExit

191+ -- XXX restore state here

190 return ()

192 return ()

191

193

192-- | Run a given context in a new thread.

194-- | Run a given context in a new thread.

193--

195--

194forkContext :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)

196forkContext :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)

195− => Context -> StateT Context m ()

197+ => AsyncT m a -> StateT Context m ()

196−forkContext context = do

198+forkContext action = do

199+ parentCtx <- get

197− child <- childContext context

200+ childCtx <- childContext parentCtx

201+ tid <- forkFinally1 childCtx action

198− tid <- forkFinally1 child (beforeExit child (childChannel context))

202+ (beforeExit childCtx (childChannel parentCtx))

199− updatePendingThreads context tid

203+ updatePendingThreads parentCtx tid

200

204

201 where

205 where

202

206

@@ -245,6 +249,7 @@ canFork context = do

245 -- special handling for those cases. Add those to

249 -- special handling for those cases. Add those to

246 -- unreclaimable list? And always execute them in an

250 -- unreclaimable list? And always execute them in an

247 -- async thread, cannot use sync for those.

251 -- async thread, cannot use sync for those.

252+ --

248 waitForOneEvent context

253 waitForOneEvent context

249 canFork context

254 canFork context

250 True -> return True

255 True -> return True

@@ -254,16 +259,15 @@ canFork context = do

254-- the current thread quota.

259-- the current thread quota.

255--

260--

256resumeContextWith :: MonadAsync m

261resumeContextWith :: MonadAsync m

257− => Context -- the context to resume

258− -> Bool -- force synchronous

262+ => Bool -- force synchronous

259− -> AsyncT m a -- the action to execute in the resumed context

263+ -> AsyncT m a

260 -> StateT Context m ()

264 -> StateT Context m ()

261−resumeContextWith context synch action = do

265+resumeContextWith synch action = do

262− let ctx = setContextMailBox context action

266+ context <- get

263 can <- liftIO $ canFork context

267 can <- liftIO $ canFork context

264 case can && (not synch) of

268 case can && (not synch) of

265− False -> runContext ctx -- run synchronously

269+ False -> runContext context action

266− True -> forkContext ctx

270+ True -> forkContext action

267

271

268-- The current model is to start a new thread for every task. The input is

272-- The current model is to start a new thread for every task. The input is

269-- provided at the time of the creation and therefore no synchronization is

273-- provided at the time of the creation and therefore no synchronization is

@@ -291,32 +295,13 @@ spawningParentDone = do

291 when (loc /= RemoteNode) $ setLocation WaitingParent

295 when (loc /= RemoteNode) $ setLocation WaitingParent

292 return Nothing

296 return Nothing

293

297

294−-- | Captures the state of the current computation at this point, starts a new

295−-- thread, using the argument passed as the return value of the computation and

296−-- continues the computation from the capture point. The end result is as if

297−-- this function just returned the value passed as argument albeit in a new

298−-- thread.

299−--

300-- If a new thread cannot be created then the computation is run in the same

298-- If a new thread cannot be created then the computation is run in the same

301-- thread, but the functional behavior remains the same.

299-- thread, but the functional behavior remains the same.

302

300

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

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

304async action = AsyncT $ do

302async action = AsyncT $ do

305− val <- takeContextMailBox

306− case val of

307− -- Child task

308− Right m -> runAsyncT m

309−

310− -- Spawning parent

311− Left ctx -> do

312− resumeContextWith ctx False action

303+ resumeContextWith False action

313−

314− -- If we started the task asynchronously in a new thread then the

315− -- parent thread reaches here, immediately after spawning the task.

316− --

317− -- However, if the task was executed synchronously then we reach

318− -- here after it is completely done.

319− spawningParentDone

304+ spawningParentDone

320

305

321-- | Makes an asyncly action from a callback setter function; can be used to

306-- | Makes an asyncly action from a callback setter function; can be used to

322-- convert existing callback style code into asyncly style code. The first

307-- convert existing callback style code into asyncly style code. The first

@@ -325,17 +310,16 @@ async action = AsyncT $ do

325-- value of type 'a'. After the computation is done the result of the action

310-- value of type 'a'. After the computation is done the result of the action

326-- in second parameter is returned to the callback.

311-- in second parameter is returned to the callback.

327--

312--

328−makeAsync :: (MonadAsync m) => ((a -> m b) -> m ()) -> m b -> AsyncT m a

313+makeAsync :: MonadAsync m => ((a -> m ()) -> m ()) -> AsyncT m a

329−makeAsync cbsetter cbret = AsyncT $ do

314+makeAsync cbsetter = AsyncT $ do

330− val <- takeContextMailBox

315+ -- XXX should we force fork a thread or keep running the context of the

331− case val of

316+ -- callback?

332− Right m -> runAsyncT m -- child

317+ ctx <- get

333− Left ctx -> do -- parent

334− lift $ cbsetter $ \a -> do

318+ lift $ cbsetter $ \a -> do

335− let s = resumeContextWith ctx False (return a)

319+ let s = resumeContextWith False (return a)

336− _ <- runStateT s ctx

320+ _ <- runStateT s ctx

337− cbret

321+ return ()

338− spawningParentDone

322+ spawningParentDone

339

323

340-- scatter

324-- scatter

341each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)

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

@@ -380,15 +364,14 @@ threads n process = AsyncT $ do

380

364

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

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

382-- thread of the computation in a list.

366-- thread of the computation in a list.

383−waitAsync :: forall m a. (MonadIO m, MonadCatch m)

367+waitAsync :: (MonadIO m, MonadCatch m)

384 => (a -> AsyncT m a) -> Maybe (IORef [Log]) -> AsyncT m a -> m ()

368 => (a -> AsyncT m a) -> Maybe (IORef [Log]) -> AsyncT m a -> m ()

385waitAsync finalizer lref m = do

369waitAsync finalizer lref m = do

386 childChan <- liftIO $ atomically newTChan

370 childChan <- liftIO $ atomically newTChan

387 pendingRef <- liftIO $ newIORef []

371 pendingRef <- liftIO $ newIORef []

388 credit <- liftIO $ newIORef maxBound

372 credit <- liftIO $ newIORef maxBound

389

373

390− let ctx = initContext (empty :: AsyncT m a) childChan pendingRef credit

374+ let ctx = initContext childChan pendingRef credit finalizer lref

391− finalizer lref

392

375

393 r <- try $ runStateT (runAsyncT $ m >>= finalizer) ctx

376 r <- try $ runStateT (runAsyncT $ m >>= finalizer) ctx

394

377

test/Main.hs+1−3

@@ -31,8 +31,6 @@ main = hspec $ do

31 it "Nested async tasks ignoring return values" $

32 (wait $ (async $ async $ return 0 :: AsyncT IO Int) >> return 1)

33 `shouldReturn` ([1] :: [Int])

34− -- XXX we should run the alternative purely based on the highest level

35− -- computation, the nested ones should not affect it.

36 it "Nested async tasks with Alternative" $

34 it "Nested async tasks with Alternative" $

37 (wait (async (async $ return 0) <|> return 1) >>= return . sort)

35 (wait (async (async $ return 0) <|> return 1) >>= return . sort)

38− `shouldReturn` ([0,1,1] :: [Int])

36+ `shouldReturn` ([0,1] :: [Int])