diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,3 @@ +### 0.1.0.0 + +- Initial release. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2021 Well-Typed + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Oleg Grenrus nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/eventlog-live.cabal b/eventlog-live.cabal new file mode 100644 --- /dev/null +++ b/eventlog-live.cabal @@ -0,0 +1,132 @@ +cabal-version: 3.12 +name: eventlog-live +version: 0.1.0.0 +synopsis: Live processing of eventlog data. +description: + This package supports live processing of eventlog data. + It consists of three libraries: + + * The @eventlog-live@ library defines + [machines](https://hackage.haskell.org/package/machines) + for processing eventlog data. + * The @eventlog-live:options@ sub-library defines + [optparse-applicative](https://hackage.haskell.org/package/optparse-applicative) + parsers for common command-line arguments. + * The @eventlog-live:socket@ sub-library defines functions to read the + eventlog from a Unix socket, which depend on the + [network](https://hackage.haskell.org/package/network) package. + + This package is primarily intended for use via the + [eventlog-live-influxdb](https://hackage.haskell.org/package/eventlog-live-influxdb) + and + [eventlog-live-otelcol](https://hackage.haskell.org/package/eventlog-live-otelcol) + packages. + + [⚠️ Warning]: + This package is experimental. + It is versioned according to the [PVP](https://pvp.haskell.org). + However, breaking changes should be expected and no effort will be + made to avoid major version bumps until at least version @1.0.0.0@. + + For more information, see [the README](https://github.com/well-typed/eventlog-live#readme). + +license: AGPL-3.0-only +license-file: LICENSE +author: Wen Kokke +maintainer: wen@well-typed.com +copyright: (c) 2021-2025 Well-Typed +build-type: Simple +category: Debug, Monitoring, System +extra-doc-files: CHANGELOG.md +tested-with: + GHC ==9.2.8 + || ==9.4.8 + || ==9.6.7 + || ==9.8.4 + || ==9.10.2 + || ==9.12.2 + +source-repository head + type: git + location: https://github.com/well-typed/eventlog-live.git + subdir: eventlog-live + +common language + ghc-options: + -Wall -Wcompat -Widentities -Wprepositive-qualified-module + -Wredundant-constraints -Wunticked-promoted-constructors + -Wunused-packages + + default-language: Haskell2010 + default-extensions: + BangPatterns + ConstraintKinds + DataKinds + DeriveFoldable + DeriveFunctor + DeriveTraversable + DerivingStrategies + DuplicateRecordFields + FlexibleContexts + FlexibleInstances + GADTs + GeneralizedNewtypeDeriving + ImportQualifiedPost + InstanceSigs + KindSignatures + LambdaCase + MultiParamTypeClasses + NamedFieldPuns + NoFieldSelectors + NumericUnderscores + OverloadedRecordDot + RankNTypes + RecordWildCards + ScopedTypeVariables + TupleSections + TypeApplications + TypeFamilies + +library + import: language + hs-source-dirs: src-machines + exposed-modules: + GHC.Eventlog.Live.Machines + GHC.Eventlog.Live.Verbosity + + other-modules: GHC.Eventlog.Live.Internal.Logger + build-depends: + , base >=4.16 && <4.22 + , bytestring >=0.11 && <0.13 + , dlist >=1.0 && <1.1 + , ghc-events >=0.20 && <0.21 + , hashable >=1.4 && <1.6 + , machines >=0.7.4 && <0.8 + , text >=1.2 && <2.2 + , transformers >=0.5.6 && <0.7 + , unordered-containers >=0.2.20 && <0.3 + +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.19 && <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:options + , ghc-events >=0.20 && <0.21 + , machines >=0.7.4 && <0.8 + , network >=3.2.7 && <3.3 + , unliftio-core >=0.2.1 && <0.3 diff --git a/src-machines/GHC/Eventlog/Live/Internal/Logger.hs b/src-machines/GHC/Eventlog/Live/Internal/Logger.hs new file mode 100644 --- /dev/null +++ b/src-machines/GHC/Eventlog/Live/Internal/Logger.hs @@ -0,0 +1,40 @@ +{-# LANGUAGE OverloadedStrings #-} + +module GHC.Eventlog.Live.Internal.Logger ( + LogSource, + logMessage, + logError, + logWarning, +) where + +import Data.Text (Text) +import Data.Text.IO qualified as TIO +import GHC.Eventlog.Live.Verbosity (Verbosity, showVerbosity, verbosityError, verbosityWarning) +import System.IO qualified as IO + +{- | +Internal helper. Denotes the source of a log message. +-} +type LogSource = Text + +{- | +Internal helper. Log messages to `IO.stderr`. +Only prints a message if its verbosity level is above the verbosity threshold. +-} +logMessage :: Verbosity -> Verbosity -> LogSource -> Text -> IO () +logMessage verbosityLevel verbosityThreshold logSource msg + | verbosityLevel >= verbosityThreshold = + TIO.hPutStrLn IO.stderr . mconcat $ [logSource, ": ", showVerbosity verbosityLevel, ": ", msg] + | otherwise = pure () + +{- | +Internal helper. Log errors to `IO.stderr`. +-} +logError :: Verbosity -> LogSource -> Text -> IO () +logError = logMessage verbosityError + +{- | +Internal helper. Log warnings to `IO.stderr`. +-} +logWarning :: Verbosity -> LogSource -> Text -> IO () +logWarning = logMessage verbosityWarning diff --git a/src-machines/GHC/Eventlog/Live/Machines.hs b/src-machines/GHC/Eventlog/Live/Machines.hs new file mode 100644 --- /dev/null +++ b/src-machines/GHC/Eventlog/Live/Machines.hs @@ -0,0 +1,1996 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# OPTIONS_GHC -Wno-name-shadowing #-} + +{- | +Module : GHC.Eventlog.Live.Machines +Description : Machines for processing eventlog data. +Stability : experimental +Portability : portable +-} +module GHC.Eventlog.Live.Machines ( + -- * Eventlog source + sourceHandleWait, + sourceHandleBatch, + defaultChunkSizeBytes, + + -- * Eventlog file sink + fileSink, + fileSinkBatch, + + -- * Event decoding + decodeEvent, + decodeEventBatch, + + -- * Event processing + + -- ** Start time + WithStartTime (..), + setWithStartTime'value, + tryGetTimeUnixNano, + withStartTime, + withStartTime', + dropStartTime, + + -- ** Main thread ID + WithMainThreadId (..), + withMainThreadId, + withMainThreadId', + dropMainThreadId, + + -- ** Supplier + Stream (..), + supplier, + + -- ** 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, + + -- * Ticks + Tick (..), + batchByTick, + batchToTick, + batchListToTick, + batchByTickList, + liftTick, + dropTick, + onlyTick, + liftBatch, + + -- * Event sorting + sortByBatch, + sortByBatchTick, + + -- * Delimiting + between, + delimit, + + -- * Heap profile breakdown + heapProfBreakdownEitherReader, + heapProfBreakdownShow, + + -- * Data + + -- ** Metrics + Metric (..), + + -- ** Spans + IsSpan, + duration, + + -- ** Attributes + Attr, + AttrKey, + AttrValue (..), + IsAttrValue (..), + (~=), +) where + +import Control.Exception (Exception, catch, throwIO) +import Control.Monad (forever, unless, when) +import Control.Monad.IO.Class (MonadIO (..)) +import Data.ByteString qualified as BS +import Data.Char (isSpace) +import Data.DList qualified as D +import Data.Either (isLeft) +import Data.Foldable (for_) +import Data.Function (fix, on) +import Data.Functor ((<&>)) +import Data.HashMap.Strict (HashMap) +import Data.HashMap.Strict qualified as M +import Data.Hashable (Hashable (..)) +import Data.Int (Int16, Int32, Int64, Int8) +import Data.List qualified as L +import Data.Machine (Is (..), MachineT (..), Moore (..), PlanT, Process, ProcessT, Step (..), asParts, await, construct, encased, mapping, repeatedly, starve, stopped, yield, (~>)) +import Data.Machine.Fanout (fanout) +import Data.Machine.Mealy (unfoldMealy) +import Data.Machine.Process (Automaton (..)) +import Data.Maybe (fromMaybe, isNothing, listToMaybe, mapMaybe) +import Data.Semigroup (Max (..)) +import Data.Text (Text) +import Data.Text qualified as T +import Data.Void (Void) +import Data.Word (Word16, Word32, Word64, Word8) +import GHC.Clock (getMonotonicTimeNSec) +import GHC.Eventlog.Live.Internal.Logger (logError, logWarning) +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.RTS.Events.Incremental (Decoder (..), decodeEventLog) +import GHC.Records (HasField (..)) +import Numeric (showHex) +import System.IO (Handle, hWaitForInput) +import System.IO.Error (isEOFError) +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) + +------------------------------------------------------------------------------- +-- Event processing +------------------------------------------------------------------------------- + +------------------------------------------------------------------------------- +-- 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) + +------------------------------------------------------------------------------- +-- Main thread ID + +{- | +Data decorated with a main thread ID. +-} +data WithMainThreadId a = WithMainThreadId + { value :: !a + , maybeMainThreadId :: !(Maybe ThreadId) + } + deriving (Functor, Show) + +{- | +Wrap every event in t`WithMainThreadId`. Every event after the first `E.RunThread` +event will have its main thread ID set to `Just` the main thread ID. +-} +withMainThreadId :: Process Event (WithMainThreadId Event) +withMainThreadId = withMainThreadId' E.evSpec WithMainThreadId + +{- | +Generalised version of `withMainThreadId` that can be adapted to work on arbitrary +types using a getter and a setter. +-} +withMainThreadId' :: (a -> EventInfo) -> (a -> Maybe ThreadId -> b) -> Process a b +withMainThreadId' getEventInfo setMainThreadId = construct start + where + start = + await >>= \case + value + | E.RunThread{thread} <- getEventInfo value -> + yield (value `setMainThreadId` Just thread) >> continue thread + | otherwise -> + yield (value `setMainThreadId` Nothing) >> start + continue thread = + mappingPlan $ \value -> + value `setMainThreadId` Just thread + +{- | +Drop the t`WithMainThreadId` wrapper. +-} +dropMainThreadId :: Process (WithMainThreadId a) a +dropMainThreadId = mapping (.value) + +------------------------------------------------------------------------------- +-- Supplier + +{- | +An infinite stream of values. +-} +data Stream a = Cons !a (Stream a) + +{- | +Supply each function with the next element from an infinite stream of values. +-} +supplier :: Stream s -> Process (s -> a) a +supplier supply = + auto $ + unfoldMealy (\(Cons x xs) f -> (f x, xs)) supply + +------------------------------------------------------------------------------- +-- 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 + +{- | +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 + +{- | +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) + +{- | +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... + liftIO . logError 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... + liftIO . 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) + +{- | +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) + +{- | +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... + liftIO . logError 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 + liftIO . 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) $ + liftIO . 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 + liftIO . 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 + liftIO . 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 + +------------------------------------------------------------------------------- +-- Reading from the socket +------------------------------------------------------------------------------- + +------------------------------------------------------------------------------- +-- 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 + +------------------------------------------------------------------------------- +-- Writing to a file +------------------------------------------------------------------------------- + +------------------------------------------------------------------------------- +-- 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 + +------------------------------------------------------------------------------- +-- 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 :: (Monoid a) => Process (Tick a) a +batchByTick = construct start + where + start = batch mempty + batch acc = + await >>= \case + Item a -> batch (a <> acc) + Tick -> yield acc >> start + +{- | +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 () + +------------------------------------------------------------------------------- +-- Decoding +------------------------------------------------------------------------------- + +------------------------------------------------------------------------------- +-- 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 + +------------------------------------------------------------------------------- +-- 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 :: (a -> Timestamp) -> Process [a] [a] +sortByBatch timestamp = construct $ go mempty + where + go old = + await >>= \case + new + | null old -> go (sortByTime new) + | otherwise -> yield before >> go after + where + -- NOTE: use of partial @maximum@ is guarded by the check @null old@. + cutoff = getMax (foldMap (Max . timestamp) old) + sorted = joinByTime old (sortByTime new) + (before, after) = L.partition ((<= cutoff) . timestamp) sorted + + -- compByTime :: a -> a -> Ordering + compByTime = compare `on` timestamp + + -- sortByTime :: [a] -> [a] + sortByTime = L.sortBy compByTime + + -- joinByTime :: [a] -> [a] -> [a] + joinByTime = go + where + go [] ys = ys + go xs [] = xs + go (x : xs) (y : ys) = case compByTime x y of + LT -> x : go xs (y : ys) + _ -> y : go (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 + +------------------------------------------------------------------------------- +-- Metrics +------------------------------------------------------------------------------- + +{- | +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) + +{- | +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 + } + +------------------------------------------------------------------------------- +-- Spans +------------------------------------------------------------------------------- + +{- | +Internal helper. A span is any type with a start and end time. +-} +type IsSpan s = (HasField "startTimeUnixNano" s Timestamp, HasField "endTimeUnixNano" s Timestamp) + +{- | +Internal helper. Determine the duration of a span. +-} +{-# SPECIALIZE duration :: CapabilityUsageSpan -> Timestamp #-} +{-# SPECIALIZE duration :: GCSpan -> Timestamp #-} +{-# SPECIALIZE duration :: MutatorSpan -> Timestamp #-} +{-# SPECIALIZE duration :: ThreadStateSpan -> Timestamp #-} +duration :: (IsSpan s) => s -> Timestamp +duration s = if s.startTimeUnixNano < s.endTimeUnixNano then s.endTimeUnixNano - s.startTimeUnixNano else 0 + +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) + +------------------------------------------------------------------------------- +-- Attributes +------------------------------------------------------------------------------- + +{- | +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 CapabilityUser where + toAttrValue :: CapabilityUser -> AttrValue + toAttrValue = toAttrValue . show + {-# INLINE toAttrValue #-} + +instance (IsAttrValue v) => IsAttrValue (Maybe v) where + toAttrValue :: Maybe v -> AttrValue + toAttrValue = maybe AttrNull toAttrValue + {-# INLINE toAttrValue #-} + +------------------------------------------------------------------------------- +-- Internal Helpers +------------------------------------------------------------------------------- + +{- | +Internal helper. Variant of `mapping` for plans. +-} +mappingPlan :: (a -> b) -> PlanT (Is a) b m a +mappingPlan f = forever (await >>= \a -> yield (f a)) + +{- | +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 diff --git a/src-machines/GHC/Eventlog/Live/Verbosity.hs b/src-machines/GHC/Eventlog/Live/Verbosity.hs new file mode 100644 --- /dev/null +++ b/src-machines/GHC/Eventlog/Live/Verbosity.hs @@ -0,0 +1,58 @@ +{-# 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, +) where + +import Data.Text (Text) + +------------------------------------------------------------------------------- +-- Verbosity +------------------------------------------------------------------------------- + +{- | +The type of logging verbosities supported by the machines +in "GHC.Eventlog.Live.Machines". +-} +data Verbosity + = VerbosityWarning + | VerbosityError + | VerbosityQuiet + deriving (Eq, Ord) + +{- | +Pretty-printer for t`Verbosity`. +-} +showVerbosity :: Verbosity -> Text +showVerbosity = \case + 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 diff --git a/src-options/GHC/Eventlog/Live/Options.hs b/src-options/GHC/Eventlog/Live/Options.hs new file mode 100644 --- /dev/null +++ b/src-options/GHC/Eventlog/Live/Options.hs @@ -0,0 +1,163 @@ +{- | +Module : GHC.Eventlog.Live.Options +Description : Command-line option parsers for eventlog machines. +Stability : experimental +Portability : portable +-} +module GHC.Eventlog.Live.Options ( + EventlogSocket (..), + eventlogSocketParser, + eventlogSocketTimeoutParser, + eventlogSocketTimeoutExponentParser, + heapProfBreakdownParser, + eventlogLogFileParser, + batchIntervalParser, + verbosityParser, +) where + +import Data.Char (toLower) +import GHC.Eventlog.Live.Machines (heapProfBreakdownEitherReader) +import GHC.Eventlog.Live.Verbosity (Verbosity, verbosityError, verbosityQuiet, verbosityWarning) +import GHC.RTS.Events (HeapProfBreakdown (..)) +import Options.Applicative qualified as O +import Text.Read (readEither) + +-------------------------------------------------------------------------------- +-- Eventlog Socket + +{- | +The type of eventlog sockets. +-} +newtype EventlogSocket + = EventlogSocketUnix FilePath + +{- | +Parser for the eventlog socket. +-} +eventlogSocketParser :: O.Parser EventlogSocket +eventlogSocketParser = socketUnixParser + where + socketUnixParser = + EventlogSocketUnix + <$> O.strOption + ( O.long "eventlog-socket" + <> O.metavar "SOCKET" + <> O.help "Eventlog 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 microseconds." + <> 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 "NUM|quiet|error|warning" + <> 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 + | otherwise -> Right verbosityWarning + -- 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 + _otherwise -> Left $ "Could not parse verbosity '" <> rawVerbosity <> "'." diff --git a/src-socket/GHC/Eventlog/Live/Socket.hs b/src-socket/GHC/Eventlog/Live/Socket.hs new file mode 100644 --- /dev/null +++ b/src-socket/GHC/Eventlog/Live/Socket.hs @@ -0,0 +1,119 @@ +{- | +Module : GHC.Eventlog.Live.Socket +Description : Utilities for running eventlog machines with sockets. +Stability : experimental +Portability : portable +-} +module GHC.Eventlog.Live.Socket ( + EventlogSocket (..), + Tick (..), + tryConnect, + runWithEventlogSocket, +) where + +import Control.Concurrent (threadDelay) +import Control.Exception qualified as E +import Control.Monad.IO.Unlift (MonadUnliftIO (..)) +import Data.Machine (ProcessT, runT_, (~>)) +import Data.Machine.Fanout (fanout) +import Data.Maybe (fromMaybe) +import Data.Void (Void) +import GHC.Eventlog.Live.Machines +import GHC.Eventlog.Live.Options (EventlogSocket (..)) +import GHC.RTS.Events (Event) +import Network.Socket qualified as S +import System.IO (Handle) +import System.IO qualified as IO + +{- | +Run an event processor with an eventlog socket. +-} +runWithEventlogSocket :: + (MonadUnliftIO m) => + -- | The eventlog socket handle. + EventlogSocket -> + -- | 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 () +runWithEventlogSocket eventlogSocket timeoutExponent initialTimeoutMcs batchIntervalMs maybeChuckSizeBytes maybeOutputFile toEventSink = do + -- TODO: Handle connection errors by waiting for the socket to be created. + withEventlogSocket timeoutExponent initialTimeoutMcs eventlogSocket $ \eventlogHandle -> do + let chuckSizeBytes = fromMaybe defaultChunkSizeBytes maybeChuckSizeBytes + let fromSocket = sourceHandleBatch batchIntervalMs chuckSizeBytes eventlogHandle + 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 `EventlogSocket`. +-} +withEventlogSocket :: + (MonadUnliftIO m) => + -- | The initial timeout in microseconds for exponential backoff. + Double -> + -- | The timeout exponent for exponential backoff. + Double -> + -- | The eventlog socket. + EventlogSocket -> + (Handle -> m ()) -> + m () +withEventlogSocket initialTimeoutMcs timeoutExponent eventlogSocket action = do + withRunInIO $ \runInIO -> + E.bracket (connectRetry initialTimeoutMcs timeoutExponent eventlogSocket) IO.hClose $ \handle -> + runInIO $ + action handle + +{- | +Connect to an `EventlogSocket` with retries and non-randomised exponential backoff. +-} +connectRetry :: + -- | The initial timeout in microseconds for exponential backoff. + Double -> + -- | The timeout exponent for exponential backoff. + Double -> + -- | The eventlog socket. + EventlogSocket -> + IO Handle +connectRetry initialTimeoutMcs timeoutExponent eventlogSocket = + connectLoop initialTimeoutMcs + where + waitFor :: Double -> IO () + waitFor timeoutMcs = threadDelay $ round $ timeoutMcs * 1_000_000 + + connectLoop :: Double -> IO Handle + connectLoop timeoutMcs = do + E.catch (tryConnect eventlogSocket) $ \(_e :: E.IOException) -> do + waitFor timeoutMcs + connectLoop (timeoutMcs * timeoutExponent) + +{- | +Try to connect to an `EventlogSocket`. +-} +tryConnect :: EventlogSocket -> IO Handle +tryConnect = \case + EventlogSocketUnix socketName -> + E.bracketOnError (S.socket S.AF_UNIX S.Stream S.defaultProtocol) S.close $ \socket -> do + S.connect socket (S.SockAddrUnix socketName) + handle <- S.socketToHandle socket IO.ReadMode + IO.hSetBuffering handle IO.NoBuffering + pure handle