9779a0b · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Pass results from children to parent

9779a0bHarendra Kumar9 years ago

src/Duct/AsyncT.hs+22−18

compare with master

unifiedsplit

@@ -41,7 +41,7 @@ import Data.Maybe (isJust, isNothing)

42import Duct.Context

44−-- import Debug.Trace

44+-- import Debug.Trace (traceM)

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

@@ -49,7 +49,9 @@ newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Maybe a) }

49-- Utilities

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

52−-- dbg x = trace x (return ())

52+-- Toggle the dbg definition for debug traces

53+--dbg :: Monad m => String -> m ()

54+--dbg = traceM

53dbg :: Monad m => a -> m ()

55dbg :: Monad m => a -> m ()

54dbg _ = return ()

56dbg _ = return ()

@@ -92,6 +94,7 @@ instance Monad m => Alternative (AsyncT m) where

92 RemoteNode -> return Nothing

94 RemoteNode -> return Nothing

93 _ -> maybe (runAsyncT y) (return . Just) mx

95 _ -> maybe (runAsyncT y) (return . Just) mx

97+-- XXX remove this, use mzero instead

95-- | A synonym of 'empty' that can be used in a monadic expression. It stops

98-- | A synonym of 'empty' that can be used in a monadic expression. It stops

96-- the computation, which allows the next computation in an 'Alternative'

99-- the computation, which allows the next computation in an 'Alternative'

97-- ('<|>') composition to run.

100-- ('<|>') composition to run.

@@ -166,24 +169,22 @@ drainChildren chan pending res =

166 ev <- liftIO $ atomically $ readTChan chan

169 ev <- liftIO $ atomically $ readTChan chan

167 case ev of

170 case ev of

168 ChildDone tid er -> do

171 ChildDone tid er -> do

169− dbg $ "Reaping child: " ++ show tid

172+ dbg $ "drainChildrenTop ChildDone, tid: " ++ show tid

170 case er of

173 case er of

171 Left e -> throwM e

174 Left e -> throwM e

172− Right r -> do

175+ Right r ->

173− let newres = case r of

174− Nothing -> res

175− Just x -> x:res

176− drainChildren chan (delete tid pending) newres

176+ drainChildren chan (delete tid pending) (r ++ res)

177− PassOnResult er ->

177+ PassOnResult er -> do

178+ dbg $ "drainChildrenTop PassOnResult"

178 case er of

179 case er of

179 Left e -> throwM e

180 Left e -> throwM e

180− Right x -> drainChildren chan pending (x:res)

181+ Right r -> drainChildren chan pending (r ++ res)

181

182

182waitForChildren :: (MonadIO m, MonadThrow m)

183waitForChildren :: (MonadIO m, MonadThrow m)

183− => TChan (ChildEvent a) -> IORef [ThreadId] -> m [a]

184+ => TChan (ChildEvent a) -> IORef [ThreadId] -> [a] -> m [a]

184−waitForChildren chan pendingRef = do

185+waitForChildren chan pendingRef results = do

185 pending <- liftIO $ readIORef pendingRef

186 pending <- liftIO $ readIORef pendingRef

186− r <- drainChildren chan pending []

187+ r <- drainChildren chan pending results

187 liftIO $ writeIORef pendingRef []

188 liftIO $ writeIORef pendingRef []

188 return r

189 return r

189

190

@@ -195,24 +196,27 @@ waitForChildren chan pendingRef = do

195-- XXX Ideally it should be a non-empty list instead.

196-- XXX Ideally it should be a non-empty list instead.

196-- | Run an 'AsyncT m' computation. Returns a list of results of the

197-- | Run an 'AsyncT m' computation. Returns a list of results of the

197-- computation or may throw an exception.

198-- computation or may throw an exception.

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

199+waitAsync :: forall m a. (MonadIO m, MonadCatch m)

199 => AsyncT m a -> m [a]

200 => AsyncT m a -> m [a]

200waitAsync m = do

201waitAsync m = do

201 childChan <- liftIO $ atomically newTChan

202 childChan <- liftIO $ atomically newTChan

