eventlog-live 0.2.0.1 → 0.3.0.0

raw patch · 30 files changed

+2814/−2756 lines, 30 filesdep −eventlog-livePVP ok

version bump matches the API change (PVP)

Dependencies removed: eventlog-live

API changes (from Hackage documentation)

Files

• CHANGELOG.md +9/−0
• eventlog-live.cabal +14/−46
• src-logger/GHC/Eventlog/Live/Internal/Logger.hs +0/−103
• src-logger/GHC/Eventlog/Live/Verbosity.hs +0/−76
• src-machines/GHC/Eventlog/Live/Data/Attribute.hs +0/−142
• src-machines/GHC/Eventlog/Live/Data/Metric.hs +0/−30
• src-machines/GHC/Eventlog/Live/Data/Span.hs +0/−25
• src-machines/GHC/Eventlog/Live/Machine.hs +0/−1151
• src-machines/GHC/Eventlog/Live/Machine/Core.hs +0/−494
• src-machines/GHC/Eventlog/Live/Machine/Decoder.hs +0/−49
• src-machines/GHC/Eventlog/Live/Machine/Sink.hs +0/−42
• src-machines/GHC/Eventlog/Live/Machine/Source.hs +0/−144
• src-machines/GHC/Eventlog/Live/Machine/WithStartTime.hs +0/−88
• src-options/GHC/Eventlog/Live/Options.hs +0/−189
• src-socket/GHC/Eventlog/Live/Socket.hs +0/−177
• src/GHC/Eventlog/Live/Data/Attribute.hs +142/−0
• src/GHC/Eventlog/Live/Data/Metric.hs +30/−0
• src/GHC/Eventlog/Live/Data/Span.hs +25/−0
• src/GHC/Eventlog/Live/Logger.hs +113/−0
• src/GHC/Eventlog/Live/Machine/Analysis/Capability.hs +477/−0
• src/GHC/Eventlog/Live/Machine/Analysis/Heap.hs +438/−0
• src/GHC/Eventlog/Live/Machine/Analysis/Thread.hs +307/−0
• src/GHC/Eventlog/Live/Machine/Core.hs +494/−0
• src/GHC/Eventlog/Live/Machine/Decoder.hs +49/−0
• src/GHC/Eventlog/Live/Machine/Sink.hs +42/−0
• src/GHC/Eventlog/Live/Machine/Source.hs +144/−0
• src/GHC/Eventlog/Live/Machine/WithStartTime.hs +88/−0
• src/GHC/Eventlog/Live/Options.hs +189/−0
• src/GHC/Eventlog/Live/Socket.hs +177/−0
• src/GHC/Eventlog/Live/Verbosity.hs +76/−0

@@ -1,3 +1,12 @@+### 0.3.0.0++- **BREAKING**: Move capability usage analysis machines to their own module.+- **BREAKING**: Move heap analysis machines to their own module.+- **BREAKING**: Move thread label analysis machines to their own module.+- **BREAKING**: Move thread state analysis machines to their own module.+- **BREAKING**: Merge all sub-libraries into the main library.+- Expose `GHC.Eventlog.Live.Logger`.+ ### 0.2.0.1  - Fix error due to incorrect formatting string.

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            eventlog-live-version:         0.2.0.1+version:         0.3.0.0 synopsis:        Live processing of eventlog data. description:   This package supports live processing of eventlog data.@@ -88,67 +88,35 @@     TypeFamilies  library-  import:             language-  hs-source-dirs:     src-machines+  import:          language+  hs-source-dirs:  src   exposed-modules:     GHC.Eventlog.Live.Data.Attribute     GHC.Eventlog.Live.Data.Metric     GHC.Eventlog.Live.Data.Span-    GHC.Eventlog.Live.Machine+    GHC.Eventlog.Live.Logger+    GHC.Eventlog.Live.Machine.Analysis.Capability+    GHC.Eventlog.Live.Machine.Analysis.Heap+    GHC.Eventlog.Live.Machine.Analysis.Thread     GHC.Eventlog.Live.Machine.Core     GHC.Eventlog.Live.Machine.Decoder     GHC.Eventlog.Live.Machine.Sink     GHC.Eventlog.Live.Machine.Source     GHC.Eventlog.Live.Machine.WithStartTime+    GHC.Eventlog.Live.Options+    GHC.Eventlog.Live.Socket+    GHC.Eventlog.Live.Verbosity -  reexported-modules: GHC.Eventlog.Live.Verbosity   build-depends:+    , ansi-terminal         >=1.1    && <1.2     , base                  >=4.16   && <4.22     , bytestring            >=0.11   && <0.13     , dlist                 >=1.0    && <1.1-    , eventlog-live:logger     , ghc-events            >=0.20   && <0.21     , hashable              >=1.4    && <1.6     , machines              >=0.7.4  && <0.8+    , network               >=3.2.7  && <3.3+    , optparse-applicative  >=0.17   && <0.20     , text                  >=1.2    && <2.2+    , unliftio-core         >=0.2.1  && <0.3     , unordered-containers  >=0.2.20 && <0.3--library logger-  import:          language-  visibility:      private-  hs-source-dirs:  src-logger-  exposed-modules:-    GHC.Eventlog.Live.Internal.Logger-    GHC.Eventlog.Live.Verbosity--  build-depends:-    , ansi-terminal  >=1.1  && <1.2-    , base           >=4.16 && <4.22-    , text           >=1.2  && <2.2--library options-  import:          language-  visibility:      public-  hs-source-dirs:  src-options-  exposed-modules: GHC.Eventlog.Live.Options-  build-depends:-    , base                  >=4.16 && <4.22-    , eventlog-live-    , ghc-events            >=0.20 && <0.21-    , optparse-applicative  >=0.17 && <0.20--library socket-  import:          language-  visibility:      public-  hs-source-dirs:  src-socket-  exposed-modules: GHC.Eventlog.Live.Socket-  build-depends:-    , base                   >=4.16  && <4.22-    , eventlog-live-    , eventlog-live:logger-    , eventlog-live:options-    , ghc-events             >=0.20  && <0.21-    , machines               >=0.7.4 && <0.8-    , network                >=3.2.7 && <3.3-    , text                   >=1.2   && <2.2-    , unliftio-core          >=0.2.1 && <0.3

