@@ -41,18 +41,26 @@ import Duct.Context
4141-- Model of computation
4242------------------------------------------------------------------------------
4343
44−-- A computation starts in a top level thread. A thread can either finish the
45−-- computation or it can wait for events to occur before it proceeds further. A
46−-- thread can spawn more threads to process an event. It is a way to compose
47−-- asynchronous events in a concurrent manner. Since it is a transformer we can
48−-- use things like pipe, conduit or any other transformer monads inside the
49−-- computations to utilize single threaded composition or data flow techniques.
44+-- A computation starts in a top level thread. If no "forking primitives" are
45+-- used then the thread finishes in a straight line flow just like the IO
46+-- monad. However, if a "forking primitive" is used it "forks" the computation
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.
51+--
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.
55+--
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.
59+--
60+-- Since AsyncT is a transformer we can use things like pipe, conduit or any
61+-- other transformer monads inside the computations to utilize single threaded
62+-- composition or data flow techniques.
5063--
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.
55−
5664------------------------------------------------------------------------------
5765-- Pick up from where we left in the previous thread
5866------------------------------------------------------------------------------
@@ -60,11 +68,11 @@ import Duct.Context
6068-- | Continue execution of the closure that we were executing when we migrated
6169-- to a new thread.
6270
63−resume :: Monad m => Context -> m ()
71+resume :: Monad m => Context -> m (Maybe a)
6472resume ctx = do
6573 let s = runAsyncT (resumeContext ctx)
66− _ <- runStateT s ctx
67− return ()
74+ (a, s) <- runStateT s ctx
75+ return a
6876
6977updateContextEvent :: Context -> a -> Context
7078updateContextEvent ctx evdata = ctx { event = Just $ unsafeCoerce evdata }
@@ -138,8 +146,10 @@ forkFinally1 action and_then =
138146 _ <- runInIO $ EL.try (restore action) >>= liftIO . and_then
139147 return ()
140148
149+-- | Run a given context in a new thread.
150+--
141151forkContextWith :: (MonadBaseControl IO m, MonadIO m)
142− => (Context -> m ()) -> Context -> m ()
152+ => (Context -> m (Maybe a)) -> Context -> m ()
143153forkContextWith runCtx context = do
144154 child <- childContext context
145155 tid <- forkFinally1 (runCtx child) (beforeExit child)
@@ -182,14 +192,16 @@ forkContextWith runCtx context = do
182192-- | A wrapper to first decide whether to run the context in the same thread or
183193-- a new thread and then run the context in that thread.
184194--
185−runContext :: (MonadBaseControl IO m, MonadIO m)
186− => (Context -> m ()) -> Context -> m ()
187−runContext runCtx context = do
195+resumeContextWith :: (MonadBaseControl IO m, MonadIO m)
196+ => (Context -> m (Maybe a)) -> Context -> m ()
197+resumeContextWith runCtx context = do
188198 gotCredit <- liftIO $ waitQSemB (threadCredit context)
189199 pending <- liftIO $ readIORef $ pendingThreads context
190200 case gotCredit of
191201 False -> case pending of
192− [] -> runCtx context -- run synchronously
202+ [] -> do
203+ _ <- runCtx context -- run synchronously
204+ return ()
193205 _ -> do
194206 -- XXX If we have unreclaimable child threads e.g.
195207 -- infinite loop, this is going to deadlock us. We need
@@ -198,7 +210,7 @@ runContext runCtx context = do
198210 -- async thread, cannot use sync for those.
199211 waitForOneChild (childChannel context)
200212 (pendingThreads context)
201− runContext runCtx context
213+ resumeContextWith runCtx context
202214 True -> forkContextWith runCtx context
203215
204216-- | 'StreamData' represents a task in a task stream being generated.
@@ -224,9 +236,10 @@ data StreamData a =
224236-- available we need a way to close the channel and wakeup all waiters so that
225237-- they can go away rather than waiting indefinitely.
226238--
227−-- | Execute the IO action resume the saved context
239+-- | Execute the IO action, resume the saved context with the output of the io
240+-- action.
228241loopContextWith :: (MonadIO m, MonadBaseControl IO m)
229− => IO (StreamData t) -> Context -> m ()
242+ => IO (StreamData t) -> Context -> m (Maybe a)
230243loopContextWith ioaction ctx = do
231244 streamData <- liftIO $ ioaction `catch`
232245 \(e :: SomeException) -> return $ SError e
@@ -234,9 +247,9 @@ loopContextWith ioaction ctx = do
234247 let ctx' = updateContextEvent ctx streamData
235248 case streamData of
236249 SMore _ -> do
237− runContext resume ctx'
250+ resumeContextWith resume ctx'
238251 loopContextWith ioaction ctx
239− _ -> resume ctx'
252+ _ -> resume ctx' -- run synchronously
240253
241254-- | Run an IO action one or more times to generate a stream of tasks. The IO
242255-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new
@@ -255,15 +268,22 @@ parallel ioaction = AsyncT $ do
255268 -- generating the event data from the ioaction.
256269 context <- get
257270 case event context of
258− -- we have already executed the ioaction and now we are continuing in a new
259− -- thread. Just return the result of the ioaction stored in 'event' field.
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
273+ -- thread. Just return the result of the ioaction stored in the 'event'
274+ -- field.
260275 j@(Just _) -> do
261276 put context { event = Nothing }
262277 return $ unsafeCoerce j
263− -- we have to execute the io action and generate an event and then continue
278+
279+ -- We have to execute the io action and generate an event and then continue
264280 -- in this thread or a new thread.
265281 Nothing -> do
266− lift $ runContext (loopContextWith ioaction) context
282+ lift $ resumeContextWith (loopContextWith ioaction) context
283+
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
286+ -- here.
267287 loc <- getLocation
268288 when (loc /= RemoteNode) $ setLocation WaitingParent
269289 return Nothing