202 pendingRef <- liftIO $ newIORef []

203 pendingRef <- liftIO $ newIORef []

203 credit <- liftIO $ newIORef maxBound

204 credit <- liftIO $ newIORef maxBound

205+ resultsRef <- liftIO $ newIORef []

204

206

205 -- XXX this should be moved to Context.hs and then we can make m

207 -- XXX this should be moved to Context.hs and then we can make m

206 -- existential and remove the unsafeCoerces

208 -- existential and remove the unsafeCoerces

207 r <- try $ runStateT (runAsyncT m) $ initContext

209 r <- try $ runStateT (runAsyncT m) $ initContext

208− (empty :: AsyncT m a) childChan pendingRef credit

210+ (empty :: AsyncT m a) childChan pendingRef credit resultsRef

209

211

212+ results <- liftIO $ readIORef resultsRef

210− case r of

213+ xs <- case r of

211 Left (exc :: SomeException) -> do

214 Left (exc :: SomeException) -> do

212 liftIO $ readIORef pendingRef >>= mapM_ killThread

215 liftIO $ readIORef pendingRef >>= mapM_ killThread

213 throwM exc

216 throwM exc

217+ Right (Nothing, _) -> return results

218+ Right ((Just x), _) -> return $ x:results

214− Right (Nothing, _) -> waitForChildren childChan pendingRef

219+ waitForChildren childChan pendingRef xs

215− Right ((Just x), _) -> return [x]

216

220

217------------------------------------------------------------------------------

221------------------------------------------------------------------------------

218-- * Extensible State: Session Data Management

222-- * Extensible State: Session Data Management

src/Duct/Context.hs+34−19

@@ -23,21 +23,22 @@ import Control.Applicative (Alternative(..))

23import Control.Concurrent (ThreadId)

24import Control.Concurrent.STM (TChan)

25import Control.Exception (SomeException)

26−import Control.Monad.State (MonadState, gets, modify, StateT, MonadPlus)

26+import Control.Monad.State (MonadState, gets, modify, StateT,

27+ MonadPlus(..), MonadIO(..))

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

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

28import Data.Dynamic (TypeRep, Typeable, typeOf)

29import Data.Dynamic (TypeRep, Typeable, typeOf)

29−import Data.IORef (IORef)

30+import Data.IORef (IORef, modifyIORef)

30import qualified Data.Map as M

31import qualified Data.Map as M

31import Unsafe.Coerce (unsafeCoerce)

32import Unsafe.Coerce (unsafeCoerce)

32−import GHC.Prim (Any)

33+import GHC.Prim (Any)

34------------------------------------------------------------------------------

35------------------------------------------------------------------------------

35-- State of a continuation

36-- State of a continuation

36------------------------------------------------------------------------------

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

38data ChildEvent a =

39data ChildEvent a =

39− ChildDone ThreadId (Either SomeException (Maybe a)) -- A child is finished

40+ ChildDone ThreadId (Either SomeException [a]) -- A child is finished

40− | PassOnResult (Either SomeException a) -- Pass on the result of a child

41+ | PassOnResult (Either SomeException [a]) -- Pass on the result of a child

42-- | Describes the context of a computation.

43-- | Describes the context of a computation.

43data Context = Context

44data Context = Context

@@ -100,6 +101,8 @@ data Context = Context

100 --

101 --

101 , threadCredit :: IORef Int

102 , threadCredit :: IORef Int

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

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

104+ , accumResults :: IORef [Any]

105+ -- ^ Accumulated results when running synchronously

103 } deriving Typeable

106 } deriving Typeable

104

107

105initContext

108initContext

@@ -107,8 +110,9 @@ initContext

107 -> TChan (ChildEvent a)

110 -> TChan (ChildEvent a)

108 -> IORef [ThreadId]

111 -> IORef [ThreadId]

109 -> IORef Int

112 -> IORef Int

113+ -> IORef [a]

110 -> Context

114 -> Context

111−initContext x childChan pending credit =

115+initContext x childChan pending credit results =

