@@ -37,6 +37,22 @@ import Unsafe.Coerce (unsafeCoerce)
3737import Duct.AsyncT
3838import Duct.Context
3939
40+------------------------------------------------------------------------------
41+-- Model of computation
42+------------------------------------------------------------------------------
43+
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.
50+--
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+
4056------------------------------------------------------------------------------
4157-- Pick up from where we left in the previous thread
4258------------------------------------------------------------------------------
@@ -117,10 +133,10 @@ forkFinally1 action and_then =
117133 return ()
118134
119135forkIt :: (MonadBaseControl IO m, MonadIO m) => Context -> (Context -> m t) -> m ()
120−forkIt current proc = do
121− child <- newChildCont current
122− tid <- forkFinally1 (proc child) (beforeExit child)
123− updatePendingThreads current tid
136+forkIt context runCtx = do
137+ child <- childContext context
138+ tid <- forkFinally1 (runCtx child) (beforeExit child)
139+ updatePendingThreads context tid
124140
125141 where
126142
@@ -131,12 +147,12 @@ forkIt current proc = do
131147 liftIO $ modifyIORef (pendingThreads cur) $ (\ts -> tid:ts)
132148 tryReclaimZombies (childChannel cur) (pendingThreads cur)
133149
134− newChildCont cur = do
150+ childContext ctx = do
135151 pendingRef <- liftIO $ newIORef []
136152 chan <- liftIO $ atomically newTChan
137153 -- shares the threadCredit of the parent by default
138− return $ cur
139− { parentChannel = Just (childChannel cur)
154+ return $ ctx
155+ { parentChannel = Just (childChannel ctx)
140156 , pendingThreads = pendingRef
141157 , childChannel = chan
142158 }
@@ -157,22 +173,22 @@ forkIt current proc = do
157173 liftIO $ atomically $ writeTChan p tid
158174
159175forkMaybe :: (MonadBaseControl IO m, MonadIO m) => Context -> (Context -> m ()) -> m ()
160−forkMaybe current toRun = do
161− gotCredit <- liftIO $ waitQSemB (threadCredit current)
162− pending <- liftIO $ readIORef $ pendingThreads current
176+forkMaybe context runCtx = do
177+ gotCredit <- liftIO $ waitQSemB (threadCredit context)
178+ pending <- liftIO $ readIORef $ pendingThreads context
163179 case gotCredit of
164180 False -> case pending of
165− [] -> toRun current -- run synchronously
181+ [] -> runCtx context -- run synchronously
166182 _ -> do
167183 -- XXX If we have unreclaimable child threads e.g.
168184 -- infinite loop, this is going to deadlock us. We need
169185 -- special handling for those cases. Add those to
170186 -- unreclaimable list? And always execute them in an
171187 -- async thread, cannot use sync for those.
172− waitForOneChild (childChannel current)
173− (pendingThreads current)
174− forkMaybe current toRun
175− True -> forkIt current toRun
188+ waitForOneChild (childChannel context)
189+ (pendingThreads context)
190+ forkMaybe context runCtx
191+ True -> forkIt context runCtx
176192
177193-- | 'StreamData' represents a task in a task stream being generated.
178194data StreamData a =
@@ -182,36 +198,43 @@ data StreamData a =
182198 | SError SomeException -- ^ An error occurred
183199 deriving (Typeable, Show,Read)
184200
201+-- The current model is to start a new thread for every task. The input is
202+-- provided at the time of the creation and therefore no synchronization is
203+-- needed compared to a pool of threads contending to get the input from a
204+-- channel. However the thread creation overhead may be more than that?
205+--
206+-- When the task is over the outputs need to be collected and that requires
207+-- synchronization irrespective of a thread pool model or per task new thread
208+-- model.
209+--
185210-- XXX instead of starting a new thread every time, reuse the existing child
186211-- threads and send them work via a shared channel. When there is no more work
187212-- available we need a way to close the channel and wakeup all waiters so that
188−-- they can go away.
213+-- they can go away rather than waiting indefinitely.
189214--
190215-- | Execute the IO action and the continuation
191216genAsyncEvents :: (MonadIO m, MonadBaseControl IO m) => Context -> IO (StreamData t) -> m ()
192−genAsyncEvents parentc rec = forkMaybe parentc loop
217+genAsyncEvents context rec = forkMaybe context loop
193218
194219 where
195220
196221 -- Execute the IO computation and then the closure-continuation
197− loop curcont = do
222+ loop ctx = do
198223 streamData <- liftIO $ rec `catch`
199224 \(e :: SomeException) -> return $ SError e
200225
201226 let run = runContWith streamData
202227 case streamData of
203228 SMore _ -> do
204− forkMaybe curcont run
205− loop curcont
206− _ -> run curcont
207−
208− where
229+ forkMaybe ctx run
230+ loop ctx
231+ _ -> run ctx
209232
210− -- pass on the io result and then run the continuation
211− runContWith dat cont = do
212− let cont' = cont { event = Just $ unsafeCoerce dat }
213− _ <- runStateT (resume cont') cont'
214− return ()
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
237+ return ()
215238
216239-- | Run an IO action one or more times to generate a stream of tasks. The IO
217240-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new
@@ -225,18 +248,20 @@ genAsyncEvents parentc rec = forkMaybe parentc loop
225248parallel :: (Monad m, MonadIO m, MonadBaseControl IO m)
226249 => IO (StreamData a) -> AsyncT m (StreamData a)
227250parallel ioaction = AsyncT $ do
228− -- receive the events via a channel instead
229− cont <- get
230− case event cont of
231− -- we have already exceuted the ioaction and now we are continuing in a new
251+ -- We retrieve the context here and pass it on so that we can resume it
252+ -- later. Control resumes at this place when we resume this context after
253+ -- generating the event data from the ioaction.
254+ context <- get
255+ case event context of
256+ -- we have already executed the ioaction and now we are continuing in a new
232257 -- thread. Just return the result of the ioaction stored in 'event' field.
233258 j@(Just _) -> do
234− put cont { event = Nothing }
259+ put context { event = Nothing }
235260 return $ unsafeCoerce j
236261 -- we have to execute the io action and generate an event and then continue
237262 -- in this thread or a new thread.
238263 Nothing -> do
239− lift $ genAsyncEvents cont ioaction
264+ lift $ genAsyncEvents context ioaction
240265 loc <- getLocation
241266 when (loc /= RemoteNode) $ setLocation WaitingParent
242267 return Nothing