c49bcc7 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Pass on exceptions and results from child to parent

c49bcc7Harendra Kumar9 years ago

duct.cabal+1

compare with master

unifiedsplit

@@ -36,6 +36,7 @@ library

36 , ghc-prim

38 -- libraries not bundled with GHC

39+ , exceptions

39 , lifted-base

40 , lifted-base

40 , monad-control

41 , monad-control

41 , mtl

42 , mtl

src/Duct/AsyncT.hs+49−37

@@ -14,17 +14,17 @@ module Duct.AsyncT

14 , waitAsync

15 , (<**)

16 , onNothing

17− , waitForChildren

18 , dbg

17 , dbg

19 )

18 )

20where

19where

22import Control.Applicative (Alternative (..))

21import Control.Applicative (Alternative (..))

23−import Control.Concurrent (ThreadId)

22+import Control.Concurrent (ThreadId, killThread)

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

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

25 readTChan)

24 readTChan)

26−import Control.Exception (SomeException, catch)

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

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

26+import Control.Monad.Catch (MonadCatch, MonadThrow, try,

27+ throwM, SomeException)

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

29 MonadState (..), StateT (..),

30 liftM, modify, runStateT, when)

@@ -157,50 +157,62 @@ instance (MonadBaseControl b m, MonadIO m) => MonadBaseControl b (AsyncT m) wher

157-- Thread management

158------------------------------------------------------------------------------

159

160−drainChildren :: MonadIO m => TChan ThreadId -> [ThreadId] -> m ()

160+drainChildren :: (MonadIO m, MonadThrow m)

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

161−drainChildren chan pending =

162+drainChildren chan pending res =

162− if length pending == 0

163+ if pending == []

163− then return ()

164+ then return res

164 else do

165 else do

165− tid <- liftIO $ atomically $ readTChan chan

166+ ev <- liftIO $ atomically $ readTChan chan

167+ case ev of

168+ ChildDone tid er -> do

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

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

170+ case er of

171+ Left e -> throwM e

172+ Right r -> do

173+ let newres = case r of

174+ Nothing -> res

175+ Just x -> x:res

167− drainChildren chan $ delete tid pending

176+ drainChildren chan (delete tid pending) newres

168−

177+ PassOnResult er ->

178+ case er of

179+ Left e -> throwM e

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

181+

169−waitForChildren :: MonadIO m => TChan ThreadId -> IORef [ThreadId] -> m ()

182+waitForChildren :: (MonadIO m, MonadThrow m)

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

170waitForChildren chan pendingRef = do

184waitForChildren chan pendingRef = do

171 pending <- liftIO $ readIORef pendingRef

185 pending <- liftIO $ readIORef pendingRef

172− drainChildren chan pending

186+ r <- drainChildren chan pending []

173 liftIO $ writeIORef pendingRef []

187 liftIO $ writeIORef pendingRef []

188+ return r

174

189

175------------------------------------------------------------------------------

190------------------------------------------------------------------------------

176-- Running the monad

191-- Running the monad

177------------------------------------------------------------------------------

192------------------------------------------------------------------------------

178

193

179−-- | Run a transient computation with a default initial state

180−runContext :: forall m a. MonadIO m => AsyncT m a -> m (Maybe a, Context)

181−runContext t = do

182− childChan <- liftIO $ atomically newTChan

183− pendingRef <- liftIO $ newIORef []

184− credit <- liftIO $ newIORef maxBound

185−

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

187− -- and remove the unsafeCoerces

188− r <- runStateT (runAsyncT t) $ initContext

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

190−

191− waitForChildren childChan pendingRef

192− return r

193−

194−-- | Run an 'AsyncT m' computation. Returns the result of the computation or

195−-- may throw an exception. The computation finishes with a 'Nothing' result

196−-- when the input sources are exhausted.

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

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

198−-- Ideally it should be a non-empty list instead.

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

199−--waitAsync :: MonadIO m => AsyncT m a -> m [a]

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

197+-- computation or may throw an exception.

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

200−waitAsync :: MonadIO m => AsyncT m a -> m (Maybe a)

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

201−waitAsync m = do

200+waitAsync m = do