112 Context { event = mempty

116 Context { event = mempty

113 , currentm = unsafeCoerce x

117 , currentm = unsafeCoerce x

114 , fstack = []

118 , fstack = []

@@ -117,7 +121,8 @@ initContext x childChan pending credit =

117 , parentChannel = Nothing

121 , parentChannel = Nothing

118 , childChannel = unsafeCoerce childChan

122 , childChannel = unsafeCoerce childChan

119 , pendingThreads = pending

123 , pendingThreads = pending

120− , threadCredit = credit }

124+ , threadCredit = credit

125+ , accumResults = unsafeCoerce results }

121

126

122------------------------------------------------------------------------------

127------------------------------------------------------------------------------

123-- Where is the computation running?

128-- Where is the computation running?

@@ -168,22 +173,32 @@ restoreContext x = do

168 return mres

173 return mres

169

174

170-- | Resume execution of the computation from a given context. We can retrieve

175-- | Resume execution of the computation from a given context. We can retrieve

171−-- a context at any point and resume that context at some later point, which

176+-- a context at any point and resume that context at some later point, upon

172−-- means start executing again from the same point where the context was

177+-- resumption we start executing again from the same point where the context

173−-- retrieved.

178+-- was retrieved.

174--

179--

175−-- Run the stack of pending functions in fstack. The types don't match. We just

176−-- coerce the types here, we know that they actually match.

177−

178−resumeContext :: MonadPlus m => Context -> m a

180+resumeContext :: (MonadIO m, MonadPlus m) => Context -> m a

179−resumeContext Context { currentm = m, fstack = fs } =

181+resumeContext Context { currentm = m

182+ , fstack = fs

183+ , accumResults = resultsRef

184+ } =

180− unsafeCoerce m >>= runfStack (unsafeCoerce fs)

185+ unsafeCoerce m >>= composefStack (unsafeCoerce fs)

181

186

182 where

187 where

183

188

189+ -- Compose the stack of pending functions in fstack using the bind

190+ -- operation of the monad. The types don't match. We just coerce the types

191+ -- here, we know that they actually match.

192+

193+ -- The last computation is always 'mzero' to stop the whole computation.

194+ -- Before returning we save the result of the computation in the context.

195+

184− -- runfStack :: Monad m => [a -> AsyncT m a] -> a -> AsyncT m b

196+ -- composefStack :: Monad m => [a -> AsyncT m a] -> a -> AsyncT m b

185− runfStack [] _ = empty

197+ composefStack [] x = do

186− runfStack (f:ff) x = f x >>= runfStack ff

198+ -- remove this, return use the modified context

199+ liftIO $ modifyIORef resultsRef $ \r -> x : r

200+ mzero

201+ composefStack (f:ff) x = f x >>= composefStack ff

187

202

188------------------------------------------------------------------------------

203------------------------------------------------------------------------------

189-- * Extensible State: Session Data Management

204-- * Extensible State: Session Data Management

src/Duct/Threads.hs+27−30

@@ -69,11 +69,12 @@ import Duct.Context

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

70-- to a new thread.

72−resume :: Monad m => Context -> m (Maybe a)

72+resume :: MonadIO m => Context -> m Context

73resume ctx = do

74 let s = runAsyncT (resumeContext ctx)

75+ -- The returned value is always 'Nothing', we just discard it

75− (a, _) <- runStateT s ctx

76+ (_, c) <- runStateT s ctx

76− return a

77+ return c

78updateContextEvent :: Context -> a -> Context

79updateContextEvent :: Context -> a -> Context

79updateContextEvent ctx evdata = ctx { event = Just $ unsafeCoerce evdata }

80updateContextEvent ctx evdata = ctx { event = Just $ unsafeCoerce evdata }

@@ -96,9 +97,11 @@ processOneEvent ev pchan pending exc = do

96 Nothing ->

97 Nothing ->

97 case ev of

98 case ev of

98 ChildDone tid res -> do

99 ChildDone tid res -> do

99− dbg $ "Reaping child: " ++ show tid

100+ dbg $ "processOneEvent ChildDone: " ++ show tid

100− handleDone res

101+ handlePass res

101− PassOnResult res -> handlePass res

