From dd5eef03489bba0ccfa4ebd79250dd05dce80eca Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Tue, 28 Apr 2026 18:06:53 +0530 Subject: [PATCH 01/14] Remove the code example from docs It is now under examples dir --- docs/TraceEventIO.md | 24 ------------------------ 1 file changed, 24 deletions(-) delete mode 100644 docs/TraceEventIO.md diff --git a/docs/TraceEventIO.md b/docs/TraceEventIO.md deleted file mode 100644 index f1e9293..0000000 --- a/docs/TraceEventIO.md +++ /dev/null @@ -1,24 +0,0 @@ -# WAI Middleware - -``` -import GHC.Conc (labelThread, myThreadId) -import Debug.Trace (traceEventIO) - -withEventLog :: Application -> Application -withEventLog app request respond = do - tid <- myThreadId - labelThread tid "server" - traceEventIO ("START:server") - app request respond1 - - where - - respond1 r = do - res <- respond r - traceEventIO ("END:server") - return res -``` - -``` -runSettings settings $ withEventLog $ app -``` From 0deffd3d058b21ccca0a94708660bdeeb5945843 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Tue, 28 Apr 2026 18:11:12 +0530 Subject: [PATCH 02/14] Rename README to more specific title "metrics" --- dev/{README.md => metrics.md} | 0 haskell-perf.cabal | 5 ++++- 2 files changed, 4 insertions(+), 1 deletion(-) rename dev/{README.md => metrics.md} (100%) diff --git a/dev/README.md b/dev/metrics.md similarity index 100% rename from dev/README.md rename to dev/metrics.md diff --git a/haskell-perf.cabal b/haskell-perf.cabal index 040a566..7787d62 100644 --- a/haskell-perf.cabal +++ b/haskell-perf.cabal @@ -23,7 +23,10 @@ build-type: Simple extra-doc-files: Changelog.md README.md - dev/GhcFlags.md dev/Methods.md dev/README.md dev/ghc-work.md + dev/GhcFlags.md + dev/Methods.md + dev/metrics.md + dev/ghc-work.md docs/*.md examples/README.md From 5958277363c1f037eb0ec1483b144d1cceaa8a45 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Tue, 28 Apr 2026 21:03:15 +0530 Subject: [PATCH 03/14] Update ghc rts stats measurement mechanism --- docs/ghc-rts-performance-analysis.md | 121 +++++++++++++++++---------- 1 file changed, 75 insertions(+), 46 deletions(-) diff --git a/docs/ghc-rts-performance-analysis.md b/docs/ghc-rts-performance-analysis.md index f114341..a4d94a2 100644 --- a/docs/ghc-rts-performance-analysis.md +++ b/docs/ghc-rts-performance-analysis.md @@ -1,28 +1,32 @@ - In a multithreaded program using RTS stats we can only tell time how much total CPU time (and allocations) the entire Haskell process (all threads) spent between two points, but we cannot tell which Haskell -thread spent how much time. ---> +thread spent how much time or how much time was actually spent by the +instructions between those two points. -# Components of a Haskell Process +## Components of a Haskell Process * An OS level process * Multiple OS level threads in the OS process @@ -30,19 +34,18 @@ thread spent how much time. threads can run on any of the available OS threads every time it is ready to run. -# A prototypical program +## A prototypical program -A simple yet comprehensive program to understand different components of +[This example](examples/console-loop-multi-thread.hs) is a +a simple yet comprehensive program to understand different components of performance analysis and stats. You can play with this to understand how things work, how the stats add up and what they mean. -See examples/console-loop-multi-thread.hs . +## How many OS threads do we have? -# How many OS threads do we have? - -To see how may OS threads a haskell process is using on Linux. -Run examples/console-loop-multi-thread.hs , note its pid printed in the output. -All of its OS threads are: +To see how many OS threads a Haskell process is using on Linux. Run +[this example](examples/console-loop-multi-thread.hs), note its pid +printed in the output. All of its OS threads can be printed by: ``` ls /proc//task ``` @@ -62,7 +65,7 @@ specified with the -N rts option. Usually we see 3 threads plus 2 threads per capability when compiled with `-threaded` option. -# GHC RTS stats +## GHC RTS stats The getRTSStats call gives us the CPU time (essentially get_clocktime or getrusage under the hood to get the CPU time) of the process and @@ -72,7 +75,7 @@ getRTSStats at point B and diff them? There are two problems with this. (1) the allocation count is recorded from the last GC which does not correspond to point A or point B unless we force a GC at both the points which is not practical and is going to -change the performance characterstics of the program drastically. (2) +change the performance characteristics of the program drastically. (2) the CPU time that we get is for the entire process which includes all the OS threads, if the program is built with `-threaded` option then this is always going to be inaccurate; even if we get OS thread level @@ -91,17 +94,10 @@ This is used in this way in micro-benchmarking programs but it is very restrictive and useless in practice. For small programs though the `+RTS -s` options is very useful to assess the -performance characterstics. I often take out the small piece that I want to +performance characteristics. I often take out the small piece that I want to measure and run it with `+RTS -s`. -# Using it - -In the haskell-perf library we do have a convenient way to wrap a function -around and use getRTSStats before and after it. If the function passes the -criterion mentioned above then we can get a decent measurement, not very -accurate but workable. We automatically perform a GC before and after. - -# Interpreting the RTS Stats +## Interpreting the RTS Stats We divide the stats in two categories. The first category is the non-gc stats, these stats are accurate up to the time of the `getRTSStats` @@ -125,20 +121,53 @@ in this category. Note that the GC cpu time should be computed by adding the `gc_cpu_ns` and `nonmoving_gc_cpu_ns` when the non-moving gc is enabled. -# Variability of Measurements +## Variability of Measurements Performance measurement is tricky and there are many factors to take care of if you want to get reliable results: -* cannot use wall-clock time, need to use process cpu time -* disable CPU frequency scaling -* Memory contention can affect the measurement, do not run other things on the - same machine. -* cache effect due to context switching can affect it, do not run other things - on the same machine. +* Disable CPU frequency scaling, can cause run-to-run or variability in the + same run. +* Do not run other things on the same machine. interrupts, kernel + activity, background daemons can also affect: + * Memory contention can affect the measurement. + * cache effects due to context switching can affect it. +* Discard first runs, first runs are usually outliers because of warm up effects, + instruction cache cold, data cache cold, page faults, branch predictor + not trained. +* Use thread affinity. Thread migration to another CPU: causes cache + invalidation, different core state, timing noise. +* Use larger measurements. In smaller one measurement overhead and + variance may dominate: timing calls (clock_gettime), counters, RTS + stats. * Different CPUs running at different frequencies can make the results unpredictable. -* The clocks of different CPUs may not be in sync. +* The clocks of different CPUs may not be perfectly in sync. + +To counter the last two factors we should use instruction count or +allocation count rather than time as a more reliable measure. Even +the instruction count might vary because of measurement overhead adds +instruction count, which can vary depending on how many times the thread +is context switched. + +## Haskell specific variability + +* Lazy evaluation, may defer work which might get evaluated later in the + context of some other measurement window. + +## Using getRTSStats with haskell-perf + +In the `haskell-perf` library we do have a convenient way to wrap a +function around to use `getRTSStats` before and after it and print the +resulting stats. If the function passes the criterion mentioned above +then we can get a decent measurement, not very accurate but workable. We +automatically perform a GC before and after. + +## Summary: -To counter the last two factors we should use instruction count or allocation -count rather than time as a reliable measure. +* `+RTS -s` option is very useful in assessing the behavior of the entire + program. +* `getRTSStats` can be used to measure the timing of a piece of code if we are + single threaded, the thread does not yield during the measurement, we are + forcing GCs for roughly correct allocation counts. GCs happening in the + middle of the measurement can add to the noise. From 519064c4fbcad80616b231688d1f56bb07a561b3 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Wed, 29 Apr 2026 00:45:57 +0530 Subject: [PATCH 04/14] Restore the previously deleted streamly-metrics modules --- haskell-perf.cabal | 113 ++++++++-- lib/Streamly/KeyValue/Type.hs | 65 ++++++ lib/Streamly/Metrics/Channel.hs | 113 ++++++++++ lib/Streamly/Metrics/Channel/Common.hs | 48 ++++ lib/Streamly/Metrics/Channel/Unbounded.hs | 79 +++++++ lib/Streamly/Metrics/Console.hs | 9 + lib/Streamly/Metrics/File.hs | 11 + lib/Streamly/Metrics/Measure.hs | 62 +++++ lib/Streamly/Metrics/Perf.hs | 120 ++++++++++ lib/Streamly/Metrics/Perf/RUsage.hsc | 229 +++++++++++++++++++ lib/Streamly/Metrics/Perf/Type.hs | 263 ++++++++++++++++++++++ lib/Streamly/Metrics/Type.hs | 208 +++++++++++++++++ src/Main.hs | 4 +- test/Main.hs | 32 +++ 14 files changed, 1329 insertions(+), 27 deletions(-) create mode 100644 lib/Streamly/KeyValue/Type.hs create mode 100644 lib/Streamly/Metrics/Channel.hs create mode 100644 lib/Streamly/Metrics/Channel/Common.hs create mode 100644 lib/Streamly/Metrics/Channel/Unbounded.hs create mode 100644 lib/Streamly/Metrics/Console.hs create mode 100644 lib/Streamly/Metrics/File.hs create mode 100644 lib/Streamly/Metrics/Measure.hs create mode 100644 lib/Streamly/Metrics/Perf.hs create mode 100644 lib/Streamly/Metrics/Perf/RUsage.hsc create mode 100644 lib/Streamly/Metrics/Perf/Type.hs create mode 100644 lib/Streamly/Metrics/Type.hs create mode 100644 test/Main.hs diff --git a/haskell-perf.cabal b/haskell-perf.cabal index 7787d62..629b667 100644 --- a/haskell-perf.cabal +++ b/haskell-perf.cabal @@ -32,6 +32,8 @@ extra-doc-files: extra-source-files: examples/*.hs + lib/Streamly/Metrics/File.hs + lib/Streamly/Metrics/Console.hs source-repository head type: git @@ -52,14 +54,13 @@ flag dev ------------------------------------------------------------------------------- common default-extensions + -- In GHC 2024 default-extensions: BangPatterns - CApiFFI ConstraintKinds DeriveDataTypeable DeriveGeneric DeriveTraversable - DoAndIfThenElse ExistentialQuantification FlexibleContexts FlexibleInstances @@ -67,24 +68,20 @@ common default-extensions InstanceSigs KindSignatures LambdaCase - MagicHash MultiParamTypeClasses - PatternSynonyms RankNTypes - RecordWildCards ScopedTypeVariables TupleSections TypeApplications - TypeFamilies - ViewPatterns + TypeOperators - -- MonoLocalBinds, enabled by TypeFamilies, causes performance - -- regressions. Disable it. This must come after TypeFamilies, - -- otherwise TypeFamilies will enable it again. - NoMonoLocalBinds - - -- UndecidableInstances -- Does not show any perf impact - -- UnboxedTuples -- interferes with (#.) + -- Not in GHC2024 + default-extensions: + CPP + CApiFFI + MagicHash + PatternSynonyms + RecordWildCards common compile-options import: default-extensions @@ -96,6 +93,7 @@ common compile-options -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missing-local-signatures + -Wno-missing-role-annotations -Wno-missing-safe-haskell-mode -Wno-missed-specialisations -Wno-all-missed-specialisations @@ -113,6 +111,71 @@ common compile-options -Wall-missed-specialisations -fno-ignore-asserts +common optimization-options + ghc-options: -O2 + -fdicts-strict + -fspec-constr-recursive=16 + -fmax-worker-args=16 + +common lib-options + import: compile-options, optimization-options + +common test-options + import: compile-options, optimization-options + ghc-options: -with-rtsopts -T + +------------------------------------------------------------------------------- +-- Library +------------------------------------------------------------------------------- + +library + import: lib-options + + hs-source-dirs: lib + exposed-modules: + Streamly.KeyValue.Type + , Streamly.Metrics.Type + , Streamly.Metrics.Perf + , Streamly.Metrics.Perf.Type + , Streamly.Metrics.Perf.RUsage + , Streamly.Metrics.Channel + , Streamly.Metrics.Channel.Unbounded + , Streamly.Metrics.Channel.Common + , Streamly.Metrics.Measure + -- , Streamly.Metrics.File + -- , Streamly.Metrics.Console + + build-depends: + base >= 4.9 && < 5 + , pretty-show >= 1.10 && < 1.11 + , stm >= 2.5.3 && < 2.6 + , streamly >= 0.11.0 && < 0.12 + , streamly-core >= 0.3.0 && < 0.4 + +------------------------------------------------------------------------------- +-- Executables +------------------------------------------------------------------------------- + +executable hperf + import: compile-options, optimization-options + hs-source-dirs: src + main-is: Main.hs + other-modules: Aggregator, EventParser + build-depends: + base >= 4.9 && < 5 + , containers < 0.9 + , format-numbers < 0.2 + , optparse-applicative < 0.20 + , streamly-core >= 0.3.0 && < 0.4 + , streamly-statistics < 0.3 + , text < 2.2 + +------------------------------------------------------------------------------- +-- Examples +------------------------------------------------------------------------------- + +-- XXX Use examples flag to compile these + executable console-loop-multi-thread import: compile-options hs-source-dirs: examples @@ -124,17 +187,17 @@ executable console-loop-multi-thread if os(windows) buildable: False -executable hperf - import: compile-options - hs-source-dirs: src +------------------------------------------------------------------------------- +-- Tests +------------------------------------------------------------------------------- + +test-suite basic + import: test-options + type: exitcode-stdio-1.0 + hs-source-dirs: test main-is: Main.hs - other-modules: Aggregator, EventParser ghc-options: -O2 -fmax-worker-args=16 -fspec-constr-recursive=16 build-depends: - base >= 4.9 && < 5 - , containers < 0.9 - , optparse-applicative < 0.20 - , streamly-core >= 0.3.0 && < 0.4 - , streamly-statistics < 0.3 - , format-numbers < 0.2 - , text < 2.2 + base + , haskell-perf + , streamly-core diff --git a/lib/Streamly/KeyValue/Type.hs b/lib/Streamly/KeyValue/Type.hs new file mode 100644 index 0000000..18d90e9 --- /dev/null +++ b/lib/Streamly/KeyValue/Type.hs @@ -0,0 +1,65 @@ +-- | +-- Module : Streamly.KeyValue.Type +-- Copyright : (c) 2021 Composewell Technologies +-- License : Apache-2.0 +-- Maintainer : streamly@composewell.com +-- Stability : experimental +-- Portability : GHC +-- +module Streamly.KeyValue.Type + ( + -- * KeyValue + KeyValue (..) + , Zip (..) + ) +where + +import Control.Exception (assert) +import Prelude hiding (zip, zipWith) + +-- XXX Use the approach like in https://hackage.haskell.org/package/keys? Or we +-- could use a lens based approach? However, the more we abstract the less +-- comprehensible it becomes. +-- +-- XXX We could also use a columnar storage (arrays) to store the keys and +-- values. Further, keys could be stored in boxed arrays whereas the values +-- could be stored in unboxed arrays. + +-- | A key value represents a sample of some value labeled with a unique key of +-- type k and having a value of type v. +-- +data KeyValue k v = KeyValue !k !v + +------------------------------------------------------------------------------- +-- Instances +------------------------------------------------------------------------------- + +instance Functor (KeyValue k) where + fmap f (KeyValue k v) = KeyValue k (f v) + +-- | Append two key values using the Semigroup instance of the underlying +-- value. +-- +instance (Eq k, Semigroup v) => Semigroup (KeyValue k v) where + {-# INLINE (<>) #-} + -- XXX Only KeyValue having the same key make sense to be combined. + -- However, matching the keys at run time would incur some cost depending + -- on the comparison function. + KeyValue k1 v1 <> KeyValue k2 v2 = + assert (k1 == k2) (KeyValue k1 (v1 <> v2)) + +-- Functors that can be Zipped. +class Functor f => Zip f where + zipWith :: (a -> b -> c) -> f a -> f b -> f c + zipWith f a b = uncurry f <$> zip a b + + zip :: f a -> f b -> f (a, b) + zip = zipWith (,) + + -- | Zip applicative + zap :: f (a -> b) -> f a -> f b + zap = zipWith id + +instance Eq k => Zip (KeyValue k) where + zipWith f (KeyValue k1 v1) (KeyValue k2 v2) = + assert (k1 == k2) $ KeyValue k1 (f v1 v2) diff --git a/lib/Streamly/Metrics/Channel.hs b/lib/Streamly/Metrics/Channel.hs new file mode 100644 index 0000000..9d01ef1 --- /dev/null +++ b/lib/Streamly/Metrics/Channel.hs @@ -0,0 +1,113 @@ +module Streamly.Metrics.Channel + ( + Channel + , newChannel + , send + , printChannel + , forkChannelPrinter + , benchOn + , benchOnWith + ) +where + +import Control.Concurrent (forkIO, ThreadId) +import Control.Concurrent.STM (atomically) +import Control.Concurrent.STM.TBQueue + (TBQueue, newTBQueue, readTBQueue, writeTBQueue) +import Control.Monad.IO.Class (liftIO, MonadIO) +import Data.Bifunctor (second) +import Data.Function ((&)) +import Data.Maybe (fromJust, isJust) +import Streamly.Data.Stream (Stream) +import Streamly.Internal.Data.Time.Clock (getTime, Clock (Monotonic)) +import Streamly.Internal.Data.Time.Units (AbsTime) +import Streamly.Metrics.Perf.Type (PerfMetrics(..)) +import Streamly.Metrics.Perf (benchWith) +import Streamly.Metrics.Type (showList, Indexable) +import Streamly.Data.Stream.Prelude (MonadAsync) + +import qualified Streamly.Data.Fold as Fold +import qualified Streamly.Internal.Data.Fold as Fold +import qualified Streamly.Data.Stream as Stream +import qualified Streamly.Internal.Data.Stream.Prelude as Stream + +import Prelude hiding (showList) + +------------------------------------------------------------------------------- +-- Event processing +------------------------------------------------------------------------------- + +-- XXX Use streamly SVar instead so that we do not need STM and we can use just +-- one channel type. + +-- | A metrics channel. +newtype Channel a = Channel (TBQueue (AbsTime, ([Char], [a]))) + +-- | Create a new metrics channel. +newChannel :: IO (Channel a) +newChannel = atomically $ do + tbq <- newTBQueue 1 + return $ Channel tbq + +-- | Send a list of metrics to a metrics channel. +-- @send channel description metrics@ +send :: MonadIO m => Channel a -> String -> [a] -> m () +send (Channel chan) desc metrics = do + -- XXX should use asyncClock + now <- liftIO $ getTime Monotonic + liftIO $ atomically $ writeTBQueue chan (now, (desc, metrics)) + +fromChan :: MonadAsync m => TBQueue a -> Stream m a +fromChan = Stream.repeatM . (liftIO . atomically . readTBQueue) + +aggregateListBy :: (MonadAsync m, Ord k, Fractional a) => + Double -> Int -> Stream m (AbsTime, (k, [a])) -> Stream m (k, [a]) +aggregateListBy timeout batchsize stream = + fmap (second fromJust) + $ Stream.filter (isJust . snd) + $ Stream.classifySessionsBy + 0.1 False (return . (> 1000)) timeout f stream + + where + + scale Nothing _ = Nothing + scale (Just xs) count = Just $ map (/ count) xs + + f = + Fold.teeWithFst + scale + (Fold.take batchsize (Fold.foldl1' (zipWith (+)))) + (Fold.lmap (const 1) Fold.sum) + +printKV :: (MonadIO m, Show k, Show a, Indexable a) => Stream m (k, [a]) -> m b +printKV stream = + let f (k, xs) = liftIO $ putStrLn $ show k ++ ":\n" ++ showList xs + in Stream.fold (Fold.drainMapM f) stream >> error "printChannel: Metrics channel closed" + +-- XXX Print actual batch size and also scale the results per event. + +-- | Forever print the metrics on a channel to the console periodically after +-- aggregating the metrics collected till now. +printChannel :: (MonadAsync m, Show a, Fractional a, Indexable a) => + Channel a -> Double -> Int -> m b +printChannel (Channel chan) timeout batchSize = + fromChan chan + & aggregateListBy timeout batchSize + & printKV + +forkChannelPrinter :: (MonadAsync m, Show a, Fractional a, Indexable a) => + Channel a -> Double -> Int -> m ThreadId +forkChannelPrinter chan timeout = liftIO . forkIO . printChannel chan timeout + +-- | Benchmark a function application and send the results to the specified +-- metrics channel. +benchOnWith :: Channel PerfMetrics -> String -> (a -> IO b) -> a -> IO b +benchOnWith chan desc f arg = do + (r, xs) <- benchWith f arg + send chan desc (Count 1 : xs) + return r + +-- | Like 'benchOnWith' but benchmark an action instead of function +-- application. +benchOn :: Channel PerfMetrics -> String -> IO b -> IO b +benchOn chan desc f = benchOnWith chan desc (const f) () diff --git a/lib/Streamly/Metrics/Channel/Common.hs b/lib/Streamly/Metrics/Channel/Common.hs new file mode 100644 index 0000000..e7c963f --- /dev/null +++ b/lib/Streamly/Metrics/Channel/Common.hs @@ -0,0 +1,48 @@ +module Streamly.Metrics.Channel.Common + ( + aggregateListBy + , printKV + ) +where + +import Control.Monad.IO.Class (liftIO, MonadIO) +import Data.Bifunctor (second) +import Data.Maybe (fromJust, isJust) +import Streamly.Internal.Data.Time.Units (AbsTime) +import Streamly.Metrics.Type (showList, Indexable) +import Streamly.Data.Stream (Stream) +import Streamly.Data.Stream.Prelude (MonadAsync) + +import qualified Streamly.Internal.Data.Fold as Fold +import qualified Streamly.Data.Stream as Stream +import qualified Streamly.Internal.Data.Stream.Prelude as Stream + +import Prelude hiding (showList) + +------------------------------------------------------------------------------- +-- Event processing +------------------------------------------------------------------------------- + +aggregateListBy :: (MonadAsync m, Ord k, Fractional a) => + Double -> Int -> Stream m (AbsTime, (k, [a])) -> Stream m (k, [a]) +aggregateListBy timeout batchsize stream = + fmap (second fromJust) + $ Stream.filter (isJust . snd) + $ Stream.classifySessionsBy + 0.1 False (return . (> 1000)) timeout f stream + + where + + scale Nothing _ = Nothing + scale (Just xs) count = Just $ map (/ count) xs + + f = + Fold.teeWithFst + scale + (Fold.take batchsize (Fold.foldl1' (zipWith (+)))) + (Fold.lmap (const 1) Fold.sum) + +printKV :: (MonadIO m, Show k, Show a, Indexable a) => Stream m (k, [a]) -> m b +printKV stream = + let f (k, xs) = liftIO $ putStrLn $ show k ++ ":\n" ++ showList xs + in Stream.fold (Fold.drainMapM f) stream >> error "printChannel: Metrics channel closed" diff --git a/lib/Streamly/Metrics/Channel/Unbounded.hs b/lib/Streamly/Metrics/Channel/Unbounded.hs new file mode 100644 index 0000000..1df3fa8 --- /dev/null +++ b/lib/Streamly/Metrics/Channel/Unbounded.hs @@ -0,0 +1,79 @@ +module Streamly.Metrics.Channel.Unbounded + ( + Channel + , newChannel + , send + , printChannel + , forkChannelPrinter + , benchOn + , benchOnWith + ) +where + +import Control.Concurrent (forkIO, ThreadId, yield) +import Control.Concurrent.Chan +import Control.Monad.IO.Class (liftIO, MonadIO) +import Data.Function ((&)) +import Streamly.Internal.Data.Time.Clock (getTime, Clock (Monotonic)) +import Streamly.Internal.Data.Time.Units (AbsTime) +import Streamly.Metrics.Perf.Type (PerfMetrics(..)) +import Streamly.Metrics.Perf (benchWith) +import Streamly.Metrics.Type (Indexable) +import Streamly.Data.Stream (Stream) +import Streamly.Data.Stream.Prelude (MonadAsync) + +import qualified Streamly.Data.Stream as Stream + +import Prelude hiding (showList) +import Streamly.Metrics.Channel.Common + +------------------------------------------------------------------------------- +-- Event processing +------------------------------------------------------------------------------- + +-- | An unbounded metrics channel. TBChannel should be preferred but this can +-- be used when STM cannot be used e.g. if TBChannel would create nested STM +-- transactions. +newtype Channel a = Channel (Chan (AbsTime, ([Char], [a]))) + +-- | Create a new metrics channel. +newChannel :: IO (Channel a) +newChannel = Channel <$> newChan + +-- | Send a list of metrics to a metrics channel. +-- @send channel description metrics@ +send :: MonadIO m => Channel a -> String -> [a] -> m () +send (Channel chan) desc metrics = do + -- XXX should use asyncClock + now <- liftIO $ getTime Monotonic + liftIO $ writeChan chan (now, (desc, metrics)) + liftIO yield + +fromChan :: MonadAsync m => Chan a -> Stream m a +fromChan = Stream.repeatM . (liftIO . readChan) + +-- | Forever print the metrics on a channel to the console periodically after +-- aggregating the metrics collected till now. +printChannel :: (MonadAsync m, Show a, Fractional a, Indexable a) => + Channel a -> Double -> Int -> m b +printChannel (Channel chan) timeout batchSize = + fromChan chan + & aggregateListBy timeout batchSize + & printKV + +forkChannelPrinter :: (MonadAsync m, Show a, Fractional a, Indexable a) => + Channel a -> Double -> Int -> m ThreadId +forkChannelPrinter chan timeout = liftIO . forkIO . printChannel chan timeout + +-- | Benchmark a function application and the send the results to the specified +-- metrics channel. +benchOnWith :: Channel PerfMetrics -> String -> (a -> IO b) -> a -> IO b +benchOnWith chan desc f arg = do + (r, xs) <- benchWith f arg + send chan desc (Count 1 : xs) + return r + +-- | Like 'benchOnWith' but benchmark an action instead of function +-- application. +benchOn :: Channel PerfMetrics -> String -> IO b -> IO b +benchOn chan desc f = benchOnWith chan desc (const f) () diff --git a/lib/Streamly/Metrics/Console.hs b/lib/Streamly/Metrics/Console.hs new file mode 100644 index 0000000..6f8a60c --- /dev/null +++ b/lib/Streamly/Metrics/Console.hs @@ -0,0 +1,9 @@ +module Streamly.Metrics.Console + ( + ) +where + +import Streamly.Metrics.Type +import Streamly.Internal.Data.Tuple.Strict (Tuple'(..)) +import Streamly.Internal.Data.Array (Array) +import Streamly.Internal.Data.Time.Units (AbsTime) diff --git a/lib/Streamly/Metrics/File.hs b/lib/Streamly/Metrics/File.hs new file mode 100644 index 0000000..c79ba0c --- /dev/null +++ b/lib/Streamly/Metrics/File.hs @@ -0,0 +1,11 @@ +module Streamly.Metrics.File + ( + ) +where + +import Streamly.Metrics.Type +import Streamly.Internal.Data.Tuple.Strict (Tuple'(..)) +import Streamly.Internal.Data.Array (Array) +import Streamly.Internal.Data.Time.Units (AbsTime) +import Data.Word (Word64) + diff --git a/lib/Streamly/Metrics/Measure.hs b/lib/Streamly/Metrics/Measure.hs new file mode 100644 index 0000000..4b6e38f --- /dev/null +++ b/lib/Streamly/Metrics/Measure.hs @@ -0,0 +1,62 @@ +module Streamly.Metrics.Measure + ( + measureWith + , measure + , tick + , timestamp + ) +where + +import Streamly.Internal.Data.Time.Units (AbsTime) + +------------------------------------------------------------------------------- +-- Useful metric generator functions +------------------------------------------------------------------------------- + +-- We can create a measurement Monad in which we provide the following +-- commands: +-- +-- * start -- start measurement +-- * end -- end measurement +-- +-- One or more perf events are generated by "end". Options/config can be +-- provided to "start" (use startWith) for controlling what kind of +-- measurements will be done and what events will be generated. The event will +-- be identified by both start and end ids. We can use end multiple times. + +-- | Apply a monadic function to its argument with pre and post action hooks +-- and return a tuple containing the result of the action as well as the result +-- of the post hook. +-- +-- The function is exception safe. If an exception occurs the post hook is run +-- and the exception is rethrown. +-- +-- Any pure computations that are dependent on @b@ are computed every time. +-- However anything pure that does not depend on @b@ will be precomputed and +-- won't take any time. If you want to measure pure computations make sure all +-- the inputs are supplied dynamically as funciton arguments. +-- +{-# INLINE measureWith #-} +measureWith :: Monad m => m a -> (a -> m d) -> (b -> m c) -> b -> m (c, d) +measureWith pre post func arg = do + r <- pre + -- When using in pure code with unsafePerformIO forcing the return value + -- could be useful. + !v <- func arg + r1 <- post r + return (v, r1) + +-- | Like 'measureWith' but using an action instead of a function and its +-- argument. +{-# INLINE measure #-} +measure :: Monad m => m a -> (a -> m d) -> m c -> m (c, d) +measure pre post action = measureWith pre post (const action) () + +-- | Return a @()@ value as side effect every time a pure value is used. +{-# INLINE tick #-} +tick :: Monad m => a -> m (a, ()) +tick a = measure (pure ()) pure (pure a) + +-- | Return the current timestamp as a side effect. +timestamp :: a -> m (a, AbsTime) +timestamp = undefined diff --git a/lib/Streamly/Metrics/Perf.hs b/lib/Streamly/Metrics/Perf.hs new file mode 100644 index 0000000..fdd0c82 --- /dev/null +++ b/lib/Streamly/Metrics/Perf.hs @@ -0,0 +1,120 @@ +module Streamly.Metrics.Perf + ( + PerfMetrics(..) + , benchWith + , bench + , preRun + , postRun + ) +where + +import Control.Monad (unless) +import Data.Maybe (catMaybes) +import GHC.Stats (getRTSStats, getRTSStatsEnabled, RTSStats(..)) +import Streamly.Internal.Data.Time.Units (NanoSecond64, fromAbsTime) +import Streamly.Metrics.Measure (measureWith) +import Streamly.Metrics.Perf.Type (PerfMetrics(..), checkMonotony) +import Streamly.Metrics.Perf.RUsage (getRuMetrics, pattern RUsageSelf) +import Text.Show.Pretty (ppShow) + +import qualified Streamly.Internal.Data.Time.Clock as Clock + +{-# INLINE getProcMetrics #-} +getProcMetrics :: IO [PerfMetrics] +getProcMetrics = do + time <- Clock.getTime Clock.Monotonic + tcpu <- Clock.getTime Clock.ThreadCPUTime + pcpu <- Clock.getTime Clock.ProcessCPUTime + + let tcpuSec = fromIntegral (fromAbsTime tcpu :: NanoSecond64) * 1e-9 + let pcpuSec = fromIntegral (fromAbsTime pcpu :: NanoSecond64) * 1e-9 + let timeSec = fromIntegral (fromAbsTime time :: NanoSecond64) * 1e-9 + return + [ MonotonicTime timeSec + , ProcessCPUTime pcpuSec + , ThreadCPUTime tcpuSec + ] + +-- Compatible with GHC 8.2 (base 4.10) onwards +{-# INLINE getGcMetrics #-} +getGcMetrics :: IO [PerfMetrics] +getGcMetrics = do + res <- getRTSStatsEnabled + if res + then do + stats <- getRTSStats + pure + [ GcAllocatedBytes (fromIntegral (allocated_bytes stats)) + , GcCopiedBytes (fromIntegral (copied_bytes stats)) + , GcMaxLiveBytes (fromIntegral (max_live_bytes stats)) + , GcMaxLargeObjectBytes (fromIntegral (max_large_objects_bytes stats)) + , GcMaxCompactBytes (fromIntegral (max_compact_bytes stats)) + , GcMaxMemInUse (fromIntegral (max_mem_in_use_bytes stats)) + , GcMutatorCpuTime (fromIntegral (mutator_cpu_ns stats) / 1e9) + , GcMutatorElapsedTime + (fromIntegral (mutator_elapsed_ns stats) / 1e9) + , GcGcCpuTime (fromIntegral (gc_cpu_ns stats) / 1e9) + , GcGcElapsedTime (fromIntegral (gc_elapsed_ns stats) / 1e9) + , GcCpuTime (fromIntegral (cpu_ns stats) / 1e9) + , GcElapsedTime (fromIntegral (elapsed_ns stats) / 1e9) + ] + else pure [] + +{-# INLINE getPerfMetrics #-} +getPerfMetrics :: IO [PerfMetrics] +getPerfMetrics = do + procMetrics <- getProcMetrics + gcMetrics <- getGcMetrics + ruMetrics <- getRuMetrics RUsageSelf + return $ concat [procMetrics, gcMetrics, ruMetrics] + +{-# INLINE preRun #-} +preRun :: IO [PerfMetrics] +preRun = do + -- XXX If we have nested perf measurement calls then it is a bad idea to + -- perform GC. + -- performGC + getPerfMetrics + +{-# INLINE postRun #-} +postRun :: [PerfMetrics] -> IO [PerfMetrics] +postRun stats = do + -- We should not leave garbage behind. Any GC work is part of the + -- function being benchmarked. We start with no garbage before we start + -- measuring and we leave no garbage behind. However, this also adds a + -- constant overhead of GC which would otherwise be lesser if we do not + -- perform GC too often. So this may show inflated cpu times but the + -- numbers would be consistent across iterations. + -- + -- XXX We can have this as an option. + -- XXX The allocations and cycles consumed by the measuring functions + -- also add to the benchmark stats. We could run a dummy function and + -- measure the overhead and deduct that overhead. However, this is pretty + -- small and should not matter when benchmarking large functions. + -- performGC + stats1 <- getPerfMetrics + let diff = zipWith (-) stats1 stats + r = + catMaybes + $ zipWith (\old new -> + if checkMonotony old new + then Nothing + else Just (new, old)) stats stats1 + unless (null r) $ do + putStrLn "Warning: perf stat diff validation failed. " + putStrLn "(New, old) =" + putStrLn $ ppShow r + putStrLn "New =" + putStrLn $ ppShow stats1 + putStrLn "Old =" + putStrLn $ ppShow stats + return diff + +-- | Benchmark a function application returning the function output and the +-- performance metrics. +benchWith :: (a -> IO b) -> a -> IO (b, [PerfMetrics]) +benchWith = measureWith preRun postRun + +-- | Like 'benchWith' but benchmark an action instead. +bench :: IO a -> IO (a, [PerfMetrics]) +bench action = benchWith (const action) () diff --git a/lib/Streamly/Metrics/Perf/RUsage.hsc b/lib/Streamly/Metrics/Perf/RUsage.hsc new file mode 100644 index 0000000..7140047 --- /dev/null +++ b/lib/Streamly/Metrics/Perf/RUsage.hsc @@ -0,0 +1,229 @@ +module Streamly.Metrics.Perf.RUsage + ( + pattern RUsageSelf + , pattern RUsageChildren +-- , RUsageThread + , getRuMetrics + , RUsage(..) + , getRUsage + ) +where + +import Control.Applicative () +import Data.Word (Word64) +import Foreign.C.Error (throwErrnoIfMinus1_) +import Foreign.C.Types (CInt(..), CLong) +import Foreign.Marshal.Alloc (alloca) +import Foreign.Ptr (Ptr) +import Foreign.Storable (Storable(..)) +import Streamly.Metrics.Perf.Type (PerfMetrics(..)) +import Streamly.Metrics.Type (GaugeMax(..), Seconds(..), Bytes(..)) + +#include +#include + +------------------------------------------------------------------------------- +-- Unsafe cast operations +------------------------------------------------------------------------------- + +clongToW64 :: CLong -> Word64 +clongToW64 = fromIntegral + +w64ToCLong :: Word64 -> CLong +w64ToCLong = fromIntegral + +------------------------------------------------------------------------------- +-- struct timeval +------------------------------------------------------------------------------- + +data TimeVal = + TimeVal + {-# UNPACK #-} !Word64 -- sec + {-# UNPACK #-} !Word64 -- usec + deriving (Show, Eq) + +instance Storable TimeVal where + alignment _ = 8 + + sizeOf _ = #const sizeof(struct timeval) + + peek p = do + s <- (#peek struct timeval, tv_sec) p + us <- (#peek struct timeval, tv_usec) p + return $ TimeVal (clongToW64 s) (clongToW64 us) + + poke p (TimeVal s us) = do + (#poke struct timeval, tv_sec) p (w64ToCLong s) + (#poke struct timeval, tv_usec) p (w64ToCLong us) + +{------------------------------------------------------------------------------ +The resource usages are returned in the structure pointed to by usage, which +has the following form: + + struct rusage { + struct timeval ru_utime; /* user CPU time used */ + struct timeval ru_stime; /* system CPU time used */ + long ru_maxrss; /* maximum resident set size */ + long ru_ixrss; /* integral shared memory size */ + long ru_idrss; /* integral unshared data size */ + long ru_isrss; /* integral unshared stack size */ + long ru_minflt; /* page reclaims (soft page faults) */ + long ru_majflt; /* page faults (hard page faults) */ + long ru_nswap; /* swaps */ + long ru_inblock; /* block input operations */ + long ru_oublock; /* block output operations */ + long ru_msgsnd; /* IPC messages sent */ + long ru_msgrcv; /* IPC messages received */ + long ru_nsignals; /* signals received */ + long ru_nvcsw; /* voluntary context switches */ + long ru_nivcsw; /* involuntary context switches */ + }; +------------------------------------------------------------------------------} + +data RUsage = RUsage + { ru_utime :: {-# UNPACK #-} !Double -- seconds + , ru_stime :: {-# UNPACK #-} !Double -- seconds + , ru_maxrss :: {-# UNPACK #-} !Word64 -- Bytes + , ru_ixrss :: {-# UNPACK #-} !Word64 -- Bytes + , ru_idrss :: {-# UNPACK #-} !Word64 -- Bytes + , ru_isrss :: {-# UNPACK #-} !Word64 -- Bytes + , ru_minflt :: {-# UNPACK #-} !Word64 + , ru_majflt :: {-# UNPACK #-} !Word64 + , ru_nswap :: {-# UNPACK #-} !Word64 + , ru_inblock :: {-# UNPACK #-} !Word64 + , ru_oublock :: {-# UNPACK #-} !Word64 + , ru_msgsnd :: {-# UNPACK #-} !Word64 + , ru_msgrcv :: {-# UNPACK #-} !Word64 + , ru_nsignals :: {-# UNPACK #-} !Word64 + , ru_nvcsw :: {-# UNPACK #-} !Word64 + , ru_nivcsw :: {-# UNPACK #-} !Word64 + } deriving (Show, Eq) + +-- | convert TimeVal to seconds +timeValToDouble :: TimeVal -> Double +timeValToDouble (TimeVal s us) = + fromIntegral s + fromIntegral us * 1e-6 + +-- | convert seconds to TimeVal +doubleToTimeVal :: Double -> TimeVal +doubleToTimeVal sec = + let (s, us) = round (sec * 1e6) `divMod` (10^(6::Int)) + in TimeVal s us + +instance Storable RUsage where + alignment _ = 8 + + sizeOf _ = #const sizeof(struct rusage) + + peek p = + RUsage + <$> (timeValToDouble <$> (#peek struct rusage, ru_utime) p) + <*> (timeValToDouble <$> (#peek struct rusage, ru_stime) p) + <*> ((* 1024) . clongToW64 <$> (#peek struct rusage, ru_maxrss) p) + <*> ((* 1024) . clongToW64 <$> (#peek struct rusage, ru_ixrss ) p) + <*> ((* 1024) . clongToW64 <$> (#peek struct rusage, ru_idrss ) p) + <*> ((* 1024) . clongToW64 <$> (#peek struct rusage, ru_isrss ) p) + <*> (clongToW64 <$> (#peek struct rusage, ru_minflt ) p) + <*> (clongToW64 <$> (#peek struct rusage, ru_majflt ) p) + <*> (clongToW64 <$> (#peek struct rusage, ru_nswap ) p) + <*> (clongToW64 <$> (#peek struct rusage, ru_inblock ) p) + <*> (clongToW64 <$> (#peek struct rusage, ru_oublock ) p) + <*> (clongToW64 <$> (#peek struct rusage, ru_msgsnd ) p) + <*> (clongToW64 <$> (#peek struct rusage, ru_msgrcv ) p) + <*> (clongToW64 <$> (#peek struct rusage, ru_nsignals) p) + <*> (clongToW64 <$> (#peek struct rusage, ru_nvcsw ) p) + <*> (clongToW64 <$> (#peek struct rusage, ru_nivcsw ) p) + + poke p RUsage{..} = do + (#poke struct rusage, ru_utime) p (doubleToTimeVal ru_utime) + (#poke struct rusage, ru_stime) p (doubleToTimeVal ru_stime) + (#poke struct rusage, ru_maxrss) p (w64ToCLong (ru_maxrss `div` 1024)) + (#poke struct rusage, ru_ixrss) p (w64ToCLong (ru_ixrss `div` 1024)) + (#poke struct rusage, ru_idrss) p (w64ToCLong (ru_idrss `div` 1024)) + (#poke struct rusage, ru_isrss) p (w64ToCLong (ru_isrss `div` 1024)) + (#poke struct rusage, ru_minflt) p (w64ToCLong ru_minflt) + (#poke struct rusage, ru_majflt) p (w64ToCLong ru_majflt) + (#poke struct rusage, ru_nswap) p (w64ToCLong ru_nswap) + (#poke struct rusage, ru_inblock) p (w64ToCLong ru_inblock) + (#poke struct rusage, ru_oublock) p (w64ToCLong ru_oublock) + (#poke struct rusage, ru_msgsnd) p (w64ToCLong ru_msgsnd) + (#poke struct rusage, ru_msgrcv) p (w64ToCLong ru_msgrcv) + (#poke struct rusage, ru_nsignals) p (w64ToCLong ru_nsignals) + (#poke struct rusage, ru_nvcsw) p (w64ToCLong ru_nvcsw) + (#poke struct rusage, ru_nivcsw) p (w64ToCLong ru_nivcsw) + +------------------------------------------------------------------------------- +-- int getrusage(int who, struct rusage *usage); +------------------------------------------------------------------------------- + +-- | See "man getrusage". +foreign import ccall unsafe "getrusage" c_getrusage :: + CInt -> Ptr RUsage -> IO CInt + +-- | "who" could be: +-- RUsageSelf +-- RUsageChildren +getRuMetrics :: CInt -> IO [PerfMetrics] +getRuMetrics who = do + alloca $ \p -> do + throwErrnoIfMinus1_ "getrusage" (c_getrusage who p) + + utime <- (#peek struct rusage, ru_utime) p + stime <- (#peek struct rusage, ru_stime) p + maxrss <- (#peek struct rusage, ru_maxrss) p + ixrss <- (#peek struct rusage, ru_ixrss) p + idrss <- (#peek struct rusage, ru_idrss) p + isrss <- (#peek struct rusage, ru_isrss) p + minflt <- (#peek struct rusage, ru_minflt) p + majflt <- (#peek struct rusage, ru_majflt) p + nswap <- (#peek struct rusage, ru_nswap) p + inblock <- (#peek struct rusage, ru_inblock) p + oublock <- (#peek struct rusage, ru_oublock) p + msgsnd <- (#peek struct rusage, ru_msgsnd) p + msgrcv <- (#peek struct rusage, ru_msgrcv) p + nsignals <- (#peek struct rusage, ru_nsignals) p + nvcsw <- (#peek struct rusage, ru_nvcsw) p + nivcsw <- (#peek struct rusage, ru_nivcsw) p + + return + [ RuUtime (Seconds (timeValToDouble utime)) + , RuStime (Seconds (timeValToDouble stime)) + , RuMaxrss (GaugeMax (Bytes (1024 * clongToW64 maxrss))) + , RuIxrss (GaugeMax (Bytes (1024 * clongToW64 ixrss))) + , RuIdrss (GaugeMax (Bytes (1024 * clongToW64 idrss))) + , RuIsrss (GaugeMax (Bytes (1024 * clongToW64 isrss))) + , RuMinflt (clongToW64 minflt) + , RuMajflt (clongToW64 majflt) + , RuNswap (clongToW64 nswap) + , RuInblock (clongToW64 inblock) + , RuOublock (clongToW64 oublock) + , RuMsgsnd (clongToW64 msgsnd) + , RuMsgrcv (clongToW64 msgrcv) + , RuNsignals (clongToW64 nsignals) + , RuNvcsw (clongToW64 nvcsw) + , RuNivcsw (clongToW64 nivcsw) + ] + +------------------------------------------------------------------------------- +-- data Who = RUsageSelf | RUsageChildren | RUsageThread +------------------------------------------------------------------------------- + +pattern RUsageSelf :: CInt +pattern RUsageSelf = (#const RUSAGE_SELF) :: CInt + +pattern RUsageChildren :: CInt +pattern RUsageChildren = (#const RUSAGE_CHILDREN) :: CInt + +{- +pattern RUsageThread :: CInt +pattern RUsageThread = (#const RUSAGE_THREAD) :: CInt +-} + +-- | "who" could be: +-- RUsageSelf +-- RUsageChildren +getRUsage :: CInt -> IO RUsage +getRUsage who = + alloca $ \ptr -> do + throwErrnoIfMinus1_ "getrusage" (c_getrusage who ptr) + peek ptr diff --git a/lib/Streamly/Metrics/Perf/Type.hs b/lib/Streamly/Metrics/Perf/Type.hs new file mode 100644 index 0000000..6524086 --- /dev/null +++ b/lib/Streamly/Metrics/Perf/Type.hs @@ -0,0 +1,263 @@ +module Streamly.Metrics.Perf.Type + ( + PerfMetrics (..) + , checkMonotony + ) +where + +import Data.Word (Word64) +import Streamly.Metrics.Type + (GaugeMax(..), Seconds(..), Bytes(..), Indexable(..)) + +-- Use Counter/Gauge as the outer constructor and Bytes/Seconds as the inner +-- constuctor. +-- +-- The order is important, related stats are grouped/sorted in that order for +-- presentation purposes. +data PerfMetrics = + -- | MonotonicTime and GcElapsedTime both should provide the same figures. + MonotonicTime !(Seconds Double) + -- XXX Make this hierarchical (include rusage data) + -- data ProcessCPUTime = total user system + -- | In a single threaded system with unbound threads 'ProcessCPUTime', + -- 'ThreadCPUTime', ('RuUtime' + 'RuStime) and 'GcCpuTime' would be the same. + -- Note that a binary built with `-threaded` and using `-N1` RTS options is + -- not the same as single threaded because the GC thread may still run in a + -- separate OS thread. + -- + -- Note that if the perf stats measurement pre and post calls straddle over a + -- blocking Haskell thread then the stats may include cpuTime for all other + -- threads that might have run until our post call stats collection occurs. + -- So it may not reflect the accurate measurement of perf stats of the call + -- in question. + -- + | ProcessCPUTime !(Seconds Double) + | ThreadCPUTime !(Seconds Double) + + -- XXX Make this hierarchical + -- data ElapsedTime = ElapsedTime total mutator gc + -- GC Memory Stats + | GcElapsedTime !(Seconds Double) + -- | 'GcMutatorElapsedTime' and 'GcGcElapsedTime' should add up to + -- 'GcElapsedTime'. + | GcMutatorElapsedTime !(Seconds Double) + | GcGcElapsedTime !(Seconds Double) + + -- XXX Make this hierarchical + -- data CPUTime = total mutator gc + | GcCpuTime !(Seconds Double) + -- | 'GcMutatorCpuTime' and 'GcGcCpuTime' should add up to 'GcCpuTime'. + -- XXX Note that GHC measures GcCputTime to be just the utime using + -- rusage, it should be utime + stime to match PROCESS_CPU_TIME clock. + | GcMutatorCpuTime !(Seconds Double) + | GcGcCpuTime !(Seconds Double) + + | GcAllocatedBytes !(Bytes Word64) + | GcCopiedBytes !(Bytes Word64) + | GcMaxLiveBytes !(GaugeMax (Bytes Word64)) + | GcMaxLargeObjectBytes !(GaugeMax (Bytes Word64)) + | GcMaxCompactBytes !(GaugeMax (Bytes Word64)) + | GcMaxMemInUse !(GaugeMax (Bytes Word64)) + + -- rusage Stats + -- | 'RuUtime' and 'RuStime' should add up to 'ProcessCPUTime'. + | RuUtime !(Seconds Double) + | RuStime !(Seconds Double) + | RuMaxrss !(GaugeMax (Bytes Word64)) + | RuIxrss !(GaugeMax (Bytes Word64)) + | RuIdrss !(GaugeMax (Bytes Word64)) + | RuIsrss !(GaugeMax (Bytes Word64)) + | RuMinflt !Word64 + | RuMajflt !Word64 + | RuNswap !Word64 + | RuInblock !Word64 + | RuOublock !Word64 + | RuMsgsnd !Word64 + | RuMsgrcv !Word64 + | RuNsignals !Word64 + | RuNvcsw !Word64 + | RuNivcsw !Word64 + | Count !Word64 + deriving (Show) + +#define UNARY_OP_ONE(constr,op) op (constr a) = constr (op a) +#define UNARY_OP(op) \ + UNARY_OP_ONE(MonotonicTime,op); \ + UNARY_OP_ONE(ProcessCPUTime,op); \ + UNARY_OP_ONE(ThreadCPUTime,op); \ + UNARY_OP_ONE(GcAllocatedBytes,op); \ + UNARY_OP_ONE(GcCopiedBytes,op); \ + UNARY_OP_ONE(GcMaxLiveBytes,op); \ + UNARY_OP_ONE(GcMaxLargeObjectBytes,op); \ + UNARY_OP_ONE(GcMaxCompactBytes,op); \ + UNARY_OP_ONE(GcMaxMemInUse,op); \ + UNARY_OP_ONE(GcMutatorCpuTime,op); \ + UNARY_OP_ONE(GcMutatorElapsedTime,op); \ + UNARY_OP_ONE(GcGcCpuTime,op); \ + UNARY_OP_ONE(GcGcElapsedTime,op); \ + UNARY_OP_ONE(GcCpuTime,op); \ + UNARY_OP_ONE(GcElapsedTime,op); \ + UNARY_OP_ONE(RuUtime,op); \ + UNARY_OP_ONE(RuStime,op); \ + UNARY_OP_ONE(RuMaxrss,op); \ + UNARY_OP_ONE(RuIxrss,op); \ + UNARY_OP_ONE(RuIdrss,op); \ + UNARY_OP_ONE(RuIsrss,op); \ + UNARY_OP_ONE(RuMinflt,op); \ + UNARY_OP_ONE(RuMajflt,op); \ + UNARY_OP_ONE(RuNswap,op); \ + UNARY_OP_ONE(RuInblock,op); \ + UNARY_OP_ONE(RuOublock,op); \ + UNARY_OP_ONE(RuMsgsnd,op); \ + UNARY_OP_ONE(RuMsgrcv,op); \ + UNARY_OP_ONE(RuNsignals,op); \ + UNARY_OP_ONE(RuNvcsw,op); \ + UNARY_OP_ONE(RuNivcsw,op); \ + UNARY_OP_ONE(Count,op); + +#define INFIX_OP_ONE(constr,op) constr a op constr b = constr (a op b) +#define FUNC_OP_ONE(constr,op) constr a `op` constr b = constr (a `op` b) + +#define INFIX_OP(op) \ + INFIX_OP_ONE(MonotonicTime,op); \ + INFIX_OP_ONE(ProcessCPUTime,op); \ + INFIX_OP_ONE(ThreadCPUTime,op); \ + INFIX_OP_ONE(GcAllocatedBytes,op); \ + INFIX_OP_ONE(GcCopiedBytes,op); \ + INFIX_OP_ONE(GcMaxLiveBytes,op); \ + INFIX_OP_ONE(GcMaxLargeObjectBytes,op); \ + INFIX_OP_ONE(GcMaxCompactBytes,op); \ + INFIX_OP_ONE(GcMaxMemInUse,op); \ + INFIX_OP_ONE(GcMutatorCpuTime,op); \ + INFIX_OP_ONE(GcMutatorElapsedTime,op); \ + INFIX_OP_ONE(GcGcCpuTime,op); \ + INFIX_OP_ONE(GcGcElapsedTime,op); \ + INFIX_OP_ONE(GcCpuTime,op); \ + INFIX_OP_ONE(GcElapsedTime,op); \ + INFIX_OP_ONE(RuUtime,op); \ + INFIX_OP_ONE(RuStime,op); \ + INFIX_OP_ONE(RuMaxrss,op); \ + INFIX_OP_ONE(RuIxrss,op); \ + INFIX_OP_ONE(RuIdrss,op); \ + INFIX_OP_ONE(RuIsrss,op); \ + INFIX_OP_ONE(RuMinflt,op); \ + INFIX_OP_ONE(RuMajflt,op); \ + INFIX_OP_ONE(RuNswap,op); \ + INFIX_OP_ONE(RuInblock,op); \ + INFIX_OP_ONE(RuOublock,op); \ + INFIX_OP_ONE(RuMsgsnd,op); \ + INFIX_OP_ONE(RuMsgrcv,op); \ + INFIX_OP_ONE(RuNsignals,op); \ + INFIX_OP_ONE(RuNvcsw,op); \ + INFIX_OP_ONE(RuNivcsw,op); \ + INFIX_OP_ONE(Count,op); \ + x1 op x2 = error $ "Cannot operate on different types of metrics" \ + ++ show x1 ++ " / " ++ show x2; + +checkMonotony :: PerfMetrics -> PerfMetrics -> Bool +-- XXX can we just use >= on Perfmetrics? +checkMonotony (MonotonicTime t1) (MonotonicTime t2) = t2 >= t1 +checkMonotony (ProcessCPUTime t1) (ProcessCPUTime t2) = t2 >= t1 +checkMonotony (ThreadCPUTime t1) (ThreadCPUTime t2) = t2 >= t1 +checkMonotony (GcElapsedTime t1) (GcElapsedTime t2) = t2 >= t1 +checkMonotony (GcMutatorElapsedTime t1) (GcMutatorElapsedTime t2) = t2 >= t1 +checkMonotony (GcGcElapsedTime t1) (GcGcElapsedTime t2) = t2 >= t1 +checkMonotony (GcCpuTime t1) (GcCpuTime t2) = t2 >= t1 +checkMonotony (GcMutatorCpuTime t1) (GcMutatorCpuTime t2) = t2 >= t1 +checkMonotony (GcGcCpuTime t1) (GcGcCpuTime t2) = t2 >= t1 +checkMonotony _ _ = True + +-- XXX Can we derive this generically? +instance Num PerfMetrics where + fromInteger val = Count (fromInteger val) + UNARY_OP(signum) + UNARY_OP(abs) + INFIX_OP(+) + INFIX_OP(-) + INFIX_OP(*) + +#define DIV_OP_ONE(constr) constr a / Count b = constr (a / b) +#define DIV_DOUBLE(a,b) (a / fromIntegral b) +#define DIV_SECONDS(constr) constr (Seconds a) / (Count b) \ + = constr (Seconds DIV_DOUBLE(a,b)) +#define DIV_ROUND(a,b) (round (fromIntegral a / fromIntegral b :: Double)) +#define DIV_BYTES(constr) constr (Bytes a) / (Count b) \ + = constr (Bytes DIV_ROUND(a,b)) +#define DIV_MAX_BYTES(constr) constr (GaugeMax (Bytes a)) / (Count _) = \ + constr (GaugeMax (Bytes a)) +#define DIV_COUNT(constr) constr a / Count b = constr DIV_ROUND(a,b) + +instance Fractional PerfMetrics where + DIV_SECONDS(MonotonicTime) + DIV_SECONDS(ProcessCPUTime) + DIV_SECONDS(ThreadCPUTime) + DIV_SECONDS(GcMutatorCpuTime) + DIV_SECONDS(GcMutatorElapsedTime) + DIV_SECONDS(GcGcCpuTime) + DIV_SECONDS(GcGcElapsedTime) + DIV_SECONDS(GcCpuTime) + DIV_SECONDS(GcElapsedTime) + DIV_BYTES(GcAllocatedBytes) + DIV_BYTES(GcCopiedBytes) + DIV_MAX_BYTES(GcMaxLiveBytes) + DIV_MAX_BYTES(GcMaxLargeObjectBytes) + DIV_MAX_BYTES(GcMaxCompactBytes) + DIV_MAX_BYTES(GcMaxMemInUse) + DIV_SECONDS(RuUtime) + DIV_SECONDS(RuStime) + DIV_MAX_BYTES(RuMaxrss) + DIV_MAX_BYTES(RuIxrss) + DIV_MAX_BYTES(RuIdrss) + DIV_MAX_BYTES(RuIsrss) + DIV_COUNT(RuMinflt) + DIV_COUNT(RuMajflt) + DIV_COUNT(RuNswap) + DIV_COUNT(RuInblock) + DIV_COUNT(RuOublock) + DIV_COUNT(RuMsgsnd) + DIV_COUNT(RuMsgrcv) + DIV_COUNT(RuNsignals) + DIV_COUNT(RuNvcsw) + DIV_COUNT(RuNivcsw) + Count a / Count _ = Count a + x1 / x2 = + error + $ "Undefined fractional operation on PerfMetrics " + ++ show x1 ++ " / " ++ show x2 + fromRational = error "fromRational: not supported for PerfMetrics" + +instance Indexable PerfMetrics where + getIndex (Count _) = 0 + getIndex (MonotonicTime _) = 1 + getIndex (GcElapsedTime _) = 2 + getIndex (GcMutatorElapsedTime _) = 3 + getIndex (GcGcElapsedTime _) = 4 + + getIndex (ProcessCPUTime _) = 5 + getIndex (RuUtime _) = 6 + getIndex (RuStime _) = 7 + getIndex (ThreadCPUTime _) = 8 + getIndex (GcCpuTime _) = 9 + getIndex (GcMutatorCpuTime _) = 10 + getIndex (GcGcCpuTime _) = 11 + + getIndex (GcAllocatedBytes _) = 12 + getIndex (GcCopiedBytes _) = 13 + getIndex (GcMaxLiveBytes _) = 14 + getIndex (GcMaxLargeObjectBytes _) = 14 + getIndex (GcMaxCompactBytes _) = 14 + getIndex (GcMaxMemInUse _) = 14 + getIndex (RuMaxrss _) = 15 + getIndex (RuIxrss _) = 16 + getIndex (RuIdrss _) = 17 + getIndex (RuIsrss _) = 18 + getIndex (RuMinflt _) = 19 + getIndex (RuMajflt _) = 20 + getIndex (RuNswap _) = 21 + getIndex (RuInblock _) = 22 + getIndex (RuOublock _) = 23 + getIndex (RuMsgsnd _) = 24 + getIndex (RuMsgrcv _) = 25 + getIndex (RuNsignals _) = 26 + getIndex (RuNvcsw _) = 27 + getIndex (RuNivcsw _) = 28 diff --git a/lib/Streamly/Metrics/Type.hs b/lib/Streamly/Metrics/Type.hs new file mode 100644 index 0000000..39fe9ff --- /dev/null +++ b/lib/Streamly/Metrics/Type.hs @@ -0,0 +1,208 @@ +{-# LANGUAGE DerivingVia #-} +-- | +-- Module : Streamly.Metrics.Type +-- Copyright : (c) 2021 Composewell Technologies +-- License : Apache-2.0 +-- Maintainer : streamly@composewell.com +-- Stability : experimental +-- Portability : GHC +-- +-- A 'Metric' represents a sample of some value labeled with a unique +-- identifier. Metrics may be ordered with respect to each other or with +-- respect to time e.g. 'Sequence' or 'Time'. Metrics may have different +-- semantics e.g. 'Counter' or 'Gauge'. +-- +module Streamly.Metrics.Type + ( + -- * Semantics + -- | A counter counts how many events of a type have occurred whereas a + -- gauge measures what is the value of a certain property at a given point + -- of time. A counter is always @monotonic@ whereas a gauge is usually + -- @volatile@. + Log (..) + , Counter (..) + , GaugeMax (..) + + -- * Ordering + -- | Metrics may be ordered with respect to each other or with respect to + -- time. + , Timestamped (..) + , Sequenced (..) + + -- * Units + , Seconds (..) + , Bytes (..) + + -- * Utilities + , Indexable(..) + , showList + ) +where + +import Data.List (sortBy) +import Streamly.Internal.Data.Time.Units (AbsTime) +import Text.Printf (printf, PrintfArg) +import Prelude hiding (showList) + +------------------------------------------------------------------------------- +-- Metric semantics +------------------------------------------------------------------------------- + +-- Semantics define what operations make sense on a metric or how you can +-- combine, aggregate or collapse multiple samples of a metric. +-- +-- Metrics can be aggregated in different ways: +-- +-- * a -> a -> a (catenate, sum) +-- * a -> Array a -> Array a (cons) +-- +-- If the type is a Monoid we can represent all interesting operations on the +-- metrici in terms of Monoid. If "a" represents a metric then "Array a" also +-- represents a metric, if "a" is a Monoid "Array a" is also a Monoid. XXX do +-- we make the monoid instance of "Array a" append the type "a" rather than the +-- array itself like the Tee type or do we need a newtype wrapper for that? + +-- | A log is the simplest metric. It has no semantics except catenation of +-- multiple logs. +-- +newtype Log a = Log a + +-- | Represents a monotonically increasing value like time or number of events. +-- +-- Counters should support addition and subtraction operations. We can diff the +-- values of a counter at two points to figure out the change that occurred in +-- that duration. On the other hand, maximum, minimum, range or average values +-- usually do not make sense for a counter. +-- +-- It does not make sense to add different snapshots of the same counter, +-- however it makes sense to aggregate counters from multiple independent +-- sources e.g. to aggregate total number of events, or to add durations. +-- Counters could be continuous of discrete. Time is an example of a continuous +-- counter whereas instruction count is an example of a discrete counter. +-- +-- Time is a fundamental counter, in fact it is a reference counter for +-- all changes, changes are measured with respect to time. + +newtype Counter a = + Counter a + deriving (Num, Fractional) + +instance Show a => Show (Counter a) where + show (Counter a) = show a + +-- | Represents the current utilization level of some resource e.g. memory +-- in use. A gauge can increase or decrease from a previous value. +-- +-- It usually does not make sense to add gauge metrics collected from multiple +-- independent sources. However, maximum, minimum, range or average values make +-- sense for multiple snapshots of a gauge or the value of a gauge from +-- multiple sources. Thus, a gauge type can be further classified into +-- maximum (e.g. peak memory) or minimum (e.g. available memory), average (e.g. +-- load average) or range. +-- +newtype GaugeMax a = GaugeMax a + +instance Show a => Show (GaugeMax a) where + show (GaugeMax a) = show a + +instance (Num a, Ord a) => Num (GaugeMax a) where + fromInteger a = GaugeMax (fromInteger a) + abs (GaugeMax a) = GaugeMax (abs a) + signum (GaugeMax a) = GaugeMax (signum a) + (GaugeMax a) * (GaugeMax b) = GaugeMax (a * b) + + -- XXX Abusing Num instance, we can use a separate type class + -- For a GaugeMax these are defined as maximum of the two values + (GaugeMax a) - (GaugeMax b) = GaugeMax (max a b) + (GaugeMax a) + (GaugeMax b) = GaugeMax (max a b) + +------------------------------------------------------------------------------- +-- Metrics ordering +------------------------------------------------------------------------------- + +-- How do we distinguish (if at all) different samples of the same metric. + +-- | A value tagged with a sequence number. Useful when we want to order the +-- samples of a value with respect to each other but do not care about the +-- ordering with respect to time. +-- +data Sequenced a = Sequenced Int a + +-- | A value tagged with a timestamp. Useful when we want to measure samples +-- with respect to time. +-- +data Timestamped a = Timestamped AbsTime a -- XXX use a tuple and derive Ord? + +-- XXX When zipping trees of metrics we need to traverse the keys like file +-- paths or a file system tree. That means a metric should probably support a +-- lookup operation like a Map. +-- +-- The simplest would be a one level Map as a metric or a metric set. Should we +-- separate these to concepts of a metric and a metric set. A metric would be +-- like a file whereas a metric set would be like a dir tree (MetricTree). +-- +-- ZipList and a single level Map are a one level metric tree. + +------------------------------------------------------------------------------- +-- Units +------------------------------------------------------------------------------- + +-- XXX Use the common time units code from streamly instead. +-- +-- | Describe a relative unit i.e. a unit in terms of another unit. A relative +-- unit has a label and a ratio which when multiplied with the unit gives us +-- the other unit. For example, if the known time unit is seconds, we can +-- describe a millisecond as @Unit "ms" (1/1000)@. +data RelativeUnit a = RelativeUnit String a deriving Show + +-- Given seconds, choose appropriate unit based on the size +secondsConverter :: (Ord a, Fractional a) => a -> RelativeUnit a +secondsConverter k + | k < 0 = secondsConverter (-k) + | k >= 1 = RelativeUnit "s" 1 + | k >= 1e-3 = RelativeUnit "ms" 1e3 + | k >= 1e-6 = RelativeUnit "μs" 1e6 + | otherwise = RelativeUnit "ns" 1e9 + +newtype Seconds a = Seconds a deriving (Num, Fractional, Eq, Ord) + +instance (Show a, Ord a, Fractional a, PrintfArg a) => Show (Seconds a) where + show (Seconds t) = + let (RelativeUnit label multiplier) = secondsConverter t + in printf "%.2f" (t * multiplier) ++ " " ++ label + +-- Given bytes, choose appropriate unit based on the size +bytesConverter :: (Num a, Ord a) => a -> RelativeUnit a +bytesConverter k + | k < 0 = bytesConverter (-k) + | k >= 2^(30 :: Int) = RelativeUnit "GiB" (2^(30 :: Int)) + | k >= 2^(20 :: Int) = RelativeUnit "MiB" (2^(20 :: Int)) + | k >= 2^(10 :: Int) = RelativeUnit "KiB" (2^(10 :: Int)) + | otherwise = RelativeUnit "Bytes" 1 + +newtype Bytes a = Bytes a deriving (Num, Eq, Ord, Fractional) + +instance (Show a, Num a, Ord a, PrintfArg a, Integral a) => Show (Bytes a) + + where + + show (Bytes b) = + let (RelativeUnit label multiplier) = bytesConverter b + n = fromIntegral b / fromIntegral multiplier :: Double + in printf "%.2f" n ++ " " ++ label + +------------------------------------------------------------------------------- +-- Utilities +------------------------------------------------------------------------------- + +-- Give indices to different values of the type 'a' +class Indexable a where + getIndex :: a -> Int + +-- Show a list of values of an Indexable type sorted in ascending index order +showList :: (Show a, Indexable a) => [a] -> String +showList xs = unlines $ show <$> sortBy f xs + + where + + f p1 p2 = compare (getIndex p1) (getIndex p2) diff --git a/src/Main.hs b/src/Main.hs index 3194f06..cf05499 100644 --- a/src/Main.hs +++ b/src/Main.hs @@ -453,8 +453,8 @@ configParser = Config optsInfo :: ParserInfo Config optsInfo = info (configParser <**> helper) ( fullDesc - <> progDesc "Analyse CPU cost, heap allocations, and Linux perf event \ - \counters for Haskell threads and user-defined code windows." + <> progDesc ("Analyse CPU cost, heap allocations, and Linux perf event " + ++ "counters for Haskell threads and user-defined code windows.") <> header "hperf - Haskell performance analysis tool" ) diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 index 0000000..8e2493c --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,32 @@ +import Control.Concurrent(threadDelay) +import Streamly.Metrics.Channel + (Channel, newChannel, forkChannelPrinter, benchOnWith) +-- import Streamly.Metrics.Channel (printChannel) +import Streamly.Metrics.Perf.Type (PerfMetrics) + +import qualified Streamly.Data.Fold as Fold +import qualified Streamly.Data.Stream as Stream +import Prelude hiding (sum) + +noop :: Channel PerfMetrics -> IO () +noop chan = do + benchOnWith chan "noop" (const (return ())) (1000000 :: Int) + +sum :: Channel PerfMetrics -> IO () +sum chan = do + _ <- benchOnWith + chan "sum" (Stream.fold Fold.sum . Stream.enumerateFromTo (1::Int)) 1000000 + return () + +main :: IO () +main = do + chan <- newChannel + _ <- forkChannelPrinter chan 10 100 + Stream.fold Fold.drain (Stream.replicateM 1000 (noop chan)) + Stream.fold Fold.drain (Stream.replicateM 1000 (sum chan)) + threadDelay 1000000 + {- + Stream.drain + ((Stream.replicateM 1000 (noop chan) <> Stream.replicateM 1000 (sum chan)) + `Stream.parallelFst` Stream.fromEffect (printChannel chan 1 10)) + -} From 8e41d216f871637206407aa2b78e03a95d781fa1 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Wed, 29 Apr 2026 03:34:18 +0530 Subject: [PATCH 05/14] Use common channel code from the Common module --- lib/Streamly/Metrics/Channel.hs | 41 ++++++++------------------------- 1 file changed, 9 insertions(+), 32 deletions(-) diff --git a/lib/Streamly/Metrics/Channel.hs b/lib/Streamly/Metrics/Channel.hs index 9d01ef1..b9a10ec 100644 --- a/lib/Streamly/Metrics/Channel.hs +++ b/lib/Streamly/Metrics/Channel.hs @@ -15,23 +15,17 @@ import Control.Concurrent.STM (atomically) import Control.Concurrent.STM.TBQueue (TBQueue, newTBQueue, readTBQueue, writeTBQueue) import Control.Monad.IO.Class (liftIO, MonadIO) -import Data.Bifunctor (second) import Data.Function ((&)) -import Data.Maybe (fromJust, isJust) import Streamly.Data.Stream (Stream) import Streamly.Internal.Data.Time.Clock (getTime, Clock (Monotonic)) import Streamly.Internal.Data.Time.Units (AbsTime) +import Streamly.Metrics.Channel.Common (aggregateListBy, printKV) import Streamly.Metrics.Perf.Type (PerfMetrics(..)) import Streamly.Metrics.Perf (benchWith) -import Streamly.Metrics.Type (showList, Indexable) +import Streamly.Metrics.Type (Indexable) import Streamly.Data.Stream.Prelude (MonadAsync) -import qualified Streamly.Data.Fold as Fold -import qualified Streamly.Internal.Data.Fold as Fold import qualified Streamly.Data.Stream as Stream -import qualified Streamly.Internal.Data.Stream.Prelude as Stream - -import Prelude hiding (showList) ------------------------------------------------------------------------------- -- Event processing @@ -60,30 +54,6 @@ send (Channel chan) desc metrics = do fromChan :: MonadAsync m => TBQueue a -> Stream m a fromChan = Stream.repeatM . (liftIO . atomically . readTBQueue) -aggregateListBy :: (MonadAsync m, Ord k, Fractional a) => - Double -> Int -> Stream m (AbsTime, (k, [a])) -> Stream m (k, [a]) -aggregateListBy timeout batchsize stream = - fmap (second fromJust) - $ Stream.filter (isJust . snd) - $ Stream.classifySessionsBy - 0.1 False (return . (> 1000)) timeout f stream - - where - - scale Nothing _ = Nothing - scale (Just xs) count = Just $ map (/ count) xs - - f = - Fold.teeWithFst - scale - (Fold.take batchsize (Fold.foldl1' (zipWith (+)))) - (Fold.lmap (const 1) Fold.sum) - -printKV :: (MonadIO m, Show k, Show a, Indexable a) => Stream m (k, [a]) -> m b -printKV stream = - let f (k, xs) = liftIO $ putStrLn $ show k ++ ":\n" ++ showList xs - in Stream.fold (Fold.drainMapM f) stream >> error "printChannel: Metrics channel closed" - -- XXX Print actual batch size and also scale the results per event. -- | Forever print the metrics on a channel to the console periodically after @@ -95,6 +65,13 @@ printChannel (Channel chan) timeout batchSize = & aggregateListBy timeout batchSize & printKV +-- | Start an async thread to print the stats received on the supplied channel +-- and print the stats on console. +-- +-- Usage: @forkChannelPrinter channel timeout batch-size@. +-- +-- Stats are printed when either as many stat samples as the batch size have +-- been received or we have not received a stat in "timeout" seconds. forkChannelPrinter :: (MonadAsync m, Show a, Fractional a, Indexable a) => Channel a -> Double -> Int -> m ThreadId forkChannelPrinter chan timeout = liftIO . forkIO . printChannel chan timeout From 150d15d47ade39381d9b6e511faa5dec9d25f2fb Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Wed, 29 Apr 2026 04:06:43 +0530 Subject: [PATCH 06/14] Refactor aggregateListBy and add some tracing --- lib/Streamly/Metrics/Channel/Common.hs | 23 ++++++++++++++--------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/lib/Streamly/Metrics/Channel/Common.hs b/lib/Streamly/Metrics/Channel/Common.hs index e7c963f..9d66b8d 100644 --- a/lib/Streamly/Metrics/Channel/Common.hs +++ b/lib/Streamly/Metrics/Channel/Common.hs @@ -23,26 +23,31 @@ import Prelude hiding (showList) -- Event processing ------------------------------------------------------------------------------- +-- XXX We can use incremental window folds instead. We do not need to collect +-- all stats and then calculate. aggregateListBy :: (MonadAsync m, Ord k, Fractional a) => Double -> Int -> Stream m (AbsTime, (k, [a])) -> Stream m (k, [a]) aggregateListBy timeout batchsize stream = fmap (second fromJust) $ Stream.filter (isJust . snd) - $ Stream.classifySessionsBy - 0.1 False (return . (> 1000)) timeout f stream + -- $ Stream.trace (\x -> liftIO $ putStrLn $ "after classify: " ++ show x) + $ Stream.classifySessionsBy 0.1 False (return . (> 1000)) timeout f + -- $ Stream.trace (liftIO . print) + $ stream where scale Nothing _ = Nothing - scale (Just xs) count = Just $ map (/ count) xs + scale (Just xs) cnt = Just $ map (/ cnt) xs - f = - Fold.teeWithFst - scale - (Fold.take batchsize (Fold.foldl1' (zipWith (+)))) - (Fold.lmap (const 1) Fold.sum) + addMetrics = Fold.foldl1' (zipWith (+)) + collectBatch = Fold.take batchsize addMetrics + count = fmap fromIntegral Fold.length + + f = Fold.teeWithFst scale collectBatch count printKV :: (MonadIO m, Show k, Show a, Indexable a) => Stream m (k, [a]) -> m b printKV stream = let f (k, xs) = liftIO $ putStrLn $ show k ++ ":\n" ++ showList xs - in Stream.fold (Fold.drainMapM f) stream >> error "printChannel: Metrics channel closed" + in Stream.fold (Fold.drainMapM f) stream + >> error "printChannel: Metrics channel closed" From 970287ceb0cc6faa899edfdcf4ad34fe3cd5d3b2 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Wed, 29 Apr 2026 06:48:55 +0530 Subject: [PATCH 07/14] Simplify, cleanup the stat collection/reporting example --- test/Main.hs | 74 ++++++++++++++++++++++++++++++++++++++-------------- 1 file changed, 55 insertions(+), 19 deletions(-) diff --git a/test/Main.hs b/test/Main.hs index 8e2493c..56cbadd 100644 --- a/test/Main.hs +++ b/test/Main.hs @@ -6,27 +6,63 @@ import Streamly.Metrics.Perf.Type (PerfMetrics) import qualified Streamly.Data.Fold as Fold import qualified Streamly.Data.Stream as Stream -import Prelude hiding (sum) -noop :: Channel PerfMetrics -> IO () -noop chan = do - benchOnWith chan "noop" (const (return ())) (1000000 :: Int) - -sum :: Channel PerfMetrics -> IO () -sum chan = do - _ <- benchOnWith - chan "sum" (Stream.fold Fold.sum . Stream.enumerateFromTo (1::Int)) 1000000 +runWithStats :: Channel PerfMetrics -> String -> (a -> IO b) -> a -> IO () +runWithStats chan label f arg = do + _ <- benchOnWith chan label f arg return () +-- A simple operation that does nothing. When we measure this operation the cpu +-- time that is spent is just the overhead of the measuring code. +noOp :: b -> IO () +noOp = (const (return ())) + +sumOp :: Int -> IO Int +sumOp = + Stream.fold Fold.sum + . Stream.enumerateFromTo (1::Int) + +{- +-- Pure code example +listSum :: Int -> Int +listSum = + sum + . enumFromTo (1::Int) +-} + +timeout :: Int +timeout = 1 + +initStats :: IO (Channel PerfMetrics) +initStats = do + chan <- newChannel + -- The channel will collect 100 samples per label, as soon as it receives + -- 100 it will print the stats and start collecting the next batch. + -- If no sample comes in "timeout" seconds then print the batch anyway. + -- @forkChannelPrinter channel timeout batch-size@. + _ <- forkChannelPrinter chan (fromIntegral timeout) 100 + return chan + main :: IO () main = do - chan <- newChannel - _ <- forkChannelPrinter chan 10 100 - Stream.fold Fold.drain (Stream.replicateM 1000 (noop chan)) - Stream.fold Fold.drain (Stream.replicateM 1000 (sum chan)) - threadDelay 1000000 - {- - Stream.drain - ((Stream.replicateM 1000 (noop chan) <> Stream.replicateM 1000 (sum chan)) - `Stream.parallelFst` Stream.fromEffect (printChannel chan 1 10)) - -} + -- Initialize a channel to send the stats to + chan <- initStats + + let withStats = runWithStats chan + + -- One shot measurement, just one call + withStats "noOpOne" noOp (1000000 :: Int) + withStats "sumOpOne" sumOp (1000000 :: Int) + + -- Run many iterations and print the stats for batches of 100 + let iterations n = Stream.fold Fold.drain . Stream.replicateM n + withStatsMany label f arg = iterations 1000 $ runWithStats chan label f arg + + -- Run the "noOp" and "sumOp" 1000 times, passing 1000000 as argument and + -- sending the stats to "chan". The stats collected for noOp and sumOp will + -- be sent to the channel and printed by it on console. + withStatsMany "noOpMany" noOp (1000000 :: Int) + withStatsMany "sumOpMany" sumOp (1000000 :: Int) + + -- Wait for the channel to drain + threadDelay ((timeout + 2) * 1000000) From 9110acb5ecc880699bb868ef8d07690fe18a5781 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Wed, 6 May 2026 11:48:03 +0530 Subject: [PATCH 08/14] Move with stats example to the examples dir --- test/Main.hs => examples/withStats.hs | 0 haskell-perf.cabal | 16 ++++++++-------- 2 files changed, 8 insertions(+), 8 deletions(-) rename test/Main.hs => examples/withStats.hs (100%) diff --git a/test/Main.hs b/examples/withStats.hs similarity index 100% rename from test/Main.hs rename to examples/withStats.hs diff --git a/haskell-perf.cabal b/haskell-perf.cabal index 629b667..3c20d18 100644 --- a/haskell-perf.cabal +++ b/haskell-perf.cabal @@ -187,17 +187,17 @@ executable console-loop-multi-thread if os(windows) buildable: False -------------------------------------------------------------------------------- --- Tests -------------------------------------------------------------------------------- - -test-suite basic +executable with-stats import: test-options - type: exitcode-stdio-1.0 - hs-source-dirs: test - main-is: Main.hs + hs-source-dirs: examples + main-is: withStats.hs ghc-options: -O2 -fmax-worker-args=16 -fspec-constr-recursive=16 build-depends: base , haskell-perf , streamly-core + +------------------------------------------------------------------------------- +-- Tests +------------------------------------------------------------------------------- + From 8500bea8f084099b20934152e7739a8839ffbce6 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Wed, 6 May 2026 11:48:26 +0530 Subject: [PATCH 09/14] Fix typo in examples/README --- examples/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/README.md b/examples/README.md index 4656fa9..e8871f8 100644 --- a/examples/README.md +++ b/examples/README.md @@ -5,7 +5,7 @@ console-loop.hs -- console echo loop delay-loop.hs -- threadDelay in a loop ffi.hs -- fork one thread + FFI c_sleep -# In memory collection +# In-memory collection threadCPUTime.hs -- single threaded stat collection From c5a99a2174d6485e3c878c34debcefebe80ca99c Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Fri, 8 May 2026 07:36:13 +0530 Subject: [PATCH 10/14] Add a note about Windows timeval struct --- lib/Streamly/Metrics/Perf/RUsage.hsc | 2 ++ 1 file changed, 2 insertions(+) diff --git a/lib/Streamly/Metrics/Perf/RUsage.hsc b/lib/Streamly/Metrics/Perf/RUsage.hsc index 7140047..4925bc3 100644 --- a/lib/Streamly/Metrics/Perf/RUsage.hsc +++ b/lib/Streamly/Metrics/Perf/RUsage.hsc @@ -36,6 +36,8 @@ w64ToCLong = fromIntegral -- struct timeval ------------------------------------------------------------------------------- +-- Note: On Windows winsock library can be used for struct timeval. + data TimeVal = TimeVal {-# UNPACK #-} !Word64 -- sec From e3d032e10b76c226f18db0e6d12c19e35a2cee7e Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Fri, 8 May 2026 09:20:27 +0530 Subject: [PATCH 11/14] Use getrusage only under posix --- haskell-perf.cabal | 5 ++++- lib/Streamly/Metrics/Perf.hs | 6 ++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/haskell-perf.cabal b/haskell-perf.cabal index 3c20d18..1f72104 100644 --- a/haskell-perf.cabal +++ b/haskell-perf.cabal @@ -137,7 +137,6 @@ library , Streamly.Metrics.Type , Streamly.Metrics.Perf , Streamly.Metrics.Perf.Type - , Streamly.Metrics.Perf.RUsage , Streamly.Metrics.Channel , Streamly.Metrics.Channel.Unbounded , Streamly.Metrics.Channel.Common @@ -145,6 +144,10 @@ library -- , Streamly.Metrics.File -- , Streamly.Metrics.Console + if !os(windows) + exposed-modules: + Streamly.Metrics.Perf.RUsage + build-depends: base >= 4.9 && < 5 , pretty-show >= 1.10 && < 1.11 diff --git a/lib/Streamly/Metrics/Perf.hs b/lib/Streamly/Metrics/Perf.hs index fdd0c82..a21a56e 100644 --- a/lib/Streamly/Metrics/Perf.hs +++ b/lib/Streamly/Metrics/Perf.hs @@ -14,7 +14,9 @@ import GHC.Stats (getRTSStats, getRTSStatsEnabled, RTSStats(..)) import Streamly.Internal.Data.Time.Units (NanoSecond64, fromAbsTime) import Streamly.Metrics.Measure (measureWith) import Streamly.Metrics.Perf.Type (PerfMetrics(..), checkMonotony) +#if !defined(mingw32_HOST_OS) import Streamly.Metrics.Perf.RUsage (getRuMetrics, pattern RUsageSelf) +#endif import Text.Show.Pretty (ppShow) import qualified Streamly.Internal.Data.Time.Clock as Clock @@ -65,8 +67,12 @@ getPerfMetrics :: IO [PerfMetrics] getPerfMetrics = do procMetrics <- getProcMetrics gcMetrics <- getGcMetrics +#if !defined(mingw32_HOST_OS) ruMetrics <- getRuMetrics RUsageSelf return $ concat [procMetrics, gcMetrics, ruMetrics] +#else + return $ concat [procMetrics, gcMetrics] +#endif {-# INLINE preRun #-} preRun :: IO [PerfMetrics] From 437251e0299463e117ad5ed029bbad691673bb41 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Fri, 8 May 2026 09:39:04 +0530 Subject: [PATCH 12/14] Move random code snippets to the test/snippets dir --- haskell-perf.cabal | 1 + {examples => test/snippets}/console-loop-multi-thread.hs | 0 {examples => test/snippets}/console-loop.hs | 0 {examples => test/snippets}/delay-loop.hs | 0 {examples => test/snippets}/ffi.hs | 0 {examples => test/snippets}/hello.hs | 0 {examples => test/snippets}/threadCPUTime.hs | 0 {examples => test/snippets}/traceEventIOWarp.hs | 0 8 files changed, 1 insertion(+) rename {examples => test/snippets}/console-loop-multi-thread.hs (100%) rename {examples => test/snippets}/console-loop.hs (100%) rename {examples => test/snippets}/delay-loop.hs (100%) rename {examples => test/snippets}/ffi.hs (100%) rename {examples => test/snippets}/hello.hs (100%) rename {examples => test/snippets}/threadCPUTime.hs (100%) rename {examples => test/snippets}/traceEventIOWarp.hs (100%) diff --git a/haskell-perf.cabal b/haskell-perf.cabal index 1f72104..50e4539 100644 --- a/haskell-perf.cabal +++ b/haskell-perf.cabal @@ -32,6 +32,7 @@ extra-doc-files: extra-source-files: examples/*.hs + test/snippets/*.hs lib/Streamly/Metrics/File.hs lib/Streamly/Metrics/Console.hs diff --git a/examples/console-loop-multi-thread.hs b/test/snippets/console-loop-multi-thread.hs similarity index 100% rename from examples/console-loop-multi-thread.hs rename to test/snippets/console-loop-multi-thread.hs diff --git a/examples/console-loop.hs b/test/snippets/console-loop.hs similarity index 100% rename from examples/console-loop.hs rename to test/snippets/console-loop.hs diff --git a/examples/delay-loop.hs b/test/snippets/delay-loop.hs similarity index 100% rename from examples/delay-loop.hs rename to test/snippets/delay-loop.hs diff --git a/examples/ffi.hs b/test/snippets/ffi.hs similarity index 100% rename from examples/ffi.hs rename to test/snippets/ffi.hs diff --git a/examples/hello.hs b/test/snippets/hello.hs similarity index 100% rename from examples/hello.hs rename to test/snippets/hello.hs diff --git a/examples/threadCPUTime.hs b/test/snippets/threadCPUTime.hs similarity index 100% rename from examples/threadCPUTime.hs rename to test/snippets/threadCPUTime.hs diff --git a/examples/traceEventIOWarp.hs b/test/snippets/traceEventIOWarp.hs similarity index 100% rename from examples/traceEventIOWarp.hs rename to test/snippets/traceEventIOWarp.hs From 15207a6593509f5ef87a95176d2273a0cf99edd1 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Fri, 8 May 2026 09:40:50 +0530 Subject: [PATCH 13/14] Rename withStats example to getRTSStats --- examples/{withStats.hs => getRTSStats.hs} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename examples/{withStats.hs => getRTSStats.hs} (100%) diff --git a/examples/withStats.hs b/examples/getRTSStats.hs similarity index 100% rename from examples/withStats.hs rename to examples/getRTSStats.hs From f36747b0de8421a8411fb3c7f50583faee18a6d8 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Fri, 8 May 2026 10:40:22 +0530 Subject: [PATCH 14/14] Fix example executables, update the eventlog doc --- docs/enabling-perf-counters.md | 19 ++++++ docs/eventlog-performance-analysis.md | 96 +++++++++++++++------------ examples/traceEventIO.hs | 36 ++++++++-- haskell-perf.cabal | 26 ++++---- 4 files changed, 118 insertions(+), 59 deletions(-) create mode 100644 docs/enabling-perf-counters.md diff --git a/docs/enabling-perf-counters.md b/docs/enabling-perf-counters.md new file mode 100644 index 0000000..b80da3f --- /dev/null +++ b/docs/enabling-perf-counters.md @@ -0,0 +1,19 @@ +## Enable Linux perf counters + +Enable unrestricted use of perf counters: + +``` +# echo -1 > /proc/sys/kernel/perf_event_paranoid +``` + +## Disable CPU scaling + +Set the scaling governer of all your cpus to `performance`: + +``` +echo performance > /sys/devices/system/cpu/cpu0/cpufreq/scaling_governor +echo performance > /sys/devices/system/cpu/cpu1/cpufreq/scaling_governor +... +... +echo performance > /sys/devices/system/cpu/cpu7/cpufreq/scaling_governor +``` diff --git a/docs/eventlog-performance-analysis.md b/docs/eventlog-performance-analysis.md index e7a444b..f5b690e 100644 --- a/docs/eventlog-performance-analysis.md +++ b/docs/eventlog-performance-analysis.md @@ -1,12 +1,13 @@ -# GHC Event logging +# Haskell Perf Analysis using Eventlog -Available in GHC 9.2.8 RTS patch. Can be ported to later GHCs. +The GHC RTS instrumentation for accurate and thread-aware event logging +is available in GHC 9.2.8 RTS patch. Can be ported to later GHCs. -Eventlog based Haskell thread aware time and allocation analysis is +Eventlog based Haskell-thread aware time and allocation analysis is possible with stock GHC but there are some limitations and drawbacks which are fixed in the RTS patch described below. The patch adds accurate timing and allocation information and hardware performance @@ -18,53 +19,57 @@ program not just the current thread. TBD: document the exact limitations and differences. --> -## Enable Linux perf counters +## Generating the eventlog -Enable unrestricted use of perf counters: +To generate the event log, we need to enable event log at compile time +(on modern GHCs it is always enabled) and the run the program with +eventlog enabled at run-time, we use the `-l` rts option to do that. -``` -# echo -1 > /proc/sys/kernel/perf_event_paranoid -``` +There are multiple ways of running your program with eventlog enabled at +run-time: -## Disable CPU scaling - -Set the scaling governer of all your cpus to `performance`: +__GHC Command Line__: +Compiling: ``` -echo performance > /sys/devices/system/cpu/cpu0/cpufreq/scaling_governor -echo performance > /sys/devices/system/cpu/cpu1/cpufreq/scaling_governor -... -... -echo performance > /sys/devices/system/cpu/cpu7/cpufreq/scaling_governor +ghc Main.hs -rtsopts ``` -## Generating the eventlog - -To generate the event log, we need to compile the program with the eventlog enabled -and run the program setting the `-l` rts option. - -There are multiple ways of doing this. - -__Using plain GHC__: - +Running: ``` -ghc Main.hs -rtsopts -eventlog ./Main +RTS -l -RTS ``` -__Using Cabal__: +You can bake in the rts options during compilation itself: +``` +ghc Main.hs -with-rtsopts=-l +``` -The `.cabal` file should contain the following ghc options +Now you can run without any explicit RTS options: ``` -ghc-options: -eventlog "-with-rtsopts=-l" +./Main ``` -If the `-threaded` option is used while compiling. You may want to use the `-N1` -rts option. +After we run the above program a "Main.eventlog" file will be generated. This +file can be analyzed using the `hperf` executable in this package to +generate an analysis report. To be able to find anything in the report you need +to first instrument your program which is described in the following sections. + +Note 1: For older compilers you need `-eventlog` GHC flag as well when building + +Note 2: If the `-threaded` option is used while compiling. You may want +to use the `-N1` rts option. -## Creating windows +## Measurement instrumentation -Helper function to create windows: +See the example in [examples/traceEventIO.hs](examples/traceEventIO.hs) . + +Use the `traceEventIO` function to log events. Add an event before and +after the code block you want to measure. The event message before the block +should be in the format "START: