@@ -21,6 +21,7 @@ import Control.Concurrent.STM (TChan, atomically, newTChan,
2121import Control.Exception (ErrorCall (..),
2222 SomeException (..), catch)
2323import qualified Control.Exception.Lifted as EL
24+import Control.Monad.Catch (MonadThrow, throwM)
2425import Control.Monad.IO.Class (MonadIO (..))
2526import Control.Monad.State (get, gets, modify, put,
2627 runStateT, when)
@@ -71,7 +72,7 @@ import Duct.Context
7172resume :: Monad m => Context -> m (Maybe a)
7273resume ctx = do
7374 let s = runAsyncT (resumeContext ctx)
74− (a, s) <- runStateT s ctx
75+ (a, _) <- runStateT s ctx
7576 return a
7677
7778updateContextEvent :: Context -> a -> Context
@@ -81,25 +82,114 @@ updateContextEvent ctx evdata = ctx { event = Just $ unsafeCoerce evdata }
8182-- Thread Management (creation, reaping and killing)
8283------------------------------------------------------------------------------
8384
84−tryReclaimZombies :: MonadIO m => TChan ThreadId -> IORef [ThreadId] -> m ()
85−tryReclaimZombies chan pendingRef = do
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 ()
138+tryReclaimZombies ctx = do
139+ let pchan = fromJust (parentChannel ctx)
140+ cchan = childChannel ctx
141+ pendingRef = pendingThreads ctx
142+
86143 pending <- liftIO $ readIORef pendingRef
87144 case pending of
88145 [] -> return ()
89146 _ -> do
90− mtid <- liftIO $ atomically $ tryReadTChan chan
91− case mtid of
147+ mev <- liftIO $ atomically $ tryReadTChan cchan
148+ case mev of
92149 Nothing -> return ()
93− Just tid -> do
94− liftIO $ writeIORef pendingRef $ delete tid pending
95− tryReclaimZombies chan pendingRef
96−
97−waitForOneChild :: MonadIO m => TChan ThreadId -> IORef [ThreadId] -> m ()
98−waitForOneChild chan pendingRef = do
150+ Just ev -> do
151+ (p, e) <- processOneEvent ev pchan pending Nothing
152+ liftIO $ writeIORef pendingRef p
153+ maybe (return ()) throwM e
154+ tryReclaimZombies ctx
155+
156+waitForOneEvent :: (MonadIO m, MonadThrow m) => Context -> m ()
157+waitForOneEvent ctx = do
99158 -- XXX assert pending must have at least one element
100159 -- assert that the tid is found in our list
101− tid <- liftIO $ atomically $ readTChan chan
102− liftIO $ modifyIORef pendingRef $ delete tid
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
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)
189+ pending <- liftIO $ readIORef pendingRef
190+ e <- drainChildren pchan (childChannel ctx) pending exc
191+ liftIO $ writeIORef pendingRef []
192+ return e
103193
104194-- | kill all the child threads associated with the continuation context
105195killChildren :: Context -> IO ()
@@ -148,7 +238,7 @@ forkFinally1 action and_then =
148238
149239-- | Run a given context in a new thread.
150240--
151−forkContextWith :: (MonadBaseControl IO m, MonadIO m)
241+forkContextWith :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)
152242 => (Context -> m (Maybe a)) -> Context -> m ()
153243forkContextWith runCtx context = do
154244 child <- childContext context
@@ -157,12 +247,13 @@ forkContextWith runCtx context = do
157247
158248 where
159249
160− updatePendingThreads :: MonadIO m => Context -> ThreadId -> m ()
161− updatePendingThreads cur tid = do
250+ updatePendingThreads :: (MonadIO m, MonadThrow m)
251+ => Context -> ThreadId -> m ()
252+ updatePendingThreads ctx tid = do
162253 -- update the new thread before reclaiming zombies so that if it exited
163254 -- already reclaim finds it in the list and does not panic.
164− liftIO $ modifyIORef (pendingThreads cur) $ (\ts -> tid:ts)
165− tryReclaimZombies (childChannel cur) (pendingThreads cur)
255+ liftIO $ modifyIORef (pendingThreads ctx) $ (\ts -> tid:ts)
256+ tryReclaimZombies ctx
166257
167258 childContext ctx = do
168259 pendingRef <- liftIO $ newIORef []
@@ -174,25 +265,27 @@ forkContextWith runCtx context = do
174265 , childChannel = chan
175266 }
176267
177− beforeExit cur res = do
178− case res of
179− Left _exc -> liftIO $ killChildren cur
180− -- propagate the exception to the parent
181− Right _ -> -- collect the results?
182− return ()
268+ beforeExit ctx res = do
269+ exc <- case res of
270+ Left e -> do
271+ liftIO $ killChildren ctx
272+ return (Just e)
273+ Right _ -> return Nothing
183274
184− waitForChildren (childChannel cur) (pendingThreads cur)
185− signalQSemB (threadCredit cur)
275+ e <- waitForChildren ctx exc
276+
277+ tid <- myThreadId
186278 -- We are guaranteed to have a parent because we have been explicitly
187279 -- forked by some parent.
188− let p = fromJust (parentChannel cur)
189− tid <- myThreadId
190− liftIO $ atomically $ writeTChan p tid
280+ let p = fromJust (parentChannel ctx)
281+ signalQSemB (threadCredit ctx)
282+ liftIO $ atomically $ writeTChan p
283+ (ChildDone tid (maybe (unsafeCoerce res) Left e))
191284
192285-- | A wrapper to first decide whether to run the context in the same thread or
193286-- a new thread and then run the context in that thread.
194287--
195−resumeContextWith :: (MonadBaseControl IO m, MonadIO m)
288+resumeContextWith :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)
196289 => (Context -> m (Maybe a)) -> Context -> m ()
197290resumeContextWith runCtx context = do
198291 gotCredit <- liftIO $ waitQSemB (threadCredit context)
@@ -208,8 +301,7 @@ resumeContextWith runCtx context = do
208301 -- special handling for those cases. Add those to
209302 -- unreclaimable list? And always execute them in an
210303 -- async thread, cannot use sync for those.
211− waitForOneChild (childChannel context)
212− (pendingThreads context)
304+ waitForOneEvent context
213305 resumeContextWith runCtx context
214306 True -> forkContextWith runCtx context
215307
@@ -238,7 +330,7 @@ data StreamData a =
238330--
239331-- | Execute the IO action, resume the saved context with the output of the io
240332-- action.
241−loopContextWith :: (MonadIO m, MonadBaseControl IO m)
333+loopContextWith :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
242334 => IO (StreamData t) -> Context -> m (Maybe a)
243335loopContextWith ioaction ctx = do
244336 streamData <- liftIO $ ioaction `catch`
@@ -260,7 +352,7 @@ loopContextWith ioaction ctx = do
260352-- Unless the maximum number of threads (set with 'threads') has been reached,
261353-- the task is generated in a new thread and the current thread returns a void
262354-- task.
263−parallel :: (Monad m, MonadIO m, MonadBaseControl IO m)
355+parallel :: (Monad m, MonadIO m, MonadBaseControl IO m, MonadThrow m)
264356 => IO (StreamData a) -> AsyncT m (StreamData a)
265357parallel ioaction = AsyncT $ do
266358 -- We retrieve the context here and pass it on so that we can resume it
@@ -291,7 +383,8 @@ parallel ioaction = AsyncT $ do
291383-- | An task stream generator that produces an infinite stream of tasks by
292384-- running an IO computation in a loop. A task is triggered carrying the output
293385-- 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
295388waitEvents io = do
296389 mr <- parallel (SMore <$> io)
297390 case mr of
@@ -301,7 +394,8 @@ waitEvents io = do
301394-- | Run an IO computation asynchronously and generate a single task carrying
302395-- the result of the computation when it completes. See 'parallel' for notes on
303396-- 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
305399async io = do
306400 mr <- parallel (SLast <$> io)
307401 case mr of
@@ -324,7 +418,8 @@ sync x = AsyncT $ do
324418-- task carries the result of the computation. A new task is generated only if
325419-- the output of the computation is different from the previous one. See
326420-- '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
328423sample action interval = do
329424 v <- liftIO action
330425 prev <- liftIO $ newIORef v