102+ PassOnResult res -> do

103+ dbg $ "processOneEvent PassOnResult"

104+ handlePass res

102 Just _ -> return exc

105 Just _ -> return exc

103 let p = case ev of

106 let p = case ev of

104 ChildDone tid _ -> delete tid pending

107 ChildDone tid _ -> delete tid pending

@@ -107,29 +110,16 @@ processOneEvent ev pchan pending exc = do

107

110

108 where

111 where

109

112

110− handleDone :: MonadIO m

111− => Either SomeException (Maybe a) -> m (Maybe SomeException)

112− handleDone res =

113− case res of

114− Left e -> do

115− liftIO $ mapM_ killThread pending

116− return (Just e)

117− Right r -> do

118− case r of

119− Nothing -> return ()

120− Just x -> liftIO $ atomically

121− $ writeTChan pchan

122− (PassOnResult (Right (unsafeCoerce x)))

123− return Nothing

124−

125 handlePass :: MonadIO m

113 handlePass :: MonadIO m

126− => Either SomeException a -> m (Maybe SomeException)

114+ => Either SomeException [a] -> m (Maybe SomeException)

127 handlePass res =

115 handlePass res =

128 case res of

116 case res of

129 Left e -> do

117 Left e -> do

118+ dbg $ "handlePass: caught exception"

130 liftIO $ mapM_ killThread pending

119 liftIO $ mapM_ killThread pending

131 return (Just e)

120 return (Just e)

132− Right _ -> do

121+ Right [] -> return Nothing

122+ Right _ -> do

133 liftIO $ atomically $ writeTChan pchan

123 liftIO $ atomically $ writeTChan pchan

134 (PassOnResult (unsafeCoerce res))

124 (PassOnResult (unsafeCoerce res))

135 return Nothing

125 return Nothing

@@ -230,10 +220,10 @@ instance Read SomeException where

230

220

231forkFinally1 :: (MonadIO m, MonadBaseControl IO m) =>

221forkFinally1 :: (MonadIO m, MonadBaseControl IO m) =>

232 m a -> (Either SomeException a -> IO ()) -> m ThreadId

222 m a -> (Either SomeException a -> IO ()) -> m ThreadId

233−forkFinally1 action and_then =

223+forkFinally1 action preExit =

234 EL.mask $ \restore ->

224 EL.mask $ \restore ->

235 liftBaseWith $ \runInIO -> forkIO $ do

225 liftBaseWith $ \runInIO -> forkIO $ do

236− _ <- runInIO $ EL.try (restore action) >>= liftIO . and_then

226+ _ <- runInIO $ EL.try (restore action) >>= liftIO . preExit

237 return ()

227 return ()

238

228

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

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

@@ -258,33 +248,38 @@ forkContextWith runCtx context = do

258 childContext ctx = do

248 childContext ctx = do

259 pendingRef <- liftIO $ newIORef []

249 pendingRef <- liftIO $ newIORef []

260 chan <- liftIO $ atomically newTChan

250 chan <- liftIO $ atomically newTChan

251+ resultsRef <- liftIO $ newIORef []

261 -- shares the threadCredit of the parent by default

252 -- shares the threadCredit of the parent by default

262 return $ ctx

253 return $ ctx

263 { parentChannel = Just (childChannel ctx)

254 { parentChannel = Just (childChannel ctx)

264 , pendingThreads = pendingRef

255 , pendingThreads = pendingRef

265 , childChannel = chan

256 , childChannel = chan

257+ , accumResults = resultsRef

266 }

258 }

267

259

268 beforeExit ctx res = do

260 beforeExit ctx res = do

261+ tid <- myThreadId

269 exc <- case res of

262 exc <- case res of

270 Left e -> do

263 Left e -> do

264+ dbg $ "beforeExit: " ++ show tid ++ " caught exception"

271 liftIO $ killChildren ctx

265 liftIO $ killChildren ctx

272 return (Just e)

266 return (Just e)

273 Right _ -> return Nothing

267 Right _ -> return Nothing

274

268

275 e <- waitForChildren ctx exc

269 e <- waitForChildren ctx exc

