git.iko.soy/streamly
on 7beefce help sign in
streamly/src/Asyncly/RunAsync.hs
157 lines 5.5 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 , for
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.Monoid ((<>))
35import Data.IORef (IORef, newIORef, readIORef)
37import Control.Monad.Trans.Recorder (MonadRecorder(..), RecorderT,
38 Recording, blank, runRecorderT)
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.
46{-
47newtype AsynclyT m a = AsynclyT { runAsynclyT :: AsyncT (StateT Context m) a }
49deriving instance Monad m => Functor (AsynclyT m)
50deriving instance Monad m => Applicative (AsynclyT m)
51deriving instance Monad m => Alternative (AsynclyT m)
52deriving instance Monad m => Monad (AsynclyT m)
53deriving instance MonadIO m => MonadIO (AsynclyT m)
54deriving instance MonadThrow m => MonadThrow (AsynclyT m)
55instance MonadTrans (AsynclyT) where
56 lift mx = AsynclyT $ AsyncT $ \c _ k -> lift mx >>= (\a -> (k a c Nothing))
58-- XXX orphan instance, use a newtype instead?
59instance (Monad m, MonadRecorder m)
60 => MonadRecorder (StateT Context m) where
61 getJournal = lift getJournal
62 putJournal = lift . putJournal
63 play = lift . play
65deriving instance (Monad m, MonadRecorder m)
66 => MonadRecorder (AsynclyT m)
68-}
70type AsynclyT = AsyncT
72------------------------------------------------------------------------------
73-- Running the monad
74------------------------------------------------------------------------------
76-- | Run an 'AsynclyT m' computation, wait for it to finish and discard the
77-- results.
78{-# INLINABLE runAsynclyLogged #-}
79runAsynclyLogged :: MonadAsync m
80 => Maybe (IORef [Recording]) -> AsynclyT m a -> m ()
81runAsynclyLogged lref m = run Nothing m
83 where
85 stop = return ()
86 {-# INLINE yield #-}
87 yield _ c r = maybe stop (\x -> (run c x)) r
88 run ct mx = (runAsyncT mx) ct stop yield
90runAsyncly :: MonadAsync m => AsynclyT m a -> m ()
91runAsyncly m = runAsynclyLogged Nothing m
93-- | Run an 'AsynclyT m' computation and collect the results generated by each
94-- thread of the computation in a list.
95{-# INLINABLE toList #-}
96toList :: MonadAsync m => AsynclyT m a -> m [a]
97toList m = run Nothing m
99 where
101 stop = return []
102 done a = return [a]
103 {-# INLINE yield #-}
104 yield a c r = maybe (done a) (\x -> liftM (a :) (run c x)) r
106 {-# INLINE run #-}
107 run ctx mx = (runAsyncT mx) ctx stop yield
109-- | Run a given function concurrently on the list and collect the results.
110{-# INLINABLE for #-}
111for :: MonadAsync m => [a] -> (a -> AsyncT m b) -> AsynclyT m b
112for xs f = foldr (<|>) empty $ map f xs
114-- XXX rename to fromList?
115{-# INLINABLE each #-}
116each :: MonadAsync m => [a] -> AsynclyT m a
117each xs = foldr (<>) empty $ map return xs
119------------------------------------------------------------------------------
120-- Logging
121------------------------------------------------------------------------------
123-- | Compose a computation using previously captured logs
124playRecording :: (MonadAsync m, MonadRecorder m)
125 => AsynclyT m a -> Recording -> AsynclyT m a
126playRecording m recording = play recording >> m
128-- | Resume an 'AsyncT' computation using previously recorded logs. The
129-- recording consists of a list of journals one for each thread in the
130-- computation.
131playRecordings :: (MonadAsync m, MonadRecorder m)
132 => AsynclyT m a -> [Recording] -> AsynclyT m a
133playRecordings m logs = each logs >>= playRecording m
135{-
136-- | Run an 'AsyncT' computation with recording enabled, wait for it to finish
137-- returning results for completed threads and recordings for paused threads.
138toListRecorded :: (MonadAsync m, MonadCatch m)
139 => AsynclyT m a -> m ([a], [Recording])
140toListRecorded m = do
141 resultsRef <- liftIO $ newIORef []
142 lref <- liftIO $ newIORef []
143 waitAsync (gatherResult resultsRef) (Just lref) m
144 res <- liftIO $ readIORef resultsRef
145 logs <- liftIO $ readIORef lref
146 return (res, logs)
147 -}
149-- | Run an 'AsyncT' computation with recording enabled, wait for it to finish
150-- and discard the results and return the recordings for paused threads, if
151-- any.
152runAsynclyRecorded :: MonadAsync m => AsynclyT (RecorderT m) a -> m [Recording]
153runAsynclyRecorded m = do
154 lref <- liftIO $ newIORef []
155 runRecorderT blank (runAsynclyLogged (Just lref) m)
156 logs <- liftIO $ readIORef lref
157 return logs