8340241 · streamly · git.iko.soy

git.iko.soy/streamly

Split the logging functionality into a separate package

commit8340241116e0287dda390ffce449a55b615bc341 authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 13 Jul 2017, 18:21 (9 years ago) parent795dd4b

6 files changed+93−171

compare with master

unifiedsplit

asyncly.cabal+2 src/Asyncly.hs+3−8 src/Asyncly/AsyncT.hs+53−53 src/Asyncly/Threads.hs+22−97 stack.yaml+2 test/logged.hs+11−13

asyncly.cabal+2

@@ -41,6 +41,7 @@ library

4141 , exceptions

4242 , lifted-base

4343 , monad-control

44+ , monad-recorder

4445 , mtl

4546 , stm

4647 , transformers-base

@@ -55,6 +56,7 @@ test-suite test

5556 , hspec >= 2.0 && < 3

5657 , containers

5758 , asyncly

59+ , monad-recorder

5860 default-language: Haskell2010

5961

6062source-repository head

src/Asyncly.hs+3−8

@@ -39,14 +39,9 @@ module Asyncly

3939 , afterFirst

4040 , (>>|)

4141

42− , Log

43− , Loggable

44− , waitLogged

45− , waitLogged_

46− , logged

47− , suspend

48− , withLog

49− , eachWithLog

42+ , waitRecord

43+ , waitRecord_

44+ , playRecordings

5045 )

5146where

5247

src/Asyncly/AsyncT.hs+53−53

@@ -14,10 +14,6 @@ module Asyncly.AsyncT

1414 ( AsyncT (..)

1515 , MonadAsync

1616 , ChildEvent(..)

17− , Loggable

18− , Log (..)

19− , CtxLog (..)

20− , LogEntry (..)

2117 , Context(..)

2218 , initContext

2319 , Location(..)

@@ -25,6 +21,7 @@ module Asyncly.AsyncT

2521 , setLocation

2622 , threads

2723 , waitForChildren

24+ , handleResult

2825

2926 , async

3027 , makeAsync

@@ -48,7 +45,8 @@ import Control.Concurrent (ThreadId, forkIO, killThread,

4845import Control.Concurrent.STM (TChan, atomically, newTChan,

4946 readTChan, tryReadTChan,

5047 writeTChan)

51−import Control.Exception (SomeException (..))

48+import Control.Exception (fromException,

49+ SomeException (..))

5250import qualified Control.Exception.Lifted as EL

5351import Control.Monad (ap, mzero, when)

5452import Control.Monad.Base (MonadBase (..), liftBaseDefault)

@@ -69,36 +67,12 @@ import Data.Maybe (isJust)

6967import GHC.Prim (Any)

7068import Unsafe.Coerce (unsafeCoerce)

7169

70+import Control.Monad.Trans.Recorder (MonadRecorder(..), Paused(..),

71+ Recording)

7272--import Debug.Trace (traceM)

7373

7474newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Maybe a) }

7575

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−

10276------------------------------------------------------------------------------

10377-- Parent child thread communication types

10478------------------------------------------------------------------------------

@@ -117,12 +91,10 @@ data Context = Context

11791

11892 -- a -> AsyncT m b

11993 continuation :: Any

120− -- Logging is done by the 'logged' primitive in this journal only if logsRef

121− -- exists.

122− , journal :: CtxLog

12394

12495 -- 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?

12698 , location :: Location

12799

128100 ---------------------------------------------------------------------------

@@ -175,11 +147,10 @@ initContext

175147 -> IORef [ThreadId]

176148 -> IORef Int

177149 -> (a -> AsyncT m a)

178− -> Maybe (IORef [Log])

150+ -> Maybe (IORef [Recording])

179151 -> Context

180152initContext childChan pending credit finalizer lref =

181153 Context { continuation = unsafeCoerce finalizer

182− , journal = CtxLog [] []

183154 , logsRef = lref

184155 , location = Worker

185156 , childChannel = unsafeCoerce childChan

@@ -318,6 +289,28 @@ signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ())

318289-- high.

319290--

320291

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+

321314forkFinally1 :: MonadAsync m

322315 => Context

323316 -> AsyncT m a

@@ -362,15 +355,9 @@ forkContext action = do

