From f2fa4f982502dfa1079392c36163c7c4d841e918 Mon Sep 17 00:00:00 2001
From: Adithya Kumar <adithya@composewell.com>
Date: Sat, 11 Nov 2023 15:23:32 +0530
Subject: [PATCH 01/10] Add an executable to collect live statitics

---
 cabal.project.user         |   5 +
 haskell-perf.cabal         |  15 ++
 stat-collector-src/Main.hs | 307 +++++++++++++++++++++++++++++++++++++
 3 files changed, 327 insertions(+)
 create mode 100644 stat-collector-src/Main.hs

diff --git a/cabal.project.user b/cabal.project.user
index a178b69..23b949c 100644
--- a/cabal.project.user
+++ b/cabal.project.user
@@ -5,3 +5,8 @@ source-repository-package
   location: https://github.com/composewell/streamly.git
   tag: master
   subdir: core
+
+source-repository-package
+  type: git
+  location: https://github.com/composewell/streamly.git
+  tag: master
diff --git a/haskell-perf.cabal b/haskell-perf.cabal
index eb93ddf..dc684a6 100644
--- a/haskell-perf.cabal
+++ b/haskell-perf.cabal
@@ -117,3 +117,18 @@ executable hperf
      , streamly-core == 0.2.0
      , format-numbers
      , text
