38b148f · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Do not use IORef for results

38b148fHarendra Kumar9 years ago

src/Duct/AsyncT.hs+11−5

compare with master

unifiedsplit

@@ -185,6 +185,13 @@ waitForChildren chan pendingRef results = do

185-- Running the monad

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

187

188+-- | Invoked to store the result of the computation in the context and finish

189+-- the computation when the computation is done

190+finishComputation :: Monad m => a -> AsyncT m b

191+finishComputation x = AsyncT $ do

192+ contextSaveResult x

193+ return Nothing

194+

188-- XXX pass a collector function and return a Traversable.

195-- XXX pass a collector function and return a Traversable.

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

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

190-- | 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

@@ -195,20 +202,19 @@ waitAsync m = do

195 childChan <- liftIO $ atomically newTChan

202 childChan <- liftIO $ atomically newTChan

196 pendingRef <- liftIO $ newIORef []

203 pendingRef <- liftIO $ newIORef []

197 credit <- liftIO $ newIORef maxBound

204 credit <- liftIO $ newIORef maxBound

198− resultsRef <- liftIO $ newIORef []

199

205

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

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

201 -- existential and remove the unsafeCoerces

207 -- existential and remove the unsafeCoerces

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

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

203− (empty :: AsyncT m a) childChan pendingRef credit resultsRef

209+ (empty :: AsyncT m a) childChan pendingRef credit

210+ finishComputation

204

211

205− results <- liftIO $ readIORef resultsRef

206 xs <- case r of

212 xs <- case r of

207 Left (exc :: SomeException) -> do

213 Left (exc :: SomeException) -> do

208 liftIO $ readIORef pendingRef >>= mapM_ killThread

214 liftIO $ readIORef pendingRef >>= mapM_ killThread

209 throwM exc

215 throwM exc

210− Right (Nothing, _) -> return results

216+ Right (Nothing, ctx) -> return $ contextGetResult ctx

211− Right ((Just x), _) -> return $ x:results

217+ Right ((Just x), ctx) -> return $ x : contextGetResult ctx

212 waitForChildren childChan pendingRef xs

218 waitForChildren childChan pendingRef xs

213

219

214------------------------------------------------------------------------------

220------------------------------------------------------------------------------

src/Duct/Context.hs+19−11

@@ -9,6 +9,8 @@ module Duct.Context

9 , saveContext

10 , restoreContext

11 , resumeContext

12+ , contextSaveResult

13+ , contextGetResult

12 , Location(..)

14 , Location(..)

13 , getLocation

15 , getLocation

14 , setLocation

16 , setLocation

@@ -50,6 +52,7 @@ data Context = Context

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

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

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

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

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

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

55+ -- XXX keep a composed computation instead

53 -- ^ List of continuations

56 -- ^ List of continuations

55 -- log only when there is a restore or if we are teleporting

58 -- log only when there is a restore or if we are teleporting

@@ -101,7 +104,7 @@ data Context = Context

101 --

104 --

102 , threadCredit :: IORef Int

105 , threadCredit :: IORef Int

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

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

104− , accumResults :: IORef [Any]

107+ , accumResults :: [Any]

105 -- ^ Accumulated results when running synchronously

108 -- ^ Accumulated results when running synchronously

106 } deriving Typeable

109 } deriving Typeable

107

110

@@ -110,19 +113,19 @@ initContext

110 -> TChan (ChildEvent a)

113 -> TChan (ChildEvent a)

111 -> IORef [ThreadId]

114 -> IORef [ThreadId]

112 -> IORef Int

115 -> IORef Int

113− -> IORef [a]

116+ -> (b -> m a)

114 -> Context

117 -> Context

115−initContext x childChan pending credit results =

118+initContext x childChan pending credit finalizer =

116− Context { event = mempty

119+ Context { event = mempty -- XXX Nothing

117 , currentm = unsafeCoerce x

120 , currentm = unsafeCoerce x

118− , fstack = []

121+ , fstack = [unsafeCoerce finalizer]

119 , location = Worker

122 , location = Worker

120 , mfData = mempty

123 , mfData = mempty

121 , parentChannel = Nothing

124 , parentChannel = Nothing

122 , childChannel = unsafeCoerce childChan

125 , childChannel = unsafeCoerce childChan

123 , pendingThreads = pending

126 , pendingThreads = pending

124 , threadCredit = credit

127 , threadCredit = credit

125− , accumResults = unsafeCoerce results }

128+ , accumResults = [] }

126

129

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

130------------------------------------------------------------------------------

128-- Where is the computation running?

131-- Where is the computation running?

@@ -165,6 +168,7 @@ restoreContext x = do

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

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

166

169

167 let mres = case x of

170 let mres = case x of

171+ -- XXX use mzero instead

