@@ -37,6 +37,22 @@ import Unsafe.Coerce (unsafeCoerce)
37import Duct.AsyncT
37import Duct.AsyncT
38import Duct.Context
38import Duct.Context
40+------------------------------------------------------------------------------
41+-- Model of computation
42+------------------------------------------------------------------------------
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.
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.
40------------------------------------------------------------------------------
56------------------------------------------------------------------------------
41-- Pick up from where we left in the previous thread
57-- Pick up from where we left in the previous thread
42------------------------------------------------------------------------------
58------------------------------------------------------------------------------
@@ -117,10 +133,10 @@ forkFinally1 action and_then =
117 return ()
133 return ()
119forkIt :: (MonadBaseControl IO m, MonadIO m) => Context -> (Context -> m t) -> m ()
135forkIt :: (MonadBaseControl IO m, MonadIO m) => Context -> (Context -> m t) -> m ()
120−forkIt current proc = do
136+forkIt context runCtx = do
121− child <- newChildCont current
137+ child <- childContext context
122− tid <- forkFinally1 (proc child) (beforeExit child)
138+ tid <- forkFinally1 (runCtx child) (beforeExit child)
123− updatePendingThreads current tid
139+ updatePendingThreads context tid
@@ -131,12 +147,12 @@ forkIt current proc = do
131 liftIO $ modifyIORef (pendingThreads cur) $ (\ts -> tid:ts)
147 liftIO $ modifyIORef (pendingThreads cur) $ (\ts -> tid:ts)
132 tryReclaimZombies (childChannel cur) (pendingThreads cur)
148 tryReclaimZombies (childChannel cur) (pendingThreads cur)
134− newChildCont cur = do
150+ childContext ctx = do
135 pendingRef <- liftIO $ newIORef []
151 pendingRef <- liftIO $ newIORef []
136 chan <- liftIO $ atomically newTChan
152 chan <- liftIO $ atomically newTChan
137 -- shares the threadCredit of the parent by default
153 -- shares the threadCredit of the parent by default
138− return $ cur
154+ return $ ctx
139− { parentChannel = Just (childChannel cur)
155+ { parentChannel = Just (childChannel ctx)
140 , pendingThreads = pendingRef
156 , pendingThreads = pendingRef
141 , childChannel = chan
157 , childChannel = chan
@@ -157,22 +173,22 @@ forkIt current proc = do
157 liftIO $ atomically $ writeTChan p tid
173 liftIO $ atomically $ writeTChan p tid
159forkMaybe :: (MonadBaseControl IO m, MonadIO m) => Context -> (Context -> m ()) -> m ()
175forkMaybe :: (MonadBaseControl IO m, MonadIO m) => Context -> (Context -> m ()) -> m ()
160−forkMaybe current toRun = do
176+forkMaybe context runCtx = do
161− gotCredit <- liftIO $ waitQSemB (threadCredit current)
177+ gotCredit <- liftIO $ waitQSemB (threadCredit context)
162− pending <- liftIO $ readIORef $ pendingThreads current
178+ pending <- liftIO $ readIORef $ pendingThreads context
163 case gotCredit of
179 case gotCredit of
164 False -> case pending of
180 False -> case pending of
165− [] -> toRun current -- run synchronously
181+ [] -> runCtx context -- run synchronously
167 -- XXX If we have unreclaimable child threads e.g.
183 -- XXX If we have unreclaimable child threads e.g.
168 -- infinite loop, this is going to deadlock us. We need
184 -- infinite loop, this is going to deadlock us. We need
169 -- special handling for those cases. Add those to
185 -- special handling for those cases. Add those to
170 -- unreclaimable list? And always execute them in an
186 -- unreclaimable list? And always execute them in an
171 -- async thread, cannot use sync for those.
187 -- async thread, cannot use sync for those.
172− waitForOneChild (childChannel current)
188+ waitForOneChild (childChannel context)
173− (pendingThreads current)
189+ (pendingThreads context)
174− forkMaybe current toRun
190+ forkMaybe context runCtx
175− True -> forkIt current toRun
191+ True -> forkIt context runCtx
177-- | 'StreamData' represents a task in a task stream being generated.
193-- | 'StreamData' represents a task in a task stream being generated.
178data StreamData a =
194data StreamData a =
@@ -182,36 +198,43 @@ data StreamData a =
182 | SError SomeException -- ^ An error occurred
198 | SError SomeException -- ^ An error occurred
183 deriving (Typeable, Show,Read)
199 deriving (Typeable, Show,Read)
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?
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
185-- XXX instead of starting a new thread every time, reuse the existing child
210-- XXX instead of starting a new thread every time, reuse the existing child
186-- threads and send them work via a shared channel. When there is no more work
211-- threads and send them work via a shared channel. When there is no more work
187-- available we need a way to close the channel and wakeup all waiters so that
212-- 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.
190-- | Execute the IO action and the continuation
215-- | Execute the IO action and the continuation
191genAsyncEvents :: (MonadIO m, MonadBaseControl IO m) => Context -> IO (StreamData t) -> m ()
216genAsyncEvents :: (MonadIO m, MonadBaseControl IO m) => Context -> IO (StreamData t) -> m ()
192−genAsyncEvents parentc rec = forkMaybe parentc loop
217+genAsyncEvents context rec = forkMaybe context loop
196 -- Execute the IO computation and then the closure-continuation
221 -- Execute the IO computation and then the closure-continuation
197− loop curcont = do
222+ loop ctx = do
198 streamData <- liftIO $ rec `catch`
223 streamData <- liftIO $ rec `catch`
199 \(e :: SomeException) -> return $ SError e
224 \(e :: SomeException) -> return $ SError e
201 let run = runContWith streamData
226 let run = runContWith streamData
202 case streamData of
227 case streamData of
203 SMore _ -> do
228 SMore _ -> do
204− forkMaybe curcont run
229+ forkMaybe ctx run
205− loop curcont
230+ loop ctx
206− _ -> run curcont
231+ _ -> run ctx
210− -- pass on the io result and then run the continuation
233+ -- resume the context with the result of the IO computation
211− runContWith dat cont = do
234+ runContWith dat ctx = do
212− let cont' = cont { event = Just $ unsafeCoerce dat }
235+ let newCtx = ctx { event = Just $ unsafeCoerce dat }
213− _ <- runStateT (resume cont') cont'
236+ _ <- runStateT (resume newCtx) newCtx
214− return ()
237+ return ()
216-- | 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
217-- 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
@@ -225,18 +248,20 @@ genAsyncEvents parentc rec = forkMaybe parentc loop
225parallel :: (Monad m, MonadIO m, MonadBaseControl IO m)
248parallel :: (Monad m, MonadIO m, MonadBaseControl IO m)
226 => IO (StreamData a) -> AsyncT m (StreamData a)
249 => IO (StreamData a) -> AsyncT m (StreamData a)
227parallel ioaction = AsyncT $ do
250parallel ioaction = AsyncT $ do
228− -- receive the events via a channel instead
251+ -- We retrieve the context here and pass it on so that we can resume it
229− cont <- get
252+ -- later. Control resumes at this place when we resume this context after
253+ -- generating the event data from the ioaction.
230− case event cont of
255+ case event context of
231− -- we have already exceuted the ioaction and now we are continuing in a new
256+ -- we have already executed the ioaction and now we are continuing in a new
232 -- thread. Just return the result of the ioaction stored in 'event' field.
257 -- thread. Just return the result of the ioaction stored in 'event' field.
233 j@(Just _) -> do
258 j@(Just _) -> do
234− put cont { event = Nothing }
259+ put context { event = Nothing }
235 return $ unsafeCoerce j
260 return $ unsafeCoerce j
236 -- we have to execute the io action and generate an event and then continue
261 -- we have to execute the io action and generate an event and then continue
237 -- in this thread or a new thread.
262 -- in this thread or a new thread.
238 Nothing -> do
263 Nothing -> do
239− lift $ genAsyncEvents cont ioaction
264+ lift $ genAsyncEvents context ioaction
240 loc <- getLocation
265 loc <- getLocation
241 when (loc /= RemoteNode) $ setLocation WaitingParent
266 when (loc /= RemoteNode) $ setLocation WaitingParent
242 return Nothing
267 return Nothing