@@ -60,8 +60,14 @@ import Duct.Context
60-- | Continue execution of the closure that we were executing when we migrated
60-- | Continue execution of the closure that we were executing when we migrated
61-- to a new thread.
61-- to a new thread.
63−resume :: Monad m => Context -> StateT Context m (Maybe a)
63+resume :: Monad m => Context -> m ()
64−resume = runAsyncT . resumeContext
65+ let s = runAsyncT (resumeContext ctx)
69+updateContextEvent :: Context -> a -> Context
70+updateContextEvent ctx evdata = ctx { event = Just $ unsafeCoerce evdata }
66------------------------------------------------------------------------------
72------------------------------------------------------------------------------
67-- Thread Management (creation, reaping and killing)
73-- Thread Management (creation, reaping and killing)
@@ -132,8 +138,9 @@ forkFinally1 action and_then =
132 _ <- runInIO $ EL.try (restore action) >>= liftIO . and_then
138 _ <- runInIO $ EL.try (restore action) >>= liftIO . and_then
133 return ()
139 return ()
135−forkIt :: (MonadBaseControl IO m, MonadIO m) => Context -> (Context -> m t) -> m ()
141+forkContext :: (MonadBaseControl IO m, MonadIO m)
142+ => Context -> (Context -> m ()) -> m ()
136−forkIt context runCtx = do
143+forkContext context runCtx = do
137 child <- childContext context
144 child <- childContext context
138 tid <- forkFinally1 (runCtx child) (beforeExit child)
145 tid <- forkFinally1 (runCtx child) (beforeExit child)
139 updatePendingThreads context tid
146 updatePendingThreads context tid
@@ -188,7 +195,7 @@ forkMaybe context runCtx = do
188 waitForOneChild (childChannel context)
195 waitForOneChild (childChannel context)
189 (pendingThreads context)
196 (pendingThreads context)
190 forkMaybe context runCtx
197 forkMaybe context runCtx
191− True -> forkIt context runCtx
198+ True -> forkContext context runCtx
193-- | 'StreamData' represents a task in a task stream being generated.
200-- | 'StreamData' represents a task in a task stream being generated.
194data StreamData a =
201data StreamData a =
@@ -212,29 +219,22 @@ data StreamData a =
212-- available we need a way to close the channel and wakeup all waiters so that
219-- available we need a way to close the channel and wakeup all waiters so that
213-- they can go away rather than waiting indefinitely.
220-- they can go away rather than waiting indefinitely.
215−-- | Execute the IO action and the continuation
222+-- | Execute the IO action resume the saved context
216genAsyncEvents :: (MonadIO m, MonadBaseControl IO m) => Context -> IO (StreamData t) -> m ()
223genAsyncEvents :: (MonadIO m, MonadBaseControl IO m) => Context -> IO (StreamData t) -> m ()
217genAsyncEvents context rec = forkMaybe context loop
224genAsyncEvents context rec = forkMaybe context loop
221− -- Execute the IO computation and then the closure-continuation
222 loop ctx = do
228 loop ctx = do
223 streamData <- liftIO $ rec `catch`
229 streamData <- liftIO $ rec `catch`
224 \(e :: SomeException) -> return $ SError e
230 \(e :: SomeException) -> return $ SError e
226− let run = runContWith streamData
232+ let ctx' = updateContextEvent ctx streamData
227 case streamData of
233 case streamData of
228 SMore _ -> do
234 SMore _ -> do
229− forkMaybe ctx run
235+ forkMaybe ctx' resume
231− _ -> run ctx
237+ _ -> resume ctx'
233− -- resume the context with the result of the IO computation
234− runContWith dat ctx = do
235− let newCtx = ctx { event = Just $ unsafeCoerce dat }
236− _ <- runStateT (resume newCtx) newCtx
239-- | Run an IO action one or more times to generate a stream of tasks. The IO
239-- | Run an IO action one or more times to generate a stream of tasks. The IO
240-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new
240-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new