362355 }

363356

364357 beforeExit ctx pchan res = do

365− tid <- myThreadId

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−

358+ r <- handleResult ctx res

373359 signalQSemB (threadCredit ctx)

360+ tid <- myThreadId

374361 liftIO $ atomically $ writeTChan pchan (ChildDone tid r)

375362

376363-- | Decide whether to resume the context in the same thread or a new thread

@@ -615,13 +602,6 @@ thenDiscard ma mb = AsyncT $ do

615602(>>*) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a

616603(>>*) = thenDiscard

617604

618−------------------------------------------------------------------------------

619−-- MonadIO

620−------------------------------------------------------------------------------

621−

622−instance MonadAsync m => MonadIO (AsyncT m) where

623− liftIO mx = AsyncT $ liftIO mx >>= return . Just

624−

625605-------------------------------------------------------------------------------

626606-- AsyncT transformer

627607-------------------------------------------------------------------------------

@@ -629,6 +609,10 @@ instance MonadAsync m => MonadIO (AsyncT m) where

629609instance MonadTrans AsyncT where

630610 lift mx = AsyncT $ lift mx >>= return . Just

631611

612+-------------------------------------------------------------------------------

613+-- monad-control

614+-------------------------------------------------------------------------------

615+

632616instance MonadTransControl AsyncT where

633617 type StT AsyncT a = (Maybe a, Context)

634618 liftWith f = AsyncT $ StateT $ \s ->

@@ -648,5 +632,21 @@ instance (MonadBaseControl b m, MonadAsync m) => MonadBaseControl b (AsyncT m) w

