@@ -1,24 +1,21 @@
1+{-# LANGUAGE ConstraintKinds #-}
12{-# LANGUAGE FlexibleContexts #-}
23{-# LANGUAGE ScopedTypeVariables #-}
34
45module Strands.Threads
5− ( parallel
6− , waitEvents
6+ ( MonadAsync
7+ , wait
8+ , wait_
9+ , threads
10+
711 , async
812 , each
9− , sample
10− , sync
1113 , gather
12− --, react
13− , threads
14−
15− , wait
16− , wait_
1714
18− , logged
19− , suspend
2015 , waitLogged
2116 , waitLogged_
17+ , logged
18+ , suspend
2219 , withLog
2320 , eachWithLog
2421 )
@@ -26,12 +23,11 @@ where
2623
2724import Control.Applicative ((<|>), empty)
2825import Control.Concurrent (ThreadId, forkIO, killThread,
29− myThreadId, threadDelay)
26+ myThreadId)
3027import Control.Concurrent.STM (TChan, atomically, newTChan,
3128 readTChan, tryReadTChan,
3229 writeTChan)
33−import Control.Exception (ErrorCall (..),
34− SomeException (..), catch)
30+import Control.Exception (SomeException (..))
3531import qualified Control.Exception.Lifted as EL
3632import Control.Monad.Catch (MonadCatch, MonadThrow, throwM,
3733 try)
@@ -49,6 +45,8 @@ import Data.Maybe (isJust)
4945import Strands.AsyncT
5046import Strands.Context
5147
48+type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
49+
5250------------------------------------------------------------------------------
5351-- Model of computation
5452------------------------------------------------------------------------------
@@ -80,11 +78,11 @@ import Strands.Context
8078-- | Continue execution of the closure that we were executing when we migrated
8179-- to a new thread.
8280
83−runContext :: MonadIO m => Context -> StateT Context m ()
81+runContext :: Monad m => Context -> StateT Context m ()
8482runContext ctx = do
85− let s = runAsyncT (composeContext ctx)
86− _ <- lift $ runStateT s ctx
87− return ()
83+ let s = runAsyncT (composeContext ctx)
84+ _ <- lift $ runStateT s ctx
85+ return ()
8886
8987------------------------------------------------------------------------------
9088-- Thread Management (creation, reaping and killing)
@@ -180,7 +178,9 @@ signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ())
180178--
181179
182180forkFinally1 :: (MonadIO m, MonadBaseControl IO m)
183− => Context -> (Either SomeException () -> IO ()) -> StateT Context m ThreadId
181+ => Context
182+ -> (Either SomeException () -> IO ())
183+ -> StateT Context m ThreadId
184184forkFinally1 ctx preExit =
185185 EL.mask $ \restore ->
186186 liftBaseWith $ \runInIO -> forkIO $ do
@@ -252,30 +252,18 @@ canFork context = do
252252-- in the same thread or in a new thread depending on the synch parameter and
253253-- the current thread quota.
254254--
255−resumeContextWith :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)
255+resumeContextWith :: MonadAsync m
256256 => Context -- the context to resume
257257 -> Bool -- force synchronous
258− -> (Context -> AsyncT m (StreamData a)) -- the action to execute in the resumed context
258+ -> AsyncT m a -- the action to execute in the resumed context
259259 -> StateT Context m ()
260260resumeContextWith context synch action = do
261− let ctx = setContextMailBox context (action context)
261+ let ctx = setContextMailBox context action
262262 can <- liftIO $ canFork context
263263 case can && (not synch) of
264264 False -> runContext ctx -- run synchronously
265265 True -> forkContext ctx
266266
267−instance Read SomeException where
268− readsPrec _n str = [(SomeException $ ErrorCall s, r)]
269− where [(s , r)] = read str
270−
271−-- | 'StreamData' represents a task in a task stream being generated.
272−data StreamData a =
273− SMore a -- ^ More tasks to come
274− | SLast a -- ^ This is the last task
275− | SDone -- ^ No more tasks, we are done
276− | SError SomeException -- ^ An error occurred
277− deriving (Show, Read)
278−
279267-- The current model is to start a new thread for every task. The input is
280268-- provided at the time of the creation and therefore no synchronization is
281269-- needed compared to a pool of threads contending to get the input from a
@@ -296,173 +284,44 @@ data StreamData a =
296284-- parent task needs to terminate. Either the task is fully done or we handed
297285-- it over to another thread, in any case the current thread is done.
298286
299−spawningParentDone :: MonadIO m => StateT Context m (Maybe (StreamData a))
287+spawningParentDone :: MonadIO m => StateT Context m (Maybe a)
300288spawningParentDone = do
301289 loc <- getLocation
302290 when (loc /= RemoteNode) $ setLocation WaitingParent
303291 return Nothing
304292
305293-- | Captures the state of the current computation at this point, starts a new
306−-- thread, passing the captured state, runs the argument computation in the new
307−-- thread, returns its value and continues the computation from the capture
308−-- point. The end result is as if this function just returned the value
309−-- generated by the argument computation albeit in a new thread.
294+-- thread, using the argument passed as the return value of the computation and
295+-- continues the computation from the capture point. The end result is as if
296+-- this function just returned the value passed as argument albeit in a new
297+-- thread.
310298--
311299-- If a new thread cannot be created then the computation is run in the same
312300-- thread, but the functional behavior remains the same.
313301
314−spawnAsyncT :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
315− => (Context -> AsyncT m (StreamData a))
316− -> AsyncT m (StreamData a)
317−spawnAsyncT func = AsyncT $ do
302+async :: MonadAsync m => AsyncT m a -> AsyncT m a
303+async action = AsyncT $ do
318304 val <- takeContextMailBox
319305 case val of
320306 -- Child task
321− Right x -> runAsyncT x
307+ Right m -> runAsyncT m
322308
323309 -- Spawning parent
324310 Left ctx -> do
325− resumeContextWith ctx False func
311+ resumeContextWith ctx False action
326312
327313 -- If we started the task asynchronously in a new thread then the
328314 -- parent thread reaches here, immediately after spawning the task.
329315 --
330− -- However, if the task was executed synchronously then we will
331− -- reach after it is completely done.
316+ -- However, if the task was executed synchronously then we reach
317+ -- here after it is completely done.
332318 spawningParentDone
333319
334−-- | Execute the specified IO action, resume the saved context returning the
335−-- output of the io action, continue this in a loop until the ioaction
336−-- indicates that its done.
337−
338−loopContextWith :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
339− => IO (StreamData a) -> Context -> AsyncT m (StreamData a)
340−loopContextWith ioaction context = AsyncT $ do
341− -- Note that the context that we are resuming may have been passed from
342− -- another thread, therefore we must inherit thread control related
343− -- parameters from the current context.
344− curCtx <- get
345− loop context
346− { pendingThreads = pendingThreads curCtx
347− , childChannel = childChannel curCtx
348− }
349−
350− where
351−
352− loop ctx = do
353− streamData <- liftIO $ ioaction `catch`
354− \(e :: SomeException) -> return $ SError e
355−
356− let resumeCtx synch = resumeContextWith ctx synch $ \_ ->
357− return streamData
358−
359− case streamData of
360− SMore _ -> resumeCtx False >> loop ctx
361− _ -> resumeCtx True >> spawningParentDone
362−
363−-- | Run an IO action one or more times to generate a stream of tasks. The IO
364−-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new
365−-- task is triggered with the result value. If the return value is 'SMore', the
366−-- action is run again to generate the next task, otherwise task creation
367−-- stops.
368−--
369−-- Unless the maximum number of threads (set with 'threads') has been reached,
370−-- the task is generated in a new thread and the current thread returns a void
371−-- task.
372−parallel :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
373− => IO (StreamData a) -> AsyncT m (StreamData a)
374−parallel ioaction = spawnAsyncT (loopContextWith ioaction)
375−
376−-- | An task stream generator that produces an infinite stream of tasks by
377−-- running an IO computation in a loop. A task is triggered carrying the output
378−-- of the computation. See 'parallel' for notes on the return value.
379−waitEvents :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
380− => IO a -> AsyncT m a
381−waitEvents io = do
382− mr <- parallel (SMore <$> io)
383− case mr of
384− SMore x -> return x
385− -- SError e -> back e
386−
387−-- | Run an IO computation asynchronously and generate a single task carrying
388−-- the result of the computation when it completes. See 'parallel' for notes on
389−-- the return value.
390−async :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
391− => IO a -> AsyncT m a
392−async io = do
393− mr <- parallel (SLast <$> io)
394− case mr of
395− SLast x -> return x
396− -- SError e -> back e
397−
320+-- scatter
398321each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
399322 => [a] -> AsyncT m a
400323each xs = foldl (<|>) empty $ map (async . return) xs
401324
402−-- | Force an async computation to run synchronously. It can be useful in an
403−-- 'Alternative' composition to run the alternative only after finishing a
404−-- computation. Note that in Applicatives it might result in an undesired
405−-- serialization.
406−sync :: MonadIO m => AsyncT m a -> AsyncT m a
407−sync x = AsyncT $ do
408− setLocation RemoteNode
409− r <- runAsyncT x
410− setLocation Worker
411− return r
412−
413−-- | An task stream generator that produces an infinite stream of tasks by
414−-- running an IO computation periodically at the specified time interval. The
415−-- task carries the result of the computation. A new task is generated only if
416−-- the output of the computation is different from the previous one. See
417−-- 'parallel' for notes on the return value.
418−sample :: (Eq a, MonadIO m, MonadBaseControl IO m, MonadThrow m)
419− => IO a -> Int -> AsyncT m a
420−sample action interval = do
421− v <- liftIO action
422− prev <- liftIO $ newIORef v
423− waitEvents (loop action prev) <|> async (return v)
424− where loop act prev = loop'
425− where loop' = do
426− threadDelay interval
427− v <- act
428− v' <- readIORef prev
429− if v /= v' then writeIORef prev v >> return v else loop'
430−
431−-- | Make a transient task generator from an asynchronous callback handler.
432−--
433−-- The first parameter is a callback. The second parameter is a value to be
434−-- returned to the callback; if the callback expects no return value it
435−-- can just be a @return ()@. The callback expects a setter function taking the
436−-- @eventdata@ as an argument and returning a value to the callback; this
437−-- function is supplied by 'react'.
438−--
439−-- Callbacks from foreign code can be wrapped into such a handler and hooked
440−-- into the transient monad using 'react'. Every time the callback is called it
441−-- generates a new task for the transient monad.
442−--
443−{-
444−react
445− :: (Monad m, MonadIO m)
446− => ((eventdata -> m response) -> m ())
447− -> IO response
448− -> AsyncT m eventdata
449−react setHandler iob = AsyncT $ do
450− context <- get
451− case event context of
452− Nothing -> do
453− lift $ setHandler $ \dat ->do
454− resume (updateContextEvent context dat)
455− liftIO iob
456− loc <- getLocation
457− when (loc /= RemoteNode) $ setLocation WaitingParent
458− return Nothing
459−
460− j@(Just _) -> do
461− put context{event=Nothing}
462− return $ unsafeCoerce j
463−
464−-}
465−
466325------------------------------------------------------------------------------
467326-- Controlling thread quota
468327------------------------------------------------------------------------------
@@ -484,12 +343,6 @@ threads n process = AsyncT $ do
484343 return r
485344
486345{-
487−-- | Run a "non transient" computation within the underlying state monad, so it
488−-- is guaranteed that the computation neither can stop nor can trigger
489−-- additional events/threads.
490−noTrans :: Monad m => StateM m x -> AsyncT m x
491−noTrans x = AsyncT $ x >>= return . Just
492−
493346-- This can be used to set, increase or decrease the existing limit. The limit
494347-- is shared by multiple threads and therefore needs to modified atomically.
495348-- Note that when there is no limit the limit is set to maxBound it can