+
+executable stat-collector
+  import: compile-options
+  hs-source-dirs: stat-collector-src
+  main-is: Main.hs
+  ghc-options: -O2 -fmax-worker-args=16 -fspec-constr-recursive=16
+  build-depends:
+       base
+     , streamly-core
+     , streamly
+     , containers
+     , ansi-terminal
+     , format-numbers
+     , unix
+     , text
diff --git a/stat-collector-src/Main.hs b/stat-collector-src/Main.hs
new file mode 100644
index 0000000..253ea23
--- /dev/null
+++ b/stat-collector-src/Main.hs
@@ -0,0 +1,307 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Main (main) where
+
+--------------------------------------------------------------------------------
+-- Imports
+--------------------------------------------------------------------------------
+
+-- import Control.Concurrent (threadDelay)
+import Data.Int (Int32, Int64)
+import System.Environment (getArgs)
+import Control.Monad.IO.Class (MonadIO(..))
+import Data.Function ((&))
+import Data.List (foldl', findIndex, sortBy, find)
+import Data.Map (Map)
+import Data.Maybe (fromMaybe, fromJust)
+import Data.Word (Word8)
+import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
+import Foreign.Storable (Storable, peek)
+import Numeric (showFFloat)
+import Streamly.Data.Array (Array)
+import Streamly.Data.Fold (Fold)
+import Streamly.Data.Stream (Stream)
+import Streamly.Internal.Data.Fold (Fold(..), Step(..))
+import Streamly.Internal.Data.Ring (slidingWindow)
+import Streamly.Internal.Data.Tuple.Strict (Tuple3Fused' (Tuple3Fused'))
+import Streamly.Unicode.String (str)
+import System.IO (hFlush, stdout, stdin)
+import Text.Read (readMaybe)
+import System.Posix.Signals (installHandler, Handler(Catch), sigINT, sigTERM)
+import Data.Text.Format.Numbers (prettyI)
+
+import qualified Data.Text as Text
+import qualified Data.Map as Map
+import qualified Streamly.Data.Fold as Fold
+import qualified Streamly.Data.Stream.Prelude as Stream
+import qualified Streamly.FileSystem.Handle as Handle
+import qualified Streamly.Internal.Data.Fold as Fold
+import qualified Streamly.Internal.Data.Ring as Ring
+import qualified Streamly.Unicode.Stream as Unicode
+import qualified System.Console.ANSI as ANSI
+
+--------------------------------------------------------------------------------
+-- Utils
+--------------------------------------------------------------------------------
+
+double :: Int -> Double
+double = fromIntegral
+
+--------------------------------------------------------------------------------
+-- Types
+--------------------------------------------------------------------------------
+
+data Boundary a b
+    = Start a
+    | End a (Maybe b)
+    deriving (Read, Show, Ord, Eq)
+
+getWindowId :: Boundary WindowId Label -> WindowId
+getWindowId (Start a) = a
+getWindowId (End a _) = a
+
+data Counter
+    = ThreadCpuTime
+    | ProcessCpuTime
+    | WallClockTime
+    | Allocated
+    | SchedOut
+    deriving (Read, Show, Ord, Eq)
+
+type WindowId = String
+type Label = String
+type ThreadId = Int32
+type Tag = String
+type Value = Int64
+
+data EventId =
+    EventId
+        { evTid :: ThreadId
+        , evCounter :: Counter
+        , evTag :: Tag
+        }
+    deriving (Eq, Ord, Show)
+
+data UnboundedEvent
+    = UEvent (Boundary WindowId Label) ThreadId Counter Value
+    deriving (Show)
+
+data Event
+    = Event EventId Value
+    deriving (Show)
+
+getEventId :: Event -> EventId
+getEventId (Event evId _) = evId
+
+getEventVal :: Event -> Value
+getEventVal (Event _ evVal) = evVal
+
+--------------------------------------------------------------------------------
+-- Folds
+--------------------------------------------------------------------------------
+
+statsLayout :: [String]
+statsLayout =
+    [ "latest", "total", "count", "avg", "minimum", "maximum", "stddev"]
+
+{-# INLINE stats #-}
+stats :: Fold IO Int64 [(String, Int)]
+stats =
+      Fold.lmap (fromIntegral :: Int64 -> Int)
+    $ Fold.distribute
+        [ fmap (\x -> ("latest", fromJust x)) Fold.latest
+        , fmap (\x -> ("total", x)) Fold.sum
+        , fmap (\x -> ("count", x)) Fold.length
+        , fmap (\x -> ("avg", round x)) (Fold.lmap double Fold.mean)
+        , fmap (\x -> ("minimum", fromJust x)) Fold.minimum
+        , fmap (\x -> ("maximum", fromJust x)) Fold.maximum
+        , fmap (\x -> ("stddev", round x)) (Fold.lmap double Fold.stdDev)
+        ]
+
+--------------------------------------------------------------------------------
+-- Parsing Input
+--------------------------------------------------------------------------------
+
+-- Event format:
+-- Start/<window-id>/<label>/<tid>/<counterName>/<value>
+-- End/<window-id>/<label>/<tid>/<counterName/<value>
+
+errorString :: String -> String -> String
+errorString line reason = [str|Error:
+Line: #{line}
+Reason: #{reason}
+|]
+
+parseLineToEvent :: Monad m => String -> m (Either String UnboundedEvent)
+parseLineToEvent line = do
+    res <-
+        Stream.fromList line
+            & Stream.foldMany (Fold.takeEndBy_ (== '/') Fold.toList)
+            & Stream.toList
+    case res of
+        ["Start", windowId, tid, counter, val] ->
+            case withParsed (UEvent (Start windowId)) tid counter val of
+                Just val -> pure $ Right val
+                Nothing -> pure $ Left $ errorString line "Not valid"
+        ["End", windowId, tid, counter, val] ->
+            case withParsed (UEvent (End windowId Nothing)) tid counter val of
+                Just val -> pure $ Right val
+                Nothing -> pure $ Left $ errorString line "Not valid"
+        ["End", windowId, label, tid, counter, val] ->
+            case withParsed (UEvent (End windowId (Just label))) tid counter val of
+                Just val -> pure $ Right val
+                Nothing -> pure $ Left $ errorString line "Not valid"
+        _ -> pure $ Left $ errorString line "Chunks /= 4"
+
+    where
+
+    withParsed
+        :: (ThreadId -> Counter -> Value -> UnboundedEvent)
+        -> String
+        -> String
+        -> String
+        -> Maybe UnboundedEvent
+    withParsed func tid counter val =
+        func <$> readMaybe tid <*> readMaybe counter <*> readMaybe val
+
+parseInputToEventStream
+    :: MonadIO m => Stream m (Array Word8) -> Stream m UnboundedEvent
+parseInputToEventStream inp =
+    Unicode.decodeUtf8Chunks inp
+        & Stream.foldMany
+              (Fold.takeEndBy_
+                   (== '\n')
+                   (Fold.rmapM parseLineToEvent Fold.toList))
+        & Stream.catRights
+
+--------------------------------------------------------------------------------
+-- Processing stats
+--------------------------------------------------------------------------------
+
+boundEvents :: Monad m => Fold m UnboundedEvent (Maybe Event)
+boundEvents = Fold step initial extract extract
+    where
+    initial = pure $ Partial (Nothing, Map.empty)
+
+    alterFunc :: UnboundedEvent -> Maybe Value -> (Maybe Event, Maybe Value)
+    alterFunc (UEvent (Start _) _ _ val) Nothing = (Nothing, Just val)
+    alterFunc (UEvent (Start _) _ _ val) (Just _) = (Nothing, Just val)
+    alterFunc (UEvent (End w Nothing) tid counter val) (Just prevVal) =
+        ( Just (Event (EventId tid counter w) (val - prevVal))
+        , Nothing
+        )
+    alterFunc (UEvent (End w (Just tag)) tid counter val) (Just prevVal) =
+        ( Just (Event (EventId tid counter (w ++ ":" ++ tag)) (val - prevVal))
+        , Just prevVal
+        )
+    alterFunc _ Nothing = (Nothing, Nothing)
+
+    step (_, mp) uev@(UEvent b tid counter _) =
+        pure $ Partial
+             $ Map.alterF (alterFunc uev) (getWindowId b, tid, counter) mp
+
+    extract (ev, _) = pure ev
+
+statCollector :: Fold IO Event (Map EventId [(String, Int)])
+statCollector =
+    Fold.demuxToMap getEventId deriveFold
+
+    where
+
+    deriveFold ev = pure (Fold.lmap getEventVal stats)
+
+--------------------------------------------------------------------------------
+-- Printing stats
+--------------------------------------------------------------------------------
+
+fill :: Int -> String  -> String
+fill i x =
+    let len = length x
+     in replicate (i - len) ' ' ++ x
+
+printTable :: [[String]] -> IO ()
+printTable rows = do
+    case map (unwords . fillRow) rows of
+        [] -> putStrLn "printTable: empty rows"
+        (header:rest) -> putStrLn $ unlines $ header:unwords separatorRow:rest
+
+    where
+
+    rowLengths = map (map length) rows -- [[Int]]
+    maxLengths = foldl' (zipWith max) (head rowLengths) rowLengths
+    separatorRow = map (\n -> replicate n '-') maxLengths
+    fillRow r = zipWith (\n x -> fill n x) maxLengths r
+
+printStatsMap
+    :: (Show a, Show b, Show c, Ord a, Ord b, Ord c)
+    => (EventId -> a)
+    -> (EventId -> b)
+    -> (EventId -> c)
+    -> Map EventId [(String, Int)]
+    -> IO ()
+printStatsMap index1 index2 index3 mp =
+    mapM_ printOneTable $ Map.toList $ anchorOnTidAndCounter mp
+
+    where
+
+    alterFunction v Nothing = Just [v]
+    alterFunction v (Just v0) = Just (v:v0)
+
+    foldingFunction mp ev v =
+        Map.alter (alterFunction (index3 ev, v)) (index1 ev, index2 ev) mp
+
+    anchorOnTidAndCounter mp =
+        Map.foldlWithKey' foldingFunction Map.empty mp
+
+    printOneTable ((i1, i2), rows) = do
+        let i1Str = show i1
+            i2Str = show i2
+            headingL1 = [str|Index1: #{i1Str}|]
+            headingL2 = [str|Index2: #{i2Str}|]
+            divider = replicate (max (length headingL2) (length headingL1)) '-'
+        putStrLn divider
+        putStrLn headingL1
+        putStrLn headingL2
+        putStrLn divider
+        putStrLn ""
+        printTable
+            $ (:) tableHeader
+            $ map (\(i3, v) -> show i3 : map (pShowInt . snd) v) rows
+        putStrLn ""
+
+    pShowInt = Text.unpack . prettyI (Just ',')
+
+    tableHeader = "Index3":statsLayout
+
+--------------------------------------------------------------------------------
+-- Main
+--------------------------------------------------------------------------------
+
+main :: IO ()
+main = do
+    statsMap <-
+        Stream.unfold Handle.chunkReader stdin
+            & parseInputToEventStream
+            & Stream.scan boundEvents
+            & Stream.catMaybes
+            & Stream.fold statCollector
+    (arg:[]) <- getArgs
+    case arg of
+        "Tag" ->
+            printStatsMap
+                (evTid)
+                (evCounter)
+                (evTag)
+                statsMap
+        "Counter" ->
+            printStatsMap
+                (evTid)
+                (evTag)
+                (evCounter)
+                statsMap
+        "ThreadId" ->
+            printStatsMap
+                (evCounter)
+                (evTag)
+                (evTid)
+                statsMap

From 6ac288cdfe703a162f6df9b367846002ba5d40a5 Mon Sep 17 00:00:00 2001
From: Adithya Obilisetty <adithya.obilisetty@juspay.in>
Date: Sat, 3 Feb 2024 01:49:49 +0530
Subject: [PATCH 02/10] Add document that describes the nomenclature of
 identifying windows

---
 dev/LiveStatWindows.md | 48 ++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 48 insertions(+)
 create mode 100644 dev/LiveStatWindows.md

diff --git a/dev/LiveStatWindows.md b/dev/LiveStatWindows.md
new file mode 100644
index 0000000..5f1e16a
--- /dev/null
+++ b/dev/LiveStatWindows.md
@@ -0,0 +1,48 @@
+# Windows
+
+In a linear flow, we have the ability to start a window anywhere and
+stop it anywhere. A winow is basically the realestate between any 2
+points in the linear code.
+
+The goal is to make it easier and convinent to create windows.  A
+point in the linear flow should have some identification. We can use
+the `ModuleName:LineNumber` to identify this. Using this has two
+advantages. One being, removal of developer overhead of thinking about
+a name and the other being extremely fast identification of whether
+the point is enabled or not based on the line number.
+
+A window can be indentified by a tag. The tag, can be derived from
+point identification. We can use
+`ModuleName:LineNumber-ModuleName:LineNumber` respectively.
+
+This nomenclature lets us create non overlapping windows and
+overlapping windows windows that are completely encapsulated.
+
+Non overlapping:
+```
+............[...........].....................................
+......................................[.............].........
+```
+
+Overlapping:
+```
+............[.......................................].........
+.....................[................].......................
+```
+
+
+Window like this aren't possible:
+```
+............[.........................].......................
+.....................[...........................]............
+```
+
+From my experience, These kind of windows are rarely needed. But for
+the 1% use-cases we can tackle this problem.
+
+To make the windows like the above possible, we can use namespaces.
+Windows can belong to certain namespaces.
+
+We can use the `NameSpace[ModuleName:LineNumber]` to identify a point,
+and `NameSpace[ModuleName:LineNumber-ModuleName:LineNumber]` to
+identify a window.

From 800e7c2e2164cd02814ba48c4d8f39b21bbf09f0 Mon Sep 17 00:00:00 2001
From: Adithya Obilisetty <adithya.obilisetty@juspay.in>
Date: Sat, 3 Feb 2024 03:03:23 +0530
Subject: [PATCH 03/10] Update the stat-collector executable using a newer
 format

---
 dev/LiveStatWindows.md     |   3 +
 haskell-perf.cabal         |   2 -
 stat-collector-src/Main.hs | 175 ++++++++++++++++++++++---------------
 3 files changed, 109 insertions(+), 71 deletions(-)

diff --git a/dev/LiveStatWindows.md b/dev/LiveStatWindows.md
index 5f1e16a..29d9e00 100644
--- a/dev/LiveStatWindows.md
+++ b/dev/LiveStatWindows.md
@@ -46,3 +46,6 @@ Windows can belong to certain namespaces.
 We can use the `NameSpace[ModuleName:LineNumber]` to identify a point,
 and `NameSpace[ModuleName:LineNumber-ModuleName:LineNumber]` to
 identify a window.
+
+We can consider the point to be in a global namespace if the namespace
+isn't mentioned.
diff --git a/haskell-perf.cabal b/haskell-perf.cabal
index dc684a6..088440e 100644
--- a/haskell-perf.cabal
+++ b/haskell-perf.cabal
@@ -128,7 +128,5 @@ executable stat-collector
      , streamly-core
      , streamly
      , containers
-     , ansi-terminal
      , format-numbers
-     , unix
      , text
diff --git a/stat-collector-src/Main.hs b/stat-collector-src/Main.hs
index 253ea23..221aa26 100644
--- a/stat-collector-src/Main.hs
+++ b/stat-collector-src/Main.hs
@@ -1,4 +1,5 @@
 {-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 
 module Main (main) where
 
@@ -9,36 +10,27 @@ module Main (main) where
 -- import Control.Concurrent (threadDelay)
 import Data.Int (Int32, Int64)
 import System.Environment (getArgs)
-import Control.Monad.IO.Class (MonadIO(..))
 import Data.Function ((&))
-import Data.List (foldl', findIndex, sortBy, find)
+import Data.List (foldl', uncons)
+import Data.Char (ord)
 import Data.Map (Map)
-import Data.Maybe (fromMaybe, fromJust)
+import Data.Maybe (fromJust)
 import Data.Word (Word8)
-import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
-import Foreign.Storable (Storable, peek)
-import Numeric (showFFloat)
 import Streamly.Data.Array (Array)
 import Streamly.Data.Fold (Fold)
 import Streamly.Data.Stream (Stream)
 import Streamly.Internal.Data.Fold (Fold(..), Step(..))
-import Streamly.Internal.Data.Ring (slidingWindow)
-import Streamly.Internal.Data.Tuple.Strict (Tuple3Fused' (Tuple3Fused'))
 import Streamly.Unicode.String (str)
-import System.IO (hFlush, stdout, stdin)
-import Text.Read (readMaybe)
-import System.Posix.Signals (installHandler, Handler(Catch), sigINT, sigTERM)
+import System.IO (stdin)
 import Data.Text.Format.Numbers (prettyI)
+import Control.Exception (catch, SomeException, displayException)
 
 import qualified Data.Text as Text
 import qualified Data.Map as Map
 import qualified Streamly.Data.Fold as Fold
 import qualified Streamly.Data.Stream.Prelude as Stream
 import qualified Streamly.FileSystem.Handle as Handle
-import qualified Streamly.Internal.Data.Fold as Fold
-import qualified Streamly.Internal.Data.Ring as Ring
 import qualified Streamly.Unicode.Stream as Unicode
-import qualified System.Console.ANSI as ANSI
 
 --------------------------------------------------------------------------------
 -- Utils
@@ -52,13 +44,15 @@ double = fromIntegral
 --------------------------------------------------------------------------------
 
 data Boundary a b
-    = Start a
-    | End a (Maybe b)
+    = Start a b
+    | Record a b
+    | End a b
     deriving (Read, Show, Ord, Eq)
 
-getWindowId :: Boundary WindowId Label -> WindowId
-getWindowId (Start a) = a
-getWindowId (End a _) = a
+getNameSpace :: Boundary NameSpace PointId -> NameSpace
+getNameSpace (Start a _) = a
+getNameSpace (Record a _) = a
+getNameSpace (End a _) = a
 
 data Counter
     = ThreadCpuTime
@@ -68,8 +62,10 @@ data Counter
     | SchedOut
     deriving (Read, Show, Ord, Eq)
 
-type WindowId = String
-type Label = String
+type NameSpace = String
+type ModuleName = String
+type LineNum = Int
+type PointId = (ModuleName, LineNum)
 type ThreadId = Int32
 type Tag = String
 type Value = Int64
@@ -83,7 +79,7 @@ data EventId =
     deriving (Eq, Ord, Show)
 
 data UnboundedEvent
-    = UEvent (Boundary WindowId Label) ThreadId Counter Value
+    = UEvent (Boundary NameSpace PointId) ThreadId Counter Value
     deriving (Show)
 
 data Event
@@ -123,8 +119,12 @@ stats =
 --------------------------------------------------------------------------------
 
 -- Event format:
--- Start/<window-id>/<label>/<tid>/<counterName>/<value>
--- End/<window-id>/<label>/<tid>/<counterName/<value>
+-- Start,<tag>,<tid>,<counterName>,<value>
+-- Record,<tag>,<tid>,<counterName>,<value>
+-- End,<tag>,<tid>,<counterName,<value>
+
+-- Tag format:
+-- NameSpace[ModuleName:LineNumber]
 
 errorString :: String -> String -> String
 errorString line reason = [str|Error:
@@ -132,40 +132,59 @@ Line: #{line}
 Reason: #{reason}
 |]
 
-parseLineToEvent :: Monad m => String -> m (Either String UnboundedEvent)
-parseLineToEvent line = do
-    res <-
-        Stream.fromList line
-            & Stream.foldMany (Fold.takeEndBy_ (== '/') Fold.toList)
-            & Stream.toList
-    case res of
-        ["Start", windowId, tid, counter, val] ->
-            case withParsed (UEvent (Start windowId)) tid counter val of
-                Just val -> pure $ Right val
-                Nothing -> pure $ Left $ errorString line "Not valid"
-        ["End", windowId, tid, counter, val] ->
-            case withParsed (UEvent (End windowId Nothing)) tid counter val of
-                Just val -> pure $ Right val
-                Nothing -> pure $ Left $ errorString line "Not valid"
-        ["End", windowId, label, tid, counter, val] ->
-            case withParsed (UEvent (End windowId (Just label))) tid counter val of
-                Just val -> pure $ Right val
-                Nothing -> pure $ Left $ errorString line "Not valid"
-        _ -> pure $ Left $ errorString line "Chunks /= 4"
+fIntegral :: (Monad m, Integral a) => Fold m Char a
+fIntegral =
+    Fold.foldl' (\b a -> 10 * b + ord1 a) 0
+    where
+    ord1 a =
+        case ord a - 48 of
+            x ->
+                if x >= 0 || x <= 9
+                then fromIntegral x
+                else error "fIntegral: NaN"
+
+fEventBoundary ::
+    Monad m => Fold m Char (NameSpace -> PointId -> Boundary NameSpace PointId)
+fEventBoundary =
+    f <$> Fold.toList
+    where
+    f "Start" = Start
+    f "Record" = Record
+    f "End" = End
+    f _ = error "fEventBoundary: undefined"
+
+fCounterName :: Monad m => Fold m Char Counter
+fCounterName = read <$> Fold.toList
+
+fTag :: Monad m => Fold m Char (NameSpace, PointId)
+fTag =
+    (\a b c -> (a, (b, c)))
+        <$> Fold.takeEndBy_ (== '[') Fold.toList
+        <*> Fold.takeEndBy_ (== ':') Fold.toList
+        <*> Fold.takeEndBy_ (== ']') fIntegral
+
+fUnboundedEvent :: Monad m => Fold m Char UnboundedEvent
+fUnboundedEvent =
+    f
+        <$> (Fold.takeEndBy_ (== ',') fEventBoundary)
+        <*> (fTag <* Fold.one) -- fTag is a terminating fold
+        <*> (Fold.takeEndBy_ (== ',') fIntegral)
+        <*> (Fold.takeEndBy_ (== ',') fCounterName)
+        <*> fIntegral
 
     where
 
-    withParsed
-        :: (ThreadId -> Counter -> Value -> UnboundedEvent)
-        -> String
-        -> String
-        -> String
-        -> Maybe UnboundedEvent
-    withParsed func tid counter val =
-        func <$> readMaybe tid <*> readMaybe counter <*> readMaybe val
+    f a b c d e = UEvent (a (fst b) (snd b)) c d e
+
+parseLineToEvent :: String -> IO (Either String UnboundedEvent)
+parseLineToEvent line =
+    catch
+        (Right <$> Stream.fold fUnboundedEvent (Stream.fromList line))
+        (\(e :: SomeException) ->
+             pure (Left (errorString line (displayException e))))
 
 parseInputToEventStream
-    :: MonadIO m => Stream m (Array Word8) -> Stream m UnboundedEvent
+    :: Stream IO (Array Word8) -> Stream IO UnboundedEvent
 parseInputToEventStream inp =
     Unicode.decodeUtf8Chunks inp
         & Stream.foldMany
@@ -183,22 +202,39 @@ boundEvents = Fold step initial extract extract
     where
     initial = pure $ Partial (Nothing, Map.empty)
 
-    alterFunc :: UnboundedEvent -> Maybe Value -> (Maybe Event, Maybe Value)
-    alterFunc (UEvent (Start _) _ _ val) Nothing = (Nothing, Just val)
-    alterFunc (UEvent (Start _) _ _ val) (Just _) = (Nothing, Just val)
-    alterFunc (UEvent (End w Nothing) tid counter val) (Just prevVal) =
-        ( Just (Event (EventId tid counter w) (val - prevVal))
-        , Nothing
-        )
-    alterFunc (UEvent (End w (Just tag)) tid counter val) (Just prevVal) =
-        ( Just (Event (EventId tid counter (w ++ ":" ++ tag)) (val - prevVal))
-        , Just prevVal
-        )
+    alterFunc
+        :: UnboundedEvent
+        -> Maybe [(PointId, Value)]
+        -> (Maybe Event, Maybe [(PointId, Value)])
+    alterFunc (UEvent (Start _ point) _ _ val) Nothing =
+        (Nothing, Just [(point, val)])
+    alterFunc (UEvent (Start _ point) _ _ val) (Just xs) =
+        (Nothing, Just ((point, val):xs))
+    alterFunc (UEvent (End ns (md, ln)) tid counter val) (Just stk) =
+        case uncons stk of
+            Just (((md1, ln1), prevVal), stk1) ->
+                let lnStr = show ln
+                    ln1Str = show ln1
+                    win = [str|#{ns}[#{md1}:#{ln1Str}-#{md}:#{lnStr}]|]
+                 in ( Just (Event (EventId tid counter win) (val - prevVal))
+                    , Just stk1
+                    )
+            Nothing -> error "boundEvents: Empty stack"
+    alterFunc (UEvent (Record ns (md, ln)) tid counter val) (Just stk) =
+        case uncons stk of
+            Just (((md1, ln1), prevVal), _) ->
+                let lnStr = show ln
+                    ln1Str = show ln1
+                    win = [str|#{ns}[#{md1}:#{ln1Str}-#{md}:#{lnStr}]|]
+                 in ( Just (Event (EventId tid counter win) (val - prevVal))
+                    , Just stk
+                    )
+            Nothing -> error "boundEvents: Empty stack"
     alterFunc _ Nothing = (Nothing, Nothing)
 
     step (_, mp) uev@(UEvent b tid counter _) =
         pure $ Partial
-             $ Map.alterF (alterFunc uev) (getWindowId b, tid, counter) mp
+             $ Map.alterF (alterFunc uev) (getNameSpace b, tid, counter) mp
 
     extract (ev, _) = pure ev
 
@@ -208,7 +244,7 @@ statCollector =
 
     where
 
-    deriveFold ev = pure (Fold.lmap getEventVal stats)
+    deriveFold _ = pure (Fold.lmap getEventVal stats)
 
 --------------------------------------------------------------------------------
 -- Printing stats
@@ -233,14 +269,14 @@ printTable rows = do
     fillRow r = zipWith (\n x -> fill n x) maxLengths r
 
 printStatsMap
-    :: (Show a, Show b, Show c, Ord a, Ord b, Ord c)
+    :: (Show a, Show b, Show c, Ord a, Ord b)
     => (EventId -> a)
     -> (EventId -> b)
     -> (EventId -> c)
     -> Map EventId [(String, Int)]
     -> IO ()
-printStatsMap index1 index2 index3 mp =
-    mapM_ printOneTable $ Map.toList $ anchorOnTidAndCounter mp
+printStatsMap index1 index2 index3 mp0 =
+    mapM_ printOneTable $ Map.toList $ anchorOnTidAndCounter mp0
 
     where
 
@@ -305,3 +341,4 @@ main = do
                 (evTag)
                 (evTid)
                 statsMap
+        _ -> error "Undefined arg."

From 619af9bda44567924e95899e23c992773faf8d11 Mon Sep 17 00:00:00 2001
From: Adithya Obilisetty <adithya.obilisetty@juspay.in>
Date: Sat, 3 Feb 2024 04:21:30 +0530
Subject: [PATCH 04/10] Add a restart tag

---
 stat-collector-src/Main.hs | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/stat-collector-src/Main.hs b/stat-collector-src/Main.hs
index 221aa26..542edc8 100644
--- a/stat-collector-src/Main.hs
+++ b/stat-collector-src/Main.hs
@@ -46,12 +46,14 @@ double = fromIntegral
 data Boundary a b
     = Start a b
     | Record a b
+    | Restart a b
     | End a b
     deriving (Read, Show, Ord, Eq)
 
 getNameSpace :: Boundary NameSpace PointId -> NameSpace
 getNameSpace (Start a _) = a
 getNameSpace (Record a _) = a
+getNameSpace (Restart a _) = a
 getNameSpace (End a _) = a
 
 data Counter
@@ -121,6 +123,7 @@ stats =
 -- Event format:
 -- Start,<tag>,<tid>,<counterName>,<value>
 -- Record,<tag>,<tid>,<counterName>,<value>
+-- Restart,<tag>,<tid>,<counterName>,<value>
 -- End,<tag>,<tid>,<counterName,<value>
 
 -- Tag format:
@@ -150,6 +153,7 @@ fEventBoundary =
     where
     f "Start" = Start
     f "Record" = Record
+    f "Restart" = Restart
     f "End" = End
     f _ = error "fEventBoundary: undefined"
 
@@ -220,6 +224,16 @@ boundEvents = Fold step initial extract extract
                     , Just stk1
                     )
             Nothing -> error "boundEvents: Empty stack"
+    alterFunc (UEvent (Restart ns point@(md, ln)) tid counter val) (Just stk) =
+        case uncons stk of
+            Just (((md1, ln1), prevVal), stk1) ->
+                let lnStr = show ln
+                    ln1Str = show ln1
+                    win = [str|#{ns}[#{md1}:#{ln1Str}-#{md}:#{lnStr}]|]
+                 in ( Just (Event (EventId tid counter win) (val - prevVal))
+                    , Just ((point ,val):stk1)
+                    )
+            Nothing -> error "boundEvents: Empty stack"
     alterFunc (UEvent (Record ns (md, ln)) tid counter val) (Just stk) =
         case uncons stk of
             Just (((md1, ln1), prevVal), _) ->

From 720553e3954e29d002f534f876d8b63b051612cf Mon Sep 17 00:00:00 2001
From: Adithya Obilisetty <adithya.obilisetty@juspay.in>
Date: Sun, 4 Feb 2024 13:55:45 +0530
Subject: [PATCH 05/10] Use throwIO instead of error

---
 stat-collector-src/Main.hs | 31 ++++++++++++++++---------------
 1 file changed, 16 insertions(+), 15 deletions(-)

diff --git a/stat-collector-src/Main.hs b/stat-collector-src/Main.hs
index 542edc8..d16fdd2 100644
--- a/stat-collector-src/Main.hs
+++ b/stat-collector-src/Main.hs
@@ -23,7 +23,8 @@ import Streamly.Internal.Data.Fold (Fold(..), Step(..))
 import Streamly.Unicode.String (str)
 import System.IO (stdin)
 import Data.Text.Format.Numbers (prettyI)
-import Control.Exception (catch, SomeException, displayException)
+import Control.Exception
+    (catch, SomeException, displayException, throwIO, ErrorCall(..))
 
 import qualified Data.Text as Text
 import qualified Data.Map as Map
@@ -135,39 +136,39 @@ Line: #{line}
 Reason: #{reason}
 |]
 
-fIntegral :: (Monad m, Integral a) => Fold m Char a
+fIntegral :: Integral a => Fold IO Char a
 fIntegral =
-    Fold.foldl' (\b a -> 10 * b + ord1 a) 0
+    Fold.foldlM' combine (pure 0)
     where
-    ord1 a =
+    combine b a =
         case ord a - 48 of
             x ->
                 if x >= 0 || x <= 9
-                then fromIntegral x
-                else error "fIntegral: NaN"
+                then pure $ fromIntegral x + b * 10
+                else throwIO $ ErrorCall "fIntegral: NaN"
 
 fEventBoundary ::
-    Monad m => Fold m Char (NameSpace -> PointId -> Boundary NameSpace PointId)
+    Fold IO Char (NameSpace -> PointId -> Boundary NameSpace PointId)
 fEventBoundary =
-    f <$> Fold.toList
+    Fold.rmapM f Fold.toList
     where
-    f "Start" = Start
-    f "Record" = Record
-    f "Restart" = Restart
-    f "End" = End
-    f _ = error "fEventBoundary: undefined"
+    f "Start" = pure Start
+    f "Record" = pure Record
+    f "Restart" = pure Restart
+    f "End" = pure End
+    f _ = throwIO $ ErrorCall "fEventBoundary: undefined"
 
 fCounterName :: Monad m => Fold m Char Counter
 fCounterName = read <$> Fold.toList
 
-fTag :: Monad m => Fold m Char (NameSpace, PointId)
+fTag :: Fold IO Char (NameSpace, PointId)
 fTag =
     (\a b c -> (a, (b, c)))
         <$> Fold.takeEndBy_ (== '[') Fold.toList
         <*> Fold.takeEndBy_ (== ':') Fold.toList
         <*> Fold.takeEndBy_ (== ']') fIntegral
 
-fUnboundedEvent :: Monad m => Fold m Char UnboundedEvent
+fUnboundedEvent :: Fold IO Char UnboundedEvent
 fUnboundedEvent =
     f
         <$> (Fold.takeEndBy_ (== ',') fEventBoundary)

From 01a4ca71a1f41b51fa704b4001cfddd0b9e1d9e8 Mon Sep 17 00:00:00 2001
From: Adithya Obilisetty <adithya.obilisetty@juspay.in>
Date: Mon, 5 Feb 2024 19:49:19 +0530
Subject: [PATCH 06/10] Add a lib that generates stats

---
 haskell-perf.cabal |  13 ++-
 lib/Stat.hs        | 227 +++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 239 insertions(+), 1 deletion(-)
 create mode 100644 lib/Stat.hs

diff --git a/haskell-perf.cabal b/haskell-perf.cabal
index 088440e..fda16eb 100644
--- a/haskell-perf.cabal
+++ b/haskell-perf.cabal
@@ -105,6 +105,17 @@ common compile-options
                     -Wall-missed-specialisations
                     -fno-ignore-asserts
 
+library
+  import: compile-options
+  hs-source-dirs: lib
+  exposed-modules: Stat
+  ghc-options: -O2 -fmax-worker-args=16 -fspec-constr-recursive=16
+  build-depends:
+       base >= 4.9 && < 5
+     , streamly-core == 0.2.2
+     , time
+     , template-haskell
+
 executable hperf
   import: compile-options
   hs-source-dirs: src
@@ -114,7 +125,7 @@ executable hperf
   build-depends:
        base >= 4.9 && < 5
      , containers
-     , streamly-core == 0.2.0
+     , streamly-core == 0.2.2
      , format-numbers
      , text
 
diff --git a/lib/Stat.hs b/lib/Stat.hs
new file mode 100644
index 0000000..b1f203d
--- /dev/null
+++ b/lib/Stat.hs
@@ -0,0 +1,227 @@
+{-# LANGUAGE CPP #-}
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE MagicHash #-}
+{-# LANGUAGE UnboxedTuples #-}
+{-# LANGUAGE UnliftedFFITypes #-}
+{-# LANGUAGE TemplateHaskell #-}
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# OPTIONS_GHC -Wno-missing-export-lists #-}
+{-# OPTIONS_GHC -Wno-implicit-lift #-}
+
+module Stat where
+
+--------------------------------------------------------------------------------
+-- Imports
+--------------------------------------------------------------------------------
+
+import Control.Monad (forM_)
+import Data.Bits (shiftR, (.&.), setBit, testBit, complement)
+-- import Data.IORef (newIORef, IORef, atomicModifyIORef)
+import Data.Word (Word8)
+import Foreign.C.Types (CInt(..))
+import GHC.Conc.Sync (myThreadId, ThreadId(..))
+import GHC.Exts (ThreadId#)
+import GHC.IO(IO(..))
+import GHC.IO.Unsafe (unsafePerformIO)
+import GHC.Int(Int64(..), Int32(..))
+import System.Environment (lookupEnv)
+import System.IO (stdout)
+-- import Streamly.Unicode.String (str)
+import System.CPUTime (getCPUTime)
+import Data.Time.Clock (getCurrentTime, UTCTime(..), diffUTCTime)
+import Streamly.Internal.Data.Array (Array(..))
+import Streamly.FileSystem.Handle as Handle
+
+import qualified Streamly.Data.StreamK as StreamK
+import qualified Streamly.Data.Stream as Stream
+import qualified Streamly.Internal.Data.Array as Array
+import qualified Streamly.Internal.Data.MutByteArray as MBA
+
+import Prelude
+import Language.Haskell.TH
+
+#ifdef PRIM_OP_AVAILABLE
+import GHC.Exts (threadCPUTime#)
+#else
+import GHC.Exts (RealWorld, Int32#, Int64#, State#)
+
+threadCPUTime# ::
+    State# RealWorld -> (# State# RealWorld, Int64#, Int64#, Int32#, Int32# #)
+threadCPUTime# = undefined
+#endif
+
+--------------------------------------------------------------------------------
+-- Thread stat
+--------------------------------------------------------------------------------
+
+type SrcLoc = Loc
+
+data EvLoc
+    = Start
+    | Record
+    | Restart
+    | End
+    deriving (Read, Show, Ord, Eq)
+$(MBA.deriveSerialize [d|instance MBA.Serialize EvLoc|])
+
+data Counter
+    = ThreadCpuTime
+    | ProcessCpuTime
+    | WallClockTime
+    | Allocated
+    | SchedOut
+    deriving (Read, Show, Ord, Eq)
+$(MBA.deriveSerialize [d|instance MBA.Serialize Counter|])
+
+data Metric =
+    Metric
+        { m_tid :: Int32
+        , m_modName :: String
+        , m_lineNum :: Int32
+        , m_counter :: Counter
+        , m_location :: EvLoc
+        , m_value :: Int64
+        }
+$(MBA.deriveSerialize [d|instance MBA.Serialize Metric|])
+
+{-# INLINE tenPow9 #-}
+tenPow9 :: Int64
+tenPow9 = 1000000000
+
+--------------------------------------------------------------------------------
+-- Window Type
+--------------------------------------------------------------------------------
+
+createBitMap :: Maybe [Int] -> IO MBA.MutByteArray
+createBitMap mLineList = do
+    let maxLines = 8000
+    arr <- MBA.new maxLines
+    forM_ [0..(maxLines - 1)]
+        $ \i -> MBA.pokeAt i arr (0 :: Word8)
+    case mLineList of
+        Nothing -> pure arr
+        Just [] -> do
+            forM_ [0..(maxLines - 1)]
+                $ \i -> MBA.pokeAt i arr (complement (0 :: Word8))
+            pure arr
+        Just lineList -> do
+            forM_ lineList $ \ln -> pokeBitOn ln arr
+            pure arr
+
+--------------------------------------------------------------------------------
+-- Helpers
+--------------------------------------------------------------------------------
+
+pokeBitOn :: Int -> MBA.MutByteArray -> IO ()
+pokeBitOn i arr = do
+    val <- MBA.peekAt (shiftR i 3) arr
+    MBA.pokeAt (shiftR i 3) arr (setBit (val :: Word8) (i .&. 7))
+
+testBitOn :: Int -> MBA.MutByteArray -> IO Bool
+testBitOn i arr = do
+    val <- MBA.peekAt (shiftR i 3) arr
+    pure $ testBit (val :: Word8) (i .&. 7)
+
+sizedSerialize :: MBA.Serialize a => a -> IO (Array Word8)
+sizedSerialize a = do
+    let len = MBA.addSizeTo 0 a + 8
+    mbarr <- MBA.new len
+    off0 <- MBA.serializeAt 0 mbarr (fromIntegral len :: Int64)
+    off1 <- MBA.serializeAt off0 mbarr a
+    pure $ Array mbarr 0 off1
+
+--------------------------------------------------------------------------------
+-- Setup Env
+--------------------------------------------------------------------------------
+
+envMeasurementWindows :: Maybe String
+envMeasurementWindows = unsafePerformIO $ lookupEnv "MEASUREMENT_WINDOWS"
+
+--------------------------------------------------------------------------------
+-- Measurement
+--------------------------------------------------------------------------------
+
+foreign import ccall unsafe "rts_getThreadId" getThreadId :: ThreadId# -> CInt
+
+{-# INLINE getThreadStatLowLevel #-}
+getThreadStatLowLevel :: IO (Int64, Int64, Int32, Int32)
+getThreadStatLowLevel = IO $ \s ->
+   case threadCPUTime# s of
+    (# s', sec, nsec, alloc, count_sched #) ->
+        (# s', (I64# sec, I64# nsec, I32# alloc, I32# count_sched) #)
+
+picoToNanoSeconds :: Integer -> Int64
+picoToNanoSeconds x = fromIntegral (x `div` 1000)
+
+epochTime :: UTCTime
+epochTime = UTCTime (toEnum 0) 0
+
+getThreadStat :: IO (Int32, Int64, Int64, Int64)
+getThreadStat = do
+    ThreadId tid <- myThreadId
+    (sec, nsec, alloc, switches) <- getThreadStatLowLevel
+    pure
+        ( fromIntegral (getThreadId tid)
+        , sec * tenPow9 + nsec
+        , fromIntegral (alloc * 8)
+        , fromIntegral switches
+        )
+
+printMetricList :: [Metric] -> IO ()
+printMetricList mList = do
+    arr <-
+        Array.fromChunksK
+            $ StreamK.mapM sizedSerialize
+            $ StreamK.fromStream
+            $ Stream.fromList mList
+    putChunk stdout arr
+
+
+eventGeneric :: (forall b. IO b -> m b) -> EvLoc -> SrcLoc -> m ()
+eventGeneric liftio evLoc srcLoc = liftio $ do
+    (a, b, c, d) <- getThreadStat
+    let modName = loc_module srcLoc
+    let lnNum = (fromIntegral :: Int -> Int32) $ fst (loc_start srcLoc)
+    pCpuTime <- picoToNanoSeconds <$> getCPUTime
+    wTimeU <- getCurrentTime
+    let wTime = round $ diffUTCTime wTimeU epochTime * 1e9
+    let mList =
+            [ Metric a modName lnNum ThreadCpuTime evLoc b
+            , Metric a modName lnNum Allocated evLoc c
+            , Metric a modName lnNum SchedOut evLoc d
+            , Metric a modName lnNum ProcessCpuTime evLoc pCpuTime
+            , Metric a modName lnNum WallClockTime evLoc wTime
+            ]
+{-
+    shouldStat <- testBitOn (fromEnum win) statEnv
+    if True
+    then printMetricList mList
+    else pure ()
+-}
+    printMetricList mList
+
+withEvLoc :: Q Exp -> Q Exp
+withEvLoc f = do
+    Loc a b c d e <- location
+    appE f [| Loc a b c d e |]
+
+start :: Q Exp
+start = do
+    Loc a b c d e <- location
+    [|\liftio -> eventGeneric liftio Start (Loc a b c d e)|]
+
+end :: Q Exp
+end = do
+    Loc a b c d e <- location
+    [|\liftio -> eventGeneric liftio End (Loc a b c d e)|]
+
+record :: Q Exp
+record = do
+    Loc a b c d e <- location
+    [|\liftio -> eventGeneric liftio Record (Loc a b c d e)|]
+
+restart :: Q Exp
+restart = do
+    Loc a b c d e <- location
+    [|\liftio -> eventGeneric liftio Restart (Loc a b c d e)|]

From ea471a2f6d7bd6ff2d0f2ad278b3e2f6eda48a2b Mon Sep 17 00:00:00 2001
From: Adithya Obilisetty <adithya.obilisetty@juspay.in>
Date: Tue, 6 Feb 2024 00:21:17 +0530
Subject: [PATCH 07/10] Update the stat-collector to work properly with the
 library

---
 haskell-perf.cabal         |   1 +
 lib/Stat.hs                |  27 ++++---
 stat-collector-src/Main.hs | 148 +++++++++----------------------------
 3 files changed, 53 insertions(+), 123 deletions(-)

diff --git a/haskell-perf.cabal b/haskell-perf.cabal
index fda16eb..3e3908a 100644
--- a/haskell-perf.cabal
+++ b/haskell-perf.cabal
@@ -141,3 +141,4 @@ executable stat-collector
      , containers
      , format-numbers
      , text
+     , haskell-perf
diff --git a/lib/Stat.hs b/lib/Stat.hs
index b1f203d..e61b84f 100644
--- a/lib/Stat.hs
+++ b/lib/Stat.hs
@@ -41,6 +41,9 @@ import qualified Streamly.Internal.Data.MutByteArray as MBA
 import Prelude
 import Language.Haskell.TH
 
+#define PRIM_OP_AVAILABLE
+-- #undef PRIM_OP_AVAILABLE
+
 #ifdef PRIM_OP_AVAILABLE
 import GHC.Exts (threadCPUTime#)
 #else
@@ -77,12 +80,14 @@ $(MBA.deriveSerialize [d|instance MBA.Serialize Counter|])
 data Metric =
     Metric
         { m_tid :: Int32
+        , m_namespace :: String
         , m_modName :: String
         , m_lineNum :: Int32
         , m_counter :: Counter
         , m_location :: EvLoc
         , m_value :: Int64
         }
+    deriving (Show)
 $(MBA.deriveSerialize [d|instance MBA.Serialize Metric|])
 
 {-# INLINE tenPow9 #-}
@@ -178,8 +183,8 @@ printMetricList mList = do
     putChunk stdout arr
 
 
-eventGeneric :: (forall b. IO b -> m b) -> EvLoc -> SrcLoc -> m ()
-eventGeneric liftio evLoc srcLoc = liftio $ do
+eventGeneric :: (forall b. IO b -> m b) -> String -> EvLoc -> SrcLoc -> m ()
+eventGeneric liftio namespace evLoc srcLoc = liftio $ do
     (a, b, c, d) <- getThreadStat
     let modName = loc_module srcLoc
     let lnNum = (fromIntegral :: Int -> Int32) $ fst (loc_start srcLoc)
@@ -187,11 +192,11 @@ eventGeneric liftio evLoc srcLoc = liftio $ do
     wTimeU <- getCurrentTime
     let wTime = round $ diffUTCTime wTimeU epochTime * 1e9
     let mList =
-            [ Metric a modName lnNum ThreadCpuTime evLoc b
-            , Metric a modName lnNum Allocated evLoc c
-            , Metric a modName lnNum SchedOut evLoc d
-            , Metric a modName lnNum ProcessCpuTime evLoc pCpuTime
-            , Metric a modName lnNum WallClockTime evLoc wTime
+            [ Metric a namespace modName lnNum ThreadCpuTime evLoc b
+            , Metric a namespace modName lnNum Allocated evLoc c
+            , Metric a namespace modName lnNum SchedOut evLoc d
+            , Metric a namespace modName lnNum ProcessCpuTime evLoc pCpuTime
+            , Metric a namespace modName lnNum WallClockTime evLoc wTime
             ]
 {-
     shouldStat <- testBitOn (fromEnum win) statEnv
@@ -209,19 +214,19 @@ withEvLoc f = do
 start :: Q Exp
 start = do
     Loc a b c d e <- location
-    [|\liftio -> eventGeneric liftio Start (Loc a b c d e)|]
+    [|eventGeneric id "g" Start (Loc a b c d e)|]
 
 end :: Q Exp
 end = do
     Loc a b c d e <- location
-    [|\liftio -> eventGeneric liftio End (Loc a b c d e)|]
+    [|eventGeneric id "g" End (Loc a b c d e)|]
 
 record :: Q Exp
 record = do
     Loc a b c d e <- location
-    [|\liftio -> eventGeneric liftio Record (Loc a b c d e)|]
+    [|eventGeneric id "g" Record (Loc a b c d e)|]
 
 restart :: Q Exp
 restart = do
     Loc a b c d e <- location
-    [|\liftio -> eventGeneric liftio Restart (Loc a b c d e)|]
+    [|eventGeneric id "g" Restart (Loc a b c d e)|]
diff --git a/stat-collector-src/Main.hs b/stat-collector-src/Main.hs
index d16fdd2..6d38e42 100644
--- a/stat-collector-src/Main.hs
+++ b/stat-collector-src/Main.hs
@@ -20,18 +20,27 @@ import Streamly.Data.Array (Array)
 import Streamly.Data.Fold (Fold)
 import Streamly.Data.Stream (Stream)
 import Streamly.Internal.Data.Fold (Fold(..), Step(..))
+import Streamly.Data.ParserK (ParserK)
 import Streamly.Unicode.String (str)
 import System.IO (stdin)
 import Data.Text.Format.Numbers (prettyI)
 import Control.Exception
     (catch, SomeException, displayException, throwIO, ErrorCall(..))
+import qualified Streamly.Internal.Data.StreamK as StreamK (unfoldrM)
 
 import qualified Data.Text as Text
 import qualified Data.Map as Map
 import qualified Streamly.Data.Fold as Fold
-import qualified Streamly.Data.Stream.Prelude as Stream
+import qualified Streamly.Data.Parser as Parser
+import qualified Streamly.Data.ParserK as ParserK
+import qualified Streamly.Data.Stream as Stream
+import qualified Streamly.Data.StreamK as StreamK
 import qualified Streamly.FileSystem.Handle as Handle
 import qualified Streamly.Unicode.Stream as Unicode
+import qualified Streamly.Internal.Data.Array as Array
+import qualified Streamly.Internal.Data.Binary.Parser as Parser
+
+import Stat
 
 --------------------------------------------------------------------------------
 -- Utils
@@ -44,30 +53,9 @@ double = fromIntegral
 -- Types
 --------------------------------------------------------------------------------
 
-data Boundary a b
-    = Start a b
-    | Record a b
-    | Restart a b
-    | End a b
-    deriving (Read, Show, Ord, Eq)
-
-getNameSpace :: Boundary NameSpace PointId -> NameSpace
-getNameSpace (Start a _) = a
-getNameSpace (Record a _) = a
-getNameSpace (Restart a _) = a
-getNameSpace (End a _) = a
-
-data Counter
-    = ThreadCpuTime
-    | ProcessCpuTime
-    | WallClockTime
-    | Allocated
-    | SchedOut
-    deriving (Read, Show, Ord, Eq)
-
 type NameSpace = String
 type ModuleName = String
-type LineNum = Int
+type LineNum = Int32
 type PointId = (ModuleName, LineNum)
 type ThreadId = Int32
 type Tag = String
@@ -81,10 +69,6 @@ data EventId =
         }
     deriving (Eq, Ord, Show)
 
-data UnboundedEvent
-    = UEvent (Boundary NameSpace PointId) ThreadId Counter Value
-    deriving (Show)
-
 data Event
     = Event EventId Value
     deriving (Show)
@@ -121,101 +105,41 @@ stats =
 -- Parsing Input
 --------------------------------------------------------------------------------
 
--- Event format:
--- Start,<tag>,<tid>,<counterName>,<value>
--- Record,<tag>,<tid>,<counterName>,<value>
--- Restart,<tag>,<tid>,<counterName>,<value>
--- End,<tag>,<tid>,<counterName,<value>
-
--- Tag format:
--- NameSpace[ModuleName:LineNumber]
-
-errorString :: String -> String -> String
-errorString line reason = [str|Error:
-Line: #{line}
-Reason: #{reason}
-|]
-
-fIntegral :: Integral a => Fold IO Char a
-fIntegral =
-    Fold.foldlM' combine (pure 0)
-    where
-    combine b a =
-        case ord a - 48 of
-            x ->
-                if x >= 0 || x <= 9
-                then pure $ fromIntegral x + b * 10
-                else throwIO $ ErrorCall "fIntegral: NaN"
-
-fEventBoundary ::
-    Fold IO Char (NameSpace -> PointId -> Boundary NameSpace PointId)
-fEventBoundary =
-    Fold.rmapM f Fold.toList
-    where
-    f "Start" = pure Start
-    f "Record" = pure Record
-    f "Restart" = pure Restart
-    f "End" = pure End
-    f _ = throwIO $ ErrorCall "fEventBoundary: undefined"
-
-fCounterName :: Monad m => Fold m Char Counter
-fCounterName = read <$> Fold.toList
-
-fTag :: Fold IO Char (NameSpace, PointId)
-fTag =
-    (\a b c -> (a, (b, c)))
-        <$> Fold.takeEndBy_ (== '[') Fold.toList
-        <*> Fold.takeEndBy_ (== ':') Fold.toList
-        <*> Fold.takeEndBy_ (== ']') fIntegral
-
-fUnboundedEvent :: Fold IO Char UnboundedEvent
-fUnboundedEvent =
-    f
-        <$> (Fold.takeEndBy_ (== ',') fEventBoundary)
-        <*> (fTag <* Fold.one) -- fTag is a terminating fold
-        <*> (Fold.takeEndBy_ (== ',') fIntegral)
-        <*> (Fold.takeEndBy_ (== ',') fCounterName)
-        <*> fIntegral
-
-    where
-
-    f a b c d e = UEvent (a (fst b) (snd b)) c d e
-
-parseLineToEvent :: String -> IO (Either String UnboundedEvent)
-parseLineToEvent line =
-    catch
-        (Right <$> Stream.fold fUnboundedEvent (Stream.fromList line))
-        (\(e :: SomeException) ->
-             pure (Left (errorString line (displayException e))))
+-- Use ParserK here?
+-- There are only 1 inner bind, ParserD should work fine.
+metricParser :: Parser.Parser Word8 IO Metric
+metricParser = do
+    size64 <- Parser.int64le
+    let size = fromIntegral size64 - 8
+    fmap Array.deserialize $ Parser.takeEQ size (Array.unsafeCreateOf size)
 
+-- parseMany can be implemented in a recursive manner using parserK
 parseInputToEventStream
-    :: Stream IO (Array Word8) -> Stream IO UnboundedEvent
+    :: Stream IO (Array Word8) -> Stream IO Metric
 parseInputToEventStream inp =
-    Unicode.decodeUtf8Chunks inp
-        & Stream.foldMany
-              (Fold.takeEndBy_
-                   (== '\n')
-                   (Fold.rmapM parseLineToEvent Fold.toList))
-        & Stream.catRights
+    fmap f $ Stream.parseMany metricParser $ Array.concat inp
+    where
+    f (Left err) = error $ show err
+    f (Right v) = v
 
 --------------------------------------------------------------------------------
 -- Processing stats
 --------------------------------------------------------------------------------
 
-boundEvents :: Monad m => Fold m UnboundedEvent (Maybe Event)
+boundEvents :: Monad m => Fold m Metric (Maybe Event)
 boundEvents = Fold step initial extract extract
     where
     initial = pure $ Partial (Nothing, Map.empty)
 
     alterFunc
-        :: UnboundedEvent
+        :: Metric
         -> Maybe [(PointId, Value)]
         -> (Maybe Event, Maybe [(PointId, Value)])
-    alterFunc (UEvent (Start _ point) _ _ val) Nothing =
-        (Nothing, Just [(point, val)])
-    alterFunc (UEvent (Start _ point) _ _ val) (Just xs) =
-        (Nothing, Just ((point, val):xs))
-    alterFunc (UEvent (End ns (md, ln)) tid counter val) (Just stk) =
+    alterFunc (Metric _ _ m l _ Start val) Nothing =
+        (Nothing, Just [((m, l), val)])
+    alterFunc (Metric _ _ m l _ Start val) (Just xs) =
+        (Nothing, Just (((m, l), val):xs))
+    alterFunc (Metric tid ns md ln counter End val) (Just stk) =
         case uncons stk of
             Just (((md1, ln1), prevVal), stk1) ->
                 let lnStr = show ln
@@ -225,17 +149,17 @@ boundEvents = Fold step initial extract extract
                     , Just stk1
                     )
             Nothing -> error "boundEvents: Empty stack"
-    alterFunc (UEvent (Restart ns point@(md, ln)) tid counter val) (Just stk) =
+    alterFunc (Metric tid ns md ln counter Restart val) (Just stk) =
         case uncons stk of
             Just (((md1, ln1), prevVal), stk1) ->
                 let lnStr = show ln
                     ln1Str = show ln1
                     win = [str|#{ns}[#{md1}:#{ln1Str}-#{md}:#{lnStr}]|]
                  in ( Just (Event (EventId tid counter win) (val - prevVal))
-                    , Just ((point ,val):stk1)
+                    , Just (((md, ln) ,val):stk1)
                     )
             Nothing -> error "boundEvents: Empty stack"
-    alterFunc (UEvent (Record ns (md, ln)) tid counter val) (Just stk) =
+    alterFunc (Metric tid ns md ln counter Record val) (Just stk) =
         case uncons stk of
             Just (((md1, ln1), prevVal), _) ->
                 let lnStr = show ln
@@ -247,9 +171,9 @@ boundEvents = Fold step initial extract extract
             Nothing -> error "boundEvents: Empty stack"
     alterFunc _ Nothing = (Nothing, Nothing)
 
-    step (_, mp) uev@(UEvent b tid counter _) =
+    step (_, mp) uev@(Metric tid ns _ _ counter _ _) =
         pure $ Partial
-             $ Map.alterF (alterFunc uev) (getNameSpace b, tid, counter) mp
+             $ Map.alterF (alterFunc uev) (ns, tid, counter) mp
 
     extract (ev, _) = pure ev
 

From af2ca75aff268d30bda795e665aac3401fb95f3a Mon Sep 17 00:00:00 2001
From: Adithya Obilisetty <adithya.obilisetty@juspay.in>
Date: Tue, 6 Feb 2024 02:26:39 +0530
Subject: [PATCH 08/10] Use a global file handle to write results into

---
 lib/Stat.hs | 41 ++++++++++++++++++++++++++++++-----------
 1 file changed, 30 insertions(+), 11 deletions(-)

diff --git a/lib/Stat.hs b/lib/Stat.hs
index e61b84f..678bb64 100644
--- a/lib/Stat.hs
+++ b/lib/Stat.hs
@@ -26,7 +26,8 @@ import GHC.IO(IO(..))
 import GHC.IO.Unsafe (unsafePerformIO)
 import GHC.Int(Int64(..), Int32(..))
 import System.Environment (lookupEnv)
-import System.IO (stdout)
+import System.IO
+    (openFile, IOMode(..), Handle, BufferMode(..), hSetBuffering, hClose)
 -- import Streamly.Unicode.String (str)
 import System.CPUTime (getCPUTime)
 import Data.Time.Clock (getCurrentTime, UTCTime(..), diffUTCTime)
@@ -94,6 +95,23 @@ $(MBA.deriveSerialize [d|instance MBA.Serialize Metric|])
 tenPow9 :: Int64
 tenPow9 = 1000000000
 
+--------------------------------------------------------------------------------
+-- Perf handle
+--------------------------------------------------------------------------------
+
+-- Perf handle
+-- Can we rely on the RTS to close the handle?
+perfHandle :: Handle
+perfHandle =
+    unsafePerformIO $ do
+        h <- openFile "perf.bin" WriteMode
+        hSetBuffering h NoBuffering
+        pure h
+
+closePerfHandle :: IO ()
+closePerfHandle = hClose perfHandle
+
+
 --------------------------------------------------------------------------------
 -- Window Type
 --------------------------------------------------------------------------------
@@ -173,18 +191,19 @@ getThreadStat = do
         , fromIntegral switches
         )
 
-printMetricList :: [Metric] -> IO ()
-printMetricList mList = do
+printMetricList :: Handle -> [Metric] -> IO ()
+printMetricList handle mList = do
     arr <-
         Array.fromChunksK
             $ StreamK.mapM sizedSerialize
             $ StreamK.fromStream
             $ Stream.fromList mList
-    putChunk stdout arr
+    putChunk handle arr
 
 
-eventGeneric :: (forall b. IO b -> m b) -> String -> EvLoc -> SrcLoc -> m ()
-eventGeneric liftio namespace evLoc srcLoc = liftio $ do
+eventGeneric ::
+    (forall b. IO b -> m b) -> String -> EvLoc -> SrcLoc -> Handle -> m ()
+eventGeneric liftio namespace evLoc srcLoc handle = liftio $ do
     (a, b, c, d) <- getThreadStat
     let modName = loc_module srcLoc
     let lnNum = (fromIntegral :: Int -> Int32) $ fst (loc_start srcLoc)
@@ -204,7 +223,7 @@ eventGeneric liftio namespace evLoc srcLoc = liftio $ do
     then printMetricList mList
     else pure ()
 -}
-    printMetricList mList
+    printMetricList handle mList
 
 withEvLoc :: Q Exp -> Q Exp
 withEvLoc f = do
@@ -214,19 +233,19 @@ withEvLoc f = do
 start :: Q Exp
 start = do
     Loc a b c d e <- location
-    [|eventGeneric id "g" Start (Loc a b c d e)|]
+    [|eventGeneric id "g" Start (Loc a b c d e) perfHandle|]
 
 end :: Q Exp
 end = do
     Loc a b c d e <- location
-    [|eventGeneric id "g" End (Loc a b c d e)|]
+    [|eventGeneric id "g" End (Loc a b c d e) perfHandle|]
 
 record :: Q Exp
 record = do
     Loc a b c d e <- location
-    [|eventGeneric id "g" Record (Loc a b c d e)|]
+    [|eventGeneric id "g" Record (Loc a b c d e) perfHandle|]
 
 restart :: Q Exp
 restart = do
     Loc a b c d e <- location
-    [|eventGeneric id "g" Restart (Loc a b c d e)|]
+    [|eventGeneric id "g" Restart (Loc a b c d e) perfHandle|]

From b6d9730e98c256c900727bda14e3d3eca6284fd4 Mon Sep 17 00:00:00 2001
From: Adithya Obilisetty <adithya.obilisetty@juspay.in>
Date: Tue, 6 Feb 2024 02:29:40 +0530
Subject: [PATCH 09/10] Make namespace of type Int32

---
 lib/Stat.hs                | 12 ++++++------
 stat-collector-src/Main.hs |  9 ++++++---
 2 files changed, 12 insertions(+), 9 deletions(-)

diff --git a/lib/Stat.hs b/lib/Stat.hs
index 678bb64..bbc4309 100644
--- a/lib/Stat.hs
+++ b/lib/Stat.hs
@@ -81,7 +81,7 @@ $(MBA.deriveSerialize [d|instance MBA.Serialize Counter|])
 data Metric =
     Metric
         { m_tid :: Int32
-        , m_namespace :: String
+        , m_namespace :: Int32
         , m_modName :: String
         , m_lineNum :: Int32
         , m_counter :: Counter
@@ -202,7 +202,7 @@ printMetricList handle mList = do
 
 
 eventGeneric ::
-    (forall b. IO b -> m b) -> String -> EvLoc -> SrcLoc -> Handle -> m ()
+    (forall b. IO b -> m b) -> Int32 -> EvLoc -> SrcLoc -> Handle -> m ()
 eventGeneric liftio namespace evLoc srcLoc handle = liftio $ do
     (a, b, c, d) <- getThreadStat
     let modName = loc_module srcLoc
@@ -233,19 +233,19 @@ withEvLoc f = do
 start :: Q Exp
 start = do
     Loc a b c d e <- location
-    [|eventGeneric id "g" Start (Loc a b c d e) perfHandle|]
+    [|eventGeneric id 0 Start (Loc a b c d e) perfHandle|]
 
 end :: Q Exp
 end = do
     Loc a b c d e <- location
-    [|eventGeneric id "g" End (Loc a b c d e) perfHandle|]
+    [|eventGeneric id 0 End (Loc a b c d e) perfHandle|]
 
 record :: Q Exp
 record = do
     Loc a b c d e <- location
-    [|eventGeneric id "g" Record (Loc a b c d e) perfHandle|]
+    [|eventGeneric id 0 Record (Loc a b c d e) perfHandle|]
 
 restart :: Q Exp
 restart = do
     Loc a b c d e <- location
-    [|eventGeneric id "g" Restart (Loc a b c d e) perfHandle|]
+    [|eventGeneric id 0 Restart (Loc a b c d e) perfHandle|]
diff --git a/stat-collector-src/Main.hs b/stat-collector-src/Main.hs
index 6d38e42..39188aa 100644
--- a/stat-collector-src/Main.hs
+++ b/stat-collector-src/Main.hs
@@ -144,7 +144,8 @@ boundEvents = Fold step initial extract extract
             Just (((md1, ln1), prevVal), stk1) ->
                 let lnStr = show ln
                     ln1Str = show ln1
-                    win = [str|#{ns}[#{md1}:#{ln1Str}-#{md}:#{lnStr}]|]
+                    nsStr = show ns
+                    win = [str|#{nsStr}[#{md1}:#{ln1Str}-#{md}:#{lnStr}]|]
                  in ( Just (Event (EventId tid counter win) (val - prevVal))
                     , Just stk1
                     )
@@ -154,7 +155,8 @@ boundEvents = Fold step initial extract extract
             Just (((md1, ln1), prevVal), stk1) ->
                 let lnStr = show ln
                     ln1Str = show ln1
-                    win = [str|#{ns}[#{md1}:#{ln1Str}-#{md}:#{lnStr}]|]
+                    nsStr = show ns
+                    win = [str|#{nsStr}[#{md1}:#{ln1Str}-#{md}:#{lnStr}]|]
                  in ( Just (Event (EventId tid counter win) (val - prevVal))
                     , Just (((md, ln) ,val):stk1)
                     )
@@ -164,7 +166,8 @@ boundEvents = Fold step initial extract extract
             Just (((md1, ln1), prevVal), _) ->
                 let lnStr = show ln
                     ln1Str = show ln1
-                    win = [str|#{ns}[#{md1}:#{ln1Str}-#{md}:#{lnStr}]|]
+                    nsStr = show ns
+                    win = [str|#{nsStr}[#{md1}:#{ln1Str}-#{md}:#{lnStr}]|]
                  in ( Just (Event (EventId tid counter win) (val - prevVal))
                     , Just stk
                     )

From 32965bc432daf8538e2b52cbbb540bd28396404a Mon Sep 17 00:00:00 2001
From: Adithya Obilisetty <adithya.obilisetty@juspay.in>
Date: Tue, 6 Feb 2024 02:34:57 +0530
Subject: [PATCH 10/10] Remove redundant imports and control CPP via cabal
 flags

---
 haskell-perf.cabal         | 22 +++++++++++++++++++++-
 lib/Stat.hs                |  3 ---
 stat-collector-src/Main.hs |  9 ---------
 3 files changed, 21 insertions(+), 13 deletions(-)

diff --git a/haskell-perf.cabal b/haskell-perf.cabal
index 3e3908a..0d9e3f7 100644
--- a/haskell-perf.cabal
+++ b/haskell-perf.cabal
@@ -34,6 +34,11 @@ flag dev
   manual: True
   default: False
 
+flag prim-op
+  description: Prim-op is available
+  manual: True
+  default: False
+
 flag fusion-plugin
   description: Use fusion plugin for benchmarks and executables
   manual: True
@@ -104,6 +109,8 @@ common compile-options
       ghc-options:  -Wmissed-specialisations
                     -Wall-missed-specialisations
                     -fno-ignore-asserts
+    if flag(prim-op)
+      cpp-options:    -DPRIM_OP_AVAILABLE
 
 library
   import: compile-options
@@ -137,7 +144,20 @@ executable stat-collector
   build-depends:
        base
      , streamly-core
-     , streamly
+     , containers
+     , format-numbers
+     , text
+     , haskell-perf
+
+test-suite stat-test
+  import: compile-options
+  type:   exitcode-stdio-1.0
+  hs-source-dirs: test
+  main-is: Main.hs
+  ghc-options: -O2 -fmax-worker-args=16 -fspec-constr-recursive=16
+  build-depends:
+       base
+     , streamly-core
      , containers
      , format-numbers
      , text
diff --git a/lib/Stat.hs b/lib/Stat.hs
index bbc4309..3807313 100644
--- a/lib/Stat.hs
+++ b/lib/Stat.hs
@@ -42,9 +42,6 @@ import qualified Streamly.Internal.Data.MutByteArray as MBA
 import Prelude
 import Language.Haskell.TH
 
-#define PRIM_OP_AVAILABLE
--- #undef PRIM_OP_AVAILABLE
-
 #ifdef PRIM_OP_AVAILABLE
 import GHC.Exts (threadCPUTime#)
 #else
diff --git a/stat-collector-src/Main.hs b/stat-collector-src/Main.hs
index 39188aa..b8a8672 100644
--- a/stat-collector-src/Main.hs
+++ b/stat-collector-src/Main.hs
@@ -12,7 +12,6 @@ import Data.Int (Int32, Int64)
 import System.Environment (getArgs)
 import Data.Function ((&))
 import Data.List (foldl', uncons)
-import Data.Char (ord)
 import Data.Map (Map)
 import Data.Maybe (fromJust)
 import Data.Word (Word8)
@@ -20,23 +19,16 @@ import Streamly.Data.Array (Array)
 import Streamly.Data.Fold (Fold)
 import Streamly.Data.Stream (Stream)
 import Streamly.Internal.Data.Fold (Fold(..), Step(..))
-import Streamly.Data.ParserK (ParserK)
 import Streamly.Unicode.String (str)
 import System.IO (stdin)
 import Data.Text.Format.Numbers (prettyI)
-import Control.Exception
-    (catch, SomeException, displayException, throwIO, ErrorCall(..))
-import qualified Streamly.Internal.Data.StreamK as StreamK (unfoldrM)
 
 import qualified Data.Text as Text
 import qualified Data.Map as Map
 import qualified Streamly.Data.Fold as Fold
 import qualified Streamly.Data.Parser as Parser
-import qualified Streamly.Data.ParserK as ParserK
 import qualified Streamly.Data.Stream as Stream
-import qualified Streamly.Data.StreamK as StreamK
 import qualified Streamly.FileSystem.Handle as Handle
-import qualified Streamly.Unicode.Stream as Unicode
 import qualified Streamly.Internal.Data.Array as Array
 import qualified Streamly.Internal.Data.Binary.Parser as Parser
 
@@ -53,7 +45,6 @@ double = fromIntegral
 -- Types
 --------------------------------------------------------------------------------
 
-type NameSpace = String
 type ModuleName = String
 type LineNum = Int32
 type PointId = (ModuleName, LineNum)