648632 {-# INLINABLE liftBaseWith #-}

649633 {-# INLINABLE restoreM #-}

650634

635+------------------------------------------------------------------------------

636+-- Standard transformer instances

637+------------------------------------------------------------------------------

638+

639+instance MonadAsync m => MonadIO (AsyncT m) where

640+ liftIO = lift . liftIO

641+

651642instance 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

66 , wait_

77 , gather

9− , waitLogged

10− , waitLogged_

11− , logged

12− , suspend

13− , withLog

14− , eachWithLog

9+ , waitRecord_

10+ , waitRecord

11+ , playRecordings

1512 )

1613where

1714

18−import Control.Concurrent (killThread)

1915import Control.Concurrent.STM (atomically, newTChan)

20−import Control.Exception (SomeException (..))

2116import Control.Monad.Catch (MonadCatch, throwM, try)

2217import Control.Monad.IO.Class (MonadIO (..))

23−import Control.Monad.State (get, mzero, put, runStateT)

18+import Control.Monad.State (mzero, runStateT)

2419import Control.Monad.Trans.Class (MonadTrans (lift))

2520import Data.IORef (IORef, atomicModifyIORef,

2621 newIORef, readIORef)

2722

23+import Control.Monad.Trans.Recorder (MonadRecorder(..), RecorderT,

24+ Recording, blank, runRecorderT)

2825import Asyncly.AsyncT

2926

3027------------------------------------------------------------------------------

@@ -34,7 +31,7 @@ import Asyncly.AsyncT

3431-- | Run an 'AsyncT m' computation and collect the results generated by each

3532-- thread of the computation in a list.

3633waitAsync :: (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 ()

3835waitAsync finalizer lref m = do

3936 childChan <- liftIO $ atomically newTChan

4037 pendingRef <- liftIO $ newIORef []

@@ -43,16 +40,8 @@ waitAsync finalizer lref m = do

4340 let ctx = initContext childChan pendingRef credit finalizer lref

4441

4542 r <- try $ runStateT (runAsyncT $ m >>= finalizer) ctx

46−

47− case r of

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 ()

43+ e <- handleResult ctx r

44+ maybe (return ()) throwM e

5645

5746-- TBD throttling of producer based on conumption rate.

5847

@@ -87,87 +76,22 @@ wait_ m = waitAsync (const mzero) Nothing m

8776-- Logging

8877------------------------------------------------------------------------------

8978

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−

15579-- | Compose a computation using previously captured logs

156−withLog :: Monad m => AsyncT m a -> Log -> AsyncT m a

157−withLog m (Log entries) = AsyncT $ do

158− ctx <- get

159− put $ ctx {journal = CtxLog [] (reverse entries)}

160− runAsyncT m

80+playRecording :: (MonadAsync m, MonadRecorder m)

81+ => AsyncT m a -> Recording -> AsyncT m a

82+playRecording m recording = play recording >> m

16183

16284-- | Compose a computation using previously captured list of logs. Composes a

16385-- group of computations, one for each log.

164−eachWithLog :: MonadAsync m => AsyncT m a -> [Log] -> AsyncT m a

165−eachWithLog m logs = each logs >>= withLog m

86+playRecordings :: (MonadAsync m, MonadRecorder m)

87+ => AsyncT m a -> [Recording] -> AsyncT m a

88+playRecordings m logs = each logs >>= playRecording m

16689

16790-- | Run an 'AsyncT m' computation with logging enabled and collect the results

16891-- or logs generated by each thread of the computation.

169−waitLogged :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m ([a], [Log])

170−waitLogged m = do

92+waitRecord :: (MonadAsync m, MonadCatch m)

93+ => AsyncT m a -> m ([a], [Recording])

94+waitRecord m = do

17195 resultsRef <- liftIO $ newIORef []

17296 lref <- liftIO $ newIORef []

17397 waitAsync (gatherResult resultsRef) (Just lref) m

@@ -178,9 +102,10 @@ waitLogged m = do

178102-- | Run an 'AsyncT m' computation with logging enabled, wait for it to finish

179103-- and discard the results. If the computation suspends collect the logs to

180104-- replay later.

181−waitLogged_ :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m [Log]

182−waitLogged_ m = do

105+waitRecord_ :: (MonadAsync m, MonadCatch m)

106+ => AsyncT (RecorderT m) a -> m [Recording]

107+waitRecord_ m = do

183108 lref <- liftIO $ newIORef []

184− waitAsync (const mzero) (Just lref) m

109+ runRecorderT blank (waitAsync (const mzero) (Just lref) m)

185110 logs <- liftIO $ readIORef lref

186111 return logs

stack.yaml+2

@@ -1,6 +1,8 @@

11resolver: lts-7.7

22packages:

33- '.'

4+- location: ../monad-recorder

5+ extra-dep: true

46extra-deps: []

57flags: {}

68extra-package-dbs: []

test/logged.hs+11−13

@@ -4,32 +4,30 @@ import Control.Applicative ((<$>), (<*>), (<|>))

44import Control.Monad.IO.Class (liftIO)

55import System.IO

66import Asyncly

7+import Control.Monad.Trans.Recorder (pause, record)

89main = do

910 hSetBuffering stdout LineBuffering

10− logs <- waitLogged_ comp

11+ logs <- waitRecord_ comp

1112

1213 putStrLn $ "\nResuming with logs:"

13− logs1 <- waitLogged_ $ eachWithLog comp logs

14+ logs1 <- waitRecord_ $ playRecordings comp logs

1415

1516 putStrLn $ "\nResuming with logs:"

16− logs2 <- waitLogged_ $ eachWithLog comp logs1

17+ logs2 <- waitRecord_ $ playRecordings comp logs1

1718

1819 putStrLn $ "\nLogs at the end must be empty:"

1920 putStrLn $ show logs2

2021

21− putStrLn $ "\nRunning the last log again using wait:"

22− wait_ $ eachWithLog comp logs1

23−

2422 where

2523

26− comp = logged $ threads 5 $ do

27− r <- logged $ each [1..3]

28− logged $ liftIO $ print ("A",r)

29− suspend

30− logged $ liftIO $ print ("B",r)

31− x <- logged $ (,) <$> (event 1 <|> event 2) <*> (event 3 <|> event 4)

32− suspend

24+ comp = record $ threads 5 $ do

25+ r <- record $ each [1..3]

26+ record $ liftIO $ print ("A",r)

27+ record pause

28+ record $ liftIO $ print ("B",r)

29+ x <- record $ (,) <$> (event 1 <|> event 2) <*> (event 3 <|> event 4)

30+ record pause

3331 liftIO $ print ("C", r, x)

3432

3533 event n = async $ liftIO $ (do putStrLn ("event" ++ show n); return n :: IO Int)