202− (r, _) <- runContext m

201+ childChan <- liftIO $ atomically newTChan

202+ pendingRef <- liftIO $ newIORef []

203+ credit <- liftIO $ newIORef maxBound

204+

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

206+ -- existential and remove the unsafeCoerces

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

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

209+

210+ case r of

211+ Left (exc :: SomeException) -> do

212+ liftIO $ readIORef pendingRef >>= mapM_ killThread

213+ throwM exc

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

203− return r

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

204

216

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

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

206-- * Extensible State: Session Data Management

218-- * Extensible State: Session Data Management

src/Duct/Context.hs+13−9

@@ -3,7 +3,8 @@

3{-# LANGUAGE RecordWildCards #-}

5module Duct.Context

6+ ( ChildEvent(..)

6− ( Context(..)

7+ , Context(..)

7 , initContext

8 , initContext

8 , saveContext

9 , saveContext

9 , restoreContext

10 , restoreContext

@@ -18,9 +19,10 @@ module Duct.Context

18 )

19 )

19where

20where

21−import Control.Applicative (Alternative(..))

22+import Control.Applicative (Alternative(..))

22import Control.Concurrent (ThreadId)

23import Control.Concurrent (ThreadId)

23import Control.Concurrent.STM (TChan)

24import Control.Concurrent.STM (TChan)

25+import Control.Exception (SomeException)

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

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

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

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

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

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

@@ -33,6 +35,10 @@ import GHC.Prim (Any)

33-- State of a continuation

35-- State of a continuation

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

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

38+data ChildEvent a =

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

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

41+

36-- | Describes the context of a computation.

42-- | Describes the context of a computation.

37data Context = Context

43data Context = Context

38 { event :: Maybe () -- untyped, XXX rename to mailbox - can we do

44 { event :: Maybe () -- untyped, XXX rename to mailbox - can we do

@@ -53,10 +59,8 @@ data Context = Context

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

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

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

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

56− -- XXX All of the following can be removed

62+ -- XXX we can pass this at the time of fork rather than keeping it here.

57− -- Even the remote status can be removed by the same logic

58−

59− , parentChannel :: Maybe (TChan ThreadId)

63+ , parentChannel :: Maybe (TChan (ChildEvent Any))

60 -- ^ Our parent thread's channel to communicate to when we die

64 -- ^ Our parent thread's channel to communicate to when we die

62 -- Thread creation and cleanup handling. When a new thread is created the

66 -- Thread creation and cleanup handling. When a new thread is created the

@@ -69,7 +73,7 @@ data Context = Context

69 -- parent thread waits on the channel until all its children are cleaned

73 -- parent thread waits on the channel until all its children are cleaned

70 -- up.

74 -- up.

72− , childChannel :: TChan ThreadId

76+ , childChannel :: TChan (ChildEvent Any)

73 -- ^ A channel for the immediate children to communicate to us when they die.

77 -- ^ A channel for the immediate children to communicate to us when they die.

74 -- Each thread has its own dedicated channel for its children

78 -- Each thread has its own dedicated channel for its children

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

100

104

101initContext

105initContext

102 :: m a

106 :: m a

103− -> TChan ThreadId

107+ -> TChan (ChildEvent a)

104 -> IORef [ThreadId]

108 -> IORef [ThreadId]

105 -> IORef Int

109 -> IORef Int

106 -> Context

110 -> Context

@@ -111,7 +115,7 @@ initContext x childChan pending credit =

111 , location = Worker

115 , location = Worker

112 , mfData = mempty

116 , mfData = mempty

113 , parentChannel = Nothing

117 , parentChannel = Nothing

114− , childChannel = childChan

118+ , childChannel = unsafeCoerce childChan

115 , pendingThreads = pending

119 , pendingThreads = pending

116 , threadCredit = credit }

120 , threadCredit = credit }

117

121

src/Duct/Threads.hs+132−37

@@ -21,6 +21,7 @@ import Control.Concurrent.STM (TChan, atomically, newTChan,

21import Control.Exception (ErrorCall (..),

22 SomeException (..), catch)

23import qualified Control.Exception.Lifted as EL

24+import Control.Monad.Catch (MonadThrow, throwM)

24import Control.Monad.IO.Class (MonadIO (..))

25import Control.Monad.IO.Class (MonadIO (..))

25import Control.Monad.State (get, gets, modify, put,

26import Control.Monad.State (get, gets, modify, put,

26 runStateT, when)

27 runStateT, when)

@@ -71,7 +72,7 @@ import Duct.Context

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

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

72resume ctx = do

73resume ctx = do

73 let s = runAsyncT (resumeContext ctx)

74 let s = runAsyncT (resumeContext ctx)

74− (a, s) <- runStateT s ctx

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

75 return a

76 return a

77updateContextEvent :: Context -> a -> Context

78updateContextEvent :: Context -> a -> Context

@@ -81,25 +82,114 @@ updateContextEvent ctx evdata = ctx { event = Just $ unsafeCoerce evdata }

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

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

82------------------------------------------------------------------------------

83------------------------------------------------------------------------------

84−tryReclaimZombies :: MonadIO m => TChan ThreadId -> IORef [ThreadId] -> m ()

85+-- XXX We are using unbounded channels so this will not block on writing to

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

87+-- on consumption rate.

88+processOneEvent :: MonadIO m

89+ => ChildEvent a

90+ -> TChan (ChildEvent a)

91+ -> [ThreadId]

92+ -> Maybe SomeException

93+ -> m ([ThreadId], Maybe SomeException)

94+processOneEvent ev pchan pending exc = do

95+ e <- case exc of

96+ Nothing ->

97+ case ev of

98+ ChildDone tid res -> do

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

100+ handleDone res

101+ PassOnResult res -> handlePass res

102+ Just _ -> return exc

103+ let p = case ev of

104+ ChildDone tid _ -> delete tid pending

105+ _ -> pending

106+ return (p, e)

107+

108+ where

109+

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

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

127+ handlePass res =

128+ case res of

129+ Left e -> do

130+ liftIO $ mapM_ killThread pending

131+ return (Just e)

132+ Right _ -> do

133+ liftIO $ atomically $ writeTChan pchan

134+ (PassOnResult (unsafeCoerce res))

135+ return Nothing

136+

137+tryReclaimZombies :: (MonadIO m, MonadThrow m) => Context -> m ()

85−tryReclaimZombies chan pendingRef = do

138+tryReclaimZombies ctx = do

139+ let pchan = fromJust (parentChannel ctx)

140+ cchan = childChannel ctx

141+ pendingRef = pendingThreads ctx

142+

86 pending <- liftIO $ readIORef pendingRef

143 pending <- liftIO $ readIORef pendingRef

87 case pending of

144 case pending of

88 [] -> return ()

145 [] -> return ()

89 _ -> do

146 _ -> do

90− mtid <- liftIO $ atomically $ tryReadTChan chan

147+ mev <- liftIO $ atomically $ tryReadTChan cchan

91− case mtid of

148+ case mev of

92 Nothing -> return ()

149 Nothing -> return ()

93− Just tid -> do

150+ Just ev -> do

151+ (p, e) <- processOneEvent ev pchan pending Nothing

94− liftIO $ writeIORef pendingRef $ delete tid pending

152+ liftIO $ writeIORef pendingRef p

153+ maybe (return ()) throwM e

95− tryReclaimZombies chan pendingRef

154+ tryReclaimZombies ctx

96−

155+

97−waitForOneChild :: MonadIO m => TChan ThreadId -> IORef [ThreadId] -> m ()

156+waitForOneEvent :: (MonadIO m, MonadThrow m) => Context -> m ()

98−waitForOneChild chan pendingRef = do

157+waitForOneEvent ctx = do

99 -- XXX assert pending must have at least one element

158 -- XXX assert pending must have at least one element

100 -- assert that the tid is found in our list

159 -- assert that the tid is found in our list

160+ let pchan = fromJust (parentChannel ctx)

161+ cchan = childChannel ctx

162+ pendingRef = pendingThreads ctx

163+

164+ ev <- liftIO $ atomically $ readTChan cchan

165+ pending <- liftIO $ readIORef pendingRef

166+ (p, e) <- processOneEvent ev pchan pending Nothing

167+ liftIO $ writeIORef pendingRef p

168+ maybe (return ()) throwM e

169+

170+drainChildren :: MonadIO m

171+ => TChan (ChildEvent a)

172+ -> TChan (ChildEvent a)

173+ -> [ThreadId]

174+ -> Maybe SomeException

175+ -> m (Maybe SomeException)

176+drainChildren pchan cchan pending exc =

177+ if pending == []

178+ then return exc

179+ else do

101− tid <- liftIO $ atomically $ readTChan chan

180+ ev <- liftIO $ atomically $ readTChan cchan

181+ (p, e) <- processOneEvent ev pchan pending exc

182+ drainChildren pchan cchan p e

183+

184+waitForChildren :: MonadIO m

185+ => Context -> Maybe SomeException -> m (Maybe SomeException)

186+waitForChildren ctx exc = do

187+ let pendingRef = pendingThreads ctx

188+ pchan = fromJust (parentChannel ctx)

102− liftIO $ modifyIORef pendingRef $ delete tid

189+ pending <- liftIO $ readIORef pendingRef

190+ e <- drainChildren pchan (childChannel ctx) pending exc

191+ liftIO $ writeIORef pendingRef []

192+ return e

103

193

104-- | kill all the child threads associated with the continuation context

194-- | kill all the child threads associated with the continuation context

105killChildren :: Context -> IO ()

195killChildren :: Context -> IO ()

@@ -148,7 +238,7 @@ forkFinally1 action and_then =

148

238

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

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

150--

240--

151−forkContextWith :: (MonadBaseControl IO m, MonadIO m)

241+forkContextWith :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)

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

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

153forkContextWith runCtx context = do

243forkContextWith runCtx context = do

154 child <- childContext context

244 child <- childContext context

@@ -157,12 +247,13 @@ forkContextWith runCtx context = do

157

247

158 where

248 where

159

249

160− updatePendingThreads :: MonadIO m => Context -> ThreadId -> m ()

250+ updatePendingThreads :: (MonadIO m, MonadThrow m)

251+ => Context -> ThreadId -> m ()

161− updatePendingThreads cur tid = do

252+ updatePendingThreads ctx tid = do

162 -- update the new thread before reclaiming zombies so that if it exited

253 -- update the new thread before reclaiming zombies so that if it exited

163 -- already reclaim finds it in the list and does not panic.

254 -- already reclaim finds it in the list and does not panic.

164− liftIO $ modifyIORef (pendingThreads cur) $ (\ts -> tid:ts)

255+ liftIO $ modifyIORef (pendingThreads ctx) $ (\ts -> tid:ts)

165− tryReclaimZombies (childChannel cur) (pendingThreads cur)

256+ tryReclaimZombies ctx

166

257

167 childContext ctx = do

258 childContext ctx = do

168 pendingRef <- liftIO $ newIORef []

259 pendingRef <- liftIO $ newIORef []

@@ -174,25 +265,27 @@ forkContextWith runCtx context = do

174 , childChannel = chan

265 , childChannel = chan

175 }