168 Nothing -> empty

172 Nothing -> empty

169 Just y -> (unsafeCoerce f) y

173 Just y -> (unsafeCoerce f) y

170

174

@@ -180,7 +184,6 @@ restoreContext x = do

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

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

181resumeContext Context { currentm = m

185resumeContext Context { currentm = m

182 , fstack = fs

186 , fstack = fs

183− , accumResults = resultsRef

184 } =

187 } =

185 unsafeCoerce m >>= composefStack (unsafeCoerce fs)

188 unsafeCoerce m >>= composefStack (unsafeCoerce fs)

186

189

@@ -194,12 +197,17 @@ resumeContext Context { currentm = m

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

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

195

198

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

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

197− composefStack [] x = do

200+ composefStack [] _ = error "Bug: this should never be reached"

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

202

203+contextSaveResult :: Monad m => a -> StateT Context m ()

204+contextSaveResult r =

205+ modify $ \ Context {accumResults = rs, ..} ->

206+ Context {accumResults = unsafeCoerce r : rs, ..}

207+

208+contextGetResult :: Context -> [a]

209+contextGetResult ctx = unsafeCoerce $ accumResults ctx

210+

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

211------------------------------------------------------------------------------

204-- * Extensible State: Session Data Management

212-- * Extensible State: Session Data Management

205------------------------------------------------------------------------------

213------------------------------------------------------------------------------

src/Duct/Threads.hs+20−8

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

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

70-- to a new thread.

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

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

73resume ctx = do

74+ -- XXX rename to buildContext or buildState?

74 let s = runAsyncT (resumeContext ctx)

75 let s = runAsyncT (resumeContext ctx)

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

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

76 (_, c) <- runStateT s ctx

77 (_, c) <- runStateT s ctx

77− return c

78+

79+ -- XXX can we pass the result directly to the root thread instead of

80+ -- passing through all the parents? We can let the parent go away and

81+ -- handle the ChildDone events as well in the root thread.

82+ case parentChannel c of

83+ Nothing -> return () -- TODO: yield the value for streaming here

84+ Just chan -> do

85+ let r = accumResults c

86+ when (length r /= 0) $

87+ -- there is only one result in case of a non-root thread

88+ -- XXX change the return type to 'a' instead of '[a]'

89+ liftIO $ atomically $ writeTChan chan (PassOnResult (Right r))

79updateContextEvent :: Context -> a -> Context

91updateContextEvent :: Context -> a -> Context

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

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

@@ -248,13 +260,12 @@ forkContextWith runCtx context = do

248 childContext ctx = do

260 childContext ctx = do

249 pendingRef <- liftIO $ newIORef []

261 pendingRef <- liftIO $ newIORef []

250 chan <- liftIO $ atomically newTChan

262 chan <- liftIO $ atomically newTChan

251− resultsRef <- liftIO $ newIORef []

252 -- shares the threadCredit of the parent by default

263 -- shares the threadCredit of the parent by default

253 return $ ctx

264 return $ ctx

254 { parentChannel = Just (childChannel ctx)

265 { parentChannel = Just (childChannel ctx)

255 , pendingThreads = pendingRef

266 , pendingThreads = pendingRef

256 , childChannel = chan

267 , childChannel = chan

257− , accumResults = resultsRef

268+ , accumResults = []

258 }

269 }

259

270

260 beforeExit ctx res = do

271 beforeExit ctx res = do

@@ -264,17 +275,18 @@ forkContextWith runCtx context = do

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

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

265 liftIO $ killChildren ctx

276 liftIO $ killChildren ctx

266 return (Just e)

277 return (Just e)

267− Right _ -> return Nothing

278+ Right Nothing -> return Nothing

279+ Right (Just _) -> error "Bug: should never happen"

268

280

269 e <- waitForChildren ctx exc

281 e <- waitForChildren ctx exc

270

282

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

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

272 -- forked by some parent.

284 -- forked by some parent.

273 let p = fromJust (parentChannel ctx)

285 let p = fromJust (parentChannel ctx)

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

275 signalQSemB (threadCredit ctx)

286 signalQSemB (threadCredit ctx)

287+ -- XXX change the return value type to Maybe SomeException

276 liftIO $ atomically $ writeTChan p

288 liftIO $ atomically $ writeTChan p

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

289+ (ChildDone tid (maybe (Right []) Left e))

278

290

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

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

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

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

@@ -337,7 +349,7 @@ loopContextWith ioaction ctx = do

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

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

338 loopContextWith ioaction ctx

350 loopContextWith ioaction ctx

339 _ -> do

351 _ -> do

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

352+ resume ctx' -- run synchronously

341 return Nothing -- we are done with the loop

353 return Nothing -- we are done with the loop

342

354

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

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