git.iko.soy/streamly
on 7b52ccf help sign in
streamly/src/Asyncly/RunAsync.hs
172 lines 6.2 kB history raw
1{-# LANGUAGE FlexibleContexts #-}
2{-# LANGUAGE FlexibleInstances #-}
3{-# LANGUAGE StandaloneDeriving #-}
4{-# LANGUAGE GeneralizedNewtypeDeriving #-}
6-- |
7-- Module : Asyncly.Threads
8-- Copyright : (c) 2017 Harendra Kumar
9--
10-- License : MIT-style
11-- Maintainer : harendra.kumar@gmail.com
12-- Stability : experimental
13-- Portability : GHC
14--
15module Asyncly.RunAsync
16 ( AsynclyT
17 , runAsyncly
18 , toList
19 , each
20 , threads
21 , runAsynclyRecorded
22-- , toListRecorded
23 , playRecordings
24 )
25where
27import Control.Applicative (Alternative (..))
28import Control.Concurrent.STM (atomically, newTChan)
29import Control.Monad (liftM)
30import Control.Monad.Catch (MonadThrow)
31import Control.Monad.IO.Class (MonadIO (..))
32import Control.Monad.Trans.Class (MonadTrans (lift))
33import Control.Monad.State (StateT(..), runStateT)
34import Data.IORef (IORef, newIORef, readIORef)
36import Control.Monad.Trans.Recorder (MonadRecorder(..), RecorderT,
37 Recording, blank, runRecorderT)
38import Asyncly.Threads
39import Asyncly.AsyncT
41-- This transformer runs AsyncT under a state to manage the threads.
42-- Separating the state from the pure ListT transformer is cleaner but it
43-- results in 2x performance degradation. At some point if that performance is
44-- really needed we can combine the two.
46newtype AsynclyT m a = AsynclyT { runAsynclyT :: AsyncT (StateT Context m) a }
48deriving instance Monad m => Functor (AsynclyT m)
49deriving instance Monad m => Applicative (AsynclyT m)
50deriving instance Monad m => Alternative (AsynclyT m)
51deriving instance Monad m => Monad (AsynclyT m)
52deriving instance MonadIO m => MonadIO (AsynclyT m)
53deriving instance MonadThrow m => MonadThrow (AsynclyT m)
54instance MonadTrans (AsynclyT) where
55 lift mx = AsynclyT $ AsyncT $ \_ k -> lift mx >>= (\a -> (k a Nothing))
57-- XXX orphan instance, use a newtype instead?
58instance (Monad m, MonadRecorder m)
59 => MonadRecorder (StateT Context m) where
60 getJournal = lift getJournal
61 putJournal = lift . putJournal
62 play = lift . play
64deriving instance (Monad m, MonadRecorder m)
65 => MonadRecorder (AsynclyT m)
67------------------------------------------------------------------------------
68-- Running the monad
69------------------------------------------------------------------------------
71getContext :: Maybe (IORef [Recording]) -> IO Context
72getContext lref = do
73 childChan <- atomically newTChan
74 pendingRef <- newIORef []
75 credit <- newIORef maxBound
76 return $ initContext childChan pendingRef credit lref
78-- | Run an 'AsynclyT m' computation, wait for it to finish and discard the
79-- results.
80{-# INLINABLE runAsynclyLogged #-}
81runAsynclyLogged :: MonadAsync m
82 => Maybe (IORef [Recording]) -> AsynclyT m a -> m ()
83runAsynclyLogged lref (AsynclyT m) = do
84 ctx <- liftIO $ getContext lref
85 _ <- runStateT (run m) ctx
86 return ()
88 where
90 stop = return ()
91 run mx = (runAsyncT mx) stop (\_ r -> maybe stop run r)
93runAsyncly :: MonadAsync m => AsynclyT m a -> m ()
94runAsyncly m = runAsynclyLogged Nothing m
96data Step a r = Stop | Done a | Yield a r
98-- | Run an 'AsynclyT m' computation and collect the results generated by each
99-- thread of the computation in a list.
100{-# INLINABLE toList #-}
101toList :: MonadAsync m => AsynclyT m a -> m [a]
102toList (AsynclyT m) = liftIO (getContext Nothing) >>= run m
104 where
106 stop = return Stop
107 done a = return (Done a)
108 yield a x = return $ Yield a x
110 run ma ctx = do
111 (res, ctx') <- runStateT
112 ((runAsyncT ma) stop (\a r -> maybe (done a) (\x -> yield a x) r))
113 ctx
114 case res of
115 Yield x mb -> liftM (x :) (run mb ctx')
116 Stop -> return []
117 Done x -> return (x : [])
119{-# INLINABLE each #-}
120each :: MonadAsync m => [a] -> AsynclyT m a
121each xs = foldr (<|>) empty $ map return xs
123------------------------------------------------------------------------------
124-- Controlling thread quota
125------------------------------------------------------------------------------
127-- | Runs a computation under a given thread limit. A limit of 0 means all new
128-- tasks start synchronously in the current thread unless overridden by
129-- 'async'.
130threads :: MonadAsync m => Int -> AsynclyT m a -> AsynclyT m a
131threads n action = AsynclyT $ AsyncT $ \stp yld ->
132 threadCtl n ((runAsyncT $ runAsynclyT action) stp yld)
134------------------------------------------------------------------------------
135-- Logging
136------------------------------------------------------------------------------
138-- | Compose a computation using previously captured logs
139playRecording :: (MonadAsync m, MonadRecorder m)
140 => AsynclyT m a -> Recording -> AsynclyT m a
141playRecording m recording = play recording >> m
143-- | Resume an 'AsyncT' computation using previously recorded logs. The
144-- recording consists of a list of journals one for each thread in the
145-- computation.
146playRecordings :: (MonadAsync m, MonadRecorder m)
147 => AsynclyT m a -> [Recording] -> AsynclyT m a
148playRecordings m logs = each logs >>= playRecording m
150{-
151-- | Run an 'AsyncT' computation with recording enabled, wait for it to finish
152-- returning results for completed threads and recordings for paused threads.
153toListRecorded :: (MonadAsync m, MonadCatch m)
154 => AsynclyT m a -> m ([a], [Recording])
155toListRecorded m = do
156 resultsRef <- liftIO $ newIORef []
157 lref <- liftIO $ newIORef []
158 waitAsync (gatherResult resultsRef) (Just lref) m
159 res <- liftIO $ readIORef resultsRef
160 logs <- liftIO $ readIORef lref
161 return (res, logs)
162 -}
164-- | Run an 'AsyncT' computation with recording enabled, wait for it to finish
165-- and discard the results and return the recordings for paused threads, if
166-- any.
167runAsynclyRecorded :: MonadAsync m => AsynclyT (RecorderT m) a -> m [Recording]
168runAsynclyRecorded m = do
169 lref <- liftIO $ newIORef []
170 runRecorderT blank (runAsynclyLogged (Just lref) m)
171 logs <- liftIO $ readIORef lref
172 return logs