@@ -1,103 +0,0 @@-{- |-Module      : GHC.Eventlog.Live.Internal.Logger-Description : /Internal module/. Logging functions.-Stability   : experimental-Portability : portable--This module is __internal__. The [PVP](https://pvp.haskell.org) __does not apply__.--}-module GHC.Eventlog.Live.Internal.Logger (-  LogSource,-  logMessage,-  logError,-  logWarning,-  logInfo,-  logDebug,-) where--import Control.Exception (bracket_)-import Control.Monad.IO.Class (MonadIO (..))-import Data.Text (Text)-import Data.Text qualified as T-import Data.Text.IO qualified as TIO-import GHC.Eventlog.Live.Verbosity (Verbosity, showVerbosity, verbosityDebug, verbosityError, verbosityInfo, verbosityWarning)-import System.Console.ANSI (Color (..), ColorIntensity (..), ConsoleLayer (..), SGR (..), hNowSupportsANSI, hSetSGR)-import System.IO qualified as IO--{- |-Internal helper. Denotes the source of a log message.--}-type LogSource = Text--{- |-Internal helper. Log messages to given handle.-Only prints a message if its verbosity level is above the verbosity threshold.--}-logMessage :: (MonadIO m) => IO.Handle -> Verbosity -> Verbosity -> LogSource -> Text -> m ()-logMessage handle verbosityLevel verbosityThreshold logSource msg-  | verbosityLevel >= verbosityThreshold = liftIO $ do-      withVerbosityColor verbosityLevel handle-        . flip TIO.hPutStrLn-        . formatMessage verbosityLevel verbosityThreshold logSource-        $ msg-      IO.hFlush handle-  | otherwise = pure ()--{- |-Internal helper. Format the message appropriately for the given verbosity level and threshold.--}-formatMessage :: Verbosity -> Verbosity -> LogSource -> Text -> Text-formatMessage verbosityLevel verbosityThreshold logSource msg-  | verbosityLevel == verbosityInfo && verbosityThreshold /= verbosityDebug = msg-  | otherwise = mconcat [showVerbosity verbosityLevel, T.pack "[", logSource, T.pack "]: ", msg]--{- |-Internal helper. Use a handle with the color set appropriately for the given verbosity level.--}-withVerbosityColor :: Verbosity -> IO.Handle -> (IO.Handle -> IO a) -> IO a-withVerbosityColor verbosity handle action = do-  supportsANSI <- hNowSupportsANSI handle-  if not supportsANSI-    then-      action handle-    else case verbosityColor verbosity of-      Nothing ->-        action handle-      Just color -> do-        let setVerbosityColor = hSetSGR handle [SetColor Foreground Vivid color]-        let setDefaultColor = hSetSGR handle [SetDefaultColor Foreground]-        bracket_ setVerbosityColor setDefaultColor $ action handle--{- |-Internal helper. Determine the ANSI color associated with a particular verbosity level.--}-verbosityColor :: Verbosity -> Maybe Color-verbosityColor verbosity-  | verbosity == verbosityError = Just Red-  | verbosity == verbosityWarning = Just Yellow-  | verbosity == verbosityDebug = Just Blue-  | otherwise = Nothing--{- |-Internal helper. Log errors to `IO.stderr`.--}-logError :: (MonadIO m) => Verbosity -> LogSource -> Text -> m ()-logError = logMessage IO.stderr verbosityError--{- |-Internal helper. Log warnings to `IO.stderr`.--}-logWarning :: (MonadIO m) => Verbosity -> LogSource -> Text -> m ()-logWarning = logMessage IO.stderr verbosityWarning--{- |-Internal helper. Log info messages to `IO.stderr`.--}-logInfo :: (MonadIO m) => Verbosity -> LogSource -> Text -> m ()-logInfo = logMessage IO.stdout verbosityInfo--{- |-Internal helper. Log debug messages to `IO.stderr`.--}-logDebug :: (MonadIO m) => Verbosity -> LogSource -> Text -> m ()-logDebug = logMessage IO.stderr verbosityDebug

@@ -1,76 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}--{- |-Module      : GHC.Eventlog.Live.Verbosity-Description : Logging verbosity for eventlog machines.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Verbosity (-  Verbosity,-  showVerbosity,-  verbosityQuiet,-  verbosityError,-  verbosityWarning,-  verbosityInfo,-  verbosityDebug,-) where--import Data.Text (Text)------------------------------------------------------------------------------------ Verbosity----------------------------------------------------------------------------------{- |-The type of logging verbosities supported by the machines-in "GHC.Eventlog.Live.Machines".--}-data Verbosity-  = VerbosityDebug-  | VerbosityInfo-  | VerbosityWarning-  | VerbosityError-  | VerbosityQuiet-  deriving (Eq, Ord)--{- |-Pretty-printer for t`Verbosity`.--}-showVerbosity :: Verbosity -> Text-showVerbosity = \case-  VerbosityDebug -> "Debug"-  VerbosityInfo -> "Info"-  VerbosityWarning -> "Warning"-  VerbosityError -> "Error"-  VerbosityQuiet -> "Quiet"--{- |-Quiet t`Verbosity`.--}-verbosityQuiet :: Verbosity-verbosityQuiet = VerbosityQuiet--{- |-Error t`Verbosity`.--}-verbosityError :: Verbosity-verbosityError = VerbosityError--{- |-Warning t`Verbosity`.--}-verbosityWarning :: Verbosity-verbosityWarning = VerbosityWarning--{- |-Info t`Verbosity`.--}-verbosityInfo :: Verbosity-verbosityInfo = VerbosityInfo--{- |-Debug t`Verbosity`.--}-verbosityDebug :: Verbosity-verbosityDebug = VerbosityDebug

@@ -1,142 +0,0 @@-{- |-Module      : GHC.Eventlog.Live.Attribute-Description : Representation for attributes.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Data.Attribute (-  Attr,-  AttrKey,-  AttrValue (..),-  IsAttrValue (..),-  (~=),-) where--import Data.Int (Int16, Int32, Int64, Int8)-import Data.Text (Text)-import Data.Text qualified as T-import Data.Word (Word16, Word32, Word64, Word8)--{- |-An attribute is a key-value pair where the key is any string and the value is-some numeric type, string, or null. Attributes should be constructed using the-`(~=)` operator, which automatically converts Haskell types to t`AttrValue`.--}-type Attr = (AttrKey, AttrValue)--{- |-Construct an t`Attr` as a pair of an t`AttrKey` and an t`AttrValue`,-constructed via the t`IsAttrValue` class.--}-(~=) :: (IsAttrValue v) => AttrKey -> v -> Attr-k ~= v = (ak, av)- where-  !ak = k-  !av = toAttrValue v-{-# INLINE (~=) #-}--{- |-The type of attribute keys.--}-type AttrKey =-  Text--{- |-The type of attribute values.--}-data AttrValue-  = AttrInt !Int-  | AttrInt8 !Int8-  | AttrInt16 !Int16-  | AttrInt32 !Int32-  | AttrInt64 !Int64-  | AttrWord !Word-  | AttrWord8 !Word8-  | AttrWord16 !Word16-  | AttrWord32 !Word32-  | AttrWord64 !Word64-  | AttrDouble !Double-  | AttrText !Text-  | AttrNull-  deriving (Show)--{- |-Utility class to help construct values of the t`AttrValue` type.--}-class IsAttrValue v where-  toAttrValue :: v -> AttrValue--instance IsAttrValue AttrValue where-  toAttrValue :: AttrValue -> AttrValue-  toAttrValue = id-  {-# INLINE toAttrValue #-}--instance IsAttrValue Int where-  toAttrValue :: Int -> AttrValue-  toAttrValue = AttrInt-  {-# INLINE toAttrValue #-}--instance IsAttrValue Int8 where-  toAttrValue :: Int8 -> AttrValue-  toAttrValue = AttrInt8-  {-# INLINE toAttrValue #-}--instance IsAttrValue Int16 where-  toAttrValue :: Int16 -> AttrValue-  toAttrValue = AttrInt16-  {-# INLINE toAttrValue #-}--instance IsAttrValue Int32 where-  toAttrValue :: Int32 -> AttrValue-  toAttrValue = AttrInt32-  {-# INLINE toAttrValue #-}--instance IsAttrValue Int64 where-  toAttrValue :: Int64 -> AttrValue-  toAttrValue = AttrInt64-  {-# INLINE toAttrValue #-}--instance IsAttrValue Word where-  toAttrValue :: Word -> AttrValue-  toAttrValue = AttrWord-  {-# INLINE toAttrValue #-}--instance IsAttrValue Word8 where-  toAttrValue :: Word8 -> AttrValue-  toAttrValue = AttrWord8-  {-# INLINE toAttrValue #-}--instance IsAttrValue Word16 where-  toAttrValue :: Word16 -> AttrValue-  toAttrValue = AttrWord16-  {-# INLINE toAttrValue #-}--instance IsAttrValue Word32 where-  toAttrValue :: Word32 -> AttrValue-  toAttrValue = AttrWord32-  {-# INLINE toAttrValue #-}--instance IsAttrValue Word64 where-  toAttrValue :: Word64 -> AttrValue-  toAttrValue = AttrWord64-  {-# INLINE toAttrValue #-}--instance IsAttrValue Double where-  toAttrValue :: Double -> AttrValue-  toAttrValue = AttrDouble-  {-# INLINE toAttrValue #-}--instance IsAttrValue String where-  toAttrValue :: String -> AttrValue-  toAttrValue = AttrText . T.pack-  {-# INLINE toAttrValue #-}--instance IsAttrValue Text where-  toAttrValue :: Text -> AttrValue-  toAttrValue = AttrText-  {-# INLINE toAttrValue #-}--instance (IsAttrValue v) => IsAttrValue (Maybe v) where-  toAttrValue :: Maybe v -> AttrValue-  toAttrValue = maybe AttrNull toAttrValue-  {-# INLINE toAttrValue #-}

@@ -1,30 +0,0 @@-{- |-Module      : GHC.Eventlog.Live.Metric-Description : Representation for metrics.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Data.Metric (-  Metric (..),-) where--import GHC.Eventlog.Live.Data.Attribute (Attr)-import GHC.RTS.Events (Timestamp)--{- |-Metrics combine a measurement with a timestamp representing the time of the-measurement, a timestamp representing the earliest possible measurment, and-a list of attributes.--}-data Metric a = Metric-  { value :: !a-  -- ^ The measurement.-  , maybeTimeUnixNano :: !(Maybe Timestamp)-  -- ^ The time at which the measurment was taken.-  , maybeStartTimeUnixNano :: !(Maybe Timestamp)-  -- ^ The earliest time at which any measurement could have been taken.-  --   Usually, this represents the start time of a process.-  , attr :: [Attr]-  -- ^ A list of attributes.-  }-  deriving (Functor, Show)

@@ -1,25 +0,0 @@-{- |-Module      : GHC.Eventlog.Live.Span-Description : Representation for spans.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Data.Span (-  IsSpan,-  duration,-) where--import GHC.RTS.Events (Timestamp)-import GHC.Records (HasField)--{- |-A span is any type with a start and end time.--}-type IsSpan s = (HasField "startTimeUnixNano" s Timestamp, HasField "endTimeUnixNano" s Timestamp)--{- |-Determine the duration of a span.--}-duration :: (IsSpan s) => s -> Timestamp-duration s = if s.startTimeUnixNano < s.endTimeUnixNano then s.endTimeUnixNano - s.startTimeUnixNano else 0-{-# INLINEABLE duration #-}

@@ -1,1151 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-{-# OPTIONS_GHC -Wno-name-shadowing #-}--{- |-Module      : GHC.Eventlog.Live.Machine-Description : Machines for processing eventlog data.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Machine (-  -- * Event processing--  -- ** Capability Usage--  -- *** Capability Usage Metrics-  processCapabilityUsageMetrics,--  -- *** Capability Usage Spans-  CapabilityUsageSpan,-  CapabilityUser (..),-  capabilityUser,-  showCapabilityUserCategory,-  processCapabilityUsageSpans,-  processCapabilityUsageSpans',--  -- *** GC Spans-  GCSpan (..),-  processGCSpans,-  processGCSpans',--  -- *** Mutator Spans-  MutatorSpan (..),-  asMutatorSpans,-  asMutatorSpans',-  processMutatorSpans,-  processMutatorSpans',--  -- ** Thread labels-  ThreadLabel (..),-  processThreadLabels,--  -- ** Thread State Spans-  ThreadState (..),-  showThreadStateCategory,-  threadStateStatus,-  threadStateCap,-  ThreadStateSpan (..),-  processThreadStateSpans,-  processThreadStateSpans',--  -- ** Heap events-  processHeapAllocatedData,-  processHeapSizeData,-  processBlocksSizeData,-  processHeapLiveData,-  MemReturnData (..),-  processMemReturnData,-  processHeapProfSampleData,--  -- * Heap profile breakdown-  heapProfBreakdownEitherReader,-  heapProfBreakdownShow,-) where--import Control.Monad (unless, when)-import Control.Monad.IO.Class (MonadIO (..))-import Data.Char (isSpace)-import Data.Either (isLeft)-import Data.Foldable (for_)-import Data.HashMap.Strict (HashMap)-import Data.HashMap.Strict qualified as M-import Data.Hashable (Hashable (..))-import Data.List qualified as L-import Data.Machine (Is (..), PlanT, Process, ProcessT, asParts, await, construct, mapping, repeatedly, yield, (~>))-import Data.Machine.Fanout (fanout)-import Data.Maybe (isNothing, listToMaybe, mapMaybe)-import Data.Text (Text)-import Data.Text qualified as T-import Data.Void (Void)-import Data.Word (Word32, Word64)-import GHC.Eventlog.Live.Data.Attribute (Attr, AttrValue, IsAttrValue (..), (~=))-import GHC.Eventlog.Live.Data.Metric (Metric (..))-import GHC.Eventlog.Live.Data.Span (duration)-import GHC.Eventlog.Live.Internal.Logger (logWarning)-import GHC.Eventlog.Live.Machine.Core (liftRouter)-import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..), setWithStartTime'value, tryGetTimeUnixNano)-import GHC.Eventlog.Live.Verbosity (Verbosity)-import GHC.RTS.Events (Event (..), EventInfo, HeapProfBreakdown (..), ThreadId, ThreadStopStatus (..), Timestamp)-import GHC.RTS.Events qualified as E-import GHC.Records (HasField (..))-import Numeric (showHex)-import Text.ParserCombinators.ReadP (readP_to_S)-import Text.ParserCombinators.ReadP qualified as P-import Text.Printf (printf)-import Text.Read (readMaybe)-import Text.Read.Lex (readHexP)------------------------------------------------------------------------------------ Capability Usage------------------------------------------------------------------------------------ Capability Usage Metrics--{- |-This machine processes t`CapabilityUsageSpan` spans and produces metrics that-contain the duration and category of each such span and each idle period in-between.--}-processCapabilityUsageMetrics ::-  forall m.-  (MonadIO m) =>-  ProcessT m (WithStartTime CapabilityUsageSpan) (Metric Timestamp)-processCapabilityUsageMetrics =-  liftRouter measure spawn- where-  measure :: WithStartTime CapabilityUsageSpan -> Maybe Int-  measure = Just . (.value.cap)--  spawn :: Int -> ProcessT m (WithStartTime CapabilityUsageSpan) (Metric Timestamp)-  spawn cap = construct $ go Nothing-   where-    go ::-      Maybe CapabilityUsageSpan ->-      PlanT (Is (WithStartTime CapabilityUsageSpan)) (Metric Timestamp) m Void-    go mi =-      await >>= \j -> do-        -- If there is a previous span, and...-        for_ mi $ \i ->-          -- ...the end time of the previous span precedes the start time of the current span, then...-          when (i.endTimeUnixNano < j.value.startTimeUnixNano) $-            -- ...yield an idle duration metric.-            yield-              Metric-                { value = j.value.startTimeUnixNano - i.endTimeUnixNano-                , maybeTimeUnixNano = Just i.endTimeUnixNano-                , maybeStartTimeUnixNano = j.maybeStartTimeUnixNano-                , attr = ["cap" ~= cap, "category" ~= ("Idle" :: Text)]-                }-        -- Yield a duration metric for the current span.-        let user = capabilityUser j.value-        yield-          Metric-            { value = duration j.value-            , maybeTimeUnixNano = Just j.value.startTimeUnixNano-            , maybeStartTimeUnixNano = j.maybeStartTimeUnixNano-            , attr = ["cap" ~= cap, "category" ~= showCapabilityUserCategory user, "user" ~= user]-            }-        go (Just j.value)--{- |-The type of process using a capability,-which is either a mutator thread or garbage collection.--}-data CapabilityUser-  = GC-  | Mutator {thread :: !ThreadId}--instance Show CapabilityUser where-  show :: CapabilityUser -> String-  show = \case-    GC -> "GC"-    Mutator{thread} -> show thread--instance IsAttrValue CapabilityUser where-  toAttrValue :: CapabilityUser -> AttrValue-  toAttrValue = toAttrValue . show-  {-# INLINE toAttrValue #-}--{- |-Get the t`CapabilityUser` associated with a t`CapabilityUsageSpan`.--}-capabilityUser :: CapabilityUsageSpan -> CapabilityUser-capabilityUser = either (const GC) (Mutator . (.thread))--{- |-Show the category of a `CapabilityUser` as either @"GC"@ or @"Mutator"@.--}-showCapabilityUserCategory :: CapabilityUser -> Text-showCapabilityUserCategory = \case-  GC{} -> "GC"-  Mutator{} -> "Mutator"------------------------------------------------------------------------------------ Capability Usage Spans--{- |-A t`CapabilityUsageSpan` is either a t`GCSpan` or a t`MutatorSpan`.--}-type CapabilityUsageSpan = Either GCSpan MutatorSpan--instance HasField "startTimeUnixNano" CapabilityUsageSpan Timestamp where-  getField :: CapabilityUsageSpan -> Timestamp-  getField = either (.startTimeUnixNano) (.startTimeUnixNano)--instance HasField "endTimeUnixNano" CapabilityUsageSpan Timestamp where-  getField :: CapabilityUsageSpan -> Timestamp-  getField = either (.endTimeUnixNano) (.endTimeUnixNano)--instance HasField "cap" CapabilityUsageSpan Int where-  getField :: CapabilityUsageSpan -> Int-  getField = either (.cap) (.cap)--{-# SPECIALIZE duration :: CapabilityUsageSpan -> Timestamp #-}--{- |-This machine runs `processGCSpans` and `processMutatorSpans` in parallel and-combines their output.--This is effectively a fanout of `processGCSpans` and `processMutatorSpans`, the-latter of which runs `processThreadStateSpans` internally. If you are running-`processThreadStateSpans` as well, then using `asMutatorSpans` and constructing-the fanout yourself is more efficient.--}-processCapabilityUsageSpans ::-  forall m.-  (MonadIO m) =>-  Verbosity ->-  ProcessT m (WithStartTime Event) (WithStartTime CapabilityUsageSpan)-processCapabilityUsageSpans verbosity =-  processCapabilityUsageSpans' tryGetTimeUnixNano (.value) setWithStartTime'value setWithStartTime'value verbosity-    ~> mapping (either (fmap Left) (fmap Right))--{- |-Generalised version of `processCapabilityUsageSpans` that can be adapted to-work on arbitrary types using a getter and a pair of lenses.--}-processCapabilityUsageSpans' ::-  forall m s t1 t2.-  (MonadIO m) =>-  (s -> Maybe Timestamp) ->-  (s -> Event) ->-  (s -> GCSpan -> t1) ->-  (s -> MutatorSpan -> t2) ->-  Verbosity ->-  ProcessT m s (Either t1 t2)-processCapabilityUsageSpans' timeUnixNano getEvent setGCSpan setMutatorSpan verbosity =-  -- NOTE:-  -- Combining this fanout with an `Either` is risky, because it-  -- has the potential to lose information if both `processGCSpans`-  -- and `processMutatorSpans` yield a value for the same input.-  -- However, this shouldn't ever happen, since the two processors-  -- process disjoint sets of events.-  fanout-    [ processGCSpans' timeUnixNano getEvent setGCSpan verbosity-        ~> mapping Left-    , processMutatorSpans' timeUnixNano getEvent setMutatorSpan verbosity-        ~> mapping Right-    ]------------------------------------------------------------------------------------ GC spans--{- |-A t`GCSpan` represents a segment of time during which the specified capability-ran GC.--}-data GCSpan = GCSpan-  { cap :: !Int-  , startTimeUnixNano :: !Timestamp-  , endTimeUnixNano :: !Timestamp-  }-  deriving (Show)--{-# SPECIALIZE duration :: GCSpan -> Timestamp #-}--{- |-This machine processes `E.StartGC` and `E.EndGC` events to produce t`GCSpan`-values that represent the segments of time a capability spent in GC.--This processor uses the following finite-state automaton:--@-      ┌─(EndGC)───┐-      │           ↓-    ┌→[   Idle    ]─┐-    │               │-(EndGC)         (StartGC)-    │               │-    └─[    GC     ]←┘-      ↑           │-      └─(StartGC)─┘-@--The transition from @GC@ to @Idle@ yields a GC span.--}-processGCSpans ::-  forall m.-  (MonadIO m) =>-  Verbosity ->-  ProcessT m (WithStartTime Event) (WithStartTime GCSpan)-processGCSpans =-  processGCSpans' tryGetTimeUnixNano (.value) setWithStartTime'value--{- |-Generalised version of `processGCSpans` that can be adapted to work on-arbitrary types using a getter and a lens.--}-processGCSpans' ::-  forall m s t.-  (MonadIO m) =>-  (s -> Maybe Timestamp) ->-  (s -> Event) ->-  (s -> GCSpan -> t) ->-  Verbosity ->-  ProcessT m s t-processGCSpans' timeUnixNano getEvent setGCSpan verbosity =-  liftRouter measure spawn- where-  getEventTime = (.evTime) . getEvent-  getEventInfo = (.evSpec) . getEvent-  getEventCap = (.evCap) . getEvent--  measure :: s -> Maybe Int-  measure i-    | accept (getEventInfo i) = getEventCap i-    | otherwise = Nothing-   where-    accept E.StartGC{} = True-    accept E.EndGC{} = True-    accept _ = False--  -- TODO: Rewrite using `MealyT`-  spawn :: Int -> ProcessT m s t-  spawn cap = construct $ go Nothing-   where-    -- The "mi" variable tracks the previous event for this capability, which-    -- is either `Nothing` or `Just` a `StartGC` or a `EndGC` event.-    go :: Maybe s -> PlanT (Is s) t m Void-    go mi =-      -- We start by awaiting the next event "j"...-      await >>= \j -> case getEventInfo j of-        -- If the next event is a `RunThread` event, and...-        E.StartGC{} -> case mi of-          Just i-            -- If the previous event was a `StartGC` event, then...-            | E.StartGC{} <- getEventInfo i ->-                -- ...continue with the oldest event.-                go (Just $ minBy getEventTime i j)-            -- If the previous event was a `EndGC` event, then...-            | E.EndGC{} <- getEventInfo i ->-                -- ...continue with the current event.-                go (Just j)-            -- If the previous event was any other event, then...-            | otherwise -> do-                -- ...emit an error, and...-                logWarning verbosity "processGCSpans" . T.pack $-                  printf-                    "Capability %d: Unsupported trace %s --> %s"-                    cap-                    (showEventInfo (getEventInfo i))-                    (showEventInfo (getEventInfo j))-                -- ...continue with the previous event.-                go (Just i)-          -- If there was no previous event, then...-          Nothing ->-            -- ...continue with the current event.-            go (Just j)-        -- If the next event is a `StopThread` event...-        E.EndGC{} -> case mi of-          Just i-            -- If the previous event was a `StartGC` event, then...-            | E.StartGC{} <- getEventInfo i-            , Just startTimeUnixNano <- timeUnixNano i-            , Just endTimeUnixNano <- timeUnixNano j -> do-                -- ...yield a GC span, and...-                yield . setGCSpan j $ GCSpan{..}-                -- ...continue with the current event.-                go (Just j)-            -- If the previous event was a `EndGC` event, then...-            | E.EndGC{} <- getEventInfo i ->-                -- ...continue with the oldest event.-                go (Just $ minBy getEventTime i j)-          -- If there was no previous event or it was any other event, then...-          _otherwise -> do-            -- ...emit an error, and...-            logWarning verbosity "processGCSpans" . T.pack $-              printf-                "Capability %d: Unsupported trace %s --> %s"-                cap-                (maybe "?" (showEventInfo . getEventInfo) mi)-                (showEventInfo (getEventInfo j))-            -- ...continue with the previous event.-            go mi-        -- If the next event is any other event, ignore it.-        _otherwise -> go mi------------------------------------------------------------------------------------ Mutator spans--{- |-A t`MutatorSpan` represents a segment of time during which the specified-capability ran the specified mutator thread.--}-data MutatorSpan = MutatorSpan-  { cap :: !Int-  , thread :: !ThreadId-  , startTimeUnixNano :: !Timestamp-  , endTimeUnixNano :: !Timestamp-  }-  deriving (Show)--{-# SPECIALIZE duration :: MutatorSpan -> Timestamp #-}--{- |-This machine processes `E.RunThread` and `E.StopThread` events to produce-t`MutatorSpan` values that represent the segments of time a capability spent-executating a mutator.--This processor uses the following finite-state automaton:--@-      ┌─(StopThread[X])─┐-      │                 ↓-    ┌→[      Idle       ]─┐-    │                     │-(StopThread[X])       (RunThread[X])-    │                     │-    └─[   Mutator[X]    ]←┘-      ↑                 │-      └─(RunThread[X])──┘-@--The transition from @Mutator[X]@ to @Idle@ yields a t`MutatorSpan`.-While in the @Mutator[X]@ state, any @RunThread[Y]@ or @StopThread[Y]@ events result in an error.-Furthermore, when a @StopThread[X]@ event with the @ThreadFinished@ status is processed,-the thread @X@ is added to a set of finished threads,-and any further @RunThread[X]@ events for that thread are ignored.-This is done because the GHC RTS frequently emits a @RunThread[X]@ event-immediately after a @StopThread[X]@ event with the @ThreadFinished@ status.--This runs `processThreadStateSpans` internally. If you are also running-`processThreadStateSpans`, then post-composing it with `asMutatorSpans`-is more efficient.--}-processMutatorSpans ::-  forall m.-  (MonadIO m) =>-  Verbosity ->-  ProcessT m (WithStartTime Event) (WithStartTime MutatorSpan)-processMutatorSpans =-  processMutatorSpans' tryGetTimeUnixNano (.value) setWithStartTime'value--{- |-Generalised version of `processMutatorSpans` that can be adapted to work on-arbitrary types using a getter and a lens.--}-processMutatorSpans' ::-  forall m s t.-  (MonadIO m) =>-  (s -> Maybe Timestamp) ->-  (s -> Event) ->-  (s -> MutatorSpan -> t) ->-  Verbosity ->-  ProcessT m s t-processMutatorSpans' timeUnixNano getEvent setMutatorSpan verbosity =-  processThreadStateSpans' timeUnixNano getEvent setThreadStateSpan verbosity ~> asParts- where-  setThreadStateSpan :: s -> ThreadStateSpan -> Maybe t-  setThreadStateSpan s threadStateSpan =-    setMutatorSpan s <$> threadStateSpanToMutatorSpan threadStateSpan--{- |-This machine converts any `Running` t`ThreadStateSpan` to a t`MutatorSpan`.--}-asMutatorSpans ::-  forall m.-  (MonadIO m) =>-  ProcessT m ThreadStateSpan MutatorSpan-asMutatorSpans = asMutatorSpans' id (const id)--{- |-Generalised version of `asMutatorSpans` that can be adapted to work on-arbitrary types using a getter and a lens.--}-asMutatorSpans' ::-  forall m s t.-  (MonadIO m) =>-  (s -> ThreadStateSpan) ->-  (s -> MutatorSpan -> t) ->-  ProcessT m s t-asMutatorSpans' getThreadStateSpan setMutatorSpan = repeatedly go- where-  go =-    await >>= \s -> do-      let threadStateSpan = getThreadStateSpan s-      let maybeMutatorSpan = threadStateSpanToMutatorSpan threadStateSpan-      for_ maybeMutatorSpan $ yield . setMutatorSpan s--{- |-Convert the `Running` t`ThreadStateSpan` to `Just` a t`MutatorSpan`.--}-threadStateSpanToMutatorSpan :: ThreadStateSpan -> Maybe MutatorSpan-threadStateSpanToMutatorSpan ThreadStateSpan{..} =-  case threadState of-    Running{..} -> Just MutatorSpan{..}-    _otherwise -> Nothing--{- |-Internal helper.-Check whether a t`ThreadStopStatus` is equal to `ThreadFinished`.-This is needed because t`ThreadStopStatus` does not define an `Eq` instance.--}-isThreadFinished :: ThreadStopStatus -> Bool-isThreadFinished = \case-  ThreadFinished -> True-  _otherwise -> False--{- |-Internal helper.-Show `EventInfo` in a condensed format suitable for logging.--}-showEventInfo :: EventInfo -> String-showEventInfo = \case-  E.RunThread{thread} -> printf "RunThread{%d}" thread-  E.StopThread{thread, status} -> printf "StopThread{%d,%s}" thread (E.showThreadStopStatus status)-  E.MigrateThread{thread} -> printf "MigrateThread{%d}" thread-  E.StartGC{} -> "StartGC"-  E.EndGC{} -> "EndGC"-  evSpec -> takeWhile (not . isSpace) . show $ evSpec------------------------------------------------------------------------------------ Thread Labels--{- |-The t`ThreadLabel` type represents the association of a label with a thread-starting at a given time.--}-data ThreadLabel-  = ThreadLabel-  { thread :: !ThreadId-  , threadlabel :: !Text-  , startTimeUnixNano :: !Timestamp-  }--{- |-This machine processes `E.ThreadLabel` events and yields t`ThreadLabel` values.--}-processThreadLabels :: Process (WithStartTime Event) ThreadLabel-processThreadLabels = repeatedly go- where-  go =-    await >>= \i -> case i.value.evSpec of-      E.ThreadLabel{..}-        | Just startTimeUnixNano <- tryGetTimeUnixNano i ->-            yield ThreadLabel{..}-      _otherwise -> pure ()------------------------------------------------------------------------------------ Thread State Spans--{- |-The execution states of a mutator thread.--}-data ThreadState-  = Running {cap :: !Int}-  | Blocked {status :: !ThreadStopStatus}-  | Finished-  deriving (Show)--{- |-Pretty-print a thread state as "Running", "Blocked", or "Finished".--}-showThreadStateCategory :: ThreadState -> Text-showThreadStateCategory = \case-  Running{} -> "Running"-  Blocked{} -> "Blocked"-  Finished{} -> "Finished"--{- |-Get the t`ThreadState` status, if the t`ThreadState` is `Blocked`.--}-threadStateStatus :: ThreadState -> Maybe ThreadStopStatus-threadStateStatus = \case-  Running{} -> Nothing-  Blocked{status} -> Just status-  Finished{} -> Nothing--{- |-Get the t`ThreadState` capability, if the `ThreadState` is `Running`.--}-threadStateCap :: ThreadState -> Maybe Int-threadStateCap = \case-  Running{cap} -> Just cap-  Blocked{} -> Nothing-  Finished{} -> Nothing--{- |-A span representing the state of a mutator thread.--}-data ThreadStateSpan-  = ThreadStateSpan-  { thread :: !ThreadId-  , threadState :: !ThreadState-  , startTimeUnixNano :: !Timestamp-  , endTimeUnixNano :: !Timestamp-  }-  deriving (Show)--{-# SPECIALIZE duration :: ThreadStateSpan -> Timestamp #-}--{- |-This machine processes `E.RunThread` and `E.StopThread` events to produce-t`ThreadStateSpan` values that represent segments of time where a thread is-running, blocked, or finished.--This processor uses the following finite-state automaton:--@-      ┌─(StopThread)─┐-      │              ↓-    ┌→[   Blocked    ]─┐-    │                  │-(StopThread)       (RunThread)-    │                  │-    └─[   Running    ]←┘-      ↑              │-      └─(RunThread)──┘-@--The transitions from @Blocked@ to @Blocked@, @Blocked@ to @Running@, and-@Running@ to @Running@ yield a t`ThreadStateSpan`. There are additional-transitions (not pictured) from either state to the final `Finished` state-with a `E.StopThread` event with the `ThreadFinished` status.--}-processThreadStateSpans ::-  (MonadIO m) =>-  Verbosity ->-  ProcessT m (WithStartTime Event) ThreadStateSpan-processThreadStateSpans =-  processThreadStateSpans' tryGetTimeUnixNano (.value) (const id)--{- |-Generalised version of `processThreadStateSpans` that can be adapted to work-on arbitrary types using a getter and a lens.--}-processThreadStateSpans' ::-  forall m s t.-  (MonadIO m) =>-  (s -> Maybe Timestamp) ->-  (s -> Event) ->-  (s -> ThreadStateSpan -> t) ->-  Verbosity ->-  ProcessT m s t-processThreadStateSpans' timeUnixNano getEvent setThreadStateSpan verbosity =-  liftRouter measure spawn- where-  getEventTime = (.evTime) . getEvent-  getEventInfo = (.evSpec) . getEvent-  getEventCap = (.evCap) . getEvent--  measure :: s -> Maybe ThreadId-  measure i = case getEventInfo i of-    E.RunThread{thread} -> Just thread-    E.StopThread{thread} -> Just thread-    _otherwise -> Nothing--  spawn :: ThreadId -> ProcessT m s t-  spawn thread = construct $ go Nothing-   where-    go :: Maybe s -> PlanT (Is s) t m Void-    go mi =-      await >>= \case-        j-          -- If the previous event was a `E.StopThread` event, and...-          | Just E.StopThread{status} <- getEventInfo <$> mi-          , --- ...it has the `ThreadFinished` status, then...-            isThreadFinished status ->-              -- ...ignore the current event.-              go mi-          ---          -- If the current event is a `E.RunThread` event, and...-          | E.RunThread{} <- getEventInfo j-          , -- ...the previous event was a `E.StopThread` event, then...-            Just E.StopThread{status} <- getEventInfo <$> mi-          , -- ...gather the end time of the previous event, and...-            Just startTimeUnixNano <- timeUnixNano =<< mi-          , -- ...gather the start time of the current event, and...-            Just endTimeUnixNano <- timeUnixNano j -> do-              -- ...yield a thread state span, and...-              yield . setThreadStateSpan j $-                ThreadStateSpan{threadState = Blocked status, ..}-              go (Just j)-          ---          -- If the current event is a `E.RunThread` event, and...-          | E.RunThread{} <- getEventInfo j-          , -- ...the previous event was a `E.RunThread` event, then...-            Just E.RunThread{} <- getEventInfo <$> mi -> do-              -- ...keep the oldest event.-              go (Just $ maybe j (minBy getEventTime j) mi)-          ---          -- If the current event is a `E.RunThread` event, and...-          | E.RunThread{} <- getEventInfo j-          , -- ...there is no previous event, then...-            isNothing mi ->-              -- ...keep the current event.-              ---              -- The reason for the additional `isNothing` test is because,-              -- otherwise, this case might silently swallow any `E.StopThread`-              -- events for which `timeUnixNano` gives `Nothing`.-              -- By excluding these, they are forwarded to the catch-all case.-              go (Just j)-          ---          -- If the current event is a `E.StopThread` event, and...-          | E.StopThread{} <- getEventInfo j-          , -- ...the previous event was a `E.StopThread` event, then...-            Just E.StopThread{status} <- getEventInfo <$> mi-          , -- ...gather the end time of the previous event, and...-            Just startTimeUnixNano <- timeUnixNano =<< mi-          , -- ...gather the start time of the current event, and...-            Just endTimeUnixNano <- timeUnixNano j -> do-              -- ...yield a thread state span, and...-              yield . setThreadStateSpan j $-                ThreadStateSpan{threadState = Blocked status, ..}-              -- ...keep the current event.-              ---              -- This causes us to adopt every `E.StopThread` event, until-              -- we hit a `E.StopThread` event with the `ThreadFinished`, at-              -- which point the first clause will cause us to stick with it.-              go (Just j)-          ---          -- If the current event is a `E.StopThread` event, and...-          | E.StopThread{} <- getEventInfo j-          , -- ...the previous event was a `E.RunThread` event, then...-            Just E.RunThread{} <- getEventInfo <$> mi-          , -- ...gather the capability of the `E.RunThread` event, and...-            Just cap <- getEventCap =<< mi-          , -- ...gather the end time of the previous event, and...-            Just startTimeUnixNano <- timeUnixNano =<< mi-          , -- ...gather the start time of the current event, and...-            Just endTimeUnixNano <- timeUnixNano j -> do-              -- ...yield a thread state span, and...-              yield . setThreadStateSpan j $-                ThreadStateSpan{threadState = Running cap, ..}-              -- ...keep the current event.-              go (Just j)-          ---          -- If the current event is any other event, then...-          | otherwise -> do-              -- ...emit an error, and...-              logWarning verbosity "processThreadStateSpans" . T.pack $-                printf-                  "Thread %d: Unexpected event %s"-                  thread-                  (showEventInfo (getEventInfo j))-              ---              -- This case may trigger for any event that isn't `E.RunThread`-              -- or `E.StopThread` and for any `E.StopThread` event that comes-              -- before the first `E.RunThread` event. It may also trigger for-              -- any event for which `timeUnixNano` returns `Nothing`.-              ---              -- ...ignore it.-              go mi------------------------------------------------------------------------------------ Heap events--------------------------------------------------------------------------------------------------------------------------------------------------------------------- HeapAllocated--{- |-This machine processes `E.HeapAllocated` events into metrics.--}-processHeapAllocatedData :: Process (WithStartTime Event) (Metric Word64)-processHeapAllocatedData =-  repeatedly $-    await >>= \case-      i-        | E.HeapAllocated{..} <- i.value.evSpec ->-            yield $-              metric i allocBytes $-                [ "evCap" ~= i.value.evCap-                , "heapCapset" ~= heapCapset-                ]-        | otherwise -> pure ()------------------------------------------------------------------------------------ HeapSize--{- |-This machine processes `E.HeapSize` events into metrics.--}-processHeapSizeData :: Process (WithStartTime Event) (Metric Word64)-processHeapSizeData = repeatedly go- where-  go =-    await >>= \case-      i-        | E.HeapSize{..} <- i.value.evSpec -> do-            yield $-              metric i sizeBytes $-                [ "evCap" ~= i.value.evCap-                , "heapCapset" ~= heapCapset-                ]-        | otherwise -> pure ()------------------------------------------------------------------------------------ BlocksSize--{- |-This machine processes `E.BlocksSize` events into metrics.--}-processBlocksSizeData :: Process (WithStartTime Event) (Metric Word64)-processBlocksSizeData =-  repeatedly $-    await >>= \case-      i-        | E.BlocksSize{..} <- i.value.evSpec -> do-            yield $-              metric i blocksSize $-                [ "evCap" ~= i.value.evCap-                , "heapCapset" ~= heapCapset-                ]-        | otherwise -> pure ()------------------------------------------------------------------------------------ HeapLive--{- |-This machine processes `E.HeapLive` events into metrics.--}-processHeapLiveData :: Process (WithStartTime Event) (Metric Word64)-processHeapLiveData =-  repeatedly $-    await >>= \case-      i-        | E.HeapLive{..} <- i.value.evSpec -> do-            yield $-              metric i liveBytes $-                [ "evCap" ~= i.value.evCap-                , "heapCapset" ~= heapCapset-                ]-        | otherwise -> pure ()------------------------------------------------------------------------------------ MemReturn--{- |-The type of data associated with a `E.MemReturn` event.--}-data MemReturnData = MemReturnData-  { current :: !Word32-  -- ^ The number of megablocks currently allocated.-  , needed :: !Word32-  -- ^ The number of megablocks currently needed.-  , returned :: !Word32-  -- ^ The number of megablocks currently being returned to the OS.-  }--{- |-This machine processes `E.MemReturn` events into metrics.--}-processMemReturnData :: Process (WithStartTime Event) (Metric MemReturnData)-processMemReturnData =-  repeatedly $-    await >>= \case-      i-        | E.MemReturn{..} <- i.value.evSpec -> do-            yield $-              metric i MemReturnData{..} $-                [ "evCap" ~= i.value.evCap-                , "heapCapset" ~= heapCapset-                ]-        | otherwise -> pure ()------------------------------------------------------------------------------------ HeapProfSample--{- |-Internal helper.-The type of info table pointers.--}-newtype InfoTablePtr = InfoTablePtr Word64-  deriving newtype (Eq, Hashable, Ord)--instance Show InfoTablePtr where-  showsPrec :: Int -> InfoTablePtr -> ShowS-  showsPrec _ (InfoTablePtr ptr) =-    showString "0x" . showHex ptr--instance Read InfoTablePtr where-  readsPrec :: Int -> ReadS InfoTablePtr-  readsPrec _ = readP_to_S (InfoTablePtr <$> (P.string "0x" *> readHexP))--{- |-Internal helper.-The type of an info table entry, as produced by the `E.InfoTableProv` event.--}-data InfoTable = InfoTable-  { infoTablePtr :: InfoTablePtr-  , infoTableName :: Text-  , infoTableClosureDesc :: Int-  , infoTableTyDesc :: Text-  , infoTableLabel :: Text-  , infoTableModule :: Text-  , infoTableSrcLoc :: Text-  }-  deriving (Show)--{- |-Internal helper.-The type of the state kept by `processHeapProfSampleData`.--}-data HeapProfSampleState = HeapProfSampleState-  { eitherShouldWarnOrHeapProfBreakdown :: Either Bool HeapProfBreakdown-  , infoTableMap :: HashMap InfoTablePtr InfoTable-  , heapProfSampleEraStack :: [Word64]-  }-  deriving (Show)--{- |-Internal helper.-Decides whether or not `processHeapProfSampleData` should track info tables.-We track info tables until (1) we learn that the RTS is not run with @-hi@,-or (2) we see the first heap profiling sample and don't yet know for sure-that the RTS is run with @-hi@.--}-shouldTrackInfoTableMap :: Either Bool HeapProfBreakdown -> Bool-shouldTrackInfoTableMap (Left _shouldWarn) = True-shouldTrackInfoTableMap (Right HeapProfBreakdownInfoTable) = True-shouldTrackInfoTableMap _ = False--{- |-Internal helper.-Checks whether a `HeapProfBreakdown` is `HeapProfBreakdownInfoTable`.-This is needed because the ghc-events package does not define an `Eq`-instance for the `HeapProfBreakdown` type.--}-isHeapProfBreakdownInfoTable :: HeapProfBreakdown -> Bool-isHeapProfBreakdownInfoTable HeapProfBreakdownInfoTable = True-isHeapProfBreakdownInfoTable _ = False--{- |-This machine processes `E.HeapProfSampleString` events into metrics.-Furthermore, it processes the `E.HeapProfBegin` and `E.ProgramArgs` events-to determine the heap profile breakdown, processes `E.InfoTableProv` events to-build an info table map, if necessary, and processes `E.HeapProfSampleBegin`-and `E.HeapProfSampleEnd` events to maintain an era stack.--}-processHeapProfSampleData ::-  (MonadIO m) =>-  Verbosity ->-  Maybe HeapProfBreakdown ->-  ProcessT m (WithStartTime Event) (Metric Word64)-processHeapProfSampleData verbosityThreshold maybeHeapProfBreakdown =-  construct $-    go-      HeapProfSampleState-        { eitherShouldWarnOrHeapProfBreakdown = maybe (Left True) Right maybeHeapProfBreakdown-        , infoTableMap = mempty-        , heapProfSampleEraStack = mempty-        }- where-  -- go :: HeapProfSampleState -> PlanT (Is (WithStartTime Event)) (Metric Word64) m Void-  go st@HeapProfSampleState{..} = do-    await >>= \i -> case i.value.evSpec of-      -- Announces the heap profile breakdown, amongst other things.-      -- This event is only emitted for code compiled with GHC >=9.14.-      E.HeapProfBegin{..}-        | isLeft eitherShouldWarnOrHeapProfBreakdown ->-            go st{eitherShouldWarnOrHeapProfBreakdown = Right heapProfBreakdown}-      -- Announces the arguments with which the program was called.-      -- This *may* include RTS options, which can be used to determine the-      -- heap profile breakdown for code compiled with GHC <9.14.-      E.ProgramArgs{..}-        | isLeft eitherShouldWarnOrHeapProfBreakdown-        , Just heapProfBreakdown <- findHeapProfBreakdown args ->-            go st{eitherShouldWarnOrHeapProfBreakdown = Right heapProfBreakdown}-      -- Announces an info table entry.-      E.InfoTableProv{..}-        | shouldTrackInfoTableMap eitherShouldWarnOrHeapProfBreakdown -> do-            let infoTablePtr = InfoTablePtr itInfo-                infoTable =-                  InfoTable-                    { infoTablePtr = infoTablePtr-                    , infoTableName = itTableName-                    , infoTableClosureDesc = itClosureDesc-                    , infoTableTyDesc = itTyDesc-                    , infoTableLabel = itLabel-                    , infoTableModule = itModule-                    , infoTableSrcLoc = itSrcLoc-                    }-            go st{infoTableMap = M.insert infoTablePtr infoTable infoTableMap}-      -- Announces the beginning of a heap profile sample.-      E.HeapProfSampleBegin{..} ->-        go st{heapProfSampleEraStack = heapProfSampleEra : heapProfSampleEraStack}-      -- Announces the end of a heap profile sample.-      E.HeapProfSampleEnd{..} ->-        case L.uncons heapProfSampleEraStack of-          Nothing -> do-            logWarning verbosityThreshold "processHeapProfSampleData" . T.pack $-              printf-                "Eventlog closed era %d, but there is no current era."-                heapProfSampleEra-            go st-          Just (currentEra, heapProfSampleEraStack') -> do-            unless (currentEra == heapProfSampleEra) $-              logWarning verbosityThreshold "processHeapProfSampleData" . T.pack $-                printf-                  "Eventlog closed era %d, but the current era is era %d."-                  heapProfSampleEra-                  currentEra-            go st{heapProfSampleEraStack = heapProfSampleEraStack'}-      -- Announces a heap profile sample.-      E.HeapProfSampleString{..}-        -- If there is no heap profile breakdown, issue a warning, then disable warnings.-        | Left True <- eitherShouldWarnOrHeapProfBreakdown -> do-            logWarning verbosityThreshold "processHeapProfSampleData" $-              "Cannot infer heap profile breakdown.\n\-              \         If your binary was compiled with a GHC version prior to 9.14,\n\-              \         you must also pass the heap profile type to this executable.\n\-              \         See: https://gitlab.haskell.org/ghc/ghc/-/commit/76d392a"-            go st{eitherShouldWarnOrHeapProfBreakdown = Left False, infoTableMap = mempty}-        -- If the heap profile breakdown is biographical, issue a warning, then disable warnings.-        | Right HeapProfBreakdownBiography <- eitherShouldWarnOrHeapProfBreakdown -> do-            logWarning verbosityThreshold "processHeapProfSampleData" . T.pack $-              printf-                "Unsupported heap profile breakdown %s"-                (heapProfBreakdownShow HeapProfBreakdownBiography)-            go st{eitherShouldWarnOrHeapProfBreakdown = Left False, infoTableMap = mempty}-        -- If there is a heap profile breakdown, handle it appropriately.-        | Right heapProfBreakdown <- eitherShouldWarnOrHeapProfBreakdown -> do-            -- If the heap profile breakdown is by info table, add the info table.-            let maybeInfoTable-                  | isHeapProfBreakdownInfoTable heapProfBreakdown = do-                      !infoTablePtr <- readMaybe (T.unpack heapProfLabel)-                      M.lookup infoTablePtr infoTableMap-                  | otherwise = Nothing-            yield $-              metric i heapProfResidency $-                [ "evCap" ~= i.value.evCap-                , "heapProfBreakdown" ~= heapProfBreakdownShow heapProfBreakdown-                , "heapProfId" ~= heapProfId-                , "heapProfLabel" ~= heapProfLabel-                , "heapProfSampleEra" ~= (fst <$> L.uncons heapProfSampleEraStack)-                , "infoTableName" ~= fmap (.infoTableName) maybeInfoTable-                , "infoTableClosureDesc" ~= fmap (.infoTableClosureDesc) maybeInfoTable-                , "infoTableTyDesc" ~= fmap (.infoTableTyDesc) maybeInfoTable-                , "infoTableLabel" ~= fmap (.infoTableLabel) maybeInfoTable-                , "infoTableModule" ~= fmap (.infoTableModule) maybeInfoTable-                , "infoTableSrcLoc" ~= fmap (.infoTableSrcLoc) maybeInfoTable-                ]-            go $ if isHeapProfBreakdownInfoTable heapProfBreakdown then st else st{infoTableMap = mempty}-      _otherwise -> go st--{- |-Parses the `HeapProfBreakdown` command-line arguments:--> heapProfBreakdownEitherReader "T" == Left HeapProfBreakdownClosureType-> heapProfBreakdownEitherReader "c" == Left HeapProfBreakdownCostCentre-> heapProfBreakdownEitherReader "m" == Left HeapProfBreakdownModule-> heapProfBreakdownEitherReader "d" == Left HeapProfBreakdownClosureDescr-> heapProfBreakdownEitherReader "y" == Left HeapProfBreakdownTypeDescr-> heapProfBreakdownEitherReader "e" == Left HeapProfBreakdownEra-> heapProfBreakdownEitherReader "r" == Left HeapProfBreakdownRetainer-> heapProfBreakdownEitherReader "b" == Left HeapProfBreakdownBiography-> heapProfBreakdownEitherReader "i" == Left HeapProfBreakdownInfoTable--}-heapProfBreakdownEitherReader :: String -> Either String HeapProfBreakdown-heapProfBreakdownEitherReader =-  \case-    "T" -> Right HeapProfBreakdownClosureType-    "c" -> Right HeapProfBreakdownCostCentre-    "m" -> Right HeapProfBreakdownModule-    "d" -> Right HeapProfBreakdownClosureDescr-    "y" -> Right HeapProfBreakdownTypeDescr-    "e" -> Right HeapProfBreakdownEra-    "r" -> Right HeapProfBreakdownRetainer-    "b" -> Right HeapProfBreakdownBiography-    "i" -> Right HeapProfBreakdownInfoTable-    str -> Left $ "Unsupported heap profile breakdown -h" <> str--{- |-Shows a `HeapProfBreakdown` as its corresponding command-line flag:--> heapProfBreakdownShow HeapProfBreakdownClosureType == "-hT"-> heapProfBreakdownShow HeapProfBreakdownCostCentre == "-hc"-> heapProfBreakdownShow HeapProfBreakdownModule == "-hm"-> heapProfBreakdownShow HeapProfBreakdownClosureDescr == "-hd"-> heapProfBreakdownShow HeapProfBreakdownTypeDescr == "-hy"-> heapProfBreakdownShow HeapProfBreakdownEra == "-he"-> heapProfBreakdownShow HeapProfBreakdownRetainer == "-hr"-> heapProfBreakdownShow HeapProfBreakdownBiography == "-hb"-> heapProfBreakdownShow HeapProfBreakdownInfoTable == "-hi"--}-heapProfBreakdownShow :: HeapProfBreakdown -> String-heapProfBreakdownShow =-  ("-h" <>) . \case-    HeapProfBreakdownClosureType -> "T"-    HeapProfBreakdownCostCentre -> "c"-    HeapProfBreakdownModule -> "m"-    HeapProfBreakdownClosureDescr -> "d"-    HeapProfBreakdownTypeDescr -> "y"-    HeapProfBreakdownEra -> "e"-    HeapProfBreakdownRetainer -> "r"-    HeapProfBreakdownBiography -> "b"-    HeapProfBreakdownInfoTable -> "i"--{- |-Internal helper.-Determine the `HeapProfBreakdown` from the list of program arguments.--__Warning__: This scan is not fully correct. It merely scans for the presence-of arguments that, as a whole, parse with `heapProfBreakdownEitherReader`.-It does not handle @-with-rtsopts@ and does not restrict its search to those-arguments between @+RTS@ and @-RTS@ tags.--}-findHeapProfBreakdown :: [Text] -> Maybe HeapProfBreakdown-findHeapProfBreakdown = listToMaybe . mapMaybe parseHeapProfBreakdown- where-  parseHeapProfBreakdown :: Text -> Maybe HeapProfBreakdown-  parseHeapProfBreakdown arg-    | "-h" `T.isPrefixOf` arg =-        either (const Nothing) Just-          . heapProfBreakdownEitherReader-          . T.unpack-          . T.drop 2-          $ arg-    | otherwise = Nothing------------------------------------------------------------------------------------ Decoding-------------------------------------------------------------------------------------------------------------------------------------------------------------------- Internal Helpers----------------------------------------------------------------------------------{- |-Internal helper. Construct a t`Metric` from an event with a start time-(t`WithStartTime` t`Event`), together with the measurement and any attributes.-This is a smart constructor that pulls the various timestamps out of the event.--}-metric ::-  WithStartTime Event ->-  v ->-  [Attr] ->-  Metric v-metric i v attr =-  Metric-    { value = v-    , maybeTimeUnixNano = tryGetTimeUnixNano i-    , maybeStartTimeUnixNano = i.maybeStartTimeUnixNano-    , attr = attr-    }--{- |-Internal helper. Return the minimal value by some projection.--}-minBy :: (Ord b) => (a -> b) -> a -> a -> a-minBy f x y = if f x < f y then x else y

@@ -1,494 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}--{- |-Module      : GHC.Eventlog.Live.Machine.Core-Description : Core machines for processing data in batches.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Machine.Core (-  -- * Ticks-  Tick (..),-  batchByTick,-  batchToTick,-  batchListToTick,-  batchByTickList,-  liftTick,-  dropTick,-  onlyTick,-  liftBatch,--  -- * Debug-  counterBy,-  counterByTick,--  -- * Routers-  liftRouter,--  -- * Event sorting-  sortByBatch,-  sortByBatchTick,--  -- * Delimiting-  between,-  delimit,--  -- * Validation-  validateInput,-  validateOrder,-) where--import Control.Monad (when)-import Control.Monad.IO.Class (MonadIO (..))-import Data.DList qualified as D-import Data.Foldable (for_)-import Data.Function (on)-import Data.Functor ((<&>))-import Data.HashMap.Strict (HashMap)-import Data.HashMap.Strict qualified as M-import Data.Hashable (Hashable (..))-import Data.List qualified as L-import Data.Machine (Is (..), MachineT (..), Moore (..), PlanT, Process, ProcessT, Step (..), await, construct, encased, mapping, repeatedly, starve, stopped, yield, (~>))-import Data.Maybe (fromMaybe)-import Data.Semigroup (Max (..))-import Data.Text (Text)-import Data.Text qualified as T-import GHC.Eventlog.Live.Internal.Logger (logDebug, logError, logWarning)-import GHC.Eventlog.Live.Verbosity (Verbosity, verbosityDebug, verbosityError, verbosityWarning)-import GHC.RTS.Events (Event (..), Timestamp)-import GHC.RTS.Events qualified as E-import Text.Printf (printf)------------------------------------------------------------------------------------ Ticks----------------------------------------------------------------------------------{- |-The type of data on a stream of items and ticks.--The t`Tick` type is isomorphic to `Maybe` modulo strictness,-but with the caveat that v`Tick` does not represent failure.--}-data Tick a = Item !a | Tick-  deriving (Eq, Functor, Foldable, Traversable, Show)--{- |-This machine batches all items between two ticks into a list.--}-batchByTickList :: Process (Tick a) [a]-batchByTickList =-  mapping (fmap D.singleton)-    ~> batchByTick-    ~> mapping D.toList--{- |-Generalised version of `batchByTickList`.--}-batchByTick ::-  forall m a.-  (Monad m, Monoid a) =>-  ProcessT m (Tick a) a-batchByTick = batchByTickWith mempty- where-  batchByTickWith :: a -> MachineT m (Is (Tick a)) a-  batchByTickWith acc = MachineT $ pure $ Await onNext Refl onStop-   where-    onNext :: Tick a -> MachineT m (Is (Tick a)) a-    onNext = \case-      Item a -> batchByTickWith (a <> acc)-      Tick -> MachineT $ pure $ Yield acc batchByTick-    onStop :: MachineT m (Is (Tick a)) a-    onStop = MachineT $ pure $ Yield acc stopped--{- |-This machine streams a list of items into a series of items-separated by ticks.--}-batchListToTick :: Process [a] (Tick a)-batchListToTick = batchToTick--{- |-Generalised version of `batchListToTick`.--}-batchToTick :: (Foldable f) => Process (f a) (Tick a)-batchToTick = repeatedly go- where-  go = await >>= \xs -> for_ xs (yield . Item) >> yield Tick--{- |-This machine drops all ticks.--}-dropTick :: Process (Tick a) a-dropTick =-  repeatedly $-    await >>= \case-      Item a -> yield a-      Tick -> pure ()--{- |-This machine drops all items.--}-onlyTick :: Process (Tick a) ()-onlyTick =-  repeatedly $-    await >>= \case-      Tick -> yield ()-      Item{} -> pure ()------------------------------------------------------------------------------------ Machine combinators----------------------------------------------------------------------------------{- |-This machine counts the number of inputs it received,-using the given function, and logs this value.--}-counterBy ::-  forall m a x.-  (MonadIO m) =>-  Verbosity ->-  Text ->-  (a -> Word) ->-  ProcessT m a x-counterBy verbosity label count-  | verbosityDebug >= verbosity = repeatedly go-  | otherwise = stopped- where-  go :: PlanT (Is a) x m ()-  go =-    await >>= \a ->-      logDebug verbosity "counterBy" ("saw " <> T.pack (show (count a)) <> " " <> label)--{- |-This machine counts the number of inputs it received,-and logs this value on every tick.--}-counterByTick ::-  forall m a x.-  (MonadIO m) =>-  Verbosity ->-  Text ->-  ProcessT m (Tick a) x-counterByTick verbosity label-  | verbosityDebug >= verbosity = construct $ go 0-  | otherwise = stopped- where-  go :: Word -> PlanT (Is (Tick a)) x m ()-  go count =-    await >>= \case-      Item _ -> go (count + 1)-      Tick -> logDebug verbosity "counterByTick" ("saw " <> T.pack (show count) <> " " <> label) >> go 0------------------------------------------------------------------------------------ Machine combinators--------------------------------------------------------------------------------------------------------------------------------------------------------------------- Lift a machine to a machine that passes on ticks unchanged--{- |-Lift a machine to a machine that passes on ticks unchanged.--Constructs the following machine:--@-           ┌─(if Tick)────────────────────┐-  [ Tick a ]                              [ Tick b ]-           └─(if Item)─( ProcessT m a b )─┘-@--}-liftTick ::-  (Monad m) =>-  ProcessT m a b ->-  ProcessT m (Tick a) (Tick b)-liftTick m =-  MachineT $-    runMachineT m <&> \case-      Stop ->-        Stop-      Yield o k ->-        Yield (Item o) (liftTick k)-      Await (onNext :: t -> ProcessT m a b) Refl onStop ->-        await'-       where-        await' = Await onNext' Refl onStop'-         where-          onNext' :: Tick a -> ProcessT m (Tick a) (Tick b)-          onNext' = \case-            Tick ->-              MachineT . pure . Yield Tick $-                MachineT . pure $-                  await'-            Item a -> liftTick (onNext a)-          onStop' :: ProcessT m (Tick a) (Tick b)-          onStop' = liftTick onStop------------------------------------------------------------------------------------- Lift a machine to a machine that operates on batches--{- |-Lift a machine that processes @a@s into @b@s to a machine that processes-batches of @a@s into batches of @b@s.--}-liftBatch ::-  forall m a b.-  (Monad m) =>-  ProcessT m a b ->-  ProcessT m [a] [b]-liftBatch = MachineT . running [] []- where-  -- The parent machine is running the child machine with the current batch.-  running :: [a] -> [b] -> ProcessT m a b -> m (Step (Is [a]) [b] (ProcessT m [a] [b]))-  running as bs m =-    runMachineT m >>= \case-      Stop ->-        pure Stop-      Yield b k ->-        running as (b : bs) k-      Await (onNext :: t -> ProcessT m a b) Refl onStop ->-        pure $ Yield (reverse bs) $ MachineT $ awaiting as onNext onStop--  -- The parent machine is awaiting new input.-  awaiting :: [a] -> (a -> ProcessT m a b) -> ProcessT m a b -> m (Step (Is [a]) [b] (ProcessT m [a] [b]))-  awaiting (a : as) onNext _onStop = running as [] $ onNext a-  awaiting [] onNext onStop = pure $ Await onNext' Refl onStop'-   where-    onNext' :: [a] -> ProcessT m [a] [b]-    onNext' as = MachineT $ awaiting as onNext onStop-    onStop' :: ProcessT m [a] [b]-    onStop' = exhausting onStop--  -- The parent machine is exhausting the child machine to gather its output.-  exhausting :: ProcessT m a b -> ProcessT m x [b]-  exhausting = MachineT . go []-   where-    go :: [b] -> ProcessT m a b -> m (Step (Is x) [b] (ProcessT m x [b]))-    go bs m =-      runMachineT m >>= \case-        Stop ->-          pure Stop-        Yield b k ->-          go (b : bs) k-        Await _onNext _Refl onStop ->-          pure $ Yield (reverse bs) $ MachineT $ go [] onStop------------------------------------------------------------------------------------- Construct a processor that spawns a separate child processor for each measure--{- |-Spawn a copy of a machine for each "measure".--Constructs the following machine:--@-    ┌─────(if measure == k0)─( ProcessT m a b )────┐-  [ a ] ──(if measure == ..)─( ProcessT m a b )─ [ b ]-    └─────(if measure == kN)─( ProcessT m a b )────┘-@--__Warning:__ The router does not currently garbage-collect terminated child processors.--}-liftRouter ::-  forall m k a b.-  (MonadIO m, Hashable k) =>-  -- | Function to measure.-  (a -> Maybe k) ->-  -- | Function to spawn child processors.-  (k -> ProcessT m a b) ->-  ProcessT m a b-liftRouter measure spawn = awaiting M.empty- where-  awaiting :: HashMap k (ProcessT m a b) -> ProcessT m a b-  awaiting st = MachineT . pure $ Await onNext Refl onStop-   where-    onNext :: a -> MachineT m (Is a) b-    onNext a = case measure a of-      Nothing -> awaiting st-      Just k -> provideThen a m $ \m' -> awaiting (M.insert k m' st)-       where-        m = fromMaybe (spawn k) (M.lookup k st)-    onStop :: MachineT m (Is a) b-    onStop = foldr starve stopped (M.elems st)--  provideThen :: a -> ProcessT m a b -> (ProcessT m a b -> ProcessT m a b) -> ProcessT m a b-  provideThen a m k =-    MachineT $-      runMachineT m >>= \case-        Stop -> runMachineT (k stopped)-        Yield o m' -> pure (Yield o (provideThen a m' k))-        Await onNext Refl _onStop -> runMachineT (exhaustThen (onNext a) k)--  exhaustThen :: ProcessT m a b -> (ProcessT m a b -> ProcessT m a b) -> ProcessT m a b-  exhaustThen m k =-    MachineT $-      runMachineT m >>= \case-        Yield o m' -> pure (Yield o (k m'))-        m' -> runMachineT (k (encased m'))------------------------------------------------------------------------------------ Event stream sorting----------------------------------------------------------------------------------{- |-Reorder events respecting ticks.--This machine caches two batches worth of events, sorts them together,-and then yields only those events whose timestamp is less than or equal-to the maximum of the first batch.--}-sortByBatch ::-  forall m a.-  (Monad m) =>-  (a -> Timestamp) ->-  ProcessT m [a] [a]-sortByBatch timestamp = sortByBatchWith Nothing- where-  sortByBatchWith :: Maybe [a] -> ProcessT m [a] [a]-  sortByBatchWith = \case-    Nothing -> MachineT $ pure $ Await onNext Refl onStop-     where-      onNext :: [a] -> ProcessT m [a] [a]-      onNext new = sortByBatchWith (Just sortedNew)-       where-        sortedNew = sortByTime new-      onStop :: ProcessT m [a] [a]-      onStop = stopped-    Just sortedOld -> MachineT $ pure $ Await onNext Refl onStop-     where-      onNext :: [a] -> ProcessT m [a] [a]-      onNext new-        | null sortedOld = sortByBatchWith $ Just sortedNew-        | otherwise = MachineT $ pure $ Yield sortedBeforeCutoff $ sortByBatchWith $ Just sortedAfterCutoff-       where-        -- NOTE: use of partial @maximum@ is guarded by the check @null old@.-        cutoff = getMax (foldMap (Max . timestamp) sortedOld)-        sortedNew = sortByTime new-        sorted = joinByTime sortedOld sortedNew-        (sortedBeforeCutoff, sortedAfterCutoff) = L.partition ((<= cutoff) . timestamp) sorted-      onStop :: ProcessT m [a] [a]-      onStop = MachineT $ pure $ Yield sortedOld $ stopped--  -- compByTime :: a -> a -> Ordering-  compByTime = compare `on` timestamp--  -- sortByTime :: [a] -> [a]-  sortByTime = L.sortBy compByTime--  -- joinByTime :: [a] -> [a] -> [a]-  joinByTime [] ys = ys-  joinByTime xs [] = xs-  joinByTime (x : xs) (y : ys) = case compByTime x y of-    LT -> x : joinByTime xs (y : ys)-    _ -> y : joinByTime (x : xs) ys--{- |-Variant of `sortByBatch` that operates on streams of items and ticks.--}-sortByBatchTick :: (a -> Timestamp) -> Process (Tick a) (Tick a)-sortByBatchTick timestamp =-  mapping (fmap (: [])) ~> batchByTick ~> sortByBatch timestamp ~> batchListToTick------------------------------------------------------------------------------------ Filtering semaphores----------------------------------------------------------------------------------{- | A simple delimiting t'Moore' machine,-which is opened by one constant marker and closed by the other one.--}-between :: Text -> Text -> Moore Text Bool-between x y = open- where-  open = Moore False open' where open' x' = if x == x' then close else open-  close = Moore True close' where close' y' = if y == y' then end else close-  end = Moore False (const end)---- | Delimit the event process.-delimit :: (Monad m) => Moore Text Bool -> ProcessT m Event Event-delimit = construct . go- where-  go :: (Monad m) => Moore Text Bool -> PlanT (Is Event) Event m ()-  go mm@(Moore s next) = do-    e <- await-    case evSpec e of-      -- on marker step the moore machine.-      E.UserMarker m -> do-        let mm'@(Moore s' _) = next m-        -- if current or next state is open (== True), emit the marker.-        when (s || s') $ yield e-        go mm'--      -- for other events, emit if the state is open.-      _ -> do-        when s $ yield e-        go mm------------------------------------------------------------------------------------ Validation----------------------------------------------------------------------------------{- |-This machine validates that there is some input.--If no input is encountered after the given number of ticks, the machine prints-a warning that directs the user to check that the @-l@ flag was set correctly.--}-validateInput ::-  (MonadIO m) =>-  Verbosity ->-  Int ->-  ProcessT m (Tick a) x-validateInput verbosity ticks-  | verbosityWarning >= verbosity = construct $ start ticks-  | otherwise = stopped- where-  start remaining-    | remaining <= 0 = liftIO $ do-        logWarning verbosity "validateInput" . T.pack $-          printf-            "No input after %d ticks. Did you pass -l to the GHC RTS?"-            ticks-    | otherwise = do-        logDebug verbosity "validateInput" $-          T.pack (show remaining) <> " ticks remaining."-        await >>= \case-          Item{} -> do-            logDebug verbosity "validateInput" "Received item."-          Tick -> do-            logDebug verbosity "validateInput" "Received tick."-            start (pred remaining)--{- |-This machine validates that the inputs are received in order.--If an out-of-order input is encountered, the machine prints an error message-that directs the user to check that the @--eventlog-flush-interval@ and the-@--batch-interval@ flags are set correctly.--}-validateOrder ::-  (MonadIO m, Show a) =>-  Verbosity ->-  (a -> Timestamp) ->-  ProcessT m a x-validateOrder verbosity timestamp-  | verbosityError >= verbosity = construct $ start Nothing-  | otherwise = stopped- where-  start maybeOld =-    await >>= \new ->-      case maybeOld of-        Just old-          | timestamp new < timestamp old -> do-              logError verbosity "validateOrder" . T.pack $-                "Encountered two out-of-order inputs.\n\-                \Did you pass --eventlog-flush-interval to the GHC RTS?\n\-                \Did you set --batch-interval to be at least as big as the value of --eventlog-flush-interval?"-              logDebug verbosity "validateOrder" . T.pack $-                printf-                  "Out-of-order inputs:\n\-                  \- %s\n\-                  \- %s"-                  (show old)-                  (show new)-              pure ()-        _otherwise -> do-          start (Just new)

@@ -1,49 +0,0 @@-{- |-Module      : GHC.Eventlog.Live.Machine.Decoder-Description : Machines for processing eventlog data.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Machine.Decoder (-  -- * Event decoding-  DecodeError (..),-  decodeEvent,-  decodeEventBatch,-) where--import Control.Exception (Exception, throwIO)-import Control.Monad.IO.Class (MonadIO (..))-import Data.ByteString qualified as BS-import Data.Machine (Is, PlanT, ProcessT, await, construct, yield)-import GHC.Eventlog.Live.Machine.Core (Tick (..), liftTick)-import GHC.RTS.Events (Event)-import GHC.RTS.Events.Incremental (Decoder (..), decodeEventLog)------------------------------------------------------------------------------------ Decoding events--{- |-Parse t'Event's from a stream of 'BS.ByteString' chunks with ticks.--Throws a t'DecodeError' on error.--}-decodeEvent :: (MonadIO m) => ProcessT m BS.ByteString Event-decodeEvent = construct $ loop decodeEventLog- where-  loop :: (MonadIO m) => Decoder a -> PlanT (Is BS.ByteString) a m ()-  loop Done{} = pure ()-  loop (Consume k) = await >>= \chunk -> loop (k chunk)-  loop (Produce a d') = yield a >> loop d'-  loop (Error _ err) = liftIO $ throwIO $ DecodeError err--{- |-Parse 'Event's from a stream of 'BS.ByteString' chunks with ticks.--Throws 'DecodeError' on error.--}-decodeEventBatch :: (MonadIO m) => ProcessT m (Tick BS.ByteString) (Tick Event)-decodeEventBatch = liftTick decodeEvent--newtype DecodeError = DecodeError String deriving (Show)--instance Exception DecodeError

@@ -1,42 +0,0 @@-{- |-Module      : GHC.Eventlog.Live.Machine.Sink-Description : Machines for processing eventlog data.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Machine.Sink (-  -- * Eventlog file sink-  fileSink,-  fileSinkBatch,-) where--import Control.Monad.IO.Class (MonadIO (..))-import Data.ByteString qualified as BS-import Data.Machine (ProcessT, await, repeatedly, (~>))-import Data.Void (Void)-import GHC.Eventlog.Live.Machine.Core (Tick (..), dropTick)-import System.IO (Handle)------------------------------------------------------------------------------------ Log file sink--{- |-File sink for optional eventlog log file.--}-fileSink ::-  (MonadIO m) =>-  Handle ->-  ProcessT m BS.ByteString Void-fileSink handle = repeatedly $ await >>= liftIO . BS.hPut handle------------------------------------------------------------------------------------ Log file sink with batches--{- |-File sink for optional eventlog log file.--}-fileSinkBatch ::-  (MonadIO m) =>-  Handle ->-  ProcessT m (Tick BS.ByteString) Void-fileSinkBatch handle = dropTick ~> fileSink handle

@@ -1,144 +0,0 @@-{- |-Module      : GHC.Eventlog.Live.Machine.Source-Description : Machines for processing eventlog data.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Machine.Source (-  -- * Eventlog source-  sourceHandleWait,-  sourceHandleBatch,-  defaultChunkSizeBytes,-) where--import Control.Exception (catch, throwIO)-import Control.Monad.IO.Class (MonadIO (..))-import Data.ByteString qualified as BS-import Data.Function (fix)-import Data.Machine (MachineT (..), construct, yield)-import Data.Word (Word64)-import GHC.Clock (getMonotonicTimeNSec)-import GHC.Eventlog.Live.Machine.Core (Tick (..))-import System.IO (Handle, hWaitForInput)-import System.IO.Error (isEOFError)------------------------------------------------------------------------------------ Socket source--{- |-A source which reads chunks from a `Handle`.-When an input is available, it yields an v`Item`.-When the timeout is reached, it yields a v`Tick`.--}-sourceHandleWait ::-  (MonadIO m) =>-  -- | The wait timeout in milliseconds.-  Int ->-  -- | The number of bytes to read.-  Int ->-  -- | The eventlog socket handle.-  Handle ->-  MachineT m k (Tick BS.ByteString)-sourceHandleWait timeoutMilli chunkSizeBytes handle =-  construct $ fix $ \loop -> do-    ready <- liftIO $ hWaitForInput' handle timeoutMilli-    case ready of-      Ready -> do-        bs <- liftIO $ BS.hGetSome handle chunkSizeBytes-        yield (Item bs)-        loop-      NotReady -> do-        yield Tick-        loop-      EOF ->-        pure ()------------------------------------------------------------------------------------ Socket source with batches--{- |-A source which reads chunks from a `Handle`.-When input is available, it yields an v`Item`.-It yields a v`Tick` at each increment of the batch interval.--}-sourceHandleBatch ::-  (MonadIO m) =>-  -- | The batch interval in milliseconds.-  Int ->-  -- | The number of bytes to read.-  Int ->-  -- | The eventlog socket handle.-  Handle ->-  MachineT m k (Tick BS.ByteString)-sourceHandleBatch batchIntervalMs chunkSizeBytes handle = construct start- where-  start = do-    startTimeMs <- liftIO getMonotonicTimeMilli-    batch startTimeMs-  batch startTimeMs = waitForInput-   where-    getRemainingTimeMilli = do-      currentTimeMilli <- liftIO getMonotonicTimeMilli-      pure $ (startTimeMs + batchIntervalMs) - currentTimeMilli-    waitForInput = do-      remainingTimeMilli <- getRemainingTimeMilli-      if remainingTimeMilli <= 0-        then do-          yield Tick-          start-        else do-          ready <- liftIO (hWaitForInput' handle remainingTimeMilli)-          case ready of-            Ready -> do-              chunk <- liftIO $ BS.hGetSome handle chunkSizeBytes-              yield (Item chunk) >> waitForInput-            NotReady -> waitForInput-            EOF -> pure ()--{- |-Eventlog chunk size in bytes.-This should be equal to the page size.--}-defaultChunkSizeBytes :: Int-defaultChunkSizeBytes = 4096--{- |-Internal helper.-Return monotonic time in milliseconds, since some unspecified starting point--}-getMonotonicTimeMilli :: IO Int-getMonotonicTimeMilli = nanoToMilli <$> getMonotonicTimeNSec--{- |-Internal helper.-Convert nanoseconds to milliseconds.-The conversion from 'Word64' to 'Int' is safe.-It cannot overflow due to the division by 1_000_000.--}-nanoToMilli :: Word64 -> Int-nanoToMilli = fromIntegral . (`div` 1_000_000)--{- |-Internal helper.-Type to represent the state of a handle.--}-data Ready = Ready | NotReady | EOF--{- |-Internal helper.-Wait for input from a `Handle` for a given number of milliseconds.--}-hWaitForInput' ::-  -- | The handle.-  Handle ->-  -- | The timeout in milliseconds.-  Int ->-  IO Ready-hWaitForInput' handle timeoutMilli =-  catch (boolToReady <$> hWaitForInput handle timeoutMilli) handleEOFError- where-  boolToReady True = Ready-  boolToReady False = NotReady-  handleEOFError err-    | isEOFError err = pure EOF-    | otherwise = throwIO err

@@ -1,88 +0,0 @@-{- |-Module      : GHC.Eventlog.Live.Machine.WithStartTime-Description : Machines for processing eventlog data.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Machine.WithStartTime (-  WithStartTime (..),-  setWithStartTime'value,-  tryGetTimeUnixNano,-  withStartTime,-  withStartTime',-  dropStartTime,-) where--import Control.Monad (forever)-import Data.Machine (Is (..), PlanT, Process, await, construct, mapping, yield)-import GHC.RTS.Events (Event (..), EventInfo, Timestamp)-import GHC.RTS.Events qualified as E------------------------------------------------------------------------------------ Start time--{- |-Data decorated with a start time in nanoseconds since the Unix epoch.--}-data WithStartTime a = WithStartTime-  { value :: !a-  , maybeStartTimeUnixNano :: !(Maybe Timestamp)-  }-  deriving (Functor, Show)--{- |-Setter for the value of a t`WithStartTime`--}-setWithStartTime'value :: WithStartTime a -> b -> WithStartTime b-setWithStartTime'value (WithStartTime _a t) b = WithStartTime b t--{- |-If the event has a start time, return `Just` the time of the event in-nanoseconds since the Unix epoch. Otherwise, return `Nothing`.--}-tryGetTimeUnixNano :: WithStartTime Event -> Maybe Timestamp-tryGetTimeUnixNano i = (i.value.evTime +) <$> i.maybeStartTimeUnixNano--{- |-Wrap every event in t`WithStartTime`. Every event after `E.WallClockTime` will-have its start time field set to `Just` the process start time.--This machine swallows the first and only `E.WallClockTime` event.--}-withStartTime :: Process Event (WithStartTime Event)-withStartTime = withStartTime' E.evSpec WithStartTime--{- |-Generalised version of `withStartTime` that can be adapted to work on arbitrary-types using a getter and a setter.--}-withStartTime' :: (a -> EventInfo) -> (a -> Maybe Timestamp -> b) -> Process a b-withStartTime' getEventInfo setStartTime = construct start- where-  start =-    await >>= \case-      value-        -- The `WallClockTime` event announces the wall-clock time at which the-        -- process was started.-        | E.WallClockTime{..} <- getEventInfo value -> do-            -- This will start overflowing on Sunday, 21 July 2554 23:34:33, UTC.-            let !startTimeNs = sec * 1_000_000_000 + fromIntegral nsec-            -- We do not re-emit the `WallClockTime` event.-            continue startTimeNs-        | otherwise ->-            yield (value `setStartTime` Nothing) >> start-  continue startTimeUnixNano =-    mappingPlan $ \value ->-      value `setStartTime` Just startTimeUnixNano--{- |-Drop the t`WithStartTime` wrapper.--}-dropStartTime :: Process (WithStartTime a) a-dropStartTime = mapping (.value)--{- |-Internal helper. Variant of `mapping` for plans.--}-mappingPlan :: (a -> b) -> PlanT (Is a) b m a-mappingPlan f = forever (await >>= \a -> yield (f a))

@@ -1,189 +0,0 @@-{- |-Module      : GHC.Eventlog.Live.Options-Description : Command-line option parsers for eventlog machines.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Options (-  EventlogSource (..),-  eventlogSourceParser,-  eventlogSocketTimeoutParser,-  eventlogSocketTimeoutExponentParser,-  heapProfBreakdownParser,-  eventlogLogFileParser,-  batchIntervalParser,-  verbosityParser,-) where--import Control.Applicative (asum)-import Data.Char (toLower)-import GHC.Eventlog.Live.Machine (heapProfBreakdownEitherReader)-import GHC.Eventlog.Live.Verbosity (Verbosity, verbosityDebug, verbosityError, verbosityInfo, verbosityQuiet, verbosityWarning)-import GHC.RTS.Events (HeapProfBreakdown (..))-import Options.Applicative qualified as O-import Text.Read (readEither)------------------------------------------------------------------------------------- Eventlog Socket--{- |-The type of eventlog sockets.--}-data EventlogSource-  = EventlogStdin-  | EventlogFile FilePath-  | EventlogSocketUnix FilePath--{- |-Parser for the eventlog socket.--}-eventlogSourceParser :: O.Parser EventlogSource-eventlogSourceParser =-  asum-    [ stdinParser-    , fileParser-    , socketUnixParser-    ]- where-  stdinParser =-    EventlogStdin-      <$ O.flag'-        ()-        ( O.long "eventlog-stdin"-            <> O.help "Read the eventlog from stdin."-        )-  fileParser =-    EventlogFile-      <$> O.strOption-        ( O.long "eventlog-file"-            <> O.metavar "FILE"-            <> O.help "Read the eventlog from a file."-        )-  socketUnixParser =-    EventlogSocketUnix-      <$> O.strOption-        ( O.long "eventlog-socket"-            <> O.metavar "SOCKET"-            <> O.help "Read the eventlog from a Unix socket."-        )--{- |-Parser for the intial timeout for exponential backoff.--}-eventlogSocketTimeoutParser :: O.Parser Double-eventlogSocketTimeoutParser =-  O.option-    O.auto-    ( O.long "eventlog-socket-timeout"-        <> O.metavar "NUM"-        <> O.help "Eventlog socket connection retry timeout in microseconds."-        <> O.value 1-    )--{- |-Parser for the exponent for exponential backoff.--}-eventlogSocketTimeoutExponentParser :: O.Parser Double-eventlogSocketTimeoutExponentParser =-  O.option-    O.auto-    ( O.long "eventlog-socket-exponent"-        <> O.metavar "NUM"-        <> O.help "Eventlog socket connection retry timeout exponent."-        <> O.value 1-    )------------------------------------------------------------------------------------- Heap Profile Breakdown--{- |-Parser for the heap profile breakdown.--}-heapProfBreakdownParser :: O.Parser HeapProfBreakdown-heapProfBreakdownParser =-  O.option-    (O.eitherReader heapProfBreakdownEitherReader)-    ( O.short 'h'-        <> O.metavar "Tcmdyrbi"-        <> O.help "Heap profile breakdown."-    )------------------------------------------------------------------------------------- Eventlog Log File--{- |-Parser for the eventlog log file.--}-eventlogLogFileParser :: O.Parser FilePath-eventlogLogFileParser =-  O.strOption-    ( O.long "eventlog-log-file"-        <> O.metavar "FILE"-        <> O.help "Use file to log binary eventlog data."-    )------------------------------------------------------------------------------------- Batch Interval--{- |-Parser for the batch interval.--}-batchIntervalParser :: O.Parser Int-batchIntervalParser =-  O.option-    O.auto-    ( O.long "batch-interval"-        <> O.metavar "NUM"-        <> O.help "Batch interval in milliseconds."-        <> O.value defaultBatchIntervalMs-    )--{- |-Internal helper.-The default batch interval in milliseconds.--}-defaultBatchIntervalMs :: Int-defaultBatchIntervalMs = 1_000------------------------------------------------------------------------------------- Verbosity--{- |-Parser for verbosities.-The default verbosity is `verbosityWarning`.--}-verbosityParser :: O.Parser Verbosity-verbosityParser =-  O.option-    (O.eitherReader readEitherVerbosity)-    ( O.short 'v'-        <> O.long "verbosity"-        <> O.metavar "quiet|error|warning|info|debug|0-4"-        <> O.help "The verbosity threshold for logging."-        <> O.value verbosityWarning-    )--{- |-Internal helper.-Parser for verbosities by number or name.-Case insensitive.--}-readEitherVerbosity :: String -> Either String Verbosity-readEitherVerbosity rawVerbosity =-  -- try to parse the verbosity as a number...-  case readEither @Word rawVerbosity of-    -- if the verbosity string is a number, map it to a verbosity...-    Right verbosityThreshold-      | verbosityThreshold <= 0 -> Right verbosityQuiet-      | verbosityThreshold == 1 -> Right verbosityError-      | verbosityThreshold == 2 -> Right verbosityWarning-      | verbosityThreshold == 3 -> Right verbosityInfo-      | otherwise -> Right verbosityDebug-    -- otherwise, match it against the literal names of the levels...-    Left _parseError -> case toLower <$> rawVerbosity of-      "quiet" -> Right verbosityQuiet-      "error" -> Right verbosityError-      "warning" -> Right verbosityWarning-      "info" -> Right verbosityInfo-      "debug" -> Right verbosityDebug-      _otherwise -> Left $ "Could not parse verbosity '" <> rawVerbosity <> "'."

@@ -1,177 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}--{- |-Module      : GHC.Eventlog.Live.Socket-Description : Utilities for running eventlog machines with sockets.-Stability   : experimental-Portability : portable--}-module GHC.Eventlog.Live.Socket (-  EventlogSource (..),-  Tick (..),-  tryConnect,-  runWithEventlogSource,-) where--import Control.Concurrent (threadDelay)-import Control.Exception (Exception (..))-import Control.Exception qualified as E-import Control.Monad.IO.Unlift (MonadUnliftIO (..))-import Data.Foldable (traverse_)-import Data.Machine (ProcessT, runT_, (~>))-import Data.Machine.Fanout (fanout)-import Data.Maybe (fromMaybe)-import Data.Text (Text)-import Data.Text qualified as T-import Data.Void (Void)-import GHC.Eventlog.Live.Internal.Logger (logDebug, logInfo)-import GHC.Eventlog.Live.Machine.Core-import GHC.Eventlog.Live.Machine.Decoder-import GHC.Eventlog.Live.Machine.Sink-import GHC.Eventlog.Live.Machine.Source-import GHC.Eventlog.Live.Options (EventlogSource (..))-import GHC.Eventlog.Live.Verbosity (Verbosity)-import GHC.RTS.Events (Event)-import Network.Socket qualified as S-import System.IO (Handle)-import System.IO qualified as IO-import Text.Printf (printf)--{- |-Run an event processor with an eventlog socket.--}-runWithEventlogSource ::-  (MonadUnliftIO m) =>-  -- | The logging verbosity.-  Verbosity ->-  -- | The eventlog socket handle.-  EventlogSource ->-  -- | The initial timeout in microseconds for exponential backoff.-  Double ->-  -- | The timeout exponent for exponential backoff.-  Double ->-  -- | The batch interval in milliseconds.-  Int ->-  -- | The number of bytes to read (defaults to 4KiB).-  Maybe Int ->-  -- | An optional file to which to stream binary eventlog data.-  Maybe FilePath ->-  -- | The event processor.-  ProcessT m (Tick Event) Void ->-  m ()-runWithEventlogSource verbosity eventlogSocket timeoutExponent initialTimeoutMcs batchIntervalMs maybeChuckSizeBytes maybeOutputFile toEventSink = do-  withEventlogSource verbosity timeoutExponent initialTimeoutMcs eventlogSocket $ \eventlogSource -> do-    let chuckSizeBytes = fromMaybe defaultChunkSizeBytes maybeChuckSizeBytes-    let fromSocket = sourceHandleBatch batchIntervalMs chuckSizeBytes eventlogSource-    case maybeOutputFile of-      Nothing ->-        runT_ $-          fromSocket ~> decodeEventBatch ~> toEventSink-      Just outputFile ->-        withRunInIO $ \runInIO ->-          IO.withFile outputFile IO.WriteMode $ \outputHandle -> do-            runInIO . runT_ $-              fromSocket-                ~> fanout-                  [ fileSinkBatch outputHandle-                  , decodeEventBatch ~> toEventSink-                  ]--{- |-Run an action with a `Handle` to an `EventlogSource`.--}-withEventlogSource ::-  (MonadUnliftIO m) =>-  -- | The logging verbosity.-  Verbosity ->-  -- | The initial timeout in microseconds for exponential backoff.-  Double ->-  -- | The timeout exponent for exponential backoff.-  Double ->-  -- | The eventlog socket.-  EventlogSource ->-  (Handle -> m ()) ->-  m ()-withEventlogSource verbosity initialTimeoutMcs timeoutExponent eventlogSource action = do-  withRunInIO $ \runInIO ->-    case eventlogSource of-      EventlogStdin -> do-        logInfo verbosity "withEventlogSource" "Reading eventlog from stdin"-        let enter = do-              maybeStdinTextEncoding <- IO.hGetEncoding IO.stdin-              IO.hSetBinaryMode IO.stdin True-              pure maybeStdinTextEncoding-        let leave maybeStdinTextEncoding = do-              traverse_ (IO.hSetEncoding IO.stdin) maybeStdinTextEncoding-              IO.hSetNewlineMode IO.stdin IO.nativeNewlineMode-        E.bracket enter leave . const . runInIO . action $ IO.stdin-      EventlogFile eventlogFile -> do-        logInfo verbosity "withEventlogSource" $ "Reading eventlog from " <> T.pack eventlogFile-        IO.withBinaryFile eventlogFile IO.ReadMode $ \handle ->-          runInIO $ action handle-      EventlogSocketUnix eventlogSocketUnix -> do-        logInfo verbosity "withEventlogSource" $ "Waiting to connect on " <> prettyEventlogSocketUnix eventlogSocketUnix-        E.bracket (connectRetry verbosity initialTimeoutMcs timeoutExponent eventlogSocketUnix) IO.hClose $ \handle ->-          runInIO $ action handle--{- |-Connect to an `EventlogSource` with retries and non-randomised exponential backoff.--}-connectRetry ::-  -- | The logging verbosity.-  Verbosity ->-  -- | The initial timeout in microseconds for exponential backoff.-  Double ->-  -- | The timeout exponent for exponential backoff.-  Double ->-  -- | The eventlog socket.-  FilePath ->-  IO Handle-connectRetry verbosity initialTimeoutMcs timeoutExponent eventlogSocketUnix =-  connectLoop initialTimeoutMcs- where-  waitFor :: Double -> IO ()-  waitFor timeoutMcs = threadDelay $ round $ timeoutMcs * 1_000_000--  connectLoop :: Double -> IO Handle-  connectLoop timeoutMcs = do-    let connect = do-          logDebug verbosity "connectRetry" $ "Trying to connect on " <> prettyEventlogSocketUnix eventlogSocketUnix-          handle <- tryConnect eventlogSocketUnix-          logInfo verbosity "connectRetry" $ "Connected on " <> prettyEventlogSocketUnix eventlogSocketUnix-          pure handle-    let cleanup (e :: E.IOException) = do-          logDebug verbosity "connectRetry" $ "Failed to connect on " <> prettyEventlogSocketUnix eventlogSocketUnix <> ": " <> T.pack (displayException e)-          logDebug verbosity "connectRetry" $ "Waiting " <> prettyTimeoutMcs timeoutMcs <> " to retry..."-          waitFor timeoutMcs-          connectLoop (timeoutMcs * timeoutExponent)-    E.catch connect cleanup--{- |-Try to connect to a Unix socket.--}-tryConnect :: FilePath -> IO Handle-tryConnect eventlogSocketUnix =-  E.bracketOnError (S.socket S.AF_UNIX S.Stream S.defaultProtocol) S.close $ \socket -> do-    S.connect socket (S.SockAddrUnix eventlogSocketUnix)-    handle <- S.socketToHandle socket IO.ReadMode-    IO.hSetBuffering handle IO.NoBuffering-    pure handle--{- |-Interal helper. Pretty-printer for timeout values in microseconds.--}-prettyTimeoutMcs :: Double -> Text-prettyTimeoutMcs timeoutMcs-  | timeoutMcs > 8.64e10 = T.pack $ printf "%.2f days" (timeoutMcs / 8.64e10)-  | timeoutMcs > 3.6e9 = T.pack $ printf "%.2f hours" (timeoutMcs / 3.6e9)-  | timeoutMcs > 6e7 = T.pack $ printf "%.2f minutes" (timeoutMcs / 6e7)-  | timeoutMcs > 1e6 = T.pack $ printf "%.2f seconds" (timeoutMcs / 1e6)-  | timeoutMcs > 1e3 = T.pack $ printf "%.2f milliseconds" (timeoutMcs / 1e3)-  | otherwise = T.pack $ printf "%.2f microseconds" timeoutMcs--{- |-Internal helper. Pretty-printer for eventlog sockets.--}-prettyEventlogSocketUnix :: FilePath -> Text-prettyEventlogSocketUnix eventlogSocketUnix = "Unix socket " <> T.pack eventlogSocketUnix

@@ -0,0 +1,142 @@+{- |+Module      : GHC.Eventlog.Live.Attribute+Description : Representation for attributes.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Data.Attribute (+  Attr,+  AttrKey,+  AttrValue (..),+  IsAttrValue (..),+  (~=),+) where++import Data.Int (Int16, Int32, Int64, Int8)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Word (Word16, Word32, Word64, Word8)++{- |+An attribute is a key-value pair where the key is any string and the value is+some numeric type, string, or null. Attributes should be constructed using the+`(~=)` operator, which automatically converts Haskell types to t`AttrValue`.+-}+type Attr = (AttrKey, AttrValue)++{- |+Construct an t`Attr` as a pair of an t`AttrKey` and an t`AttrValue`,+constructed via the t`IsAttrValue` class.+-}+(~=) :: (IsAttrValue v) => AttrKey -> v -> Attr+k ~= v = (ak, av)+ where+  !ak = k+  !av = toAttrValue v+{-# INLINE (~=) #-}++{- |+The type of attribute keys.+-}+type AttrKey =+  Text++{- |+The type of attribute values.+-}+data AttrValue+  = AttrInt !Int+  | AttrInt8 !Int8+  | AttrInt16 !Int16+  | AttrInt32 !Int32+  | AttrInt64 !Int64+  | AttrWord !Word+  | AttrWord8 !Word8+  | AttrWord16 !Word16+  | AttrWord32 !Word32+  | AttrWord64 !Word64+  | AttrDouble !Double+  | AttrText !Text+  | AttrNull+  deriving (Show)++{- |+Utility class to help construct values of the t`AttrValue` type.+-}+class IsAttrValue v where+  toAttrValue :: v -> AttrValue++instance IsAttrValue AttrValue where+  toAttrValue :: AttrValue -> AttrValue+  toAttrValue = id+  {-# INLINE toAttrValue #-}++instance IsAttrValue Int where+  toAttrValue :: Int -> AttrValue+  toAttrValue = AttrInt+  {-# INLINE toAttrValue #-}++instance IsAttrValue Int8 where+  toAttrValue :: Int8 -> AttrValue+  toAttrValue = AttrInt8+  {-# INLINE toAttrValue #-}++instance IsAttrValue Int16 where+  toAttrValue :: Int16 -> AttrValue+  toAttrValue = AttrInt16+  {-# INLINE toAttrValue #-}++instance IsAttrValue Int32 where+  toAttrValue :: Int32 -> AttrValue+  toAttrValue = AttrInt32+  {-# INLINE toAttrValue #-}++instance IsAttrValue Int64 where+  toAttrValue :: Int64 -> AttrValue+  toAttrValue = AttrInt64+  {-# INLINE toAttrValue #-}++instance IsAttrValue Word where+  toAttrValue :: Word -> AttrValue+  toAttrValue = AttrWord+  {-# INLINE toAttrValue #-}++instance IsAttrValue Word8 where+  toAttrValue :: Word8 -> AttrValue+  toAttrValue = AttrWord8+  {-# INLINE toAttrValue #-}++instance IsAttrValue Word16 where+  toAttrValue :: Word16 -> AttrValue+  toAttrValue = AttrWord16+  {-# INLINE toAttrValue #-}++instance IsAttrValue Word32 where+  toAttrValue :: Word32 -> AttrValue+  toAttrValue = AttrWord32+  {-# INLINE toAttrValue #-}++instance IsAttrValue Word64 where+  toAttrValue :: Word64 -> AttrValue+  toAttrValue = AttrWord64+  {-# INLINE toAttrValue #-}++instance IsAttrValue Double where+  toAttrValue :: Double -> AttrValue+  toAttrValue = AttrDouble+  {-# INLINE toAttrValue #-}++instance IsAttrValue String where+  toAttrValue :: String -> AttrValue+  toAttrValue = AttrText . T.pack+  {-# INLINE toAttrValue #-}++instance IsAttrValue Text where+  toAttrValue :: Text -> AttrValue+  toAttrValue = AttrText+  {-# INLINE toAttrValue #-}++instance (IsAttrValue v) => IsAttrValue (Maybe v) where+  toAttrValue :: Maybe v -> AttrValue+  toAttrValue = maybe AttrNull toAttrValue+  {-# INLINE toAttrValue #-}

@@ -0,0 +1,30 @@+{- |+Module      : GHC.Eventlog.Live.Metric+Description : Representation for metrics.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Data.Metric (+  Metric (..),+) where++import GHC.Eventlog.Live.Data.Attribute (Attr)+import GHC.RTS.Events (Timestamp)++{- |+Metrics combine a measurement with a timestamp representing the time of the+measurement, a timestamp representing the earliest possible measurment, and+a list of attributes.+-}+data Metric a = Metric+  { value :: !a+  -- ^ The measurement.+  , maybeTimeUnixNano :: !(Maybe Timestamp)+  -- ^ The time at which the measurment was taken.+  , maybeStartTimeUnixNano :: !(Maybe Timestamp)+  -- ^ The earliest time at which any measurement could have been taken.+  --   Usually, this represents the start time of a process.+  , attr :: [Attr]+  -- ^ A list of attributes.+  }+  deriving (Functor, Show)

@@ -0,0 +1,25 @@+{- |+Module      : GHC.Eventlog.Live.Span+Description : Representation for spans.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Data.Span (+  IsSpan,+  duration,+) where++import GHC.RTS.Events (Timestamp)+import GHC.Records (HasField)++{- |+A span is any type with a start and end time.+-}+type IsSpan s = (HasField "startTimeUnixNano" s Timestamp, HasField "endTimeUnixNano" s Timestamp)++{- |+Determine the duration of a span.+-}+duration :: (IsSpan s) => s -> Timestamp+duration s = if s.startTimeUnixNano < s.endTimeUnixNano then s.endTimeUnixNano - s.startTimeUnixNano else 0+{-# INLINEABLE duration #-}

@@ -0,0 +1,113 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live..Logger+Description : Logging functions.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Logger (+  logError,+  logWarning,+  logInfo,+  logDebug,+) where++import Control.Exception (bracket_)+import Control.Monad.IO.Class (MonadIO (..))+import Data.List qualified as L+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.IO qualified as TIO+import GHC.Eventlog.Live.Verbosity (Verbosity, showVerbosity, verbosityDebug, verbosityError, verbosityInfo, verbosityWarning)+import GHC.Stack (CallStack, HasCallStack, SrcLoc (..), callStack, getCallStack)+import System.Console.ANSI (Color (..), ColorIntensity (..), ConsoleLayer (..), SGR (..), hNowSupportsANSI, hSetSGR)+import System.IO qualified as IO++{- |+Log messages to given handle.+Only prints a message if its verbosity level is above the verbosity threshold.+-}+logMessage :: (MonadIO m) => IO.Handle -> CallStack -> Verbosity -> Verbosity -> Text -> m ()+logMessage handle theCallStack verbosityLevel verbosityThreshold msg+  | verbosityLevel >= verbosityThreshold = liftIO $ do+      withVerbosityColor verbosityLevel handle+        . flip TIO.hPutStrLn+        . formatMessage verbosityLevel verbosityThreshold theCallStack+        $ msg+      IO.hFlush handle+  | otherwise = pure ()++{- |+Internal helper.+Format the `CallStack`.+-}+formatCallStack :: CallStack -> Text+formatCallStack theCallStack =+  maybe T.empty (formatSrcLoc . snd . fst) (L.uncons (getCallStack theCallStack))+ where+  formatSrcLoc :: SrcLoc -> Text+  formatSrcLoc srcLoc =+    mconcat [T.pack srcLoc.srcLocFile, ":", T.pack (show srcLoc.srcLocStartLine), ":", T.pack (show srcLoc.srcLocStartCol)]++{- |+Internal helper.+Format the message appropriately for the given verbosity level and threshold.+-}+formatMessage :: Verbosity -> Verbosity -> CallStack -> Text -> Text+formatMessage verbosityLevel verbosityThreshold theCallStack msg+  | verbosityLevel == verbosityInfo && verbosityThreshold /= verbosityDebug = msg+  | otherwise = mconcat [showVerbosity verbosityLevel, " (", formatCallStack theCallStack, "): ", msg]++{- |+Internal helper.+Use a handle with the color set appropriately for the given verbosity level.+-}+withVerbosityColor :: Verbosity -> IO.Handle -> (IO.Handle -> IO a) -> IO a+withVerbosityColor verbosity handle action = do+  supportsANSI <- hNowSupportsANSI handle+  if not supportsANSI+    then+      action handle+    else case verbosityColor verbosity of+      Nothing ->+        action handle+      Just color -> do+        let setVerbosityColor = hSetSGR handle [SetColor Foreground Vivid color]+        let setDefaultColor = hSetSGR handle [SetDefaultColor Foreground]+        bracket_ setVerbosityColor setDefaultColor $ action handle++{- |+Internal helper.+Determine the ANSI color associated with a particular verbosity level.+-}+verbosityColor :: Verbosity -> Maybe Color+verbosityColor verbosity+  | verbosity == verbosityError = Just Red+  | verbosity == verbosityWarning = Just Yellow+  | verbosity == verbosityDebug = Just Blue+  | otherwise = Nothing++{- |+Log errors to `IO.stderr`.+-}+logError :: (HasCallStack, MonadIO m) => Verbosity -> Text -> m ()+logError = logMessage IO.stderr callStack verbosityError++{- |+Log warnings to `IO.stderr`.+-}+logWarning :: (HasCallStack, MonadIO m) => Verbosity -> Text -> m ()+logWarning = logMessage IO.stderr callStack verbosityWarning++{- |+Log info messages to `IO.stderr`.+-}+logInfo :: (HasCallStack, MonadIO m) => Verbosity -> Text -> m ()+logInfo = logMessage IO.stdout callStack verbosityInfo++{- |+Log debug messages to `IO.stderr`.+-}+logDebug :: (HasCallStack, MonadIO m) => Verbosity -> Text -> m ()+logDebug = logMessage IO.stderr callStack verbosityDebug

@@ -0,0 +1,477 @@+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -Wno-name-shadowing #-}++{- |+Module      : GHC.Eventlog.Live.Machine+Description : Machines for processing eventlog data.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Machine.Analysis.Capability (+  -- * Capability Usage++  -- ** Capability Usage Metrics+  processCapabilityUsageMetrics,++  -- ** Capability Usage Spans+  CapabilityUsageSpan,+  CapabilityUser (..),+  capabilityUser,+  showCapabilityUserCategory,+  processCapabilityUsageSpans,+  processCapabilityUsageSpans',++  -- ** GC Spans+  GCSpan (..),+  processGCSpans,+  processGCSpans',++  -- ** Mutator Spans+  MutatorSpan (..),+  asMutatorSpans,+  asMutatorSpans',+  processMutatorSpans,+  processMutatorSpans',+) where++import Control.Monad (when)+import Control.Monad.IO.Class (MonadIO (..))+import Data.Char (isSpace)+import Data.Foldable (for_)+import Data.Machine (Is (..), PlanT, ProcessT, asParts, await, construct, mapping, repeatedly, yield, (~>))+import Data.Machine.Fanout (fanout)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Void (Void)+import GHC.Eventlog.Live.Data.Attribute (AttrValue, IsAttrValue (..), (~=))+import GHC.Eventlog.Live.Data.Metric (Metric (..))+import GHC.Eventlog.Live.Data.Span (duration)+import GHC.Eventlog.Live.Logger (logWarning)+import GHC.Eventlog.Live.Machine.Analysis.Thread (ThreadState (..), ThreadStateSpan (..), processThreadStateSpans')+import GHC.Eventlog.Live.Machine.Core (liftRouter)+import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..), setWithStartTime'value, tryGetTimeUnixNano)+import GHC.Eventlog.Live.Verbosity (Verbosity)+import GHC.RTS.Events (Event (..), EventInfo, ThreadId, Timestamp)+import GHC.RTS.Events qualified as E+import GHC.Records (HasField (..))+import Text.Printf (printf)++-------------------------------------------------------------------------------+-- Capability Usage Metrics++{- |+This machine processes t`CapabilityUsageSpan` spans and produces metrics that+contain the duration and category of each such span and each idle period in+between.+-}+processCapabilityUsageMetrics ::+  forall m.+  (MonadIO m) =>+  ProcessT m (WithStartTime CapabilityUsageSpan) (Metric Timestamp)+processCapabilityUsageMetrics =+  liftRouter measure spawn+ where+  measure :: WithStartTime CapabilityUsageSpan -> Maybe Int+  measure = Just . (.value.cap)++  spawn :: Int -> ProcessT m (WithStartTime CapabilityUsageSpan) (Metric Timestamp)+  spawn cap = construct $ go Nothing+   where+    go ::+      Maybe CapabilityUsageSpan ->+      PlanT (Is (WithStartTime CapabilityUsageSpan)) (Metric Timestamp) m Void+    go mi =+      await >>= \j -> do+        -- If there is a previous span, and...+        for_ mi $ \i ->+          -- ...the end time of the previous span precedes the start time of the current span, then...+          when (i.endTimeUnixNano < j.value.startTimeUnixNano) $+            -- ...yield an idle duration metric.+            yield+              Metric+                { value = j.value.startTimeUnixNano - i.endTimeUnixNano+                , maybeTimeUnixNano = Just i.endTimeUnixNano+                , maybeStartTimeUnixNano = j.maybeStartTimeUnixNano+                , attr = ["cap" ~= cap, "category" ~= ("Idle" :: Text)]+                }+        -- Yield a duration metric for the current span.+        let user = capabilityUser j.value+        yield+          Metric+            { value = duration j.value+            , maybeTimeUnixNano = Just j.value.startTimeUnixNano+            , maybeStartTimeUnixNano = j.maybeStartTimeUnixNano+            , attr = ["cap" ~= cap, "category" ~= showCapabilityUserCategory user, "user" ~= user]+            }+        go (Just j.value)++{- |+The type of process using a capability,+which is either a mutator thread or garbage collection.+-}+data CapabilityUser+  = GC+  | Mutator {thread :: !ThreadId}++instance Show CapabilityUser where+  show :: CapabilityUser -> String+  show = \case+    GC -> "GC"+    Mutator{thread} -> show thread++instance IsAttrValue CapabilityUser where+  toAttrValue :: CapabilityUser -> AttrValue+  toAttrValue = toAttrValue . show+  {-# INLINE toAttrValue #-}++{- |+Get the t`CapabilityUser` associated with a t`CapabilityUsageSpan`.+-}+capabilityUser :: CapabilityUsageSpan -> CapabilityUser+capabilityUser = either (const GC) (Mutator . (.thread))++{- |+Show the category of a `CapabilityUser` as either @"GC"@ or @"Mutator"@.+-}+showCapabilityUserCategory :: CapabilityUser -> Text+showCapabilityUserCategory = \case+  GC{} -> "GC"+  Mutator{} -> "Mutator"++-------------------------------------------------------------------------------+-- Capability Usage Spans++{- |+A t`CapabilityUsageSpan` is either a t`GCSpan` or a t`MutatorSpan`.+-}+type CapabilityUsageSpan = Either GCSpan MutatorSpan++instance HasField "startTimeUnixNano" CapabilityUsageSpan Timestamp where+  getField :: CapabilityUsageSpan -> Timestamp+  getField = either (.startTimeUnixNano) (.startTimeUnixNano)++instance HasField "endTimeUnixNano" CapabilityUsageSpan Timestamp where+  getField :: CapabilityUsageSpan -> Timestamp+  getField = either (.endTimeUnixNano) (.endTimeUnixNano)++instance HasField "cap" CapabilityUsageSpan Int where+  getField :: CapabilityUsageSpan -> Int+  getField = either (.cap) (.cap)++{-# SPECIALIZE duration :: CapabilityUsageSpan -> Timestamp #-}++{- |+This machine runs `processGCSpans` and `processMutatorSpans` in parallel and+combines their output.++This is effectively a fanout of `processGCSpans` and `processMutatorSpans`, the+latter of which runs `processThreadStateSpans` internally. If you are running+`processThreadStateSpans` as well, then using `asMutatorSpans` and constructing+the fanout yourself is more efficient.+-}+processCapabilityUsageSpans ::+  forall m.+  (MonadIO m) =>+  Verbosity ->+  ProcessT m (WithStartTime Event) (WithStartTime CapabilityUsageSpan)+processCapabilityUsageSpans verbosity =+  processCapabilityUsageSpans' tryGetTimeUnixNano (.value) setWithStartTime'value setWithStartTime'value verbosity+    ~> mapping (either (fmap Left) (fmap Right))++{- |+Generalised version of `processCapabilityUsageSpans` that can be adapted to+work on arbitrary types using a getter and a pair of lenses.+-}+processCapabilityUsageSpans' ::+  forall m s t1 t2.+  (MonadIO m) =>+  (s -> Maybe Timestamp) ->+  (s -> Event) ->+  (s -> GCSpan -> t1) ->+  (s -> MutatorSpan -> t2) ->+  Verbosity ->+  ProcessT m s (Either t1 t2)+processCapabilityUsageSpans' timeUnixNano getEvent setGCSpan setMutatorSpan verbosity =+  -- NOTE:+  -- Combining this fanout with an `Either` is risky, because it+  -- has the potential to lose information if both `processGCSpans`+  -- and `processMutatorSpans` yield a value for the same input.+  -- However, this shouldn't ever happen, since the two processors+  -- process disjoint sets of events.+  fanout+    [ processGCSpans' timeUnixNano getEvent setGCSpan verbosity+        ~> mapping Left+    , processMutatorSpans' timeUnixNano getEvent setMutatorSpan verbosity+        ~> mapping Right+    ]++-------------------------------------------------------------------------------+-- GC spans++{- |+A t`GCSpan` represents a segment of time during which the specified capability+ran GC.+-}+data GCSpan = GCSpan+  { cap :: !Int+  , startTimeUnixNano :: !Timestamp+  , endTimeUnixNano :: !Timestamp+  }+  deriving (Show)++{-# SPECIALIZE duration :: GCSpan -> Timestamp #-}++{- |+This machine processes `E.StartGC` and `E.EndGC` events to produce t`GCSpan`+values that represent the segments of time a capability spent in GC.++This processor uses the following finite-state automaton:++@+      ┌─(EndGC)───┐+      │           ↓+    ┌→[   Idle    ]─┐+    │               │+(EndGC)         (StartGC)+    │               │+    └─[    GC     ]←┘+      ↑           │+      └─(StartGC)─┘+@++The transition from @GC@ to @Idle@ yields a GC span.+-}+processGCSpans ::+  forall m.+  (MonadIO m) =>+  Verbosity ->+  ProcessT m (WithStartTime Event) (WithStartTime GCSpan)+processGCSpans =+  processGCSpans' tryGetTimeUnixNano (.value) setWithStartTime'value++{- |+Generalised version of `processGCSpans` that can be adapted to work on+arbitrary types using a getter and a lens.+-}+processGCSpans' ::+  forall m s t.+  (MonadIO m) =>+  (s -> Maybe Timestamp) ->+  (s -> Event) ->+  (s -> GCSpan -> t) ->+  Verbosity ->+  ProcessT m s t+processGCSpans' timeUnixNano getEvent setGCSpan verbosity =+  liftRouter measure spawn+ where+  getEventTime = (.evTime) . getEvent+  getEventInfo = (.evSpec) . getEvent+  getEventCap = (.evCap) . getEvent++  measure :: s -> Maybe Int+  measure i+    | accept (getEventInfo i) = getEventCap i+    | otherwise = Nothing+   where+    accept E.StartGC{} = True+    accept E.EndGC{} = True+    accept _ = False++  -- TODO: Rewrite using `MealyT`+  spawn :: Int -> ProcessT m s t+  spawn cap = construct $ go Nothing+   where+    -- The "mi" variable tracks the previous event for this capability, which+    -- is either `Nothing` or `Just` a `StartGC` or a `EndGC` event.+    go :: Maybe s -> PlanT (Is s) t m Void+    go mi =+      -- We start by awaiting the next event "j"...+      await >>= \j -> case getEventInfo j of+        -- If the next event is a `RunThread` event, and...+        E.StartGC{} -> case mi of+          Just i+            -- If the previous event was a `StartGC` event, then...+            | E.StartGC{} <- getEventInfo i ->+                -- ...continue with the oldest event.+                go (Just $ minBy getEventTime i j)+            -- If the previous event was a `EndGC` event, then...+            | E.EndGC{} <- getEventInfo i ->+                -- ...continue with the current event.+                go (Just j)+            -- If the previous event was any other event, then...+            | otherwise -> do+                -- ...emit an error, and...+                logWarning verbosity . T.pack $+                  printf+                    "Capability %d: Unsupported trace %s --> %s"+                    cap+                    (showEventInfo (getEventInfo i))+                    (showEventInfo (getEventInfo j))+                -- ...continue with the previous event.+                go (Just i)+          -- If there was no previous event, then...+          Nothing ->+            -- ...continue with the current event.+            go (Just j)+        -- If the next event is a `StopThread` event...+        E.EndGC{} -> case mi of+          Just i+            -- If the previous event was a `StartGC` event, then...+            | E.StartGC{} <- getEventInfo i+            , Just startTimeUnixNano <- timeUnixNano i+            , Just endTimeUnixNano <- timeUnixNano j -> do+                -- ...yield a GC span, and...+                yield . setGCSpan j $ GCSpan{..}+                -- ...continue with the current event.+                go (Just j)+            -- If the previous event was a `EndGC` event, then...+            | E.EndGC{} <- getEventInfo i ->+                -- ...continue with the oldest event.+                go (Just $ minBy getEventTime i j)+          -- If there was no previous event or it was any other event, then...+          _otherwise -> do+            -- ...emit an error, and...+            logWarning verbosity . T.pack $+              printf+                "Capability %d: Unsupported trace %s --> %s"+                cap+                (maybe "?" (showEventInfo . getEventInfo) mi)+                (showEventInfo (getEventInfo j))+            -- ...continue with the previous event.+            go mi+        -- If the next event is any other event, ignore it.+        _otherwise -> go mi++-------------------------------------------------------------------------------+-- Mutator spans++{- |+A t`MutatorSpan` represents a segment of time during which the specified+capability ran the specified mutator thread.+-}+data MutatorSpan = MutatorSpan+  { cap :: !Int+  , thread :: !ThreadId+  , startTimeUnixNano :: !Timestamp+  , endTimeUnixNano :: !Timestamp+  }+  deriving (Show)++{-# SPECIALIZE duration :: MutatorSpan -> Timestamp #-}++{- |+This machine processes `E.RunThread` and `E.StopThread` events to produce+t`MutatorSpan` values that represent the segments of time a capability spent+executating a mutator.++This processor uses the following finite-state automaton:++@+      ┌─(StopThread[X])─┐+      │                 ↓+    ┌→[      Idle       ]─┐+    │                     │+(StopThread[X])       (RunThread[X])+    │                     │+    └─[   Mutator[X]    ]←┘+      ↑                 │+      └─(RunThread[X])──┘+@++The transition from @Mutator[X]@ to @Idle@ yields a t`MutatorSpan`.+While in the @Mutator[X]@ state, any @RunThread[Y]@ or @StopThread[Y]@ events result in an error.+Furthermore, when a @StopThread[X]@ event with the @ThreadFinished@ status is processed,+the thread @X@ is added to a set of finished threads,+and any further @RunThread[X]@ events for that thread are ignored.+This is done because the GHC RTS frequently emits a @RunThread[X]@ event+immediately after a @StopThread[X]@ event with the @ThreadFinished@ status.++This runs `processThreadStateSpans` internally. If you are also running+`processThreadStateSpans`, then post-composing it with `asMutatorSpans`+is more efficient.+-}+processMutatorSpans ::+  forall m.+  (MonadIO m) =>+  Verbosity ->+  ProcessT m (WithStartTime Event) (WithStartTime MutatorSpan)+processMutatorSpans =+  processMutatorSpans' tryGetTimeUnixNano (.value) setWithStartTime'value++{- |+Generalised version of `processMutatorSpans` that can be adapted to work on+arbitrary types using a getter and a lens.+-}+processMutatorSpans' ::+  forall m s t.+  (MonadIO m) =>+  (s -> Maybe Timestamp) ->+  (s -> Event) ->+  (s -> MutatorSpan -> t) ->+  Verbosity ->+  ProcessT m s t+processMutatorSpans' timeUnixNano getEvent setMutatorSpan verbosity =+  processThreadStateSpans' timeUnixNano getEvent setThreadStateSpan verbosity ~> asParts+ where+  setThreadStateSpan :: s -> ThreadStateSpan -> Maybe t+  setThreadStateSpan s threadStateSpan =+    setMutatorSpan s <$> threadStateSpanToMutatorSpan threadStateSpan++{- |+This machine converts any `Running` t`ThreadStateSpan` to a t`MutatorSpan`.+-}+asMutatorSpans ::+  forall m.+  (MonadIO m) =>+  ProcessT m ThreadStateSpan MutatorSpan+asMutatorSpans = asMutatorSpans' id (const id)++{- |+Generalised version of `asMutatorSpans` that can be adapted to work on+arbitrary types using a getter and a lens.+-}+asMutatorSpans' ::+  forall m s t.+  (MonadIO m) =>+  (s -> ThreadStateSpan) ->+  (s -> MutatorSpan -> t) ->+  ProcessT m s t+asMutatorSpans' getThreadStateSpan setMutatorSpan = repeatedly go+ where+  go =+    await >>= \s -> do+      let threadStateSpan = getThreadStateSpan s+      let maybeMutatorSpan = threadStateSpanToMutatorSpan threadStateSpan+      for_ maybeMutatorSpan $ yield . setMutatorSpan s++{- |+Convert the `Running` t`ThreadStateSpan` to `Just` a t`MutatorSpan`.+-}+threadStateSpanToMutatorSpan :: ThreadStateSpan -> Maybe MutatorSpan+threadStateSpanToMutatorSpan ThreadStateSpan{..} =+  case threadState of+    Running{..} -> Just MutatorSpan{..}+    _otherwise -> Nothing++-------------------------------------------------------------------------------+-- Internal Helpers+-------------------------------------------------------------------------------++{- |+Internal helper.+Show `EventInfo` in a condensed format suitable for logging.+-}+showEventInfo :: EventInfo -> String+showEventInfo = \case+  E.RunThread{thread} -> printf "RunThread{%d}" thread+  E.StopThread{thread, status} -> printf "StopThread{%d,%s}" thread (E.showThreadStopStatus status)+  E.MigrateThread{thread} -> printf "MigrateThread{%d}" thread+  E.StartGC{} -> "StartGC"+  E.EndGC{} -> "EndGC"+  evSpec -> takeWhile (not . isSpace) . show $ evSpec++{- |+Internal helper. Return the minimal value by some projection.+-}+minBy :: (Ord b) => (a -> b) -> a -> a -> a+minBy f x y = if f x < f y then x else y

@@ -0,0 +1,438 @@+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -Wno-name-shadowing #-}++{- |+Module      : GHC.Eventlog.Live.Machine.Analysis.Heap+Description : Machines for processing eventlog data.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Machine.Analysis.Heap (+  -- * Heap Usage+  processHeapAllocatedData,+  processHeapSizeData,+  processBlocksSizeData,+  processHeapLiveData,+  MemReturnData (..),+  processMemReturnData,+  processHeapProfSampleData,++  -- ** Heap Profile Breakdown+  heapProfBreakdownEitherReader,+  heapProfBreakdownShow,+) where++import Control.Monad (unless)+import Control.Monad.IO.Class (MonadIO (..))+import Data.Either (isLeft)+import Data.HashMap.Strict (HashMap)+import Data.HashMap.Strict qualified as M+import Data.Hashable (Hashable (..))+import Data.List qualified as L+import Data.Machine (Process, ProcessT, await, construct, repeatedly, yield)+import Data.Maybe (listToMaybe, mapMaybe)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Word (Word32, Word64)+import GHC.Eventlog.Live.Data.Attribute (Attr, (~=))+import GHC.Eventlog.Live.Data.Metric (Metric (..))+import GHC.Eventlog.Live.Logger (logWarning)+import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..), tryGetTimeUnixNano)+import GHC.Eventlog.Live.Verbosity (Verbosity)+import GHC.RTS.Events (Event (..), HeapProfBreakdown (..))+import GHC.RTS.Events qualified as E+import Numeric (showHex)+import Text.ParserCombinators.ReadP (readP_to_S)+import Text.ParserCombinators.ReadP qualified as P+import Text.Printf (printf)+import Text.Read (readMaybe)+import Text.Read.Lex (readHexP)++-------------------------------------------------------------------------------+-- Heap events+-------------------------------------------------------------------------------++--------------------------------------------------------------------------------+-- HeapAllocated++{- |+This machine processes `E.HeapAllocated` events into metrics.+-}+processHeapAllocatedData :: Process (WithStartTime Event) (Metric Word64)+processHeapAllocatedData =+  repeatedly $+    await >>= \case+      i+        | E.HeapAllocated{..} <- i.value.evSpec ->+            yield $+              metric i allocBytes $+                [ "evCap" ~= i.value.evCap+                , "heapCapset" ~= heapCapset+                ]+        | otherwise -> pure ()++-------------------------------------------------------------------------------+-- HeapSize++{- |+This machine processes `E.HeapSize` events into metrics.+-}+processHeapSizeData :: Process (WithStartTime Event) (Metric Word64)+processHeapSizeData = repeatedly go+ where+  go =+    await >>= \case+      i+        | E.HeapSize{..} <- i.value.evSpec -> do+            yield $+              metric i sizeBytes $+                [ "evCap" ~= i.value.evCap+                , "heapCapset" ~= heapCapset+                ]+        | otherwise -> pure ()++-------------------------------------------------------------------------------+-- BlocksSize++{- |+This machine processes `E.BlocksSize` events into metrics.+-}+processBlocksSizeData :: Process (WithStartTime Event) (Metric Word64)+processBlocksSizeData =+  repeatedly $+    await >>= \case+      i+        | E.BlocksSize{..} <- i.value.evSpec -> do+            yield $+              metric i blocksSize $+                [ "evCap" ~= i.value.evCap+                , "heapCapset" ~= heapCapset+                ]+        | otherwise -> pure ()++-------------------------------------------------------------------------------+-- HeapLive++{- |+This machine processes `E.HeapLive` events into metrics.+-}+processHeapLiveData :: Process (WithStartTime Event) (Metric Word64)+processHeapLiveData =+  repeatedly $+    await >>= \case+      i+        | E.HeapLive{..} <- i.value.evSpec -> do+            yield $+              metric i liveBytes $+                [ "evCap" ~= i.value.evCap+                , "heapCapset" ~= heapCapset+                ]+        | otherwise -> pure ()++-------------------------------------------------------------------------------+-- MemReturn++{- |+The type of data associated with a `E.MemReturn` event.+-}+data MemReturnData = MemReturnData+  { current :: !Word32+  -- ^ The number of megablocks currently allocated.+  , needed :: !Word32+  -- ^ The number of megablocks currently needed.+  , returned :: !Word32+  -- ^ The number of megablocks currently being returned to the OS.+  }++{- |+This machine processes `E.MemReturn` events into metrics.+-}+processMemReturnData :: Process (WithStartTime Event) (Metric MemReturnData)+processMemReturnData =+  repeatedly $+    await >>= \case+      i+        | E.MemReturn{..} <- i.value.evSpec -> do+            yield $+              metric i MemReturnData{..} $+                [ "evCap" ~= i.value.evCap+                , "heapCapset" ~= heapCapset+                ]+        | otherwise -> pure ()++-------------------------------------------------------------------------------+-- HeapProfSample++{- |+Internal helper.+The type of info table pointers.+-}+newtype InfoTablePtr = InfoTablePtr Word64+  deriving newtype (Eq, Hashable, Ord)++instance Show InfoTablePtr where+  showsPrec :: Int -> InfoTablePtr -> ShowS+  showsPrec _ (InfoTablePtr ptr) =+    showString "0x" . showHex ptr++instance Read InfoTablePtr where+  readsPrec :: Int -> ReadS InfoTablePtr+  readsPrec _ = readP_to_S (InfoTablePtr <$> (P.string "0x" *> readHexP))++{- |+Internal helper.+The type of an info table entry, as produced by the `E.InfoTableProv` event.+-}+data InfoTable = InfoTable+  { infoTablePtr :: InfoTablePtr+  , infoTableName :: Text+  , infoTableClosureDesc :: Int+  , infoTableTyDesc :: Text+  , infoTableLabel :: Text+  , infoTableModule :: Text+  , infoTableSrcLoc :: Text+  }+  deriving (Show)++{- |+Internal helper.+The type of the state kept by `processHeapProfSampleData`.+-}+data HeapProfSampleState = HeapProfSampleState+  { eitherShouldWarnOrHeapProfBreakdown :: Either Bool HeapProfBreakdown+  , infoTableMap :: HashMap InfoTablePtr InfoTable+  , heapProfSampleEraStack :: [Word64]+  }+  deriving (Show)++{- |+Internal helper.+Decides whether or not `processHeapProfSampleData` should track info tables.+We track info tables until (1) we learn that the RTS is not run with @-hi@,+or (2) we see the first heap profiling sample and don't yet know for sure+that the RTS is run with @-hi@.+-}+shouldTrackInfoTableMap :: Either Bool HeapProfBreakdown -> Bool+shouldTrackInfoTableMap (Left _shouldWarn) = True+shouldTrackInfoTableMap (Right HeapProfBreakdownInfoTable) = True+shouldTrackInfoTableMap _ = False++{- |+Internal helper.+Checks whether a `HeapProfBreakdown` is `HeapProfBreakdownInfoTable`.+This is needed because the ghc-events package does not define an `Eq`+instance for the `HeapProfBreakdown` type.+-}+isHeapProfBreakdownInfoTable :: HeapProfBreakdown -> Bool+isHeapProfBreakdownInfoTable HeapProfBreakdownInfoTable = True+isHeapProfBreakdownInfoTable _ = False++{- |+This machine processes `E.HeapProfSampleString` events into metrics.+Furthermore, it processes the `E.HeapProfBegin` and `E.ProgramArgs` events+to determine the heap profile breakdown, processes `E.InfoTableProv` events to+build an info table map, if necessary, and processes `E.HeapProfSampleBegin`+and `E.HeapProfSampleEnd` events to maintain an era stack.+-}+processHeapProfSampleData ::+  (MonadIO m) =>+  Verbosity ->+  Maybe HeapProfBreakdown ->+  ProcessT m (WithStartTime Event) (Metric Word64)+processHeapProfSampleData verbosityThreshold maybeHeapProfBreakdown =+  construct $+    go+      HeapProfSampleState+        { eitherShouldWarnOrHeapProfBreakdown = maybe (Left True) Right maybeHeapProfBreakdown+        , infoTableMap = mempty+        , heapProfSampleEraStack = mempty+        }+ where+  -- go :: HeapProfSampleState -> PlanT (Is (WithStartTime Event)) (Metric Word64) m Void+  go st@HeapProfSampleState{..} = do+    await >>= \i -> case i.value.evSpec of+      -- Announces the heap profile breakdown, amongst other things.+      -- This event is only emitted for code compiled with GHC >=9.14.+      E.HeapProfBegin{..}+        | isLeft eitherShouldWarnOrHeapProfBreakdown ->+            go st{eitherShouldWarnOrHeapProfBreakdown = Right heapProfBreakdown}+      -- Announces the arguments with which the program was called.+      -- This *may* include RTS options, which can be used to determine the+      -- heap profile breakdown for code compiled with GHC <9.14.+      E.ProgramArgs{..}+        | isLeft eitherShouldWarnOrHeapProfBreakdown+        , Just heapProfBreakdown <- findHeapProfBreakdown args ->+            go st{eitherShouldWarnOrHeapProfBreakdown = Right heapProfBreakdown}+      -- Announces an info table entry.+      E.InfoTableProv{..}+        | shouldTrackInfoTableMap eitherShouldWarnOrHeapProfBreakdown -> do+            let infoTablePtr = InfoTablePtr itInfo+                infoTable =+                  InfoTable+                    { infoTablePtr = infoTablePtr+                    , infoTableName = itTableName+                    , infoTableClosureDesc = itClosureDesc+                    , infoTableTyDesc = itTyDesc+                    , infoTableLabel = itLabel+                    , infoTableModule = itModule+                    , infoTableSrcLoc = itSrcLoc+                    }+            go st{infoTableMap = M.insert infoTablePtr infoTable infoTableMap}+      -- Announces the beginning of a heap profile sample.+      E.HeapProfSampleBegin{..} ->+        go st{heapProfSampleEraStack = heapProfSampleEra : heapProfSampleEraStack}+      -- Announces the end of a heap profile sample.+      E.HeapProfSampleEnd{..} ->+        case L.uncons heapProfSampleEraStack of+          Nothing -> do+            logWarning verbosityThreshold . T.pack $+              printf+                "Eventlog closed era %d, but there is no current era."+                heapProfSampleEra+            go st+          Just (currentEra, heapProfSampleEraStack') -> do+            unless (currentEra == heapProfSampleEra) $+              logWarning verbosityThreshold . T.pack $+                printf+                  "Eventlog closed era %d, but the current era is era %d."+                  heapProfSampleEra+                  currentEra+            go st{heapProfSampleEraStack = heapProfSampleEraStack'}+      -- Announces a heap profile sample.+      E.HeapProfSampleString{..}+        -- If there is no heap profile breakdown, issue a warning, then disable warnings.+        | Left True <- eitherShouldWarnOrHeapProfBreakdown -> do+            logWarning verbosityThreshold $+              "Cannot infer heap profile breakdown.\n\+              \         If your binary was compiled with a GHC version prior to 9.14,\n\+              \         you must also pass the heap profile type to this executable.\n\+              \         See: https://gitlab.haskell.org/ghc/ghc/-/commit/76d392a"+            go st{eitherShouldWarnOrHeapProfBreakdown = Left False, infoTableMap = mempty}+        -- If the heap profile breakdown is biographical, issue a warning, then disable warnings.+        | Right HeapProfBreakdownBiography <- eitherShouldWarnOrHeapProfBreakdown -> do+            logWarning verbosityThreshold . T.pack $+              printf+                "Unsupported heap profile breakdown %s"+                (heapProfBreakdownShow HeapProfBreakdownBiography)+            go st{eitherShouldWarnOrHeapProfBreakdown = Left False, infoTableMap = mempty}+        -- If there is a heap profile breakdown, handle it appropriately.+        | Right heapProfBreakdown <- eitherShouldWarnOrHeapProfBreakdown -> do+            -- If the heap profile breakdown is by info table, add the info table.+            let maybeInfoTable+                  | isHeapProfBreakdownInfoTable heapProfBreakdown = do+                      !infoTablePtr <- readMaybe (T.unpack heapProfLabel)+                      M.lookup infoTablePtr infoTableMap+                  | otherwise = Nothing+            yield $+              metric i heapProfResidency $+                [ "evCap" ~= i.value.evCap+                , "heapProfBreakdown" ~= heapProfBreakdownShow heapProfBreakdown+                , "heapProfId" ~= heapProfId+                , "heapProfLabel" ~= heapProfLabel+                , "heapProfSampleEra" ~= (fst <$> L.uncons heapProfSampleEraStack)+                , "infoTableName" ~= fmap (.infoTableName) maybeInfoTable+                , "infoTableClosureDesc" ~= fmap (.infoTableClosureDesc) maybeInfoTable+                , "infoTableTyDesc" ~= fmap (.infoTableTyDesc) maybeInfoTable+                , "infoTableLabel" ~= fmap (.infoTableLabel) maybeInfoTable+                , "infoTableModule" ~= fmap (.infoTableModule) maybeInfoTable+                , "infoTableSrcLoc" ~= fmap (.infoTableSrcLoc) maybeInfoTable+                ]+            go $ if isHeapProfBreakdownInfoTable heapProfBreakdown then st else st{infoTableMap = mempty}+      _otherwise -> go st++{- |+Parses the `HeapProfBreakdown` command-line arguments:++> heapProfBreakdownEitherReader "T" == Left HeapProfBreakdownClosureType+> heapProfBreakdownEitherReader "c" == Left HeapProfBreakdownCostCentre+> heapProfBreakdownEitherReader "m" == Left HeapProfBreakdownModule+> heapProfBreakdownEitherReader "d" == Left HeapProfBreakdownClosureDescr+> heapProfBreakdownEitherReader "y" == Left HeapProfBreakdownTypeDescr+> heapProfBreakdownEitherReader "e" == Left HeapProfBreakdownEra+> heapProfBreakdownEitherReader "r" == Left HeapProfBreakdownRetainer+> heapProfBreakdownEitherReader "b" == Left HeapProfBreakdownBiography+> heapProfBreakdownEitherReader "i" == Left HeapProfBreakdownInfoTable+-}+heapProfBreakdownEitherReader :: String -> Either String HeapProfBreakdown+heapProfBreakdownEitherReader =+  \case+    "T" -> Right HeapProfBreakdownClosureType+    "c" -> Right HeapProfBreakdownCostCentre+    "m" -> Right HeapProfBreakdownModule+    "d" -> Right HeapProfBreakdownClosureDescr+    "y" -> Right HeapProfBreakdownTypeDescr+    "e" -> Right HeapProfBreakdownEra+    "r" -> Right HeapProfBreakdownRetainer+    "b" -> Right HeapProfBreakdownBiography+    "i" -> Right HeapProfBreakdownInfoTable+    str -> Left $ "Unsupported heap profile breakdown -h" <> str++{- |+Shows a `HeapProfBreakdown` as its corresponding command-line flag:++> heapProfBreakdownShow HeapProfBreakdownClosureType == "-hT"+> heapProfBreakdownShow HeapProfBreakdownCostCentre == "-hc"+> heapProfBreakdownShow HeapProfBreakdownModule == "-hm"+> heapProfBreakdownShow HeapProfBreakdownClosureDescr == "-hd"+> heapProfBreakdownShow HeapProfBreakdownTypeDescr == "-hy"+> heapProfBreakdownShow HeapProfBreakdownEra == "-he"+> heapProfBreakdownShow HeapProfBreakdownRetainer == "-hr"+> heapProfBreakdownShow HeapProfBreakdownBiography == "-hb"+> heapProfBreakdownShow HeapProfBreakdownInfoTable == "-hi"+-}+heapProfBreakdownShow :: HeapProfBreakdown -> String+heapProfBreakdownShow =+  ("-h" <>) . \case+    HeapProfBreakdownClosureType -> "T"+    HeapProfBreakdownCostCentre -> "c"+    HeapProfBreakdownModule -> "m"+    HeapProfBreakdownClosureDescr -> "d"+    HeapProfBreakdownTypeDescr -> "y"+    HeapProfBreakdownEra -> "e"+    HeapProfBreakdownRetainer -> "r"+    HeapProfBreakdownBiography -> "b"+    HeapProfBreakdownInfoTable -> "i"++{- |+Internal helper.+Determine the `HeapProfBreakdown` from the list of program arguments.++__Warning__: This scan is not fully correct. It merely scans for the presence+of arguments that, as a whole, parse with `heapProfBreakdownEitherReader`.+It does not handle @-with-rtsopts@ and does not restrict its search to those+arguments between @+RTS@ and @-RTS@ tags.+-}+findHeapProfBreakdown :: [Text] -> Maybe HeapProfBreakdown+findHeapProfBreakdown = listToMaybe . mapMaybe parseHeapProfBreakdown+ where+  parseHeapProfBreakdown :: Text -> Maybe HeapProfBreakdown+  parseHeapProfBreakdown arg+    | "-h" `T.isPrefixOf` arg =+        either (const Nothing) Just+          . heapProfBreakdownEitherReader+          . T.unpack+          . T.drop 2+          $ arg+    | otherwise = Nothing++-------------------------------------------------------------------------------+-- Internal Helpers+-------------------------------------------------------------------------------++{- |+Internal helper. Construct a t`Metric` from an event with a start time+(t`WithStartTime` t`Event`), together with the measurement and any attributes.+This is a smart constructor that pulls the various timestamps out of the event.+-}+metric ::+  WithStartTime Event ->+  v ->+  [Attr] ->+  Metric v+metric i v attr =+  Metric+    { value = v+    , maybeTimeUnixNano = tryGetTimeUnixNano i+    , maybeStartTimeUnixNano = i.maybeStartTimeUnixNano+    , attr = attr+    }

@@ -0,0 +1,307 @@+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -Wno-name-shadowing #-}++{- |+Module      : GHC.Eventlog.Live.Machine.Analysis.Thread+Description : Machines for processing eventlog data.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Machine.Analysis.Thread (+  -- * Thread Analysis++  -- ** Thread Labels+  ThreadLabel (..),+  processThreadLabels,++  -- ** Thread State Spans+  ThreadState (..),+  showThreadStateCategory,+  threadStateStatus,+  threadStateCap,+  ThreadStateSpan (..),+  processThreadStateSpans,+  processThreadStateSpans',+) where++import Control.Monad.IO.Class (MonadIO (..))+import Data.Char (isSpace)+import Data.Machine (Is (..), PlanT, Process, ProcessT, await, construct, repeatedly, yield)+import Data.Maybe (isNothing)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Void (Void)+import GHC.Eventlog.Live.Data.Span (duration)+import GHC.Eventlog.Live.Logger (logWarning)+import GHC.Eventlog.Live.Machine.Core (liftRouter)+import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..), tryGetTimeUnixNano)+import GHC.Eventlog.Live.Verbosity (Verbosity)+import GHC.RTS.Events (Event (..), EventInfo, ThreadId, ThreadStopStatus (..), Timestamp)+import GHC.RTS.Events qualified as E+import Text.Printf (printf)++-------------------------------------------------------------------------------+-- Thread Labels++{- |+The t`ThreadLabel` type represents the association of a label with a thread+starting at a given time.+-}+data ThreadLabel+  = ThreadLabel+  { thread :: !ThreadId+  , threadlabel :: !Text+  , startTimeUnixNano :: !Timestamp+  }++{- |+This machine processes `E.ThreadLabel` events and yields t`ThreadLabel` values.+-}+processThreadLabels :: Process (WithStartTime Event) ThreadLabel+processThreadLabels = repeatedly go+ where+  go =+    await >>= \i -> case i.value.evSpec of+      E.ThreadLabel{..}+        | Just startTimeUnixNano <- tryGetTimeUnixNano i ->+            yield ThreadLabel{..}+      _otherwise -> pure ()++-------------------------------------------------------------------------------+-- Thread State Spans++{- |+The execution states of a mutator thread.+-}+data ThreadState+  = Running {cap :: !Int}+  | Blocked {status :: !ThreadStopStatus}+  | Finished+  deriving (Show)++{- |+Pretty-print a thread state as "Running", "Blocked", or "Finished".+-}+showThreadStateCategory :: ThreadState -> Text+showThreadStateCategory = \case+  Running{} -> "Running"+  Blocked{} -> "Blocked"+  Finished{} -> "Finished"++{- |+Get the t`ThreadState` status, if the t`ThreadState` is `Blocked`.+-}+threadStateStatus :: ThreadState -> Maybe ThreadStopStatus+threadStateStatus = \case+  Running{} -> Nothing+  Blocked{status} -> Just status+  Finished{} -> Nothing++{- |+Get the t`ThreadState` capability, if the `ThreadState` is `Running`.+-}+threadStateCap :: ThreadState -> Maybe Int+threadStateCap = \case+  Running{cap} -> Just cap+  Blocked{} -> Nothing+  Finished{} -> Nothing++{- |+A span representing the state of a mutator thread.+-}+data ThreadStateSpan+  = ThreadStateSpan+  { thread :: !ThreadId+  , threadState :: !ThreadState+  , startTimeUnixNano :: !Timestamp+  , endTimeUnixNano :: !Timestamp+  }+  deriving (Show)++{-# SPECIALIZE duration :: ThreadStateSpan -> Timestamp #-}++{- |+This machine processes `E.RunThread` and `E.StopThread` events to produce+t`ThreadStateSpan` values that represent segments of time where a thread is+running, blocked, or finished.++This processor uses the following finite-state automaton:++@+      ┌─(StopThread)─┐+      │              ↓+    ┌→[   Blocked    ]─┐+    │                  │+(StopThread)       (RunThread)+    │                  │+    └─[   Running    ]←┘+      ↑              │+      └─(RunThread)──┘+@++The transitions from @Blocked@ to @Blocked@, @Blocked@ to @Running@, and+@Running@ to @Running@ yield a t`ThreadStateSpan`. There are additional+transitions (not pictured) from either state to the final `Finished` state+with a `E.StopThread` event with the `ThreadFinished` status.+-}+processThreadStateSpans ::+  (MonadIO m) =>+  Verbosity ->+  ProcessT m (WithStartTime Event) ThreadStateSpan+processThreadStateSpans =+  processThreadStateSpans' tryGetTimeUnixNano (.value) (const id)++{- |+Generalised version of `processThreadStateSpans` that can be adapted to work+on arbitrary types using a getter and a lens.+-}+processThreadStateSpans' ::+  forall m s t.+  (MonadIO m) =>+  (s -> Maybe Timestamp) ->+  (s -> Event) ->+  (s -> ThreadStateSpan -> t) ->+  Verbosity ->+  ProcessT m s t+processThreadStateSpans' timeUnixNano getEvent setThreadStateSpan verbosity =+  liftRouter measure spawn+ where+  getEventTime = (.evTime) . getEvent+  getEventInfo = (.evSpec) . getEvent+  getEventCap = (.evCap) . getEvent++  measure :: s -> Maybe ThreadId+  measure i = case getEventInfo i of+    E.RunThread{thread} -> Just thread+    E.StopThread{thread} -> Just thread+    _otherwise -> Nothing++  spawn :: ThreadId -> ProcessT m s t+  spawn thread = construct $ go Nothing+   where+    go :: Maybe s -> PlanT (Is s) t m Void+    go mi =+      await >>= \case+        j+          -- If the previous event was a `E.StopThread` event, and...+          | Just E.StopThread{status} <- getEventInfo <$> mi+          , --- ...it has the `ThreadFinished` status, then...+            isThreadFinished status ->+              -- ...ignore the current event.+              go mi+          --+          -- If the current event is a `E.RunThread` event, and...+          | E.RunThread{} <- getEventInfo j+          , -- ...the previous event was a `E.StopThread` event, then...+            Just E.StopThread{status} <- getEventInfo <$> mi+          , -- ...gather the end time of the previous event, and...+            Just startTimeUnixNano <- timeUnixNano =<< mi+          , -- ...gather the start time of the current event, and...+            Just endTimeUnixNano <- timeUnixNano j -> do+              -- ...yield a thread state span, and...+              yield . setThreadStateSpan j $+                ThreadStateSpan{threadState = Blocked status, ..}+              go (Just j)+          --+          -- If the current event is a `E.RunThread` event, and...+          | E.RunThread{} <- getEventInfo j+          , -- ...the previous event was a `E.RunThread` event, then...+            Just E.RunThread{} <- getEventInfo <$> mi -> do+              -- ...keep the oldest event.+              go (Just $ maybe j (minBy getEventTime j) mi)+          --+          -- If the current event is a `E.RunThread` event, and...+          | E.RunThread{} <- getEventInfo j+          , -- ...there is no previous event, then...+            isNothing mi ->+              -- ...keep the current event.+              --+              -- The reason for the additional `isNothing` test is because,+              -- otherwise, this case might silently swallow any `E.StopThread`+              -- events for which `timeUnixNano` gives `Nothing`.+              -- By excluding these, they are forwarded to the catch-all case.+              go (Just j)+          --+          -- If the current event is a `E.StopThread` event, and...+          | E.StopThread{} <- getEventInfo j+          , -- ...the previous event was a `E.StopThread` event, then...+            Just E.StopThread{status} <- getEventInfo <$> mi+          , -- ...gather the end time of the previous event, and...+            Just startTimeUnixNano <- timeUnixNano =<< mi+          , -- ...gather the start time of the current event, and...+            Just endTimeUnixNano <- timeUnixNano j -> do+              -- ...yield a thread state span, and...+              yield . setThreadStateSpan j $+                ThreadStateSpan{threadState = Blocked status, ..}+              -- ...keep the current event.+              --+              -- This causes us to adopt every `E.StopThread` event, until+              -- we hit a `E.StopThread` event with the `ThreadFinished`, at+              -- which point the first clause will cause us to stick with it.+              go (Just j)+          --+          -- If the current event is a `E.StopThread` event, and...+          | E.StopThread{} <- getEventInfo j+          , -- ...the previous event was a `E.RunThread` event, then...+            Just E.RunThread{} <- getEventInfo <$> mi+          , -- ...gather the capability of the `E.RunThread` event, and...+            Just cap <- getEventCap =<< mi+          , -- ...gather the end time of the previous event, and...+            Just startTimeUnixNano <- timeUnixNano =<< mi+          , -- ...gather the start time of the current event, and...+            Just endTimeUnixNano <- timeUnixNano j -> do+              -- ...yield a thread state span, and...+              yield . setThreadStateSpan j $+                ThreadStateSpan{threadState = Running cap, ..}+              -- ...keep the current event.+              go (Just j)+          --+          -- If the current event is any other event, then...+          | otherwise -> do+              -- ...emit an error, and...+              logWarning verbosity . T.pack $+                printf+                  "Thread %d: Unexpected event %s"+                  thread+                  (showEventInfo (getEventInfo j))+              --+              -- This case may trigger for any event that isn't `E.RunThread`+              -- or `E.StopThread` and for any `E.StopThread` event that comes+              -- before the first `E.RunThread` event. It may also trigger for+              -- any event for which `timeUnixNano` returns `Nothing`.+              --+              -- ...ignore it.+              go mi++-------------------------------------------------------------------------------+-- Internal Helpers+-------------------------------------------------------------------------------++{- |+Internal helper.+Check whether a t`ThreadStopStatus` is equal to `ThreadFinished`.+This is needed because t`ThreadStopStatus` does not define an `Eq` instance.+-}+isThreadFinished :: ThreadStopStatus -> Bool+isThreadFinished = \case+  ThreadFinished -> True+  _otherwise -> False++{- |+Internal helper.+Show `EventInfo` in a condensed format suitable for logging.+-}+showEventInfo :: EventInfo -> String+showEventInfo = \case+  E.RunThread{thread} -> printf "RunThread{%d}" thread+  E.StopThread{thread, status} -> printf "StopThread{%d,%s}" thread (E.showThreadStopStatus status)+  E.MigrateThread{thread} -> printf "MigrateThread{%d}" thread+  E.StartGC{} -> "StartGC"+  E.EndGC{} -> "EndGC"+  evSpec -> takeWhile (not . isSpace) . show $ evSpec++{- |+Internal helper. Return the minimal value by some projection.+-}+minBy :: (Ord b) => (a -> b) -> a -> a -> a+minBy f x y = if f x < f y then x else y

@@ -0,0 +1,494 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live.Machine.Core+Description : Core machines for processing data in batches.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Machine.Core (+  -- * Ticks+  Tick (..),+  batchByTick,+  batchToTick,+  batchListToTick,+  batchByTickList,+  liftTick,+  dropTick,+  onlyTick,+  liftBatch,++  -- * Debug+  counterBy,+  counterByTick,++  -- * Routers+  liftRouter,++  -- * Event sorting+  sortByBatch,+  sortByBatchTick,++  -- * Delimiting+  between,+  delimit,++  -- * Validation+  validateInput,+  validateOrder,+) where++import Control.Monad (when)+import Control.Monad.IO.Class (MonadIO (..))+import Data.DList qualified as D+import Data.Foldable (for_)+import Data.Function (on)+import Data.Functor ((<&>))+import Data.HashMap.Strict (HashMap)+import Data.HashMap.Strict qualified as M+import Data.Hashable (Hashable (..))+import Data.List qualified as L+import Data.Machine (Is (..), MachineT (..), Moore (..), PlanT, Process, ProcessT, Step (..), await, construct, encased, mapping, repeatedly, starve, stopped, yield, (~>))+import Data.Maybe (fromMaybe)+import Data.Semigroup (Max (..))+import Data.Text (Text)+import Data.Text qualified as T+import GHC.Eventlog.Live.Logger (logDebug, logError, logWarning)+import GHC.Eventlog.Live.Verbosity (Verbosity, verbosityDebug, verbosityError, verbosityWarning)+import GHC.RTS.Events (Event (..), Timestamp)+import GHC.RTS.Events qualified as E+import Text.Printf (printf)++-------------------------------------------------------------------------------+-- Ticks+-------------------------------------------------------------------------------++{- |+The type of data on a stream of items and ticks.++The t`Tick` type is isomorphic to `Maybe` modulo strictness,+but with the caveat that v`Tick` does not represent failure.+-}+data Tick a = Item !a | Tick+  deriving (Eq, Functor, Foldable, Traversable, Show)++{- |+This machine batches all items between two ticks into a list.+-}+batchByTickList :: Process (Tick a) [a]+batchByTickList =+  mapping (fmap D.singleton)+    ~> batchByTick+    ~> mapping D.toList++{- |+Generalised version of `batchByTickList`.+-}+batchByTick ::+  forall m a.+  (Monad m, Monoid a) =>+  ProcessT m (Tick a) a+batchByTick = batchByTickWith mempty+ where+  batchByTickWith :: a -> MachineT m (Is (Tick a)) a+  batchByTickWith acc = MachineT $ pure $ Await onNext Refl onStop+   where+    onNext :: Tick a -> MachineT m (Is (Tick a)) a+    onNext = \case+      Item a -> batchByTickWith (a <> acc)+      Tick -> MachineT $ pure $ Yield acc batchByTick+    onStop :: MachineT m (Is (Tick a)) a+    onStop = MachineT $ pure $ Yield acc stopped++{- |+This machine streams a list of items into a series of items+separated by ticks.+-}+batchListToTick :: Process [a] (Tick a)+batchListToTick = batchToTick++{- |+Generalised version of `batchListToTick`.+-}+batchToTick :: (Foldable f) => Process (f a) (Tick a)+batchToTick = repeatedly go+ where+  go = await >>= \xs -> for_ xs (yield . Item) >> yield Tick++{- |+This machine drops all ticks.+-}+dropTick :: Process (Tick a) a+dropTick =+  repeatedly $+    await >>= \case+      Item a -> yield a+      Tick -> pure ()++{- |+This machine drops all items.+-}+onlyTick :: Process (Tick a) ()+onlyTick =+  repeatedly $+    await >>= \case+      Tick -> yield ()+      Item{} -> pure ()++-------------------------------------------------------------------------------+-- Machine combinators+-------------------------------------------------------------------------------++{- |+This machine counts the number of inputs it received,+using the given function, and logs this value.+-}+counterBy ::+  forall m a x.+  (MonadIO m) =>+  Verbosity ->+  Text ->+  (a -> Word) ->+  ProcessT m a x+counterBy verbosity label count+  | verbosityDebug >= verbosity = repeatedly go+  | otherwise = stopped+ where+  go :: PlanT (Is a) x m ()+  go =+    await >>= \a ->+      logDebug verbosity ("saw " <> T.pack (show (count a)) <> " " <> label)++{- |+This machine counts the number of inputs it received,+and logs this value on every tick.+-}+counterByTick ::+  forall m a x.+  (MonadIO m) =>+  Verbosity ->+  Text ->+  ProcessT m (Tick a) x+counterByTick verbosity label+  | verbosityDebug >= verbosity = construct $ go 0+  | otherwise = stopped+ where+  go :: Word -> PlanT (Is (Tick a)) x m ()+  go count =+    await >>= \case+      Item _ -> go (count + 1)+      Tick -> logDebug verbosity ("saw " <> T.pack (show count) <> " " <> label) >> go 0++-------------------------------------------------------------------------------+-- Machine combinators+-------------------------------------------------------------------------------++--------------------------------------------------------------------------------+-- Lift a machine to a machine that passes on ticks unchanged++{- |+Lift a machine to a machine that passes on ticks unchanged.++Constructs the following machine:++@+           ┌─(if Tick)────────────────────┐+  [ Tick a ]                              [ Tick b ]+           └─(if Item)─( ProcessT m a b )─┘+@+-}+liftTick ::+  (Monad m) =>+  ProcessT m a b ->+  ProcessT m (Tick a) (Tick b)+liftTick m =+  MachineT $+    runMachineT m <&> \case+      Stop ->+        Stop+      Yield o k ->+        Yield (Item o) (liftTick k)+      Await (onNext :: t -> ProcessT m a b) Refl onStop ->+        await'+       where+        await' = Await onNext' Refl onStop'+         where+          onNext' :: Tick a -> ProcessT m (Tick a) (Tick b)+          onNext' = \case+            Tick ->+              MachineT . pure . Yield Tick $+                MachineT . pure $+                  await'+            Item a -> liftTick (onNext a)+          onStop' :: ProcessT m (Tick a) (Tick b)+          onStop' = liftTick onStop++--------------------------------------------------------------------------------+-- Lift a machine to a machine that operates on batches++{- |+Lift a machine that processes @a@s into @b@s to a machine that processes+batches of @a@s into batches of @b@s.+-}+liftBatch ::+  forall m a b.+  (Monad m) =>+  ProcessT m a b ->+  ProcessT m [a] [b]+liftBatch = MachineT . running [] []+ where+  -- The parent machine is running the child machine with the current batch.+  running :: [a] -> [b] -> ProcessT m a b -> m (Step (Is [a]) [b] (ProcessT m [a] [b]))+  running as bs m =+    runMachineT m >>= \case+      Stop ->+        pure Stop+      Yield b k ->+        running as (b : bs) k+      Await (onNext :: t -> ProcessT m a b) Refl onStop ->+        pure $ Yield (reverse bs) $ MachineT $ awaiting as onNext onStop++  -- The parent machine is awaiting new input.+  awaiting :: [a] -> (a -> ProcessT m a b) -> ProcessT m a b -> m (Step (Is [a]) [b] (ProcessT m [a] [b]))+  awaiting (a : as) onNext _onStop = running as [] $ onNext a+  awaiting [] onNext onStop = pure $ Await onNext' Refl onStop'+   where+    onNext' :: [a] -> ProcessT m [a] [b]+    onNext' as = MachineT $ awaiting as onNext onStop+    onStop' :: ProcessT m [a] [b]+    onStop' = exhausting onStop++  -- The parent machine is exhausting the child machine to gather its output.+  exhausting :: ProcessT m a b -> ProcessT m x [b]+  exhausting = MachineT . go []+   where+    go :: [b] -> ProcessT m a b -> m (Step (Is x) [b] (ProcessT m x [b]))+    go bs m =+      runMachineT m >>= \case+        Stop ->+          pure Stop+        Yield b k ->+          go (b : bs) k+        Await _onNext _Refl onStop ->+          pure $ Yield (reverse bs) $ MachineT $ go [] onStop++--------------------------------------------------------------------------------+-- Construct a processor that spawns a separate child processor for each measure++{- |+Spawn a copy of a machine for each "measure".++Constructs the following machine:++@+    ┌─────(if measure == k0)─( ProcessT m a b )────┐+  [ a ] ──(if measure == ..)─( ProcessT m a b )─ [ b ]+    └─────(if measure == kN)─( ProcessT m a b )────┘+@++__Warning:__ The router does not currently garbage-collect terminated child processors.+-}+liftRouter ::+  forall m k a b.+  (MonadIO m, Hashable k) =>+  -- | Function to measure.+  (a -> Maybe k) ->+  -- | Function to spawn child processors.+  (k -> ProcessT m a b) ->+  ProcessT m a b+liftRouter measure spawn = awaiting M.empty+ where+  awaiting :: HashMap k (ProcessT m a b) -> ProcessT m a b+  awaiting st = MachineT . pure $ Await onNext Refl onStop+   where+    onNext :: a -> MachineT m (Is a) b+    onNext a = case measure a of+      Nothing -> awaiting st+      Just k -> provideThen a m $ \m' -> awaiting (M.insert k m' st)+       where+        m = fromMaybe (spawn k) (M.lookup k st)+    onStop :: MachineT m (Is a) b+    onStop = foldr starve stopped (M.elems st)++  provideThen :: a -> ProcessT m a b -> (ProcessT m a b -> ProcessT m a b) -> ProcessT m a b+  provideThen a m k =+    MachineT $+      runMachineT m >>= \case+        Stop -> runMachineT (k stopped)+        Yield o m' -> pure (Yield o (provideThen a m' k))+        Await onNext Refl _onStop -> runMachineT (exhaustThen (onNext a) k)++  exhaustThen :: ProcessT m a b -> (ProcessT m a b -> ProcessT m a b) -> ProcessT m a b+  exhaustThen m k =+    MachineT $+      runMachineT m >>= \case+        Yield o m' -> pure (Yield o (k m'))+        m' -> runMachineT (k (encased m'))++-------------------------------------------------------------------------------+-- Event stream sorting+-------------------------------------------------------------------------------++{- |+Reorder events respecting ticks.++This machine caches two batches worth of events, sorts them together,+and then yields only those events whose timestamp is less than or equal+to the maximum of the first batch.+-}+sortByBatch ::+  forall m a.+  (Monad m) =>+  (a -> Timestamp) ->+  ProcessT m [a] [a]+sortByBatch timestamp = sortByBatchWith Nothing+ where+  sortByBatchWith :: Maybe [a] -> ProcessT m [a] [a]+  sortByBatchWith = \case+    Nothing -> MachineT $ pure $ Await onNext Refl onStop+     where+      onNext :: [a] -> ProcessT m [a] [a]+      onNext new = sortByBatchWith (Just sortedNew)+       where+        sortedNew = sortByTime new+      onStop :: ProcessT m [a] [a]+      onStop = stopped+    Just sortedOld -> MachineT $ pure $ Await onNext Refl onStop+     where+      onNext :: [a] -> ProcessT m [a] [a]+      onNext new+        | null sortedOld = sortByBatchWith $ Just sortedNew+        | otherwise = MachineT $ pure $ Yield sortedBeforeCutoff $ sortByBatchWith $ Just sortedAfterCutoff+       where+        -- NOTE: use of partial @maximum@ is guarded by the check @null old@.+        cutoff = getMax (foldMap (Max . timestamp) sortedOld)+        sortedNew = sortByTime new+        sorted = joinByTime sortedOld sortedNew+        (sortedBeforeCutoff, sortedAfterCutoff) = L.partition ((<= cutoff) . timestamp) sorted+      onStop :: ProcessT m [a] [a]+      onStop = MachineT $ pure $ Yield sortedOld $ stopped++  -- compByTime :: a -> a -> Ordering+  compByTime = compare `on` timestamp++  -- sortByTime :: [a] -> [a]+  sortByTime = L.sortBy compByTime++  -- joinByTime :: [a] -> [a] -> [a]+  joinByTime [] ys = ys+  joinByTime xs [] = xs+  joinByTime (x : xs) (y : ys) = case compByTime x y of+    LT -> x : joinByTime xs (y : ys)+    _ -> y : joinByTime (x : xs) ys++{- |+Variant of `sortByBatch` that operates on streams of items and ticks.+-}+sortByBatchTick :: (a -> Timestamp) -> Process (Tick a) (Tick a)+sortByBatchTick timestamp =+  mapping (fmap (: [])) ~> batchByTick ~> sortByBatch timestamp ~> batchListToTick++-------------------------------------------------------------------------------+-- Filtering semaphores+-------------------------------------------------------------------------------++{- | A simple delimiting t'Moore' machine,+which is opened by one constant marker and closed by the other one.+-}+between :: Text -> Text -> Moore Text Bool+between x y = open+ where+  open = Moore False open' where open' x' = if x == x' then close else open+  close = Moore True close' where close' y' = if y == y' then end else close+  end = Moore False (const end)++-- | Delimit the event process.+delimit :: (Monad m) => Moore Text Bool -> ProcessT m Event Event+delimit = construct . go+ where+  go :: (Monad m) => Moore Text Bool -> PlanT (Is Event) Event m ()+  go mm@(Moore s next) = do+    e <- await+    case evSpec e of+      -- on marker step the moore machine.+      E.UserMarker m -> do+        let mm'@(Moore s' _) = next m+        -- if current or next state is open (== True), emit the marker.+        when (s || s') $ yield e+        go mm'++      -- for other events, emit if the state is open.+      _ -> do+        when s $ yield e+        go mm++-------------------------------------------------------------------------------+-- Validation+-------------------------------------------------------------------------------++{- |+This machine validates that there is some input.++If no input is encountered after the given number of ticks, the machine prints+a warning that directs the user to check that the @-l@ flag was set correctly.+-}+validateInput ::+  (MonadIO m) =>+  Verbosity ->+  Int ->+  ProcessT m (Tick a) x+validateInput verbosity ticks+  | verbosityWarning >= verbosity = construct $ start ticks+  | otherwise = stopped+ where+  start remaining+    | remaining <= 0 = liftIO $ do+        logWarning verbosity . T.pack $+          printf+            "No input after %d ticks. Did you pass -l to the GHC RTS?"+            ticks+    | otherwise = do+        logDebug verbosity $+          T.pack (show remaining) <> " ticks remaining."+        await >>= \case+          Item{} -> do+            logDebug verbosity "Received item."+          Tick -> do+            logDebug verbosity "Received tick."+            start (pred remaining)++{- |+This machine validates that the inputs are received in order.++If an out-of-order input is encountered, the machine prints an error message+that directs the user to check that the @--eventlog-flush-interval@ and the+@--batch-interval@ flags are set correctly.+-}+validateOrder ::+  (MonadIO m, Show a) =>+  Verbosity ->+  (a -> Timestamp) ->+  ProcessT m a x+validateOrder verbosity timestamp+  | verbosityError >= verbosity = construct $ start Nothing+  | otherwise = stopped+ where+  start maybeOld =+    await >>= \new ->+      case maybeOld of+        Just old+          | timestamp new < timestamp old -> do+              logError verbosity . T.pack $+                "Encountered two out-of-order inputs.\n\+                \Did you pass --eventlog-flush-interval to the GHC RTS?\n\+                \Did you set --batch-interval to be at least as big as the value of --eventlog-flush-interval?"+              logDebug verbosity . T.pack $+                printf+                  "Out-of-order inputs:\n\+                  \- %s\n\+                  \- %s"+                  (show old)+                  (show new)+              pure ()+        _otherwise -> do+          start (Just new)

@@ -0,0 +1,49 @@+{- |+Module      : GHC.Eventlog.Live.Machine.Decoder+Description : Machines for processing eventlog data.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Machine.Decoder (+  -- * Event decoding+  DecodeError (..),+  decodeEvent,+  decodeEventBatch,+) where++import Control.Exception (Exception, throwIO)+import Control.Monad.IO.Class (MonadIO (..))+import Data.ByteString qualified as BS+import Data.Machine (Is, PlanT, ProcessT, await, construct, yield)+import GHC.Eventlog.Live.Machine.Core (Tick (..), liftTick)+import GHC.RTS.Events (Event)+import GHC.RTS.Events.Incremental (Decoder (..), decodeEventLog)++-------------------------------------------------------------------------------+-- Decoding events++{- |+Parse t'Event's from a stream of 'BS.ByteString' chunks with ticks.++Throws a t'DecodeError' on error.+-}+decodeEvent :: (MonadIO m) => ProcessT m BS.ByteString Event+decodeEvent = construct $ loop decodeEventLog+ where+  loop :: (MonadIO m) => Decoder a -> PlanT (Is BS.ByteString) a m ()+  loop Done{} = pure ()+  loop (Consume k) = await >>= \chunk -> loop (k chunk)+  loop (Produce a d') = yield a >> loop d'+  loop (Error _ err) = liftIO $ throwIO $ DecodeError err++{- |+Parse 'Event's from a stream of 'BS.ByteString' chunks with ticks.++Throws 'DecodeError' on error.+-}+decodeEventBatch :: (MonadIO m) => ProcessT m (Tick BS.ByteString) (Tick Event)+decodeEventBatch = liftTick decodeEvent++newtype DecodeError = DecodeError String deriving (Show)++instance Exception DecodeError

@@ -0,0 +1,42 @@+{- |+Module      : GHC.Eventlog.Live.Machine.Sink+Description : Machines for processing eventlog data.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Machine.Sink (+  -- * Eventlog file sink+  fileSink,+  fileSinkBatch,+) where++import Control.Monad.IO.Class (MonadIO (..))+import Data.ByteString qualified as BS+import Data.Machine (ProcessT, await, repeatedly, (~>))+import Data.Void (Void)+import GHC.Eventlog.Live.Machine.Core (Tick (..), dropTick)+import System.IO (Handle)++-------------------------------------------------------------------------------+-- Log file sink++{- |+File sink for optional eventlog log file.+-}+fileSink ::+  (MonadIO m) =>+  Handle ->+  ProcessT m BS.ByteString Void+fileSink handle = repeatedly $ await >>= liftIO . BS.hPut handle++-------------------------------------------------------------------------------+-- Log file sink with batches++{- |+File sink for optional eventlog log file.+-}+fileSinkBatch ::+  (MonadIO m) =>+  Handle ->+  ProcessT m (Tick BS.ByteString) Void+fileSinkBatch handle = dropTick ~> fileSink handle

@@ -0,0 +1,144 @@+{- |+Module      : GHC.Eventlog.Live.Machine.Source+Description : Machines for processing eventlog data.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Machine.Source (+  -- * Eventlog source+  sourceHandleWait,+  sourceHandleBatch,+  defaultChunkSizeBytes,+) where++import Control.Exception (catch, throwIO)+import Control.Monad.IO.Class (MonadIO (..))+import Data.ByteString qualified as BS+import Data.Function (fix)+import Data.Machine (MachineT (..), construct, yield)+import Data.Word (Word64)+import GHC.Clock (getMonotonicTimeNSec)+import GHC.Eventlog.Live.Machine.Core (Tick (..))+import System.IO (Handle, hWaitForInput)+import System.IO.Error (isEOFError)++-------------------------------------------------------------------------------+-- Socket source++{- |+A source which reads chunks from a `Handle`.+When an input is available, it yields an v`Item`.+When the timeout is reached, it yields a v`Tick`.+-}+sourceHandleWait ::+  (MonadIO m) =>+  -- | The wait timeout in milliseconds.+  Int ->+  -- | The number of bytes to read.+  Int ->+  -- | The eventlog socket handle.+  Handle ->+  MachineT m k (Tick BS.ByteString)+sourceHandleWait timeoutMilli chunkSizeBytes handle =+  construct $ fix $ \loop -> do+    ready <- liftIO $ hWaitForInput' handle timeoutMilli+    case ready of+      Ready -> do+        bs <- liftIO $ BS.hGetSome handle chunkSizeBytes+        yield (Item bs)+        loop+      NotReady -> do+        yield Tick+        loop+      EOF ->+        pure ()++-------------------------------------------------------------------------------+-- Socket source with batches++{- |+A source which reads chunks from a `Handle`.+When input is available, it yields an v`Item`.+It yields a v`Tick` at each increment of the batch interval.+-}+sourceHandleBatch ::+  (MonadIO m) =>+  -- | The batch interval in milliseconds.+  Int ->+  -- | The number of bytes to read.+  Int ->+  -- | The eventlog socket handle.+  Handle ->+  MachineT m k (Tick BS.ByteString)+sourceHandleBatch batchIntervalMs chunkSizeBytes handle = construct start+ where+  start = do+    startTimeMs <- liftIO getMonotonicTimeMilli+    batch startTimeMs+  batch startTimeMs = waitForInput+   where+    getRemainingTimeMilli = do+      currentTimeMilli <- liftIO getMonotonicTimeMilli+      pure $ (startTimeMs + batchIntervalMs) - currentTimeMilli+    waitForInput = do+      remainingTimeMilli <- getRemainingTimeMilli+      if remainingTimeMilli <= 0+        then do+          yield Tick+          start+        else do+          ready <- liftIO (hWaitForInput' handle remainingTimeMilli)+          case ready of+            Ready -> do+              chunk <- liftIO $ BS.hGetSome handle chunkSizeBytes+              yield (Item chunk) >> waitForInput+            NotReady -> waitForInput+            EOF -> pure ()++{- |+Eventlog chunk size in bytes.+This should be equal to the page size.+-}+defaultChunkSizeBytes :: Int+defaultChunkSizeBytes = 4096++{- |+Internal helper.+Return monotonic time in milliseconds, since some unspecified starting point+-}+getMonotonicTimeMilli :: IO Int+getMonotonicTimeMilli = nanoToMilli <$> getMonotonicTimeNSec++{- |+Internal helper.+Convert nanoseconds to milliseconds.+The conversion from 'Word64' to 'Int' is safe.+It cannot overflow due to the division by 1_000_000.+-}+nanoToMilli :: Word64 -> Int+nanoToMilli = fromIntegral . (`div` 1_000_000)++{- |+Internal helper.+Type to represent the state of a handle.+-}+data Ready = Ready | NotReady | EOF++{- |+Internal helper.+Wait for input from a `Handle` for a given number of milliseconds.+-}+hWaitForInput' ::+  -- | The handle.+  Handle ->+  -- | The timeout in milliseconds.+  Int ->+  IO Ready+hWaitForInput' handle timeoutMilli =+  catch (boolToReady <$> hWaitForInput handle timeoutMilli) handleEOFError+ where+  boolToReady True = Ready+  boolToReady False = NotReady+  handleEOFError err+    | isEOFError err = pure EOF+    | otherwise = throwIO err

@@ -0,0 +1,88 @@+{- |+Module      : GHC.Eventlog.Live.Machine.WithStartTime+Description : Machines for processing eventlog data.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Machine.WithStartTime (+  WithStartTime (..),+  setWithStartTime'value,+  tryGetTimeUnixNano,+  withStartTime,+  withStartTime',+  dropStartTime,+) where++import Control.Monad (forever)+import Data.Machine (Is (..), PlanT, Process, await, construct, mapping, yield)+import GHC.RTS.Events (Event (..), EventInfo, Timestamp)+import GHC.RTS.Events qualified as E++-------------------------------------------------------------------------------+-- Start time++{- |+Data decorated with a start time in nanoseconds since the Unix epoch.+-}+data WithStartTime a = WithStartTime+  { value :: !a+  , maybeStartTimeUnixNano :: !(Maybe Timestamp)+  }+  deriving (Functor, Show)++{- |+Setter for the value of a t`WithStartTime`+-}+setWithStartTime'value :: WithStartTime a -> b -> WithStartTime b+setWithStartTime'value (WithStartTime _a t) b = WithStartTime b t++{- |+If the event has a start time, return `Just` the time of the event in+nanoseconds since the Unix epoch. Otherwise, return `Nothing`.+-}+tryGetTimeUnixNano :: WithStartTime Event -> Maybe Timestamp+tryGetTimeUnixNano i = (i.value.evTime +) <$> i.maybeStartTimeUnixNano++{- |+Wrap every event in t`WithStartTime`. Every event after `E.WallClockTime` will+have its start time field set to `Just` the process start time.++This machine swallows the first and only `E.WallClockTime` event.+-}+withStartTime :: Process Event (WithStartTime Event)+withStartTime = withStartTime' E.evSpec WithStartTime++{- |+Generalised version of `withStartTime` that can be adapted to work on arbitrary+types using a getter and a setter.+-}+withStartTime' :: (a -> EventInfo) -> (a -> Maybe Timestamp -> b) -> Process a b+withStartTime' getEventInfo setStartTime = construct start+ where+  start =+    await >>= \case+      value+        -- The `WallClockTime` event announces the wall-clock time at which the+        -- process was started.+        | E.WallClockTime{..} <- getEventInfo value -> do+            -- This will start overflowing on Sunday, 21 July 2554 23:34:33, UTC.+            let !startTimeNs = sec * 1_000_000_000 + fromIntegral nsec+            -- We do not re-emit the `WallClockTime` event.+            continue startTimeNs+        | otherwise ->+            yield (value `setStartTime` Nothing) >> start+  continue startTimeUnixNano =+    mappingPlan $ \value ->+      value `setStartTime` Just startTimeUnixNano++{- |+Drop the t`WithStartTime` wrapper.+-}+dropStartTime :: Process (WithStartTime a) a+dropStartTime = mapping (.value)++{- |+Internal helper. Variant of `mapping` for plans.+-}+mappingPlan :: (a -> b) -> PlanT (Is a) b m a+mappingPlan f = forever (await >>= \a -> yield (f a))

@@ -0,0 +1,189 @@+{- |+Module      : GHC.Eventlog.Live.Options+Description : Command-line option parsers for eventlog machines.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Options (+  EventlogSource (..),+  eventlogSourceParser,+  eventlogSocketTimeoutParser,+  eventlogSocketTimeoutExponentParser,+  heapProfBreakdownParser,+  eventlogLogFileParser,+  batchIntervalParser,+  verbosityParser,+) where++import Control.Applicative (asum)+import Data.Char (toLower)+import GHC.Eventlog.Live.Machine.Analysis.Heap (heapProfBreakdownEitherReader)+import GHC.Eventlog.Live.Verbosity (Verbosity, verbosityDebug, verbosityError, verbosityInfo, verbosityQuiet, verbosityWarning)+import GHC.RTS.Events (HeapProfBreakdown (..))+import Options.Applicative qualified as O+import Text.Read (readEither)++--------------------------------------------------------------------------------+-- Eventlog Socket++{- |+The type of eventlog sockets.+-}+data EventlogSource+  = EventlogStdin+  | EventlogFile FilePath+  | EventlogSocketUnix FilePath++{- |+Parser for the eventlog socket.+-}+eventlogSourceParser :: O.Parser EventlogSource+eventlogSourceParser =+  asum+    [ stdinParser+    , fileParser+    , socketUnixParser+    ]+ where+  stdinParser =+    EventlogStdin+      <$ O.flag'+        ()+        ( O.long "eventlog-stdin"+            <> O.help "Read the eventlog from stdin."+        )+  fileParser =+    EventlogFile+      <$> O.strOption+        ( O.long "eventlog-file"+            <> O.metavar "FILE"+            <> O.help "Read the eventlog from a file."+        )+  socketUnixParser =+    EventlogSocketUnix+      <$> O.strOption+        ( O.long "eventlog-socket"+            <> O.metavar "SOCKET"+            <> O.help "Read the eventlog from a Unix socket."+        )++{- |+Parser for the intial timeout for exponential backoff.+-}+eventlogSocketTimeoutParser :: O.Parser Double+eventlogSocketTimeoutParser =+  O.option+    O.auto+    ( O.long "eventlog-socket-timeout"+        <> O.metavar "NUM"+        <> O.help "Eventlog socket connection retry timeout in microseconds."+        <> O.value 1+    )++{- |+Parser for the exponent for exponential backoff.+-}+eventlogSocketTimeoutExponentParser :: O.Parser Double+eventlogSocketTimeoutExponentParser =+  O.option+    O.auto+    ( O.long "eventlog-socket-exponent"+        <> O.metavar "NUM"+        <> O.help "Eventlog socket connection retry timeout exponent."+        <> O.value 1+    )++--------------------------------------------------------------------------------+-- Heap Profile Breakdown++{- |+Parser for the heap profile breakdown.+-}+heapProfBreakdownParser :: O.Parser HeapProfBreakdown+heapProfBreakdownParser =+  O.option+    (O.eitherReader heapProfBreakdownEitherReader)+    ( O.short 'h'+        <> O.metavar "Tcmdyrbi"+        <> O.help "Heap profile breakdown."+    )++--------------------------------------------------------------------------------+-- Eventlog Log File++{- |+Parser for the eventlog log file.+-}+eventlogLogFileParser :: O.Parser FilePath+eventlogLogFileParser =+  O.strOption+    ( O.long "eventlog-log-file"+        <> O.metavar "FILE"+        <> O.help "Use file to log binary eventlog data."+    )++--------------------------------------------------------------------------------+-- Batch Interval++{- |+Parser for the batch interval.+-}+batchIntervalParser :: O.Parser Int+batchIntervalParser =+  O.option+    O.auto+    ( O.long "batch-interval"+        <> O.metavar "NUM"+        <> O.help "Batch interval in milliseconds."+        <> O.value defaultBatchIntervalMs+    )++{- |+Internal helper.+The default batch interval in milliseconds.+-}+defaultBatchIntervalMs :: Int+defaultBatchIntervalMs = 1_000++--------------------------------------------------------------------------------+-- Verbosity++{- |+Parser for verbosities.+The default verbosity is `verbosityWarning`.+-}+verbosityParser :: O.Parser Verbosity+verbosityParser =+  O.option+    (O.eitherReader readEitherVerbosity)+    ( O.short 'v'+        <> O.long "verbosity"+        <> O.metavar "quiet|error|warning|info|debug|0-4"+        <> O.help "The verbosity threshold for logging."+        <> O.value verbosityWarning+    )++{- |+Internal helper.+Parser for verbosities by number or name.+Case insensitive.+-}+readEitherVerbosity :: String -> Either String Verbosity+readEitherVerbosity rawVerbosity =+  -- try to parse the verbosity as a number...+  case readEither @Word rawVerbosity of+    -- if the verbosity string is a number, map it to a verbosity...+    Right verbosityThreshold+      | verbosityThreshold <= 0 -> Right verbosityQuiet+      | verbosityThreshold == 1 -> Right verbosityError+      | verbosityThreshold == 2 -> Right verbosityWarning+      | verbosityThreshold == 3 -> Right verbosityInfo+      | otherwise -> Right verbosityDebug+    -- otherwise, match it against the literal names of the levels...+    Left _parseError -> case toLower <$> rawVerbosity of+      "quiet" -> Right verbosityQuiet+      "error" -> Right verbosityError+      "warning" -> Right verbosityWarning+      "info" -> Right verbosityInfo+      "debug" -> Right verbosityDebug+      _otherwise -> Left $ "Could not parse verbosity '" <> rawVerbosity <> "'."

@@ -0,0 +1,177 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live.Socket+Description : Utilities for running eventlog machines with sockets.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Socket (+  EventlogSource (..),+  Tick (..),+  tryConnect,+  runWithEventlogSource,+) where++import Control.Concurrent (threadDelay)+import Control.Exception (Exception (..))+import Control.Exception qualified as E+import Control.Monad.IO.Unlift (MonadUnliftIO (..))+import Data.Foldable (traverse_)+import Data.Machine (ProcessT, runT_, (~>))+import Data.Machine.Fanout (fanout)+import Data.Maybe (fromMaybe)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Void (Void)+import GHC.Eventlog.Live.Logger (logDebug, logInfo)+import GHC.Eventlog.Live.Machine.Core+import GHC.Eventlog.Live.Machine.Decoder+import GHC.Eventlog.Live.Machine.Sink+import GHC.Eventlog.Live.Machine.Source+import GHC.Eventlog.Live.Options (EventlogSource (..))+import GHC.Eventlog.Live.Verbosity (Verbosity)+import GHC.RTS.Events (Event)+import Network.Socket qualified as S+import System.IO (Handle)+import System.IO qualified as IO+import Text.Printf (printf)++{- |+Run an event processor with an eventlog socket.+-}+runWithEventlogSource ::+  (MonadUnliftIO m) =>+  -- | The logging verbosity.+  Verbosity ->+  -- | The eventlog socket handle.+  EventlogSource ->+  -- | The initial timeout in microseconds for exponential backoff.+  Double ->+  -- | The timeout exponent for exponential backoff.+  Double ->+  -- | The batch interval in milliseconds.+  Int ->+  -- | The number of bytes to read (defaults to 4KiB).+  Maybe Int ->+  -- | An optional file to which to stream binary eventlog data.+  Maybe FilePath ->+  -- | The event processor.+  ProcessT m (Tick Event) Void ->+  m ()+runWithEventlogSource verbosity eventlogSocket timeoutExponent initialTimeoutMcs batchIntervalMs maybeChuckSizeBytes maybeOutputFile toEventSink = do+  withEventlogSource verbosity timeoutExponent initialTimeoutMcs eventlogSocket $ \eventlogSource -> do+    let chuckSizeBytes = fromMaybe defaultChunkSizeBytes maybeChuckSizeBytes+    let fromSocket = sourceHandleBatch batchIntervalMs chuckSizeBytes eventlogSource+    case maybeOutputFile of+      Nothing ->+        runT_ $+          fromSocket ~> decodeEventBatch ~> toEventSink+      Just outputFile ->+        withRunInIO $ \runInIO ->+          IO.withFile outputFile IO.WriteMode $ \outputHandle -> do+            runInIO . runT_ $+              fromSocket+                ~> fanout+                  [ fileSinkBatch outputHandle+                  , decodeEventBatch ~> toEventSink+                  ]++{- |+Run an action with a `Handle` to an `EventlogSource`.+-}+withEventlogSource ::+  (MonadUnliftIO m) =>+  -- | The logging verbosity.+  Verbosity ->+  -- | The initial timeout in microseconds for exponential backoff.+  Double ->+  -- | The timeout exponent for exponential backoff.+  Double ->+  -- | The eventlog socket.+  EventlogSource ->+  (Handle -> m ()) ->+  m ()+withEventlogSource verbosity initialTimeoutMcs timeoutExponent eventlogSource action = do+  withRunInIO $ \runInIO ->+    case eventlogSource of+      EventlogStdin -> do+        logInfo verbosity "Reading eventlog from stdin"+        let enter = do+              maybeStdinTextEncoding <- IO.hGetEncoding IO.stdin+              IO.hSetBinaryMode IO.stdin True+              pure maybeStdinTextEncoding+        let leave maybeStdinTextEncoding = do+              traverse_ (IO.hSetEncoding IO.stdin) maybeStdinTextEncoding+              IO.hSetNewlineMode IO.stdin IO.nativeNewlineMode+        E.bracket enter leave . const . runInIO . action $ IO.stdin+      EventlogFile eventlogFile -> do+        logInfo verbosity $ "Reading eventlog from " <> T.pack eventlogFile+        IO.withBinaryFile eventlogFile IO.ReadMode $ \handle ->+          runInIO $ action handle+      EventlogSocketUnix eventlogSocketUnix -> do+        logInfo verbosity $ "Waiting to connect on " <> prettyEventlogSocketUnix eventlogSocketUnix+        E.bracket (connectRetry verbosity initialTimeoutMcs timeoutExponent eventlogSocketUnix) IO.hClose $ \handle ->+          runInIO $ action handle++{- |+Connect to an `EventlogSource` with retries and non-randomised exponential backoff.+-}+connectRetry ::+  -- | The logging verbosity.+  Verbosity ->+  -- | The initial timeout in microseconds for exponential backoff.+  Double ->+  -- | The timeout exponent for exponential backoff.+  Double ->+  -- | The eventlog socket.+  FilePath ->+  IO Handle+connectRetry verbosity initialTimeoutMcs timeoutExponent eventlogSocketUnix =+  connectLoop initialTimeoutMcs+ where+  waitFor :: Double -> IO ()+  waitFor timeoutMcs = threadDelay $ round $ timeoutMcs * 1_000_000++  connectLoop :: Double -> IO Handle+  connectLoop timeoutMcs = do+    let connect = do+          logDebug verbosity $ "Trying to connect on " <> prettyEventlogSocketUnix eventlogSocketUnix+          handle <- tryConnect eventlogSocketUnix+          logInfo verbosity $ "Connected on " <> prettyEventlogSocketUnix eventlogSocketUnix+          pure handle+    let cleanup (e :: E.IOException) = do+          logDebug verbosity $ "Failed to connect on " <> prettyEventlogSocketUnix eventlogSocketUnix <> ": " <> T.pack (displayException e)+          logDebug verbosity $ "Waiting " <> prettyTimeoutMcs timeoutMcs <> " to retry..."+          waitFor timeoutMcs+          connectLoop (timeoutMcs * timeoutExponent)+    E.catch connect cleanup++{- |+Try to connect to a Unix socket.+-}+tryConnect :: FilePath -> IO Handle+tryConnect eventlogSocketUnix =+  E.bracketOnError (S.socket S.AF_UNIX S.Stream S.defaultProtocol) S.close $ \socket -> do+    S.connect socket (S.SockAddrUnix eventlogSocketUnix)+    handle <- S.socketToHandle socket IO.ReadMode+    IO.hSetBuffering handle IO.NoBuffering+    pure handle++{- |+Interal helper. Pretty-printer for timeout values in microseconds.+-}+prettyTimeoutMcs :: Double -> Text+prettyTimeoutMcs timeoutMcs+  | timeoutMcs > 8.64e10 = T.pack $ printf "%.2f days" (timeoutMcs / 8.64e10)+  | timeoutMcs > 3.6e9 = T.pack $ printf "%.2f hours" (timeoutMcs / 3.6e9)+  | timeoutMcs > 6e7 = T.pack $ printf "%.2f minutes" (timeoutMcs / 6e7)+  | timeoutMcs > 1e6 = T.pack $ printf "%.2f seconds" (timeoutMcs / 1e6)+  | timeoutMcs > 1e3 = T.pack $ printf "%.2f milliseconds" (timeoutMcs / 1e3)+  | otherwise = T.pack $ printf "%.2f microseconds" timeoutMcs++{- |+Internal helper. Pretty-printer for eventlog sockets.+-}+prettyEventlogSocketUnix :: FilePath -> Text+prettyEventlogSocketUnix eventlogSocketUnix = "Unix socket " <> T.pack eventlogSocketUnix

@@ -0,0 +1,76 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live.Verbosity+Description : Logging verbosity for eventlog machines.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Verbosity (+  Verbosity,+  showVerbosity,+  verbosityQuiet,+  verbosityError,+  verbosityWarning,+  verbosityInfo,+  verbosityDebug,+) where++import Data.Text (Text)++-------------------------------------------------------------------------------+-- Verbosity+-------------------------------------------------------------------------------++{- |+The type of logging verbosities supported by the machines+in "GHC.Eventlog.Live.Machines".+-}+data Verbosity+  = VerbosityDebug+  | VerbosityInfo+  | VerbosityWarning+  | VerbosityError+  | VerbosityQuiet+  deriving (Eq, Ord)++{- |+Pretty-printer for t`Verbosity`.+-}+showVerbosity :: Verbosity -> Text+showVerbosity = \case+  VerbosityDebug -> "Debug"+  VerbosityInfo -> "Info"+  VerbosityWarning -> "Warning"+  VerbosityError -> "Error"+  VerbosityQuiet -> "Quiet"++{- |+Quiet t`Verbosity`.+-}+verbosityQuiet :: Verbosity+verbosityQuiet = VerbosityQuiet++{- |+Error t`Verbosity`.+-}+verbosityError :: Verbosity+verbosityError = VerbosityError++{- |+Warning t`Verbosity`.+-}+verbosityWarning :: Verbosity+verbosityWarning = VerbosityWarning++{- |+Info t`Verbosity`.+-}+verbosityInfo :: Verbosity+verbosityInfo = VerbosityInfo++{- |+Debug t`Verbosity`.+-}+verbosityDebug :: Verbosity+verbosityDebug = VerbosityDebug