@@ -41,18 +41,26 @@ import Duct.Context
41-- Model of computation
41-- Model of computation
42------------------------------------------------------------------------------
42------------------------------------------------------------------------------
44−-- A computation starts in a top level thread. A thread can either finish the
44+-- A computation starts in a top level thread. If no "forking primitives" are
45−-- computation or it can wait for events to occur before it proceeds further. A
45+-- used then the thread finishes in a straight line flow just like the IO
46−-- thread can spawn more threads to process an event. It is a way to compose
46+-- monad. However, if a "forking primitive" is used it "forks" the computation
47−-- asynchronous events in a concurrent manner. Since it is a transformer we can
47+-- into multiple flows. The "forked" computations may run concurrently (and
48+-- therefore in parallel when possible) or serially. When multiple forking
49+-- primitives are composed it results into a tree of computations where each
50+-- branch of the tree can run concurrently.
52+-- A forking primitive may create multiple forks at that point, each fork
53+-- provides a specific input value to be used in the forked computation, this
54+-- value defines the fork.
56+-- The final result of the computation is the collection of all the values
57+-- generated by all the leaf level forks. These values are then propagated up
58+-- the tree and collected at the root of the tree.
60+-- Since AsyncT is a transformer we can use things like pipe, conduit or any
48−-- use things like pipe, conduit or any other transformer monads inside the
61+-- other transformer monads inside the computations to utilize single threaded
49−-- computations to utilize single threaded composition or data flow techniques.
62+-- composition or data flow techniques.
51−-- Usually none of the computations in this monad need to return a result as
52−-- the computation can be handed over to the next one until the final effect is
53−-- produced. However, if needed to be embedded in a different programming model
54−-- the monad can return values in a stream like a ListT.
56------------------------------------------------------------------------------
64------------------------------------------------------------------------------
57-- Pick up from where we left in the previous thread
65-- Pick up from where we left in the previous thread
58------------------------------------------------------------------------------
66------------------------------------------------------------------------------
@@ -60,11 +68,11 @@ import Duct.Context
60-- | Continue execution of the closure that we were executing when we migrated
68-- | Continue execution of the closure that we were executing when we migrated
61-- to a new thread.
69-- to a new thread.
63−resume :: Monad m => Context -> m ()
71+resume :: Monad m => Context -> m (Maybe a)
64resume ctx = do
72resume ctx = do
65 let s = runAsyncT (resumeContext ctx)
73 let s = runAsyncT (resumeContext ctx)
66− _ <- runStateT s ctx
74+ (a, s) <- runStateT s ctx
67− return ()
75+ return a
69updateContextEvent :: Context -> a -> Context
77updateContextEvent :: Context -> a -> Context
70updateContextEvent ctx evdata = ctx { event = Just $ unsafeCoerce evdata }
78updateContextEvent ctx evdata = ctx { event = Just $ unsafeCoerce evdata }
@@ -138,8 +146,10 @@ forkFinally1 action and_then =
138 _ <- runInIO $ EL.try (restore action) >>= liftIO . and_then
146 _ <- runInIO $ EL.try (restore action) >>= liftIO . and_then
139 return ()
147 return ()
149+-- | Run a given context in a new thread.
141forkContextWith :: (MonadBaseControl IO m, MonadIO m)
151forkContextWith :: (MonadBaseControl IO m, MonadIO m)
142− => (Context -> m ()) -> Context -> m ()
152+ => (Context -> m (Maybe a)) -> Context -> m ()
143forkContextWith runCtx context = do
153forkContextWith runCtx context = do
144 child <- childContext context
154 child <- childContext context
145 tid <- forkFinally1 (runCtx child) (beforeExit child)
155 tid <- forkFinally1 (runCtx child) (beforeExit child)
@@ -182,14 +192,16 @@ forkContextWith runCtx context = do
182-- | A wrapper to first decide whether to run the context in the same thread or
192-- | A wrapper to first decide whether to run the context in the same thread or
183-- a new thread and then run the context in that thread.
193-- a new thread and then run the context in that thread.
185−runContext :: (MonadBaseControl IO m, MonadIO m)
195+resumeContextWith :: (MonadBaseControl IO m, MonadIO m)
186− => (Context -> m ()) -> Context -> m ()
196+ => (Context -> m (Maybe a)) -> Context -> m ()
187−runContext runCtx context = do
197+resumeContextWith runCtx context = do
188 gotCredit <- liftIO $ waitQSemB (threadCredit context)
198 gotCredit <- liftIO $ waitQSemB (threadCredit context)
189 pending <- liftIO $ readIORef $ pendingThreads context
199 pending <- liftIO $ readIORef $ pendingThreads context
190 case gotCredit of
200 case gotCredit of
191 False -> case pending of
201 False -> case pending of
192− [] -> runCtx context -- run synchronously
203+ _ <- runCtx context -- run synchronously
194 -- XXX If we have unreclaimable child threads e.g.
206 -- XXX If we have unreclaimable child threads e.g.
195 -- infinite loop, this is going to deadlock us. We need
207 -- infinite loop, this is going to deadlock us. We need
@@ -198,7 +210,7 @@ runContext runCtx context = do
198 -- async thread, cannot use sync for those.
210 -- async thread, cannot use sync for those.
199 waitForOneChild (childChannel context)
211 waitForOneChild (childChannel context)
200 (pendingThreads context)
212 (pendingThreads context)
201− runContext runCtx context
213+ resumeContextWith runCtx context
202 True -> forkContextWith runCtx context
214 True -> forkContextWith runCtx context
204-- | 'StreamData' represents a task in a task stream being generated.
216-- | 'StreamData' represents a task in a task stream being generated.
@@ -224,9 +236,10 @@ data StreamData a =
224-- available we need a way to close the channel and wakeup all waiters so that
236-- available we need a way to close the channel and wakeup all waiters so that
225-- they can go away rather than waiting indefinitely.
237-- they can go away rather than waiting indefinitely.
227−-- | Execute the IO action resume the saved context
239+-- | Execute the IO action, resume the saved context with the output of the io
228loopContextWith :: (MonadIO m, MonadBaseControl IO m)
241loopContextWith :: (MonadIO m, MonadBaseControl IO m)
229− => IO (StreamData t) -> Context -> m ()
242+ => IO (StreamData t) -> Context -> m (Maybe a)
230loopContextWith ioaction ctx = do
243loopContextWith ioaction ctx = do
231 streamData <- liftIO $ ioaction `catch`
244 streamData <- liftIO $ ioaction `catch`
232 \(e :: SomeException) -> return $ SError e
245 \(e :: SomeException) -> return $ SError e
@@ -234,9 +247,9 @@ loopContextWith ioaction ctx = do
234 let ctx' = updateContextEvent ctx streamData
247 let ctx' = updateContextEvent ctx streamData
235 case streamData of
248 case streamData of
236 SMore _ -> do
249 SMore _ -> do
237− runContext resume ctx'
250+ resumeContextWith resume ctx'
238 loopContextWith ioaction ctx
251 loopContextWith ioaction ctx
239− _ -> resume ctx'
252+ _ -> resume ctx' -- run synchronously
241-- | Run an IO action one or more times to generate a stream of tasks. The IO
254-- | Run an IO action one or more times to generate a stream of tasks. The IO
242-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new
255-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new
@@ -255,15 +268,22 @@ parallel ioaction = AsyncT $ do
255 -- generating the event data from the ioaction.
268 -- generating the event data from the ioaction.
256 context <- get
269 context <- get
257 case event context of
270 case event context of
258− -- we have already executed the ioaction and now we are continuing in a new
271+ -- We have already executed the ioaction in the parent thread and put the
272+ -- result in the context and now we are continuing the context in a new
259− -- thread. Just return the result of the ioaction stored in 'event' field.
273+ -- thread. Just return the result of the ioaction stored in the 'event'
260 j@(Just _) -> do
275 j@(Just _) -> do
261 put context { event = Nothing }
276 put context { event = Nothing }
262 return $ unsafeCoerce j
277 return $ unsafeCoerce j
263− -- we have to execute the io action and generate an event and then continue
279+ -- We have to execute the io action and generate an event and then continue
264 -- in this thread or a new thread.
280 -- in this thread or a new thread.
265 Nothing -> do
281 Nothing -> do
266− lift $ runContext (loopContextWith ioaction) context
282+ lift $ resumeContextWith (loopContextWith ioaction) context
284+ -- We will never reach here if we continued the context in the same
285+ -- thread. If we started a new thread then the parent thread reaches
267 loc <- getLocation
287 loc <- getLocation
268 when (loc /= RemoteNode) $ setLocation WaitingParent
288 when (loc /= RemoteNode) $ setLocation WaitingParent
269 return Nothing
289 return Nothing