bf5a118 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add logging APIs

bf5a118Harendra Kumar9 years ago

src/Strands.hs+6

compare with master

unifiedsplit

@@ -10,11 +10,17 @@

11module Strands

12 ( wait

13+ , waitLogged

13 , gather

14 , gather

15+ , gatherLogged

14 , async

16 , async

15 , sample

17 , sample

16 , threads

18 , threads

19+ , logged

20+ , suspend

21+ , Log

17 )

22 )

18where

23where

25+import Strands.Context (Log, Loggable)

20import Strands.Threads

26import Strands.Threads

src/Strands/AsyncT.hs+1−5

@@ -81,12 +81,8 @@ instance Monad m => Alternative (AsyncT m) where

81instance Monad m => Monad (AsyncT m) where

82 return = pure

84− -- Inner bind-operations in 'm' add their 'f' to fstack. If we migrate the

85− -- context to a new thread, somewhere in the middle, we unwind the fstack

86− -- and run these functions when we resume the context after migration.

87 m >>= f = AsyncT $ do

84 m >>= f = AsyncT $ do

88− saveContext m f

89− runAsyncT m >>= restoreContext >>= runAsyncT

85+ saveContext m f >> runAsyncT m >>= restoreContext >>= runAsyncT

91instance Monad m => MonadPlus (AsyncT m) where

87instance Monad m => MonadPlus (AsyncT m) where

92 mzero = empty

88 mzero = empty

src/Strands/Context.hs+48−12

@@ -1,9 +1,14 @@

