@@ -7,7 +7,7 @@ module Duct.Threads
@@ -88,9 +88,6 @@ resume ctx = do
88 -- XXX change the return type to 'a' instead of '[a]'
88 -- XXX change the return type to 'a' instead of '[a]'
89 liftIO $ atomically $ writeTChan chan (PassOnResult (Right r))
89 liftIO $ atomically $ writeTChan chan (PassOnResult (Right r))
91−updateContextEvent :: Context -> a -> Context
92−updateContextEvent ctx evdata = ctx { event = Just $ unsafeCoerce evdata }
94------------------------------------------------------------------------------
91------------------------------------------------------------------------------
95-- Thread Management (creation, reaping and killing)
92-- Thread Management (creation, reaping and killing)
96------------------------------------------------------------------------------
93------------------------------------------------------------------------------
@@ -343,7 +340,7 @@ loopContextWith ioaction ctx = do
343 streamData <- liftIO $ ioaction `catch`
340 streamData <- liftIO $ ioaction `catch`
344 \(e :: SomeException) -> return $ SError e
341 \(e :: SomeException) -> return $ SError e
346− let ctx' = updateContextEvent ctx streamData
343+ let ctx' = setContextMailBox ctx streamData
347 case streamData of
344 case streamData of
348 SMore _ -> do
345 SMore _ -> do
349 resumeContextWith (\x -> resume x >> return Nothing) ctx'
346 resumeContextWith (\x -> resume x >> return Nothing) ctx'
@@ -364,30 +361,30 @@ loopContextWith ioaction ctx = do
364parallel :: (Monad m, MonadIO m, MonadBaseControl IO m, MonadThrow m)
361parallel :: (Monad m, MonadIO m, MonadBaseControl IO m, MonadThrow m)
365 => IO (StreamData a) -> AsyncT m (StreamData a)
362 => IO (StreamData a) -> AsyncT m (StreamData a)
366parallel ioaction = AsyncT $ do
363parallel ioaction = AsyncT $ do
367− -- We retrieve the context here and pass it on so that we can resume it
364+ -- We retrieve the context here and pass it on so that we can resume it
368− -- later. Control resumes at this place when we resume this context after
365+ -- later. Control resumes at this place when we resume this context in a
369− -- generating the event data from the ioaction.
366+ -- new thread or in the same thread after generating the event data from
370− context <- get
367+ -- the ioaction.
368+ mb <- takeContextMailBox
371− case event context of
369+ case mb of
372− -- We have already executed the ioaction in the parent thread and put its
370+ -- We have already executed the ioaction in the parent thread and put
373− -- result in the context and now we are continuing the context in a new
371+ -- its result in the context and now we are continuing the context in a
374− -- thread. Just return the result of the ioaction stored in the 'event'
372+ -- new thread. Just return the result of the ioaction stored in the
375− -- field.
373+ -- 'event' field.
376− j@(Just _) -> do
374+ Right x -> return (Just x)
377− put context { event = Nothing }
375+
378− return $ unsafeCoerce j
380− -- We have to execute the io action and generate an event and then continue
376+ -- We have to execute the io action, generate event data, put it in the
377+ -- context mailbox and then continue from the point where this context
381− -- in this thread or a new thread.
378+ -- was retreieved, in this thread or a new thread.
382− Nothing -> do
379+ Left ctx -> do
383− lift $ resumeContextWith (loopContextWith ioaction) context
380+ lift $ resumeContextWith (loopContextWith ioaction) ctx
385− -- We will never reach here if we continued the context in the same
382+ -- We will never reach here if we continued the context in the same
386− -- thread. If we started a new thread then the parent thread reaches
383+ -- thread. If we started a new thread then the parent thread
387− -- here.
384+ -- reaches here.
388− loc <- getLocation
385+ loc <- getLocation
389− when (loc /= RemoteNode) $ setLocation WaitingParent
386+ when (loc /= RemoteNode) $ setLocation WaitingParent
390− return Nothing
387+ return Nothing
392-- | An task stream generator that produces an infinite stream of tasks by
389-- | An task stream generator that produces an infinite stream of tasks by
393-- running an IO computation in a loop. A task is triggered carrying the output
390-- running an IO computation in a loop. A task is triggered carrying the output
@@ -452,6 +449,7 @@ sample action interval = do
452-- into the transient monad using 'react'. Every time the callback is called it
449-- into the transient monad using 'react'. Every time the callback is called it
453-- generates a new task for the transient monad.
450-- generates a new task for the transient monad.
456 :: (Monad m, MonadIO m)
454 :: (Monad m, MonadIO m)
457 => ((eventdata -> m response) -> m ())
455 => ((eventdata -> m response) -> m ())
@@ -472,6 +470,8 @@ react setHandler iob = AsyncT $ do
472 put context{event=Nothing}
470 put context{event=Nothing}
473 return $ unsafeCoerce j
471 return $ unsafeCoerce j
475------------------------------------------------------------------------------
475------------------------------------------------------------------------------
476-- Controlling thread quota
476-- Controlling thread quota
477------------------------------------------------------------------------------
477------------------------------------------------------------------------------