38b148f · streamly · git.iko.soy

git.iko.soy/streamly

Do not use IORef for results

commit38b148f7c5bde35f40a4863a21ab25a6f4893950 authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 22 Jun 2017, 17:44 (9 years ago) parentcad6f19

3 files changed+50−24

compare with master

unifiedsplit

src/Duct/AsyncT.hs+11−5 src/Duct/Context.hs+19−11 src/Duct/Threads.hs+20−8

src/Duct/AsyncT.hs+11−5

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

185185-- Running the monad

186186------------------------------------------------------------------------------

187187

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+

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

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

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

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

195202 childChan <- liftIO $ atomically newTChan

196203 pendingRef <- liftIO $ newIORef []

197204 credit <- liftIO $ newIORef maxBound

198− resultsRef <- liftIO $ newIORef []

199205

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

201207 -- existential and remove the unsafeCoerces

202208 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

204211

205− results <- liftIO $ readIORef resultsRef

206212 xs <- case r of

207213 Left (exc :: SomeException) -> do

208214 liftIO $ readIORef pendingRef >>= mapM_ killThread

209215 throwM exc

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

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

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

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

212218 waitForChildren childChan pendingRef xs

213219

214220------------------------------------------------------------------------------

src/Duct/Context.hs+19−11

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

99 , saveContext

1010 , restoreContext

1111 , resumeContext

12+ , contextSaveResult

13+ , contextGetResult

1214 , Location(..)

1315 , getLocation

1416 , setLocation

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

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

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

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

55+ -- XXX keep a composed computation instead

5356 -- ^ List of continuations

5457

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

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

101104 --

102105 , threadCredit :: IORef Int

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

104− , accumResults :: IORef [Any]

107+ , accumResults :: [Any]

105108 -- ^ Accumulated results when running synchronously

106109 } deriving Typeable

107110

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

110113 -> TChan (ChildEvent a)

111114 -> IORef [ThreadId]

112115 -> IORef Int

113− -> IORef [a]

116+ -> (b -> m a)

114117 -> Context

115−initContext x childChan pending credit results =

116− Context { event = mempty

118+initContext x childChan pending credit finalizer =

119+ Context { event = mempty -- XXX Nothing

117120 , currentm = unsafeCoerce x

118− , fstack = []

121+ , fstack = [unsafeCoerce finalizer]

119122 , location = Worker

120123 , mfData = mempty

121124 , parentChannel = Nothing

122125 , childChannel = unsafeCoerce childChan

123126 , pendingThreads = pending

124127 , threadCredit = credit

125− , accumResults = unsafeCoerce results }

128+ , accumResults = [] }

126129

127130------------------------------------------------------------------------------

128131-- Where is the computation running?

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

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

166169

167170 let mres = case x of

171+ -- XXX use mzero instead

168172 Nothing -> empty

169173 Just y -> (unsafeCoerce f) y

170174

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

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

181185resumeContext Context { currentm = m

182186 , fstack = fs

183− , accumResults = resultsRef

184187 } =

185188 unsafeCoerce m >>= composefStack (unsafeCoerce fs)

186189

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

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

195198

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

197− composefStack [] x = do

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

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

200− mzero

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

201201 composefStack (f:ff) x = f x >>= composefStack ff

202202

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+

203211------------------------------------------------------------------------------

204212-- * Extensible State: Session Data Management

205213------------------------------------------------------------------------------

src/Duct/Threads.hs+20−8

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

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

7070-- to a new thread.

7171

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

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

7373resume ctx = do

74+ -- XXX rename to buildContext or buildState?

7475 let s = runAsyncT (resumeContext ctx)

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

7677 (_, 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))

7890

7991updateContextEvent :: Context -> a -> Context

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

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

248260 childContext ctx = do

249261 pendingRef <- liftIO $ newIORef []

250262 chan <- liftIO $ atomically newTChan

251− resultsRef <- liftIO $ newIORef []

252263 -- shares the threadCredit of the parent by default

253264 return $ ctx

254265 { parentChannel = Just (childChannel ctx)

255266 , pendingThreads = pendingRef

256267 , childChannel = chan

257− , accumResults = resultsRef

268+ , accumResults = []

258269 }

259270

260271 beforeExit ctx res = do

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

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

265276 liftIO $ killChildren ctx

266277 return (Just e)

267− Right _ -> return Nothing

278+ Right Nothing -> return Nothing

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

268280

269281 e <- waitForChildren ctx exc

270282

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

272284 -- forked by some parent.

273285 let p = fromJust (parentChannel ctx)

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

275286 signalQSemB (threadCredit ctx)

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

276288 liftIO $ atomically $ writeTChan p

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

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

278290

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

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

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

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

338350 loopContextWith ioaction ctx

339351 _ -> do

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

352+ resume ctx' -- run synchronously

341353 return Nothing -- we are done with the loop

342354

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