266 }

176

267

177− beforeExit cur res = do

268+ beforeExit ctx res = do

178− case res of

269+ exc <- case res of

270+ Left e -> do

179− Left _exc -> liftIO $ killChildren cur

271+ liftIO $ killChildren ctx

180− -- propagate the exception to the parent

181− Right _ -> -- collect the results?

182− return ()

272+ return (Just e)

273+ Right _ -> return Nothing

183

274

184− waitForChildren (childChannel cur) (pendingThreads cur)

275+ e <- waitForChildren ctx exc

185− signalQSemB (threadCredit cur)

276+

277+ tid <- myThreadId

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

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

187 -- forked by some parent.

279 -- forked by some parent.

188− let p = fromJust (parentChannel cur)

280+ let p = fromJust (parentChannel ctx)

189− tid <- myThreadId

281+ signalQSemB (threadCredit ctx)

190− liftIO $ atomically $ writeTChan p tid

282+ liftIO $ atomically $ writeTChan p

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

191

284

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

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

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

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

194--

287--

195−resumeContextWith :: (MonadBaseControl IO m, MonadIO m)

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

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

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

197resumeContextWith runCtx context = do

290resumeContextWith runCtx context = do

198 gotCredit <- liftIO $ waitQSemB (threadCredit context)

291 gotCredit <- liftIO $ waitQSemB (threadCredit context)

