git.iko.soy/streamly
on master help sign in
Split the logging functionality into a separate package
8340241Harendra Kumar
asyncly.cabal+2
compare with master
unifiedsplit
@@ -41,6 +41,7 @@ library
41 , exceptions
41 , exceptions
42 , lifted-base
42 , lifted-base
43 , monad-control
43 , monad-control
44+ , monad-recorder
44 , mtl
45 , mtl
45 , stm
46 , stm
46 , transformers-base
47 , transformers-base
@@ -55,6 +56,7 @@ test-suite test
55 , hspec >= 2.0 && < 3
56 , hspec >= 2.0 && < 3
56 , containers
57 , containers
57 , asyncly
58 , asyncly
59+ , monad-recorder
58 default-language: Haskell2010
60 default-language: Haskell2010
59
61
60source-repository head
62source-repository head
src/Asyncly.hs+3−8
@@ -39,14 +39,9 @@ module Asyncly
39 , afterFirst
39 , afterFirst
40 , (>>|)
40 , (>>|)
41
41
42− , Log
42+ , waitRecord
43− , Loggable
43+ , waitRecord_
44− , waitLogged
44+ , playRecordings
45− , waitLogged_
46− , logged
47− , suspend
48− , withLog
49− , eachWithLog
50 )
45 )
51where
46where
52
47
src/Asyncly/AsyncT.hs+53−53
@@ -14,10 +14,6 @@ module Asyncly.AsyncT
14 ( AsyncT (..)
14 ( AsyncT (..)
15 , MonadAsync
15 , MonadAsync
16 , ChildEvent(..)
16 , ChildEvent(..)
17− , Loggable
18− , Log (..)
19− , CtxLog (..)
20− , LogEntry (..)
21 , Context(..)
17 , Context(..)
22 , initContext
18 , initContext
23 , Location(..)
19 , Location(..)
@@ -25,6 +21,7 @@ module Asyncly.AsyncT
25 , setLocation
21 , setLocation
26 , threads
22 , threads
27 , waitForChildren
23 , waitForChildren
24+ , handleResult
28
25
29 , async
26 , async
30 , makeAsync
27 , makeAsync
@@ -48,7 +45,8 @@ import Control.Concurrent (ThreadId, forkIO, killThread,
48import Control.Concurrent.STM (TChan, atomically, newTChan,
45import Control.Concurrent.STM (TChan, atomically, newTChan,
49 readTChan, tryReadTChan,
46 readTChan, tryReadTChan,
50 writeTChan)
47 writeTChan)
51−import Control.Exception (SomeException (..))
48+import Control.Exception (fromException,
49+ SomeException (..))
52import qualified Control.Exception.Lifted as EL
50import qualified Control.Exception.Lifted as EL
53import Control.Monad (ap, mzero, when)
51import Control.Monad (ap, mzero, when)
54import Control.Monad.Base (MonadBase (..), liftBaseDefault)
52import Control.Monad.Base (MonadBase (..), liftBaseDefault)
@@ -69,36 +67,12 @@ import Data.Maybe (isJust)
69import GHC.Prim (Any)
67import GHC.Prim (Any)
70import Unsafe.Coerce (unsafeCoerce)
68import Unsafe.Coerce (unsafeCoerce)
71
69
70+import Control.Monad.Trans.Recorder (MonadRecorder(..), Paused(..),
71+ Recording)
72--import Debug.Trace (traceM)
72--import Debug.Trace (traceM)
73
73
74newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Maybe a) }
74newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Maybe a) }
75
75
76−------------------------------------------------------------------------------
77−-- Log types
78−------------------------------------------------------------------------------
79−
80−-- Logging is done using the 'logged' combinator. It remembers the last value
81−-- returned by a given computation and replays it the next time the computation
82−-- is resumed. However this could be problematic if we do not annotate all
83−-- impure computations. The program can take a different path due to a
84−-- non-logged computation returning a different value. In that case we may
85−-- replay a wrong value. To detect this we can use a unique id for each logging
86−-- site and abort if the id does not match on replay.
87−
88−-- | Constraint type synonym for a value that can be logged.
89−type Loggable a = (Show a, Read a)
90−
91−-- XXX remove the Maybe as we never log Nothing
92−data LogEntry =
93− Executing -- we are inside this computation, not yet done
94− | Result (Maybe String) -- computation done, we have the result to replay
95− deriving (Read, Show)
96−
97−data Log = Log [LogEntry] deriving Show
98−
99−-- log entries and replay entries
100−data CtxLog = CtxLog [LogEntry] [LogEntry] deriving (Read, Show)
101−
102------------------------------------------------------------------------------
76------------------------------------------------------------------------------
103-- Parent child thread communication types
77-- Parent child thread communication types
104------------------------------------------------------------------------------
78------------------------------------------------------------------------------
@@ -117,12 +91,10 @@ data Context = Context
117
91
118 -- a -> AsyncT m b
92 -- a -> AsyncT m b
119 continuation :: Any
93 continuation :: Any
120− -- Logging is done by the 'logged' primitive in this journal only if logsRef
121− -- exists.
122− , journal :: CtxLog
123
94
124 -- When we suspend we save the logs in this IORef and exit.
95 -- When we suspend we save the logs in this IORef and exit.
125− , logsRef :: Maybe (IORef [Log])
96+ , logsRef :: Maybe (IORef [Recording])
97+ -- XXX this functionality can be in a separate layer?
126 , location :: Location
98 , location :: Location
127
99
128 ---------------------------------------------------------------------------
100 ---------------------------------------------------------------------------
@@ -175,11 +147,10 @@ initContext
175 -> IORef [ThreadId]
147 -> IORef [ThreadId]
176 -> IORef Int
148 -> IORef Int
177 -> (a -> AsyncT m a)
149 -> (a -> AsyncT m a)
178− -> Maybe (IORef [Log])
150+ -> Maybe (IORef [Recording])
179 -> Context
151 -> Context
180initContext childChan pending credit finalizer lref =
152initContext childChan pending credit finalizer lref =
181 Context { continuation = unsafeCoerce finalizer
153 Context { continuation = unsafeCoerce finalizer
182− , journal = CtxLog [] []
183 , logsRef = lref
154 , logsRef = lref
184 , location = Worker
155 , location = Worker
185 , childChannel = unsafeCoerce childChan
156 , childChannel = unsafeCoerce childChan
@@ -318,6 +289,28 @@ signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ())
318-- high.
289-- high.
319--
290--
320
291
292+handleResult :: MonadIO m
293+ => Context -> Either SomeException a -> m (Maybe SomeException)
294+handleResult ctx res =
295+ case res of
296+ Left e -> do
297+ case fromException e of
298+ Just (Paused recording) ->
299+ maybe (passException e)
300+ (handleRecording recording) (logsRef ctx)
301+ Nothing -> passException e
302+ Right _ -> waitForChildren ctx
303+
304+ where
305+
306+ handleRecording recording ref = do
307+ liftIO $ atomicModifyIORef ref $ \logs -> (recording : logs, ())
308+ waitForChildren ctx
309+
310+ passException e = do
311+ liftIO $ readIORef (pendingThreads ctx) >>= mapM_ killThread
312+ return (Just e)
313+
321forkFinally1 :: MonadAsync m
314forkFinally1 :: MonadAsync m
322 => Context
315 => Context
323 -> AsyncT m a
316 -> AsyncT m a
@@ -362,15 +355,9 @@ forkContext action = do
362 }
355 }
363
356
364 beforeExit ctx pchan res = do
357 beforeExit ctx pchan res = do
365− tid <- myThreadId
358+ r <- handleResult ctx res
366− r <- case res of
367− Left e -> do
368− dbg $ "beforeExit: " ++ show tid ++ " caught exception"
369− liftIO $ readIORef (pendingThreads ctx) >>= mapM_ killThread
370− return (Just e)
371− Right _ -> waitForChildren ctx
372−
373 signalQSemB (threadCredit ctx)
359 signalQSemB (threadCredit ctx)
360+ tid <- myThreadId
374 liftIO $ atomically $ writeTChan pchan (ChildDone tid r)
361 liftIO $ atomically $ writeTChan pchan (ChildDone tid r)
375
362
376-- | Decide whether to resume the context in the same thread or a new thread
363-- | Decide whether to resume the context in the same thread or a new thread
@@ -615,13 +602,6 @@ thenDiscard ma mb = AsyncT $ do
615(>>*) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
602(>>*) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
616(>>*) = thenDiscard
603(>>*) = thenDiscard
617
604
618−------------------------------------------------------------------------------
619−-- MonadIO
620−------------------------------------------------------------------------------
621−
622−instance MonadAsync m => MonadIO (AsyncT m) where
623− liftIO mx = AsyncT $ liftIO mx >>= return . Just
624−
625-------------------------------------------------------------------------------
605-------------------------------------------------------------------------------
626-- AsyncT transformer
606-- AsyncT transformer
627-------------------------------------------------------------------------------
607-------------------------------------------------------------------------------
@@ -629,6 +609,10 @@ instance MonadAsync m => MonadIO (AsyncT m) where
629instance MonadTrans AsyncT where
609instance MonadTrans AsyncT where
630 lift mx = AsyncT $ lift mx >>= return . Just
610 lift mx = AsyncT $ lift mx >>= return . Just
631
611
612+-------------------------------------------------------------------------------
613+-- monad-control
614+-------------------------------------------------------------------------------
615+
632instance MonadTransControl AsyncT where
616instance MonadTransControl AsyncT where
633 type StT AsyncT a = (Maybe a, Context)
617 type StT AsyncT a = (Maybe a, Context)
634 liftWith f = AsyncT $ StateT $ \s ->
618 liftWith f = AsyncT $ StateT $ \s ->
@@ -648,5 +632,21 @@ instance (MonadBaseControl b m, MonadAsync m) => MonadBaseControl b (AsyncT m) w
648 {-# INLINABLE liftBaseWith #-}
632 {-# INLINABLE liftBaseWith #-}
649 {-# INLINABLE restoreM #-}
633 {-# INLINABLE restoreM #-}
650
634
635+------------------------------------------------------------------------------
636+-- Standard transformer instances
637+------------------------------------------------------------------------------
638+
639+instance MonadAsync m => MonadIO (AsyncT m) where
640+ liftIO = lift . liftIO
641+
651instance MonadAsync m => MonadThrow (AsyncT m) where
642instance MonadAsync m => MonadThrow (AsyncT m) where
652− throwM e = lift $ throwM e
643+ throwM = lift . throwM
644+
645+------------------------------------------------------------------------------
646+-- MonadRecorder
647+------------------------------------------------------------------------------
648+
649+instance (MonadAsync m, MonadRecorder m) => MonadRecorder (AsyncT m) where
650+ getJournal = lift getJournal
651+ putJournal = lift . putJournal
652+ play = lift . play
src/Asyncly/Threads.hs+22−97
@@ -6,25 +6,22 @@ module Asyncly.Threads
6 , wait_
6 , wait_
7 , gather
7 , gather
8
8
9− , waitLogged
9+ , waitRecord_
10− , waitLogged_
10+ , waitRecord
11− , logged
11+ , playRecordings
12− , suspend
13− , withLog
14− , eachWithLog
15 )
12 )
16where
13where
17
14
18−import Control.Concurrent (killThread)
19import Control.Concurrent.STM (atomically, newTChan)
15import Control.Concurrent.STM (atomically, newTChan)
20−import Control.Exception (SomeException (..))
21import Control.Monad.Catch (MonadCatch, throwM, try)
16import Control.Monad.Catch (MonadCatch, throwM, try)
22import Control.Monad.IO.Class (MonadIO (..))
17import Control.Monad.IO.Class (MonadIO (..))
23−import Control.Monad.State (get, mzero, put, runStateT)
18+import Control.Monad.State (mzero, runStateT)
24import Control.Monad.Trans.Class (MonadTrans (lift))
19import Control.Monad.Trans.Class (MonadTrans (lift))
25import Data.IORef (IORef, atomicModifyIORef,
20import Data.IORef (IORef, atomicModifyIORef,
26 newIORef, readIORef)
21 newIORef, readIORef)
27
22
23+import Control.Monad.Trans.Recorder (MonadRecorder(..), RecorderT,
24+ Recording, blank, runRecorderT)
28import Asyncly.AsyncT
25import Asyncly.AsyncT
29
26
30------------------------------------------------------------------------------
27------------------------------------------------------------------------------
@@ -34,7 +31,7 @@ import Asyncly.AsyncT
34-- | Run an 'AsyncT m' computation and collect the results generated by each
31-- | Run an 'AsyncT m' computation and collect the results generated by each
35-- thread of the computation in a list.
32-- thread of the computation in a list.
36waitAsync :: (MonadAsync m, MonadCatch m)
33waitAsync :: (MonadAsync m, MonadCatch m)
37− => (a -> AsyncT m a) -> Maybe (IORef [Log]) -> AsyncT m a -> m ()
34+ => (a -> AsyncT m a) -> Maybe (IORef [Recording]) -> AsyncT m a -> m ()
38waitAsync finalizer lref m = do
35waitAsync finalizer lref m = do
39 childChan <- liftIO $ atomically newTChan
36 childChan <- liftIO $ atomically newTChan
40 pendingRef <- liftIO $ newIORef []
37 pendingRef <- liftIO $ newIORef []
@@ -43,16 +40,8 @@ waitAsync finalizer lref m = do
43 let ctx = initContext childChan pendingRef credit finalizer lref
40 let ctx = initContext childChan pendingRef credit finalizer lref
44
41
45 r <- try $ runStateT (runAsyncT $ m >>= finalizer) ctx
42 r <- try $ runStateT (runAsyncT $ m >>= finalizer) ctx
46−
43+ e <- handleResult ctx r
47− case r of
44+ maybe (return ()) throwM e
48− Left (exc :: SomeException) -> do
49− liftIO $ readIORef pendingRef >>= mapM_ killThread
50− throwM exc
51− Right _ -> do
52− e <- waitForChildren ctx
53− case e of
54− Just (exc :: SomeException) -> throwM exc
55− Nothing -> return ()
56
45
57-- TBD throttling of producer based on conumption rate.
46-- TBD throttling of producer based on conumption rate.
58
47
@@ -87,87 +76,22 @@ wait_ m = waitAsync (const mzero) Nothing m
87-- Logging
76-- Logging
88------------------------------------------------------------------------------
77------------------------------------------------------------------------------
89
78
90−logged :: (Loggable a, MonadAsync m) => AsyncT m a -> AsyncT m a
91−logged m = AsyncT $ do
92− ctx <- get
93− case journal ctx of
94− -- no replay
95− CtxLog ls [] ->
96− case logsRef ctx of
97− Nothing -> runAsyncT m
98− Just _ -> do
99− put $ ctx {journal = CtxLog (Executing : ls) []}
100− runAndLogResult m
101−
102− -- replaying the log
103− CtxLog ls (r:rs) -> do
104− -- dbg $ "Replay: j: " ++ show j
105− case r of
106− Executing -> do
107− put $ ctx {journal = CtxLog (r : ls) rs}
108− runAndLogResult m
109− Result val -> do
110− let x = fmap read val
111− put $ ctx {journal = CtxLog (r : ls) rs}
112− return x
113−
114− where
115−
116− runAndLogResult action = runAsyncT $ do
117− x <- action
118− logResult x
119− return x
120−
121− -- replaces the head of the log with the supplied result
122− logResult x = AsyncT $ do
123− ctx@Context{journal = CtxLog (_ : ls) _} <- get
124− let entry = Result (fmap show (Just x))
125− put $ ctx {journal = CtxLog (entry : ls) []}
126− return $ Just ()
127−
128−-- XXX enable or disable the suspension points by sending an exception to the
129−-- computation.
130−--
131−-- | Suspend works as an exit point with current thread state (partially done)
132−-- saved. The computation exits only if all the threads in the computation hit
133−-- the suspend point or exit normally. The threads which hit the suspension and
134−-- were suspended can be re-started by replaying the log. If there are threads
135−-- which did not suspend or did not exit, the computation will not exit. For
136−-- completed threads output is returned and for suspended threads the logs to
137−-- resume them are returned.
138−
139−suspend :: MonadAsync m => AsyncT m ()
140−suspend = logged $ AsyncT $ do
141− ctx <- get
142− case logsRef ctx of
143− Nothing -> return $ Just ()
144− Just ref ->
145− case journal ctx of
146− CtxLog ls [] -> do
147− -- replace the "Executing" entry at the head of the log
148− -- with a "()" so that we do not enter suspend on replay
149− liftIO $ atomicModifyIORef ref $ \logs ->
150− (Log (logResult (Just ()) : tail ls) : logs, ())
151− return Nothing
152− _ -> error "Bug: replay inside suspend"
153− where logResult x = Result (fmap show x)
154−
155-- | Compose a computation using previously captured logs
79-- | Compose a computation using previously captured logs
80+playRecording :: (MonadAsync m, MonadRecorder m)
156−withLog :: Monad m => AsyncT m a -> Log -> AsyncT m a
81+ => AsyncT m a -> Recording -> AsyncT m a
157−withLog m (Log entries) = AsyncT $ do
82+playRecording m recording = play recording >> m
158− ctx <- get
159− put $ ctx {journal = CtxLog [] (reverse entries)}
160− runAsyncT m
161
83
162-- | Compose a computation using previously captured list of logs. Composes a
84-- | Compose a computation using previously captured list of logs. Composes a
163-- group of computations, one for each log.
85-- group of computations, one for each log.
86+playRecordings :: (MonadAsync m, MonadRecorder m)
164−eachWithLog :: MonadAsync m => AsyncT m a -> [Log] -> AsyncT m a
87+ => AsyncT m a -> [Recording] -> AsyncT m a
165−eachWithLog m logs = each logs >>= withLog m
88+playRecordings m logs = each logs >>= playRecording m
166
89
167-- | Run an 'AsyncT m' computation with logging enabled and collect the results
90-- | Run an 'AsyncT m' computation with logging enabled and collect the results
168-- or logs generated by each thread of the computation.
91-- or logs generated by each thread of the computation.
169−waitLogged :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m ([a], [Log])
92+waitRecord :: (MonadAsync m, MonadCatch m)
170−waitLogged m = do
93+ => AsyncT m a -> m ([a], [Recording])
94+waitRecord m = do
171 resultsRef <- liftIO $ newIORef []
95 resultsRef <- liftIO $ newIORef []
172 lref <- liftIO $ newIORef []
96 lref <- liftIO $ newIORef []
173 waitAsync (gatherResult resultsRef) (Just lref) m
97 waitAsync (gatherResult resultsRef) (Just lref) m
@@ -178,9 +102,10 @@ waitLogged m = do
178-- | Run an 'AsyncT m' computation with logging enabled, wait for it to finish
102-- | Run an 'AsyncT m' computation with logging enabled, wait for it to finish
179-- and discard the results. If the computation suspends collect the logs to
103-- and discard the results. If the computation suspends collect the logs to
180-- replay later.
104-- replay later.
181−waitLogged_ :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m [Log]
105+waitRecord_ :: (MonadAsync m, MonadCatch m)
182−waitLogged_ m = do
106+ => AsyncT (RecorderT m) a -> m [Recording]
107+waitRecord_ m = do
183 lref <- liftIO $ newIORef []
108 lref <- liftIO $ newIORef []
184− waitAsync (const mzero) (Just lref) m
109+ runRecorderT blank (waitAsync (const mzero) (Just lref) m)
185 logs <- liftIO $ readIORef lref
110 logs <- liftIO $ readIORef lref
186 return logs
111 return logs
stack.yaml+2
@@ -1,6 +1,8 @@
1resolver: lts-7.7
1resolver: lts-7.7
2packages:
2packages:
3- '.'
3- '.'
4+- location: ../monad-recorder
5+ extra-dep: true
4extra-deps: []
6extra-deps: []
5flags: {}
7flags: {}
6extra-package-dbs: []
8extra-package-dbs: []
test/logged.hs+11−13
@@ -4,32 +4,30 @@ import Control.Applicative ((<$>), (<*>), (<|>))
4import Control.Monad.IO.Class (liftIO)
4import Control.Monad.IO.Class (liftIO)
5import System.IO
5import System.IO
6import Asyncly
6import Asyncly
7+import Control.Monad.Trans.Recorder (pause, record)
7
8
8main = do
9main = do
9 hSetBuffering stdout LineBuffering
10 hSetBuffering stdout LineBuffering
10− logs <- waitLogged_ comp
11+ logs <- waitRecord_ comp
11
12
12 putStrLn $ "\nResuming with logs:"
13 putStrLn $ "\nResuming with logs:"
13− logs1 <- waitLogged_ $ eachWithLog comp logs
14+ logs1 <- waitRecord_ $ playRecordings comp logs
14
15
15 putStrLn $ "\nResuming with logs:"
16 putStrLn $ "\nResuming with logs:"
16− logs2 <- waitLogged_ $ eachWithLog comp logs1
17+ logs2 <- waitRecord_ $ playRecordings comp logs1
17
18
18 putStrLn $ "\nLogs at the end must be empty:"
19 putStrLn $ "\nLogs at the end must be empty:"
19 putStrLn $ show logs2
20 putStrLn $ show logs2
20
21
21− putStrLn $ "\nRunning the last log again using wait:"
22− wait_ $ eachWithLog comp logs1
23−
24 where
22 where
25
23
26− comp = logged $ threads 5 $ do
24+ comp = record $ threads 5 $ do
27− r <- logged $ each [1..3]
25+ r <- record $ each [1..3]
28− logged $ liftIO $ print ("A",r)
26+ record $ liftIO $ print ("A",r)
29− suspend
27+ record pause
30− logged $ liftIO $ print ("B",r)
28+ record $ liftIO $ print ("B",r)
31− x <- logged $ (,) <$> (event 1 <|> event 2) <*> (event 3 <|> event 4)
29+ x <- record $ (,) <$> (event 1 <|> event 2) <*> (event 3 <|> event 4)
32− suspend
30+ record pause
33 liftIO $ print ("C", r, x)
31 liftIO $ print ("C", r, x)
34
32
35 event n = async $ liftIO $ (do putStrLn ("event" ++ show n); return n :: IO Int)
33 event n = async $ liftIO $ (do putStrLn ("event" ++ show n); return n :: IO Int)