276

270

277− tid <- myThreadId

278 -- We are guaranteed to have a parent because we have been explicitly

271 -- We are guaranteed to have a parent because we have been explicitly

279 -- forked by some parent.

272 -- forked by some parent.

280 let p = fromJust (parentChannel ctx)

273 let p = fromJust (parentChannel ctx)

274+ results <- liftIO $ readIORef (accumResults ctx)

281 signalQSemB (threadCredit ctx)

275 signalQSemB (threadCredit ctx)

282 liftIO $ atomically $ writeTChan p

276 liftIO $ atomically $ writeTChan p

283− (ChildDone tid (maybe (unsafeCoerce res) Left e))

277+ (ChildDone tid (maybe (unsafeCoerce (Right results)) Left e))

284

278

285-- | A wrapper to first decide whether to run the context in the same thread or

279-- | A wrapper to first decide whether to run the context in the same thread or

286-- a new thread and then run the context in that thread.

280-- a new thread and then run the context in that thread.

287--

281--

282+-- XXX create a canFork function

288resumeContextWith :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)

283resumeContextWith :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)

289 => (Context -> m (Maybe a)) -> Context -> m ()

284 => (Context -> m (Maybe a)) -> Context -> m ()

290resumeContextWith runCtx context = do

285resumeContextWith runCtx context = do

@@ -339,9 +334,11 @@ loopContextWith ioaction ctx = do

339 let ctx' = updateContextEvent ctx streamData

334 let ctx' = updateContextEvent ctx streamData

340 case streamData of

335 case streamData of

341 SMore _ -> do

336 SMore _ -> do

342− resumeContextWith resume ctx'

337+ resumeContextWith (\x -> resume x >> return Nothing) ctx'

343 loopContextWith ioaction ctx

338 loopContextWith ioaction ctx

339+ _ -> do

344− _ -> resume ctx' -- run synchronously

340+ _ <- resume ctx' -- run synchronously

341+ return Nothing -- we are done with the loop

345

342

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

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

347-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new

344-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new

@@ -360,7 +357,7 @@ parallel ioaction = AsyncT $ do

360 -- generating the event data from the ioaction.

357 -- generating the event data from the ioaction.

361 context <- get

358 context <- get

362 case event context of

359 case event context of

363− -- We have already executed the ioaction in the parent thread and put the

360+ -- We have already executed the ioaction in the parent thread and put its

364 -- result in the context and now we are continuing the context in a new

361 -- result in the context and now we are continuing the context in a new

365 -- thread. Just return the result of the ioaction stored in the 'event'

362 -- thread. Just return the result of the ioaction stored in the 'event'

366 -- field.

363 -- field.

test/3-async.hs+15−11

@@ -5,16 +5,20 @@ import System.IO

6import Duct

8+main = do

9+ hSetBuffering stdout LineBuffering

8−main = waitAsync $ do

10+ xs <- waitAsync $ threads 100 $ do

9− liftIO $ hSetBuffering stdout LineBuffering

11+ liftIO $ hSetBuffering stdout LineBuffering

10− mainThread <- liftIO myThreadId

12+ mainThread <- liftIO myThreadId

11− liftIO $ putStrLn $ "Main thread: " ++ show mainThread

13+ liftIO $ putStrLn $ "Main thread: " ++ show mainThread

12− x <- async (randomIO :: IO Int)

14+ x <- async (randomIO :: IO Int)

13− liftIO $ putStrLn $ show x

15+ liftIO $ putStrLn $ show x

15− y <- async (randomIO :: IO Int)

17+ y <- async (randomIO :: IO Int)

17− liftIO $ threadDelay 1000000

19+ liftIO $ threadDelay 1000000

18− evThread <- liftIO myThreadId

20+ evThread <- liftIO myThreadId

19− liftIO $ putStrLn $ "Event thread: " ++ show evThread

21+ liftIO $ putStrLn $ "Event thread: " ++ show evThread

20− liftIO $ putStrLn $ show x

22+ liftIO $ putStrLn $ show y

23+ return (x, y)

24+ putStrLn $ show xs