1+{-# LANGUAGE ConstraintKinds #-}

1{-# LANGUAGE ExistentialQuantification #-}

2{-# LANGUAGE ExistentialQuantification #-}

2{-# LANGUAGE FlexibleContexts #-}

3{-# LANGUAGE FlexibleContexts #-}

3{-# LANGUAGE RecordWildCards #-}

4{-# LANGUAGE RecordWildCards #-}

5module Strands.Context

6module Strands.Context

6 ( ChildEvent(..)

7 ( ChildEvent(..)

8+ , Loggable

9+ , Log (..)

10+ , CtxLog (..)

11+ , LogEntry (..)

7 , Context(..)

12 , Context(..)

8 , initContext

13 , initContext

9 , saveContext

14 , saveContext

@@ -28,11 +33,40 @@ import Unsafe.Coerce (unsafeCoerce)

28import GHC.Prim (Any)

33import GHC.Prim (Any)

30------------------------------------------------------------------------------

35------------------------------------------------------------------------------

31−-- State of a continuation

36+-- Log types

37+------------------------------------------------------------------------------

38+

39+-- Logging is done using the 'logged' combinator. It remembers the last value

40+-- returned by a given computation and replays it the next time the computation

41+-- is resumed. However this could be problematic if we do not annotate all

42+-- impure computations. The program can take a different path due to a

43+-- non-logged computation returning a different value. In that case we may

44+-- replay a wrong value. To detect this we can use a unique id for each logging

45+-- site and abort if the id does not match on replay.

46+

47+-- | Constraint type synonym for a value that can be logged.

48+type Loggable a = (Show a, Read a)

49+

50+data LogEntry =

51+ Executing -- we are inside this computation, not yet done

52+ | Result (Maybe String) -- computation done, we have the result to replay

53+ deriving (Read, Show)

54+

55+data Log = Log [LogEntry] deriving Show

56+

57+-- log entries and replay entries

58+data CtxLog = CtxLog [LogEntry] [LogEntry] deriving (Read, Show)

59+

60+------------------------------------------------------------------------------

61+-- Parent child thread communication types

32------------------------------------------------------------------------------

62------------------------------------------------------------------------------

34data ChildEvent a = ChildDone ThreadId (Maybe SomeException)

64data ChildEvent a = ChildDone ThreadId (Maybe SomeException)

66+------------------------------------------------------------------------------

67+-- State threaded around the monad

68+------------------------------------------------------------------------------

69+

36-- | Describes the context of a computation.

70-- | Describes the context of a computation.

37data Context = Context

71data Context = Context

38 {

72 {

@@ -53,9 +87,12 @@ data Context = Context

53 -- computation has nested binds we need to store the next 'f' for each

87 -- computation has nested binds we need to store the next 'f' for each

54 -- layer of nesting.

88 -- layer of nesting.

56− -- log only when there is a restore or if we are teleporting

90+ -- Logging is done by the 'logged' primitive in this journal only if logsRef

57− -- We can use a HasLog constraint to statically enable/disable logging

91+ -- exists.

92+ , journal :: CtxLog

93+

94+ -- When we suspend we save the logs in this IORef and exit.

58− -- , journal :: Maybe Log

95+ , logsRef :: Maybe (IORef [Log])

59 , location :: Location

96 , location :: Location

61 ---------------------------------------------------------------------------

98 ---------------------------------------------------------------------------

@@ -116,11 +153,14 @@ initContext

116 -> IORef [ThreadId]

153 -> IORef [ThreadId]

117 -> IORef Int

154 -> IORef Int

118 -> (b -> m a)

155 -> (b -> m a)

156+ -> Maybe (IORef [Log])

119 -> Context

157 -> Context

120−initContext x childChan pending credit finalizer =

158+initContext x childChan pending credit finalizer lref =

121 Context { mailBox = Nothing

159 Context { mailBox = Nothing

122 , currentm = unsafeCoerce x

160 , currentm = unsafeCoerce x

123 , fstack = [unsafeCoerce finalizer]

161 , fstack = [unsafeCoerce finalizer]

162+ , journal = CtxLog [] []

163+ , logsRef = lref

124 , location = Worker

164 , location = Worker

125 -- , mfData = mempty

165 -- , mfData = mempty

126 , childChannel = unsafeCoerce childChan

166 , childChannel = unsafeCoerce childChan

@@ -169,14 +209,10 @@ restoreContext :: (MonadPlus m1, Monad m) => Maybe a -> StateT Context m (m1 b)

169restoreContext x = do

209restoreContext x = do

170 -- XXX fstack must be non-empty when this is called.

210 -- XXX fstack must be non-empty when this is called.

171 ctx@Context { fstack = f:fs } <- Lazy.get

211 ctx@Context { fstack = f:fs } <- Lazy.get

212+ let mnext = maybe mzero (unsafeCoerce f) x

213+ Lazy.put ctx { currentm = unsafeCoerce mnext, fstack = fs }

214+ return mnext

172

215

173− let mres = case x of

174− Nothing -> mzero

175− Just y -> (unsafeCoerce f) y

176−

177− Lazy.put ctx { currentm = unsafeCoerce mres, fstack = fs }

178−

179− return mres

180

216

181-- | We can retrieve a context at any point and resume that context at some

217-- | We can retrieve a context at any point and resume that context at some

182-- later point, upon resumption we start executing again from the same point

218-- later point, upon resumption we start executing again from the same point

src/Strands/Log.hsdeleted−25

@@ -1,25 +0,0 @@

1−{-# LANGUAGE ConstraintKinds #-}

2−

3−module Strands.Log

4− ( Loggable

5− , LogEntry (..)

6− , Replaying

7− , Log (..)

8− )

9−where

10−

11−------------------------------------------------------------------------------

12−-- Log types

13−------------------------------------------------------------------------------

14−

15−-- | Constraint type synonym for a value that can be logged.

16−type Loggable a = (Show a, Read a)

17−

18−data LogEntry =

19− Executing -- we are inside this computation, not yet done

20− | Maybe String -- computation is done and we have the result to replay

21− deriving (Read, Show)

22−

23−type Replaying = Bool

24−

25−data Log = Log Replaying [LogEntry] deriving (Read, Show)

src/Strands/Threads.hs+101−6

@@ -12,6 +12,11 @@ module Strands.Threads

13 , wait

14 , gather

15+

16+ , logged

17+ , suspend

18+ , waitLogged

19+ , gatherLogged

15 )

20 )

16where

21where

@@ -27,7 +32,7 @@ import qualified Control.Exception.Lifted as EL

27import Control.Monad.Catch (MonadCatch, MonadThrow, throwM,

32import Control.Monad.Catch (MonadCatch, MonadThrow, throwM,

28 try)

33 try)

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

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

30−import Control.Monad.State (get, gets, modify, mzero,

35+import Control.Monad.State (get, gets, modify, mzero, put,

31 runStateT, when, StateT)

36 runStateT, when, StateT)

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

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

33import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)

38import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)

@@ -496,14 +501,14 @@ noTrans x = AsyncT $ x >>= return . Just

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

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

497-- thread of the computation in a list.

502-- thread of the computation in a list.

498waitAsync :: forall m a. (MonadIO m, MonadCatch m)

503waitAsync :: forall m a. (MonadIO m, MonadCatch m)

499− => (a -> AsyncT m a) -> AsyncT m a -> m ()

504+ => (a -> AsyncT m a) -> Maybe (IORef [Log]) -> AsyncT m a -> m ()

500−waitAsync finalizer m = do

505+waitAsync finalizer lref m = do

501 childChan <- liftIO $ atomically newTChan

506 childChan <- liftIO $ atomically newTChan

502 pendingRef <- liftIO $ newIORef []

507 pendingRef <- liftIO $ newIORef []

503 credit <- liftIO $ newIORef maxBound

508 credit <- liftIO $ newIORef maxBound

504

509

505 let ctx = initContext (empty :: AsyncT m a) childChan pendingRef credit

510 let ctx = initContext (empty :: AsyncT m a) childChan pendingRef credit

506− finalizer

511+ finalizer lref

507

512

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

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

509

514

@@ -532,11 +537,101 @@ gather :: forall m a. (MonadIO m, MonadCatch m)

532 => AsyncT m a -> m [a]

537 => AsyncT m a -> m [a]

533gather m = do

538gather m = do

534 resultsRef <- liftIO $ newIORef []

539 resultsRef <- liftIO $ newIORef []

535− waitAsync (gatherResult resultsRef) m

540+ waitAsync (gatherResult resultsRef) Nothing m

536 liftIO $ readIORef resultsRef

541 liftIO $ readIORef resultsRef

537

542

538-- | Run an 'AsyncT m' computation, wait for it to finish and discard the

543-- | Run an 'AsyncT m' computation, wait for it to finish and discard the

539-- results.

544-- results.

540wait :: forall m a. (MonadIO m, MonadCatch m)

545wait :: forall m a. (MonadIO m, MonadCatch m)

541 => AsyncT m a -> m ()

546 => AsyncT m a -> m ()

542−wait m = waitAsync (const mzero) m

547+wait m = waitAsync (const mzero) Nothing m

548+

549+------------------------------------------------------------------------------

550+-- Logging

551+------------------------------------------------------------------------------

552+

553+logged :: (Loggable a, MonadIO m) => AsyncT m a -> AsyncT m a

554+logged m = AsyncT $ do

555+ ctx <- get

556+ case logsRef ctx of

557+ Nothing -> runAsyncT m

558+ Just _ -> withLog ctx m

559+ where

560+

561+ withLog ctx action =

562+ case journal ctx of

563+ -- no replay

564+ j@(CtxLog _ []) -> do

565+ preLog ctx j

566+ x <- runAsyncT action

567+ postLog ctx j x

568+ return x

569+

570+ -- replaying the log

571+ CtxLog ls (r:rs) -> do

572+ let j = CtxLog ls rs

573+ preLog ctx j

574+ case r of

575+ Executing -> do

576+ x <- runAsyncT action

577+ postLog ctx (CtxLog ls []) x

578+ return x

579+ Result val -> do

580+ let x = fmap read val

581+ postLog ctx j x

582+ return x

583+

584+ -- add an Executing entry at the head of the log

585+ preLog = appendLog Executing

586+ postLog ctx j x = appendLog (Result (fmap show x)) ctx j

587+ appendLog l ctx (CtxLog ls rs) = put $ ctx {journal = CtxLog (l : ls) rs}

588+

589+-- XXX enable or disable the suspension points by sending an exception to the

590+-- computation.

591+--

592+-- | Suspend works as an exit point with current thread state (partially done)

593+-- saved. The computation exits only if all the threads in the computation hit

594+-- the suspend point or exit normally. The threads which hit the suspension and

595+-- were suspended can be re-started by replaying the log. If there are threads

596+-- which did not suspend or did not exit, the computation will not exit. For

597+-- completed threads output is returned and for suspended threads the logs to

598+-- resume them are returned.

599+

600+suspend :: MonadIO m => AsyncT m ()

601+suspend = logged $ AsyncT $ do

602+ ctx <- get

603+ case logsRef ctx of

604+ Nothing -> return $ Just ()

605+ Just ref ->

606+ case journal ctx of

607+ CtxLog ls [] -> do

608+ -- replace the "Executing" entry at the head of the log

609+ -- with a "()" so that we do not enter suspend on replay

610+ liftIO $ atomicModifyIORef ref $ \logs ->

611+ (Log (logResult (Just ()) : tail ls) : logs, ())

612+ return Nothing

613+ _ -> error "Bug: replay inside suspend"

614+ where logResult x = Result (fmap show x)

615+

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

617+-- or logs generated by each thread of the computation.

618+gatherLogged :: forall m a. (MonadIO m, MonadCatch m)

619+ => AsyncT m a -> m ([a], [Log])

620+gatherLogged m = do

621+ resultsRef <- liftIO $ newIORef []

622+ lref <- liftIO $ newIORef []

623+ waitAsync (gatherResult resultsRef) (Just lref) m

624+ res <- liftIO $ readIORef resultsRef

625+ logs <- liftIO $ readIORef lref

626+ return (res, logs)

627+

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

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

630+-- replay later.

631+waitLogged :: forall m a. (MonadIO m, MonadCatch m)

632+ => AsyncT m a -> m [Log]

633+waitLogged m = do

634+ lref <- liftIO $ newIORef []

635+ waitAsync (const mzero) (Just lref) m

636+ logs <- liftIO $ readIORef lref

637+ return logs

strands.cabal−1

@@ -21,7 +21,6 @@ library

21 hs-source-dirs: src

22 exposed-modules: Strands

23 , Strands.Context

24− , Strands.Log

25 , Strands.AsyncT

24 , Strands.AsyncT

26 , Strands.Threads

25 , Strands.Threads

test/logged.hsnew file+12

@@ -0,0 +1,12 @@

1+import Strands

2+import Control.Monad.IO.Class (liftIO)

4+main = do

5+ xs <- waitLogged $ logged $ threads 0 $ do

6+ r <- logged $ return (5::Int)

7+ logged $ liftIO $ print ("A",r)

8+ suspend

9+ logged $ liftIO $ print ("B",r)

10+ suspend

11+ liftIO $ print ("C",r)

12+ putStrLn $ show xs