@@ -208,8 +301,7 @@ resumeContextWith runCtx context = do

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

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

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

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

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

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

211− waitForOneChild (childChannel context)

304+ waitForOneEvent context

212− (pendingThreads context)

213 resumeContextWith runCtx context

305 resumeContextWith runCtx context

214 True -> forkContextWith runCtx context

306 True -> forkContextWith runCtx context

215

307

@@ -238,7 +330,7 @@ data StreamData a =

238--

330--

239-- | Execute the IO action, resume the saved context with the output of the io

331-- | Execute the IO action, resume the saved context with the output of the io

240-- action.

332-- action.

241−loopContextWith :: (MonadIO m, MonadBaseControl IO m)

333+loopContextWith :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)

242 => IO (StreamData t) -> Context -> m (Maybe a)

334 => IO (StreamData t) -> Context -> m (Maybe a)

243loopContextWith ioaction ctx = do

335loopContextWith ioaction ctx = do

244 streamData <- liftIO $ ioaction `catch`

336 streamData <- liftIO $ ioaction `catch`

@@ -260,7 +352,7 @@ loopContextWith ioaction ctx = do

260-- Unless the maximum number of threads (set with 'threads') has been reached,

352-- Unless the maximum number of threads (set with 'threads') has been reached,

261-- the task is generated in a new thread and the current thread returns a void

353-- the task is generated in a new thread and the current thread returns a void

262-- task.

354-- task.

263−parallel :: (Monad m, MonadIO m, MonadBaseControl IO m)

355+parallel :: (Monad m, MonadIO m, MonadBaseControl IO m, MonadThrow m)

264 => IO (StreamData a) -> AsyncT m (StreamData a)

356 => IO (StreamData a) -> AsyncT m (StreamData a)

265parallel ioaction = AsyncT $ do

357parallel ioaction = AsyncT $ do

266 -- We retrieve the context here and pass it on so that we can resume it

358 -- We retrieve the context here and pass it on so that we can resume it

@@ -291,7 +383,8 @@ parallel ioaction = AsyncT $ do

291-- | An task stream generator that produces an infinite stream of tasks by

383-- | An task stream generator that produces an infinite stream of tasks by

292-- running an IO computation in a loop. A task is triggered carrying the output

384-- running an IO computation in a loop. A task is triggered carrying the output

293-- of the computation. See 'parallel' for notes on the return value.

385-- of the computation. See 'parallel' for notes on the return value.

294−waitEvents :: (MonadIO m, MonadBaseControl IO m) => IO a -> AsyncT m a

386+waitEvents :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)

387+ => IO a -> AsyncT m a

295waitEvents io = do

388waitEvents io = do

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

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

297 case mr of

390 case mr of

@@ -301,7 +394,8 @@ waitEvents io = do

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

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

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

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

303-- the return value.

396-- the return value.

304−async :: (MonadIO m, MonadBaseControl IO m) => IO a -> AsyncT m a

397+async :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)

398+ => IO a -> AsyncT m a

305async io = do

399async io = do

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

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

307 case mr of

401 case mr of

@@ -324,7 +418,8 @@ sync x = AsyncT $ do

324-- task carries the result of the computation. A new task is generated only if

418-- task carries the result of the computation. A new task is generated only if

325-- the output of the computation is different from the previous one. See

419-- the output of the computation is different from the previous one. See

326-- 'parallel' for notes on the return value.

420-- 'parallel' for notes on the return value.

327−sample :: (Eq a, MonadIO m, MonadBaseControl IO m) => IO a -> Int -> AsyncT m a

421+sample :: (Eq a, MonadIO m, MonadBaseControl IO m, MonadThrow m)

422+ => IO a -> Int -> AsyncT m a

328sample action interval = do

423sample action interval = do

329 v <- liftIO action

424 v <- liftIO action

330 prev <- liftIO $ newIORef v

425 prev <- liftIO $ newIORef v