eventlog-live 0.4.0.0 → 0.5.0.0

raw patch · 28 files changed

+2830/−925 lines, 28 filesdep +asyncdep +clockdep +co-log-coredep −unliftio-corePVP ok

version bump matches the API change (PVP)

Dependencies added: async, clock, co-log-core, concurrent-machines, containers, ghc-stack-profiler-core, lifted-async, monad-control, stm, time, transformers, transformers-base, vector

Dependencies removed: unliftio-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +32/−0
• eventlog-live.cabal +91/−56
• src/GHC/Eventlog/Live/Data/Attribute.hs +15/−1
• src/GHC/Eventlog/Live/Data/LogRecord.hs +29/−0
• src/GHC/Eventlog/Live/Data/Metric.hs +88/−2
• src/GHC/Eventlog/Live/Data/Severity.hs +84/−0
• src/GHC/Eventlog/Live/Logger.hs +196/−63
• src/GHC/Eventlog/Live/Machine/Analysis/Capability.hs +44/−40
• src/GHC/Eventlog/Live/Machine/Analysis/Heap.hs +56/−40
• src/GHC/Eventlog/Live/Machine/Analysis/Log.hs +77/−0
• src/GHC/Eventlog/Live/Machine/Analysis/Profile.hs +274/−0
• src/GHC/Eventlog/Live/Machine/Analysis/Thread.hs +22/−17
• src/GHC/Eventlog/Live/Machine/Core.hs +571/−271
• src/GHC/Eventlog/Live/Machine/Decoder.hs +17/−13
• src/GHC/Eventlog/Live/Machine/Source.hs +39/−93
• src/GHC/Eventlog/Live/Options.hs +46/−76
• src/GHC/Eventlog/Live/Socket.hs +0/−177
• src/GHC/Eventlog/Live/Source.hs +207/−0
• src/GHC/Eventlog/Live/Source/Core.hs +114/−0
• src/GHC/Eventlog/Live/Verbosity.hs +0/−76
• vendor/concurrent-machines-0.3.1.5/src/Data/Machine/Concurrent.hs +125/−0
• vendor/concurrent-machines-0.3.1.5/src/Data/Machine/Concurrent/AsyncStep.hs +63/−0
• vendor/concurrent-machines-0.3.1.5/src/Data/Machine/Concurrent/Buffer.hs +138/−0
• vendor/concurrent-machines-0.3.1.5/src/Data/Machine/Concurrent/Fanout.hs +87/−0
• vendor/concurrent-machines-0.3.1.5/src/Data/Machine/Concurrent/Scatter.hs +239/−0
• vendor/concurrent-machines-0.3.1.5/src/Data/Machine/Concurrent/Tee.hs +37/−0
• vendor/concurrent-machines-0.3.1.5/src/Data/Machine/Concurrent/Wye.hs +115/−0
• vendor/concurrent-machines-0.3.1.5/src/Data/Machine/Regulated.hs +24/−0

@@ -1,3 +1,35 @@+### 0.5.0.0++- Overhaul documentation for `GHC.Eventlog.Live.Machine.Core`.+- **BREAKING**: Drop `counterBy` and `counterByTick`.+- **BREAKING**: Drop `batchListToTick` and `batchListToTicks`.+- **BREAKING**: Drop `batchToTick` and `batchToTicks`.+- **BREAKING**: Drop `aggregateByTick` and `aggregateByTicks`.+- **BREAKING**: Drop `liftBatch`.+- **BREAKING**: Rename `sortByBatchTick` to `sortByTick`.+- **BREAKING**: Change `batchByTick` and `batchByTicks` to preserve ticks.+- **BREAKING**: Generalise `sortByBatch`, `sortByTick`, and `validateOrder` to work on arbitrary keys.+- **BREAKING**: Rename `between` to `betweenFirst`.+- **BREAKING**: Generalise `betweenFirst` and `delimit` to work on arbitrary items.+- Add `betweenEach`.+- **BREAKING**: Change `liftRouter` to ignore inputs after the child process stops.+- Add `fanoutTick`.+- **BREAKING**: Add implicit `TickInfo` to each `Tick` via `HasTickInfo`.+- **BREAKING**: Change `onlyTick` to yield actual `Tick` values.+- **BREAKING**: Change interval argument for `runWithEventlogSource` and `sourceHandleBatch` to batch in milliseconds.+- **BREAKING**: Add `AttrBoot` to `AttrValue`.+- Add support for user messages and markers.+- **BREAKING**: Replace logging with `co-log-core` contravariant `LogAction`.+- Add support for concurrent `fanoutTickCC` and `mergeWithTickCC`.+- **BREAKING**: Rename `EventlogSource` to `EventlogSourceOptions`.+- **BREAKING**: Rename `GHC.Eventlog.Live.Socket` to `GHC.Eventlog.Live.Source`.+- Add `GHC.Eventlog.Live.Source.Core`.+- **BREAKING**: Move `EventlogSourceOptions` to `GHC.Eventlog.Live.Source.Core`.+- Add `EventlogSourceHandle`.+- **BREAKING**: Rename `runWithEventlogSource` to `runWithEventlogSourceOptions`.+- Add `withEventlogSourceHandle` and `runWithEventlogSourceHandle` to `GHC.Eventlog.Live.Source`.+- **BREAKING**: Rename `sourceHandleBatch` to `eventlogSourceTick`.+ ### 0.4.0.0  - Add parser for `--stats` flag (`statsParser`).

@@ -1,26 +1,10 @@-cabal-version:   3.0-name:            eventlog-live-version:         0.4.0.0-synopsis:        Live processing of eventlog data.+cabal-version: 3.0+name: eventlog-live+version: 0.5.0.0+synopsis: Live processing of eventlog data. description:-  This package supports live processing of eventlog data.-  It consists of three libraries:--  *   The @eventlog-live@ library defines-      [machines](https://hackage.haskell.org/package/machines)-      for processing eventlog data.-  *   The @eventlog-live:options@ sub-library defines-      [optparse-applicative](https://hackage.haskell.org/package/optparse-applicative)-      parsers for common command-line arguments.-  *   The @eventlog-live:socket@ sub-library defines functions to read the-      eventlog from a Unix socket, which depend on the-      [network](https://hackage.haskell.org/package/network) package.--  This package is primarily intended for use via the-  [eventlog-live-influxdb](https://hackage.haskell.org/package/eventlog-live-influxdb)-  and-  [eventlog-live-otelcol](https://hackage.haskell.org/package/eventlog-live-otelcol)-  packages.+  This package provides [machines](https://hackage.haskell.org/package/machines) for processing live eventlog data.+  It is primarily intended for use via the [eventlog-live-otelcol](https://hackage.haskell.org/package/eventlog-live-otelcol) package.    [⚠️ Warning]:       This package is experimental.@@ -30,34 +14,47 @@    For more information, see [the README](https://github.com/well-typed/eventlog-live#readme). -license:         BSD-3-Clause-license-file:    LICENSE-author:          Wen Kokke-maintainer:      wen@well-typed.com-copyright:       (c) 2021-2025 Well-Typed-build-type:      Simple-category:        Debug, Monitoring, System+license: BSD-3-Clause+license-file: LICENSE+author: Wen Kokke+maintainer: wen@well-typed.com+copyright: (c) 2021-2025 Well-Typed+build-type: Simple+category: Debug, Monitoring, System extra-doc-files: CHANGELOG.md tested-with:-  GHC ==9.2.8-   || ==9.4.8-   || ==9.6.7-   || ==9.8.4-   || ==9.10.2-   || ==9.12.2+  ghc ==9.2.8 || ==9.4.8 || ==9.6.7 || ==9.8.4 || ==9.10.2 || ==9.12.2  source-repository head-  type:     git+  type: git   location: https://github.com/well-typed/eventlog-live.git-  subdir:   eventlog-live+  subdir: eventlog-live +source-repository this+  type: git+  location: https://github.com/well-typed/eventlog-live.git+  tag: eventlog-live-0.5.0.0+  subdir: eventlog-live++-- 2025-12-09:+-- This flag should enable switching between concurrent-machines and the+-- vendored copy of the packages, as it currently does not easily build.+flag use-concurrent-machines+  description: Use concurrent-machines in place of the vendored copy+  default: False+  manual: False+ common language   ghc-options:-    -Wall -Wcompat -Widentities -Wprepositive-qualified-module-    -Wredundant-constraints -Wunticked-promoted-constructors+    -Wall+    -Wcompat+    -Widentities+    -Wprepositive-qualified-module+    -Wredundant-constraints+    -Wunticked-promoted-constructors     -Wunused-packages -  default-language:   Haskell2010+  default-language: Haskell2010   default-extensions:     BangPatterns     ConstraintKinds@@ -82,25 +79,32 @@     NoFieldSelectors     NumericUnderscores     OverloadedRecordDot+    PatternSynonyms     RankNTypes     RecordWildCards     ScopedTypeVariables     StandaloneDeriving+    StandaloneKindSignatures     TupleSections     TypeApplications     TypeFamilies+    ViewPatterns  library-  import:          language-  hs-source-dirs:  src+  import: language+  hs-source-dirs: src   exposed-modules:     GHC.Eventlog.Live.Data.Attribute     GHC.Eventlog.Live.Data.Group+    GHC.Eventlog.Live.Data.LogRecord     GHC.Eventlog.Live.Data.Metric+    GHC.Eventlog.Live.Data.Severity     GHC.Eventlog.Live.Data.Span     GHC.Eventlog.Live.Logger     GHC.Eventlog.Live.Machine.Analysis.Capability     GHC.Eventlog.Live.Machine.Analysis.Heap+    GHC.Eventlog.Live.Machine.Analysis.Log+    GHC.Eventlog.Live.Machine.Analysis.Profile     GHC.Eventlog.Live.Machine.Analysis.Thread     GHC.Eventlog.Live.Machine.Core     GHC.Eventlog.Live.Machine.Decoder@@ -108,19 +112,50 @@     GHC.Eventlog.Live.Machine.Source     GHC.Eventlog.Live.Machine.WithStartTime     GHC.Eventlog.Live.Options-    GHC.Eventlog.Live.Socket-    GHC.Eventlog.Live.Verbosity+    GHC.Eventlog.Live.Source+    GHC.Eventlog.Live.Source.Core    build-depends:-    , ansi-terminal         >=1.1    && <1.2-    , base                  >=4.16   && <4.22-    , bytestring            >=0.11   && <0.13-    , dlist                 >=1.0    && <1.1-    , ghc-events            >=0.20   && <0.21-    , hashable              >=1.4    && <1.6-    , machines              >=0.7.4  && <0.8-    , network               >=3.2.7  && <3.3-    , optparse-applicative  >=0.17   && <0.20-    , text                  >=1.2    && <2.2-    , unliftio-core         >=0.2.1  && <0.3-    , unordered-containers  >=0.2.20 && <0.3+    ansi-terminal >=1.1 && <1.2,+    base >=4.16 && <4.22,+    bytestring >=0.11 && <0.13,+    clock >=0.8 && <0.9,+    co-log-core >=0.3 && <0.4,+    dlist >=1.0 && <1.1,+    ghc-events >=0.20 && <0.21,+    ghc-stack-profiler-core >=0.2 && <0.3,+    hashable >=1.4 && <1.6,+    machines >=0.7.4 && <0.8,+    monad-control >=1.0 && <1.1,+    network >=3.2.7 && <3.3,+    optparse-applicative >=0.17 && <0.20,+    stm >=2.5 && <2.6,+    text >=1.2 && <2.2,+    transformers >=0.2 && <0.7,+    unordered-containers >=0.2.20 && <0.3,+    vector >=0.11 && <0.14,++  -- 2025-12-09:+  -- This configures the build requirements for the vendored copy of+  -- the concurrent-machines package version 0.3.1.5.+  if flag(use-concurrent-machines)+    build-depends: concurrent-machines >=0.1 && <0.4+  else+    ghc-options: -Wno-prepositive-qualified-module+    hs-source-dirs: vendor/concurrent-machines-0.3.1.5/src+    other-modules:+      Data.Machine.Concurrent+      Data.Machine.Concurrent.AsyncStep+      Data.Machine.Concurrent.Buffer+      Data.Machine.Concurrent.Fanout+      Data.Machine.Concurrent.Scatter+      Data.Machine.Concurrent.Tee+      Data.Machine.Concurrent.Wye+      Data.Machine.Regulated++    build-depends:+      async >=2.0.1 && <2.3,+      containers >=0.5 && <0.8,+      lifted-async >=0.10 && <0.12,+      time >=1.4 && <1.16,+      transformers-base >=0.4 && <0.5,

@@ -6,6 +6,7 @@ -} module GHC.Eventlog.Live.Data.Attribute (   Attrs,+  lookup,   toList,   Attr,   AttrKey,@@ -23,6 +24,7 @@ import Data.Word (Word16, Word32, Word64, Word8) import GHC.Generics (Generic) import GHC.IsList (IsList (..))+import Prelude hiding (lookup)  {- | A set of attributes is a t`HashMap`@@ -30,12 +32,18 @@ newtype Attrs = Attrs {attrMap :: HashMap AttrKey AttrValue}   deriving (Eq, Generic, Show) +lookup :: AttrKey -> Attrs -> Maybe AttrValue+lookup attrKey attrs = M.lookup attrKey attrs.attrMap+ instance Hashable Attrs  instance Semigroup Attrs where   (<>) :: Attrs -> Attrs -> Attrs   x <> y = Attrs{attrMap = x.attrMap <> y.attrMap} +instance Monoid Attrs where+  mempty = Attrs mempty+ instance IsList Attrs where   type Item Attrs = Attr @@ -73,7 +81,8 @@ The type of attribute values. -} data AttrValue-  = AttrInt !Int+  = AttrBool !Bool+  | AttrInt !Int   | AttrInt8 !Int8   | AttrInt16 !Int16   | AttrInt32 !Int32@@ -99,6 +108,11 @@ instance IsAttrValue AttrValue where   toAttrValue :: AttrValue -> AttrValue   toAttrValue = id+  {-# INLINE toAttrValue #-}++instance IsAttrValue Bool where+  toAttrValue :: Bool -> AttrValue+  toAttrValue = AttrBool   {-# INLINE toAttrValue #-}  instance IsAttrValue Int where

@@ -0,0 +1,29 @@+{- |+Module      : GHC.Eventlog.Live.LogRecord+Description : Representation for metrics.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Data.LogRecord (+  LogRecord (..),+) where++import Data.Text (Text)+import GHC.Eventlog.Live.Data.Attribute (Attrs)+import GHC.Eventlog.Live.Data.Severity (Severity)+import GHC.RTS.Events (Timestamp)++{- |+LogRecords combine a timestamp, message and a severity.+-}+data LogRecord = LogRecord+  { body :: !Text+  -- ^ The log message.+  , maybeTimeUnixNano :: !(Maybe Timestamp)+  -- ^ The time at which the log was created.+  , maybeSeverity :: !(Maybe Severity)+  -- ^ The severity of the log.+  , attrs :: Attrs+  -- ^ A set of attributes.+  }+  deriving (Show)

@@ -6,23 +6,32 @@ -} module GHC.Eventlog.Live.Data.Metric (   Metric (..),++  -- * Existential wrapper+  SomeMetric (..),+  SMetricType (..),+  KnownMetricType (..), ) where  import Control.Exception (assert)+import Data.Int (Int16, Int32, Int64, Int8)+import Data.Kind (Type)+import Data.Proxy (Proxy)+import Data.Word (Word16, Word32, Word64, Word8) import GHC.Eventlog.Live.Data.Attribute (Attrs) import GHC.Eventlog.Live.Data.Group (GroupBy (..)) import GHC.RTS.Events (Timestamp)  {- | Metrics combine a measurement with a timestamp representing the time of the-measurement, a timestamp representing the earliest possible measurment, and+measurement, a timestamp representing the earliest possible measurement, and a list of attributes. -} data Metric a = Metric   { value :: !a   -- ^ The measurement.   , maybeTimeUnixNano :: !(Maybe Timestamp)-  -- ^ The time at which the measurment was taken.+  -- ^ The time at which the measurement was taken.   , maybeStartTimeUnixNano :: !(Maybe Timestamp)   -- ^ The earliest time at which any measurement could have been taken.   --   Usually, this represents the start time of a process.@@ -46,3 +55,80 @@         , maybeStartTimeUnixNano = x.maybeStartTimeUnixNano `min` y.maybeStartTimeUnixNano         , attrs = x.attrs         }++--------------------------------------------------------------------------------+-- Existential wrapper for Metrics+--------------------------------------------------------------------------------++data SomeMetric+  = forall metricType.+  (KnownMetricType metricType) =>+  SomeMetric+  { metricName :: String+  , metric :: Metric metricType+  }++data SMetricType (a :: Type) where+  SMetricTypeFloat :: SMetricType Float+  SMetricTypeDouble :: SMetricType Double+  SMetricTypeWord :: SMetricType Word+  SMetricTypeWord8 :: SMetricType Word8+  SMetricTypeWord16 :: SMetricType Word16+  SMetricTypeWord32 :: SMetricType Word32+  SMetricTypeWord64 :: SMetricType Word64+  SMetricTypeInt :: SMetricType Int+  SMetricTypeInt8 :: SMetricType Int8+  SMetricTypeInt16 :: SMetricType Int16+  SMetricTypeInt32 :: SMetricType Int32+  SMetricTypeInt64 :: SMetricType Int64++class (Num a) => KnownMetricType a where+  metricTypeSing :: Proxy a -> SMetricType a++instance KnownMetricType Float where+  metricTypeSing :: Proxy Float -> SMetricType Float+  metricTypeSing _proxy = SMetricTypeFloat++instance KnownMetricType Double where+  metricTypeSing :: Proxy Double -> SMetricType Double+  metricTypeSing _proxy = SMetricTypeDouble++instance KnownMetricType Word where+  metricTypeSing :: Proxy Word -> SMetricType Word+  metricTypeSing _proxy = SMetricTypeWord++instance KnownMetricType Word8 where+  metricTypeSing :: Proxy Word8 -> SMetricType Word8+  metricTypeSing _proxy = SMetricTypeWord8++instance KnownMetricType Word16 where+  metricTypeSing :: Proxy Word16 -> SMetricType Word16+  metricTypeSing _proxy = SMetricTypeWord16++instance KnownMetricType Word32 where+  metricTypeSing :: Proxy Word32 -> SMetricType Word32+  metricTypeSing _proxy = SMetricTypeWord32++instance KnownMetricType Word64 where+  metricTypeSing :: Proxy Word64 -> SMetricType Word64+  metricTypeSing _proxy = SMetricTypeWord64++instance KnownMetricType Int where+  metricTypeSing :: Proxy Int -> SMetricType Int+  metricTypeSing _proxy = SMetricTypeInt++instance KnownMetricType Int8 where+  metricTypeSing :: Proxy Int8 -> SMetricType Int8+  metricTypeSing _proxy = SMetricTypeInt8++instance KnownMetricType Int16 where+  metricTypeSing :: Proxy Int16 -> SMetricType Int16+  metricTypeSing _proxy = SMetricTypeInt16++instance KnownMetricType Int32 where+  metricTypeSing :: Proxy Int32 -> SMetricType Int32+  metricTypeSing _proxy = SMetricTypeInt32++instance KnownMetricType Int64 where+  metricTypeSing :: Proxy Int64 -> SMetricType Int64+  metricTypeSing _proxy = SMetricTypeInt64

@@ -0,0 +1,84 @@+{- |+Module      : GHC.Eventlog.Live.Severity+Description : Representation for metrics.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Data.Severity (+  Severity (..),+  SeverityNumber (..),+  toSeverityNumber,+  fromSeverityNumber,+  toSeverityString,+  fromSeverityString,+) where++import Data.Char (toUpper)+import Data.Ix (Ix)+import Text.Read (readMaybe)++{- |+The severity number as specified by the OpenTelemetry specification.++See: https://opentelemetry.io/docs/specs/otel/logs/data-model/#field-severitynumber+-}+newtype SeverityNumber = SeverityNumber {value :: Int}++{- |+The severity as specified by the OpenTelemetry specification.++See: https://opentelemetry.io/docs/specs/otel/logs/data-model/#displaying-severity+-}+data Severity+  = TRACE+  | TRACE2+  | TRACE3+  | TRACE4+  | DEBUG+  | DEBUG2+  | DEBUG3+  | DEBUG4+  | INFO+  | INFO2+  | INFO3+  | INFO4+  | WARN+  | WARN2+  | WARN3+  | WARN4+  | ERROR+  | ERROR2+  | ERROR3+  | ERROR4+  | FATAL+  | FATAL2+  | FATAL3+  | FATAL4+  deriving (Bounded, Enum, Eq, Ord, Read, Show, Ix)++{- |+Convert from a `Severity` to a `SeverityNumber`.+-}+toSeverityNumber :: Severity -> SeverityNumber+toSeverityNumber = SeverityNumber . (+ 1) . fromEnum++{- |+Convert from a `SeverityNumber` to a `Severity`.+-}+fromSeverityNumber :: SeverityNumber -> Maybe Severity+fromSeverityNumber severityNumber+  | 1 <= severityNumber.value && severityNumber.value <= 24 =+      Just (toEnum $ severityNumber.value - 1)+  | otherwise = Nothing++{- |+Convert from a `Severity` to a `String`.+-}+toSeverityString :: Severity -> String+toSeverityString = show++{- |+Convert from a `String` to a `Severity`.+-}+fromSeverityString :: String -> Maybe Severity+fromSeverityString = readMaybe . fmap toUpper

@@ -1,3 +1,4 @@+{-# LANGUAGE OverloadedLists #-} {-# LANGUAGE OverloadedStrings #-}  {- |@@ -7,111 +8,243 @@ Portability : portable -} module GHC.Eventlog.Live.Logger (-  logError,-  logWarning,-  logInfo,-  logDebug,+  Logger,+  MyTelemetryData (..),+  writeLog,+  writeException,+  writeMetric,+  filterBySeverity,+  stderrLogger,+  handleLogger,+  chanLogger,+  chanSource, ) where -import Control.Exception (bracket_)+import Colog.Core.Action (cfilter, (<&))+import Colog.Core.Action qualified as CCA (LogAction (..))+import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TChan (TChan, readTChan, writeTChan)+import Control.Exception (Exception (..), bracket_)+import Control.Monad ((<=<)) import Control.Monad.IO.Class (MonadIO (..))-import Data.List qualified as L+import Data.Ix (Ix (..))+import Data.Machine (SourceT, repeatedly, yield)+import Data.Maybe (isNothing) import Data.Text (Text) import Data.Text qualified as T import Data.Text.IO qualified as TIO-import GHC.Eventlog.Live.Verbosity (Verbosity, showVerbosity, verbosityDebug, verbosityError, verbosityInfo, verbosityWarning)-import GHC.Stack (CallStack, HasCallStack, SrcLoc (..), callStack, getCallStack)+import Data.Text.Lazy qualified as TL+import Data.Text.Lazy.Builder qualified as TLB+import GHC.Eventlog.Live.Data.Attribute (AttrValue (..), (~=))+import GHC.Eventlog.Live.Data.Attribute qualified as A+import GHC.Eventlog.Live.Data.LogRecord (LogRecord (..))+import GHC.Eventlog.Live.Data.Metric (KnownMetricType, Metric (..), SomeMetric (..))+import GHC.Eventlog.Live.Data.Severity (Severity (..), toSeverityString)+import GHC.RTS.Events (Timestamp)+import GHC.Stack (callStack, prettyCallStack, withFrozenCallStack)+import GHC.Stack.Types (HasCallStack)+import System.Clock (Clock (..), TimeSpec (..), getTime) import System.Console.ANSI (Color (..), ColorIntensity (..), ConsoleLayer (..), SGR (..), hNowSupportsANSI, hSetSGR) import System.IO qualified as IO+import Prelude hiding (log) +type Logger m = CCA.LogAction m MyTelemetryData+ {- |-Log messages to given handle.-Only prints a message if its verbosity level is above the verbosity threshold.+The type of internal telemetry data. -}-logMessage :: (MonadIO m) => IO.Handle -> CallStack -> Verbosity -> Verbosity -> Text -> m ()-logMessage handle theCallStack verbosityLevel verbosityThreshold msg-  | verbosityLevel >= verbosityThreshold = liftIO $ do-      withVerbosityColor verbosityLevel handle-        . flip TIO.hPutStrLn-        . formatMessage verbosityLevel verbosityThreshold theCallStack-        $ msg-      IO.hFlush handle-  | otherwise = pure ()+data MyTelemetryData+  = MyTelemetryData'LogRecord {logRecord :: !LogRecord}+  | MyTelemetryData'Metric {metric :: !SomeMetric}  {- |-Internal helper.-Format the `CallStack`.+Use a `Logger` to log a message with a severity. -}-formatCallStack :: CallStack -> Text-formatCallStack theCallStack =-  maybe T.empty (formatSrcLoc . snd . fst) (L.uncons (getCallStack theCallStack))+writeLog :: (HasCallStack) => Logger m -> Severity -> Text -> m ()+writeLog logger severity body =+  withFrozenCallStack $+    logger+      <& MyTelemetryData'LogRecord+        { logRecord =+            LogRecord+              { body+              , maybeSeverity = Just severity+              , maybeTimeUnixNano = Nothing+              , attrs = ["call-stack" ~= prettyCallStack callStack]+              }+        }++{- |+Use a `Logger` to log an exception.+-}+writeException :: (Exception e) => Logger m -> e -> m ()+writeException logger e =+  writeLog logger ERROR (T.pack $ displayException e)++{- |+Use a `Logger` to log an internal metric.+-}+writeMetric ::+  forall m metricType.+  (KnownMetricType metricType) =>+  Logger m ->+  -- | The metric name.+  String ->+  metricType ->+  m ()+writeMetric logger metricName value =+  logger+    <& MyTelemetryData'Metric+      SomeMetric+        { metricName+        , metric =+            Metric+              { value+              , maybeTimeUnixNano = Nothing+              , maybeStartTimeUnixNano = Nothing+              , attrs = []+              }+        }++{- |+A `Logger` that writes each `LogRecord` to a `IO.stderr` and ignores all other telemetry data.++__TODO:__ Support the remaining telemetry data.+-}+stderrLogger :: Logger IO+stderrLogger = handleLogger IO.stderr++{- |+A `Logger` that writes each `LogRecord` to a `IO.Handle` and ignores all other telemetry data.++__TODO:__ Support the remaining telemetry data.+-}+handleLogger ::+  IO.Handle ->+  Logger IO+handleLogger handle = CCA.LogAction $ \case+  MyTelemetryData'LogRecord logRecord -> liftIO $ do+    withSeverityColor logRecord.maybeSeverity handle $ \handleWithColor ->+      TIO.hPutStrLn handleWithColor $ formatLogRecord logRecord+    IO.hFlush handle+  MyTelemetryData'Metric{} -> pure ()++{- |+Filter a @`Logger` m@ by a `Severity`.+-}+filterBySeverity ::+  (Applicative m) =>+  Severity ->+  Logger m ->+  Logger m+filterBySeverity severityThreshold =+  cfilter severityFilter  where-  formatSrcLoc :: SrcLoc -> Text-  formatSrcLoc srcLoc =-    mconcat [T.pack srcLoc.srcLocFile, ":", T.pack (show srcLoc.srcLocStartLine), ":", T.pack (show srcLoc.srcLocStartCol)]+  severityFilter = \case+    MyTelemetryData'LogRecord{..} ->+      maybe False (>= severityThreshold) logRecord.maybeSeverity+    _otherwise -> True  {- | Internal helper. Format the message appropriately for the given verbosity level and threshold. -}-formatMessage :: Verbosity -> Verbosity -> CallStack -> Text -> Text-formatMessage verbosityLevel verbosityThreshold theCallStack msg-  | verbosityLevel == verbosityInfo && verbosityThreshold /= verbosityDebug = msg-  | otherwise = mconcat [showVerbosity verbosityLevel, " (", formatCallStack theCallStack, "): ", msg]+formatLogRecord :: LogRecord -> Text+formatLogRecord logRecord =+  TL.toStrict . TLB.toLazyText . mconcat $+    [ -- format the severity+      maybe "" (\severity -> "[" <> TLB.fromString (toSeverityString severity) <> "] ") logRecord.maybeSeverity+    , -- format the body+      TLB.fromText logRecord.body+    , -- format the call-stack, if any+      case A.lookup "call-stack" logRecord.attrs of+        Just (AttrText theCallStack)+          | maybe False (>= ERROR) logRecord.maybeSeverity ->+              "\n" <> TLB.fromText theCallStack+        _otherwise -> ""+    ]  {- | Internal helper.-Use a handle with the color set appropriately for the given verbosity level.+Determine the ANSI color and intensity associated with a particular `Severity`. -}-withVerbosityColor :: Verbosity -> IO.Handle -> (IO.Handle -> IO a) -> IO a-withVerbosityColor verbosity handle action = do+severityColor :: Severity -> Maybe (Color, ColorIntensity)+severityColor severity+  | inRange (TRACE, TRACE4) severity = Just (Blue, Dull)+  | inRange (DEBUG, DEBUG4) severity = Just (Blue, Vivid)+  | inRange (WARN, WARN4) severity = Just (Yellow, Vivid)+  | inRange (ERROR, ERROR4) severity = Just (Red, Dull)+  | inRange (FATAL, FATAL4) severity = Just (Red, Vivid)+  | otherwise = Nothing++{- |+Internal helper.+Use a handle with the color set appropriately for the given `Severity`.+-}+withSeverityColor :: Maybe Severity -> IO.Handle -> (IO.Handle -> IO a) -> IO a+withSeverityColor maybeSeverity handle action = do   supportsANSI <- hNowSupportsANSI handle   if not supportsANSI     then       action handle-    else case verbosityColor verbosity of+    else case severityColor =<< maybeSeverity of       Nothing ->         action handle-      Just color -> do-        let setVerbosityColor = hSetSGR handle [SetColor Foreground Vivid color]+      Just (color, intensity) -> do+        let setVerbosityColor = hSetSGR handle [SetColor Foreground intensity color]         let setDefaultColor = hSetSGR handle [SetDefaultColor Foreground]         bracket_ setVerbosityColor setDefaultColor $ action handle  {- |-Internal helper.-Determine the ANSI color associated with a particular verbosity level.+A `Logger` that writes the internal telemetry data to a channel. -}-verbosityColor :: Verbosity -> Maybe Color-verbosityColor verbosity-  | verbosity == verbosityError = Just Red-  | verbosity == verbosityWarning = Just Yellow-  | verbosity == verbosityDebug = Just Blue-  | otherwise = Nothing+chanLogger :: TChan MyTelemetryData -> Logger IO+chanLogger chan =+  CCA.LogAction $+    atomically . writeTChan chan <=< addTimeUnixNano  {- |-Log errors to `IO.stderr`.+A `Souce` that reads the data from a channel. -}-logError :: (HasCallStack, MonadIO m) => Verbosity -> Text -> m ()-logError = logMessage IO.stderr callStack verbosityError+chanSource :: (MonadIO m) => TChan a -> SourceT m a+chanSource chan = repeatedly $ do+  a <- liftIO $ atomically $ readTChan chan+  yield a  {- |-Log warnings to `IO.stderr`.+Add the current Unix timestamp in nanoseconds to telemetry data. -}-logWarning :: (HasCallStack, MonadIO m) => Verbosity -> Text -> m ()-logWarning = logMessage IO.stderr callStack verbosityWarning+addTimeUnixNano :: MyTelemetryData -> IO MyTelemetryData+addTimeUnixNano myTelemetryData =+  case myTelemetryData of+    MyTelemetryData'LogRecord{logRecord = LogRecord{..}}+      | isNothing maybeTimeUnixNano -> do+          timeUnixNano <- getTimeUnixNano+          pure $+            MyTelemetryData'LogRecord+              LogRecord{maybeTimeUnixNano = Just timeUnixNano, ..}+      | otherwise -> pure myTelemetryData+    MyTelemetryData'Metric{metric = SomeMetric{metricName, metric = Metric{..}}}+      | isNothing maybeTimeUnixNano -> do+          timeUnixNano <- getTimeUnixNano+          pure $+            MyTelemetryData'Metric+              SomeMetric+                { metricName+                , metric = Metric{maybeTimeUnixNano = Just timeUnixNano, ..}+                }+      | otherwise -> pure myTelemetryData  {- |-Log info messages to `IO.stderr`.--}-logInfo :: (HasCallStack, MonadIO m) => Verbosity -> Text -> m ()-logInfo verbosityThreshold = logMessage handle callStack verbosityInfo verbosityThreshold- where-  handle-    | verbosityThreshold <= verbosityDebug = IO.stderr-    | otherwise = IO.stdout+Get the current Unix time in nanoseconds. -{- |-Log debug messages to `IO.stderr`.+__Warning:__ This will start overflowing in the year 2554. -}-logDebug :: (HasCallStack, MonadIO m) => Verbosity -> Text -> m ()-logDebug = logMessage IO.stderr callStack verbosityDebug+getTimeUnixNano :: IO Timestamp+getTimeUnixNano = toNanos <$> getTime Realtime+ where+  -- NOTE: This will overflow if @t.sec > (2^64 - 1) `div` 10^9@,+  --       which means you're running this code in the year 2554.+  --       What's that like?+  toNanos :: TimeSpec -> Timestamp+  toNanos t = 1_000_000_000 * fromIntegral t.sec + fromIntegral t.nsec

@@ -36,7 +36,7 @@ ) where  import Control.Monad (when)-import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Trans.Class (MonadTrans (..)) import Data.Char (isSpace) import Data.Foldable (for_) import Data.Machine (Is (..), PlanT, ProcessT, asParts, await, construct, mapping, repeatedly, yield, (~>))@@ -46,12 +46,12 @@ import Data.Void (Void) import GHC.Eventlog.Live.Data.Attribute (AttrValue, IsAttrValue (..), (~=)) import GHC.Eventlog.Live.Data.Metric (Metric (..))+import GHC.Eventlog.Live.Data.Severity (Severity (..)) import GHC.Eventlog.Live.Data.Span (duration)-import GHC.Eventlog.Live.Logger (logWarning)+import GHC.Eventlog.Live.Logger (Logger, writeLog) import GHC.Eventlog.Live.Machine.Analysis.Thread (ThreadState (..), ThreadStateSpan (..), processThreadStateSpans') import GHC.Eventlog.Live.Machine.Core (liftRouter) import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..), setWithStartTime'value, tryGetTimeUnixNano)-import GHC.Eventlog.Live.Verbosity (Verbosity) import GHC.RTS.Events (Event (..), EventInfo, ThreadId, Timestamp) import GHC.RTS.Events qualified as E import GHC.Records (HasField (..))@@ -67,7 +67,7 @@ -} processCapabilityUsageMetrics ::   forall m.-  (MonadIO m) =>+  (Monad m) =>   ProcessT m (WithStartTime CapabilityUsageSpan) (Metric Timestamp) processCapabilityUsageMetrics =   liftRouter measure spawn@@ -172,11 +172,11 @@ -} processCapabilityUsageSpans ::   forall m.-  (MonadIO m) =>-  Verbosity ->+  (Monad m) =>+  Logger m ->   ProcessT m (WithStartTime Event) (WithStartTime CapabilityUsageSpan)-processCapabilityUsageSpans verbosity =-  processCapabilityUsageSpans' tryGetTimeUnixNano (.value) setWithStartTime'value setWithStartTime'value verbosity+processCapabilityUsageSpans logger =+  processCapabilityUsageSpans' tryGetTimeUnixNano (.value) setWithStartTime'value setWithStartTime'value logger     ~> mapping (either (fmap Left) (fmap Right))  {- |@@ -185,14 +185,14 @@ -} processCapabilityUsageSpans' ::   forall m s t1 t2.-  (MonadIO m) =>+  (Monad m) =>   (s -> Maybe Timestamp) ->   (s -> Event) ->   (s -> GCSpan -> t1) ->   (s -> MutatorSpan -> t2) ->-  Verbosity ->+  Logger m ->   ProcessT m s (Either t1 t2)-processCapabilityUsageSpans' timeUnixNano getEvent setGCSpan setMutatorSpan verbosity =+processCapabilityUsageSpans' timeUnixNano getEvent setGCSpan setMutatorSpan logger =   -- NOTE:   -- Combining this fanout with an `Either` is risky, because it   -- has the potential to lose information if both `processGCSpans`@@ -200,9 +200,9 @@   -- However, this shouldn't ever happen, since the two processors   -- process disjoint sets of events.   fanout-    [ processGCSpans' timeUnixNano getEvent setGCSpan verbosity+    [ processGCSpans' timeUnixNano getEvent setGCSpan logger         ~> mapping Left-    , processMutatorSpans' timeUnixNano getEvent setMutatorSpan verbosity+    , processMutatorSpans' timeUnixNano getEvent setMutatorSpan logger         ~> mapping Right     ] @@ -244,8 +244,8 @@ -} processGCSpans ::   forall m.-  (MonadIO m) =>-  Verbosity ->+  (Monad m) =>+  Logger m ->   ProcessT m (WithStartTime Event) (WithStartTime GCSpan) processGCSpans =   processGCSpans' tryGetTimeUnixNano (.value) setWithStartTime'value@@ -256,13 +256,13 @@ -} processGCSpans' ::   forall m s t.-  (MonadIO m) =>+  (Monad m) =>   (s -> Maybe Timestamp) ->   (s -> Event) ->   (s -> GCSpan -> t) ->-  Verbosity ->+  Logger m ->   ProcessT m s t-processGCSpans' timeUnixNano getEvent setGCSpan verbosity =+processGCSpans' timeUnixNano getEvent setGCSpan logger =   liftRouter measure spawn  where   getEventTime = (.evTime) . getEvent@@ -301,13 +301,15 @@                 go (Just j)             -- If the previous event was any other event, then...             | otherwise -> do-                -- ...emit an error, and...-                logWarning verbosity . T.pack $-                  printf-                    "Capability %d: Unsupported trace %s --> %s"-                    cap-                    (showEventInfo (getEventInfo i))-                    (showEventInfo (getEventInfo j))+                -- ...emit a warning, and...+                let msg =+                      T.pack $+                        printf+                          "Capability %d: Unsupported trace %s --> %s"+                          cap+                          (showEventInfo (getEventInfo i))+                          (showEventInfo (getEventInfo j))+                lift $ writeLog logger WARN $ msg                 -- ...continue with the previous event.                 go (Just i)           -- If there was no previous event, then...@@ -331,13 +333,15 @@                 go (Just $ minBy getEventTime i j)           -- If there was no previous event or it was any other event, then...           _otherwise -> do-            -- ...emit an error, and...-            logWarning verbosity . T.pack $-              printf-                "Capability %d: Unsupported trace %s --> %s"-                cap-                (maybe "?" (showEventInfo . getEventInfo) mi)-                (showEventInfo (getEventInfo j))+            -- ...emit a warning, and...+            let msg =+                  T.pack $+                    printf+                      "Capability %d: Unsupported trace %s --> %s"+                      cap+                      (maybe "?" (showEventInfo . getEventInfo) mi)+                      (showEventInfo (getEventInfo j))+            lift $ writeLog logger WARN $ msg             -- ...continue with the previous event.             go mi         -- If the next event is any other event, ignore it.@@ -393,8 +397,8 @@ -} processMutatorSpans ::   forall m.-  (MonadIO m) =>-  Verbosity ->+  (Monad m) =>+  Logger m ->   ProcessT m (WithStartTime Event) (WithStartTime MutatorSpan) processMutatorSpans =   processMutatorSpans' tryGetTimeUnixNano (.value) setWithStartTime'value@@ -405,14 +409,14 @@ -} processMutatorSpans' ::   forall m s t.-  (MonadIO m) =>+  (Monad m) =>   (s -> Maybe Timestamp) ->   (s -> Event) ->   (s -> MutatorSpan -> t) ->-  Verbosity ->+  Logger m ->   ProcessT m s t-processMutatorSpans' timeUnixNano getEvent setMutatorSpan verbosity =-  processThreadStateSpans' timeUnixNano getEvent setThreadStateSpan verbosity ~> asParts+processMutatorSpans' timeUnixNano getEvent setMutatorSpan logger =+  processThreadStateSpans' timeUnixNano getEvent setThreadStateSpan logger ~> asParts  where   setThreadStateSpan :: s -> ThreadStateSpan -> Maybe t   setThreadStateSpan s threadStateSpan =@@ -423,7 +427,7 @@ -} asMutatorSpans ::   forall m.-  (MonadIO m) =>+  (Monad m) =>   ProcessT m ThreadStateSpan MutatorSpan asMutatorSpans = asMutatorSpans' id (const id) @@ -433,7 +437,7 @@ -} asMutatorSpans' ::   forall m s t.-  (MonadIO m) =>+  (Monad m) =>   (s -> ThreadStateSpan) ->   (s -> MutatorSpan -> t) ->   ProcessT m s t

@@ -23,10 +23,16 @@   -- ** Heap Profile Breakdown   heapProfBreakdownEitherReader,   heapProfBreakdownShow,++  -- ** Things fendor doesn't want to reimplement+  InfoTable (..),+  InfoTablePtr (..),+  HeapProfBreakdown,+  metric, ) where  import Control.Monad (unless, when)-import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Trans.Class (MonadTrans (..)) import Data.Either (isLeft) import Data.Foldable (for_) import Data.HashMap.Strict (HashMap)@@ -41,9 +47,9 @@ import GHC.Eventlog.Live.Data.Attribute (Attrs, (~=)) import GHC.Eventlog.Live.Data.Group (GroupBy (..)) import GHC.Eventlog.Live.Data.Metric (Metric (..))-import GHC.Eventlog.Live.Logger (logWarning)+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (Logger, writeLog) import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..), tryGetTimeUnixNano)-import GHC.Eventlog.Live.Verbosity (Verbosity) import GHC.RTS.Events (Event (..), HeapProfBreakdown (..)) import GHC.RTS.Events qualified as E import Numeric (showHex)@@ -195,18 +201,19 @@ -} insertHeapProfSampleString ::   forall m.-  (MonadIO m) =>-  Verbosity ->+  (Monad m) =>+  Logger m ->   Text ->   Metric Word64 ->   HeapProfSampleData ->   m HeapProfSampleData-insertHeapProfSampleString verbosityThreshold heapProfLabel heapProfSample heapProfSamples = do+insertHeapProfSampleString logger heapProfLabel heapProfSample heapProfSamples = do   let insert :: Maybe (Metric Word64) -> m (Maybe (Metric Word64))       insert heapProfSample' = do-        when (isJust heapProfSample') $-          logWarning verbosityThreshold $-            "Duplicate HeapProfSampleString for " <> heapProfLabel <> " within the same garbage collection pass."+        when (isJust heapProfSample') $ do+          let msg = "Duplicate HeapProfSampleString for " <> heapProfLabel <> " within the same garbage collection pass."+          writeLog logger WARN $ msg+         pure (Just heapProfSample)   heapProfSampleMap' <- M.alterF insert heapProfLabel heapProfSamples.heapProfSampleMap   pure HeapProfSampleData{heapProfSampleMap = heapProfSampleMap'}@@ -240,7 +247,7 @@   , infoTableModule :: !Text   , infoTableSrcLoc :: !Text   }-  deriving (Show)+  deriving (Show, Eq, Ord)  {- | Internal helper.@@ -284,11 +291,11 @@ and `E.HeapProfSampleEnd` events to maintain an era stack. -} processHeapProfSampleData ::-  (MonadIO m) =>-  Verbosity ->+  (Monad m) =>+  Logger m ->   Maybe HeapProfBreakdown ->   ProcessT m (WithStartTime Event) HeapProfSampleData-processHeapProfSampleData verbosityThreshold maybeHeapProfBreakdown =+processHeapProfSampleData logger maybeHeapProfBreakdown =   construct $     go       HeapProfSampleState@@ -332,10 +339,11 @@       E.HeapProfSampleBegin{..} -> do         -- Check that maybeHeapProfSampleData is Nothing.         for_ st.maybeHeapProfSampleData $ \heapProfSampleData -> do-          logWarning-            verbosityThreshold-            "Unexpected event HeapProfSampleBegin while previous garbage collection pass was left open.\n\-            \This may indicate that the eventlog is not properly ordered or that its semantics have changed."+          let msg =+                "Unexpected event HeapProfSampleBegin while previous garbage collection pass was left open.\n\+                \This may indicate that the eventlog is not properly ordered or that its semantics have changed."+          lift $ writeLog logger WARN $ msg+           -- Yield the previous sample data anyway.           yield heapProfSampleData         -- Start a new garbage collection pass.@@ -352,18 +360,22 @@         heapProfSampleEraStack' <-           case L.uncons heapProfSampleEraStack of             Nothing -> do-              logWarning verbosityThreshold . T.pack $-                printf-                  "Eventlog closed era %d, but there is no current era."-                  heapProfSampleEra+              let msg =+                    T.pack $+                      printf+                        "Eventlog closed era %d, but there is no current era."+                        heapProfSampleEra+              lift $ writeLog logger WARN $ msg               pure heapProfSampleEraStack             Just (currentEra, heapProfSampleEraStack') -> do-              unless (currentEra == heapProfSampleEra) $-                logWarning verbosityThreshold . T.pack $-                  printf-                    "Eventlog closed era %d, but the current era is era %d."-                    heapProfSampleEra-                    currentEra+              unless (currentEra == heapProfSampleEra) $ do+                let msg =+                      T.pack $+                        printf+                          "Eventlog closed era %d, but the current era is era %d."+                          heapProfSampleEra+                          currentEra+                lift $ writeLog logger WARN $ msg               pure heapProfSampleEraStack'         go           st@@ -374,18 +386,21 @@       E.HeapProfSampleString{..}         -- If there is no heap profile breakdown, issue a warning, then disable warnings.         | Left True <- eitherShouldWarnOrHeapProfBreakdown -> do-            logWarning verbosityThreshold $-              "Cannot infer heap profile breakdown.\n\-              \         If your binary was compiled with a GHC version prior to 9.14,\n\-              \         you must also pass the heap profile type to this executable.\n\-              \         See: https://gitlab.haskell.org/ghc/ghc/-/commit/76d392a"+            let msg =+                  "Cannot infer heap profile breakdown.\n\+                  \         If your binary was compiled with a GHC version prior to 9.14,\n\+                  \         you must also pass the heap profile type to this executable.\n\+                  \         See: https://gitlab.haskell.org/ghc/ghc/-/commit/76d392a"+            lift $ writeLog logger WARN $ msg             go st{eitherShouldWarnOrHeapProfBreakdown = Left False, infoTableMap = mempty}         -- If the heap profile breakdown is biographical, issue a warning, then disable warnings.         | Right HeapProfBreakdownBiography <- eitherShouldWarnOrHeapProfBreakdown -> do-            logWarning verbosityThreshold . T.pack $-              printf-                "Unsupported heap profile breakdown %s"-                (heapProfBreakdownShow HeapProfBreakdownBiography)+            let msg =+                  T.pack $+                    printf+                      "Unsupported heap profile breakdown %s"+                      (heapProfBreakdownShow HeapProfBreakdownBiography)+            lift $ writeLog logger WARN $ msg             go st{eitherShouldWarnOrHeapProfBreakdown = Left False, infoTableMap = mempty}         -- If there is a heap profile breakdown, handle it appropriately.         | Right heapProfBreakdown <- eitherShouldWarnOrHeapProfBreakdown -> do@@ -399,9 +414,10 @@             heapProfSampleData <-               case st.maybeHeapProfSampleData of                 Nothing -> do-                  logWarning verbosityThreshold $-                    "Unexpected event HeapProfSampleString out of scope of HeapProfSampleBegin and HeapProfSampleEnd.\n\-                    \This may indicate that the eventlog is not properly ordered or that its semantics have changed."+                  let msg =+                        "Unexpected event HeapProfSampleString out of scope of HeapProfSampleBegin and HeapProfSampleEnd.\n\+                        \This may indicate that the eventlog is not properly ordered or that its semantics have changed."+                  lift $ writeLog logger WARN $ msg                   pure mempty                 Just heapProfSampleData ->                   pure heapProfSampleData@@ -421,7 +437,7 @@                     , "infoTableSrcLoc" ~= fmap (.infoTableSrcLoc) maybeInfoTable                     ]             heapProfSampleData' <--              insertHeapProfSampleString verbosityThreshold heapProfLabel heapProfSample heapProfSampleData+              lift $ insertHeapProfSampleString logger heapProfLabel heapProfSample heapProfSampleData             -- Continue with the updated HeapProfSampleState             go               st

@@ -0,0 +1,77 @@+{-# LANGUAGE OverloadedLists #-}+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live.Machine.Analysis.Log+Description : Machines for processing eventlog data.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Machine.Analysis.Log where++import Data.Machine (Process, await, repeatedly, yield)+import Data.Text (Text)+import GHC.Eventlog.Live.Data.Attribute (Attrs, (~=))+import GHC.Eventlog.Live.Data.LogRecord (LogRecord (..))+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..), tryGetTimeUnixNano)+import GHC.RTS.Events (Event)+import GHC.RTS.Events qualified as E++--------------------------------------------------------------------------------+-- UserMessage++{- |+This machine processes `E.UserMessage` events into logs.+-}+processUserMessageData :: Process (WithStartTime Event) LogRecord+processUserMessageData =+  repeatedly $+    await >>= \case+      i+        | E.UserMessage{..} <- i.value.evSpec ->+            yield $+              logRecord i msg (Just DEBUG) $+                [ "evCap" ~= i.value.evCap+                , "kind" ~= ("UserMessage" :: Text)+                ]+        | otherwise -> pure ()++--------------------------------------------------------------------------------+-- UserMarker++{- |+This machine processes `E.UserMarker` events into logs.+-}+processUserMarkerData :: Process (WithStartTime Event) LogRecord+processUserMarkerData =+  repeatedly $+    await >>= \case+      i+        | E.UserMarker{..} <- i.value.evSpec ->+            yield $+              logRecord i markername (Just TRACE) $+                [ "evCap" ~= i.value.evCap+                , "kind" ~= ("UserMarker" :: Text)+                ]+        | otherwise -> pure ()++{- |+Internal helper.+Construct a t`LogRecord` from an event with a start time, a message, and any+set of attributes. This is a smart constructor that pulls the timestamps out+of the event.+-}+logRecord ::+  WithStartTime Event ->+  Text ->+  Maybe Severity ->+  Attrs ->+  LogRecord+logRecord i body maybeSeverity attrs =+  LogRecord+    { body = body+    , maybeTimeUnixNano = tryGetTimeUnixNano i+    , maybeSeverity = maybeSeverity+    , attrs = attrs+    }

@@ -0,0 +1,274 @@+module GHC.Eventlog.Live.Machine.Analysis.Profile (+  StackProfSampleData (..),+  ThreadId (..),+  CapabilityId (..),+  CallStackData (..),+  StackItemData (..),+  CostCentreId (..),+  CostCentre (..),+  processStackProfSampleData,+  processCostCentreProfSampleData,+  stackProfSamples,+)+where++import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Trans.Class (MonadTrans (..))+import Data.ByteString.Lazy qualified as LBS+import Data.Foldable (for_)+import Data.HashMap.Strict (HashMap)+import Data.HashMap.Strict qualified as HashMap+import Data.HashMap.Strict qualified as M+import Data.Hashable qualified as Hashable+import Data.List qualified as List+import Data.List.NonEmpty (NonEmpty ((:|)))+import Data.List.NonEmpty qualified as NonEmpty+import Data.Machine (ProcessT, await, construct, yield)+import Data.Maybe (mapMaybe)+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Vector.Unboxed qualified as UVector+import Data.Word+import GHC.Eventlog.Live.Data.Metric+import GHC.Eventlog.Live.Logger (Logger, writeException)+import GHC.Eventlog.Live.Machine.Analysis.Heap (InfoTable (..), InfoTablePtr (..), metric)+import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..))+import GHC.Generics (Generic)+import GHC.RTS.Events (Event (..))+import GHC.RTS.Events qualified as E+import GHC.Stack.Profiler.Core.Eventlog qualified as SPCE+import GHC.Stack.Profiler.Core.SymbolTable qualified as SPCS+import GHC.Stack.Profiler.Core.ThreadSample qualified as SPCT++data StackProfSampleState = StackProfSampleState+  { infoTableMap :: !(HashMap InfoTablePtr InfoTable)+  , -- TODO: this should probably be a maybe?+    -- We could report when interleaved messages are present+    stackProfSampleChunk :: ![SPCE.BinaryCallStackMessage]+  , stackProfSymbolTableReader :: !SPCS.IntMapTable+  , maybeStackProfSampleData :: !(Maybe StackProfSampleData)+  }+  deriving (Generic)++newtype CostCentreProfSampleState = CostCentreProfSampleState+  { costCentreMap :: HashMap CostCentreId CostCentre+  }+  deriving (Generic)++newtype StackProfSampleData = StackProfSampleData+  { stackProfSample :: Metric CallStackData+  }+  deriving (Show, Generic)++newtype ThreadId = ThreadId+  { value :: Word64+  }+  deriving (Show, Eq, Ord, Generic)++newtype CapabilityId = CapabilityId+  { value :: Word64+  }+  deriving (Show, Eq, Ord, Generic)++data CallStackData = CallStackData+  { threadId :: !(Maybe ThreadId)+  , capabilityId :: !CapabilityId+  , stack :: [StackItemData]+  }+  deriving (Show, Eq, Ord, Generic)++newtype CostCentreId = CostCentreId+  { id :: Word64+  }+  deriving (Show, Eq, Ord, Generic)+  deriving newtype (Hashable.Hashable)++data CostCentre = CostCentre+  { costCentreId :: !CostCentreId+  , costCentreLabel :: !Text+  , costCentreModule :: !Text+  , costCentreSrcLoc :: !Text+  -- , heapProfFlags :: !HeapProfFlags+  }+  deriving (Show, Eq, Ord, Generic)++data StackItemData+  = IpeData !InfoTable+  | UserMessageData !Text+  | SourceLocationData !SPCT.SourceLocation+  | CostCentreData !CostCentre+  deriving (Show, Eq, Ord, Generic)++shouldTrackInfoTableMap :: Bool+shouldTrackInfoTableMap = True++shouldTrackCostCentreMap :: Bool+shouldTrackCostCentreMap = True++-- ----------------------------------------------------------------------------+-- `cost centre stack` processor+-- ----------------------------------------------------------------------------++{- |+This machine processes `E.UserBinaryMessage` events into metrics.+Furthermore, it processes the `E.InfoTableProv` events to+-}+processCostCentreProfSampleData ::+  (MonadIO m) =>+  Logger m ->+  ProcessT m (WithStartTime Event) StackProfSampleData+processCostCentreProfSampleData _logger =+  construct $+    go+      CostCentreProfSampleState+        { costCentreMap = mempty+        }+ where+  go st = do+    await >>= \i -> case i.value.evSpec of+      -- Announces an info table entry.+      E.HeapProfCostCentre{..}+        | shouldTrackCostCentreMap -> do+            let costCentreId = CostCentreId $ fromIntegral heapProfCostCentreId+                costCentre =+                  CostCentre+                    { costCentreId = costCentreId+                    , costCentreLabel = heapProfLabel+                    , costCentreModule = heapProfModule+                    , costCentreSrcLoc = heapProfSrcLoc+                    }+            go st{costCentreMap = M.insert costCentreId costCentre st.costCentreMap}+      E.ProfSampleCostCentre{..} -> do+        let lookupCostCentreStackById :: Word32 -> Maybe StackItemData+            lookupCostCentreStackById costCentreId32 =+              CostCentreData <$> HashMap.lookup (CostCentreId $ fromIntegral costCentreId32) st.costCentreMap++            callStackMessage =+              CallStackData+                { threadId = Nothing+                , capabilityId = CapabilityId $ fromIntegral profCap+                , stack =+                    -- TODO: log if we are encountering unknown cost centre ids+                    mapMaybe lookupCostCentreStackById (UVector.toList profCcsStack)+                }++            stackProfSample =+              metric i callStackMessage mempty++        yield $ StackProfSampleData stackProfSample+        go st+      _otherwise -> go st++-- ----------------------------------------------------------------------------+-- `ghc-stack-profiler` processor+-- ----------------------------------------------------------------------------++{- |+This machine processes `E.UserBinaryMessage` events into metrics.+Furthermore, it processes the `E.InfoTableProv` events to+-}+processStackProfSampleData ::+  (MonadIO m) =>+  Logger m ->+  ProcessT m (WithStartTime Event) StackProfSampleData+processStackProfSampleData logger =+  construct $+    go+      StackProfSampleState+        { infoTableMap = mempty+        , stackProfSampleChunk = mempty+        , stackProfSymbolTableReader = SPCS.emptyIntMapTable+        , maybeStackProfSampleData = Nothing+        }+ where+  go st = do+    await >>= \i -> case i.value.evSpec of+      -- Announces an info table entry.+      E.InfoTableProv{..}+        | shouldTrackInfoTableMap -> do+            let infoTablePtr = InfoTablePtr itInfo+                infoTable =+                  InfoTable+                    { infoTablePtr = infoTablePtr+                    , infoTableName = itTableName+                    , infoTableClosureDesc = itClosureDesc+                    , infoTableTyDesc = itTyDesc+                    , infoTableLabel = itLabel+                    , infoTableModule = itModule+                    , infoTableSrcLoc = itSrcLoc+                    }+            go st{infoTableMap = M.insert infoTablePtr infoTable st.infoTableMap}+      E.UserBinaryMessage{payload} ->+        case SPCT.deserializeEventlogMessage $ LBS.fromStrict payload of+          Left _err ->+            go st+          Right evMsg -> case evMsg of+            SPCE.CallStackFinal msg -> do+              let (callStackMessage, st', callStackDecodeErrors) = hydrateBinaryEventlog st msg+              for_ callStackDecodeErrors (lift . writeException logger)+              let stackProfSample = metric i callStackMessage mempty+              yield $ StackProfSampleData stackProfSample+              go st'+            SPCE.CallStackChunk msg ->+              go st{stackProfSampleChunk = msg : st.stackProfSampleChunk}+            SPCE.StringDef msg ->+              go st{stackProfSymbolTableReader = SPCS.insertTextMessage msg st.stackProfSymbolTableReader}+            SPCE.SourceLocationDef msg -> do+              let old = st.stackProfSymbolTableReader+              let errOrnew = SPCS.insertSourceLocationMessage msg old+              new <- either (\err -> lift $ writeException logger err >> pure old) pure errOrnew+              go st{stackProfSymbolTableReader = new}+      _otherwise -> go st++hydrateBinaryEventlog ::+  StackProfSampleState ->+  SPCE.BinaryCallStackMessage ->+  (CallStackData, StackProfSampleState, [SPCT.BinaryCallStackDecodeError])+hydrateBinaryEventlog spst msg = (callStackData, spst{stackProfSampleChunk = []}, callStackDecodeErrors)+ where+  chunks = spst.stackProfSampleChunk+  -- Why reverse?+  -- When decoding the stack, we walk the stack from the top down.+  -- Afterwards, the stack is chunked to fit into a single eventlog line,+  -- and the chunks are written in ascending order to the eventlog.+  -- When we pick up these messages one after another, they are prepended to+  -- 'stackProfSampleChunk', thus we are essentially storing the chunks in reverse+  -- order, as the first chunk we encounter is the top of the stack, etc...+  --+  -- Concrete example, assuming a stack @[1,2,3,4,5,6]@ and chunk size of 2:+  --+  -- 1. Chunk it: @[1,2] [3,4] [5,6]@+  -- 2. Write it to the eventlog in this order, so the messages are:+  --    [1,2]+  --    [3,4]+  --    [5,6]+  -- 3. When reading the eventlog, we store prepend later messages, resulting in:+  --    [5,6] [3,4] [1,2]+  -- 4. One reverse later: @[1,2] [3,4] [5,6]@+  -- 5. Now we can finally concat the stack frame chunks.+  orderedChunks = NonEmpty.reverse $ msg :| chunks+  fullBinaryCallStackMessage = SPCT.catCallStackMessage orderedChunks+  (callStackMessage, callStackDecodeErrors) =+    SPCT.hydrateEventlogCallStackMessage+      (SPCS.mkIntMapSymbolTableReader spst.stackProfSymbolTableReader)+      fullBinaryCallStackMessage+  callStackData =+    CallStackData+      { threadId = Just $ ThreadId $ SPCT.callThreadId callStackMessage+      , capabilityId = CapabilityId $ SPCE.getCapabilityId $ SPCT.callCapabilityId callStackMessage+      , stack =+          -- TODO: log if we are encountering unknown ipe ids+          mapMaybe (toStackItemData spst.infoTableMap) $ SPCT.callStack callStackMessage+      }++toStackItemData :: HashMap InfoTablePtr InfoTable -> SPCT.StackItem -> Maybe StackItemData+toStackItemData tbl = \case+  SPCT.IpeId iid -> IpeData <$> HashMap.lookup (InfoTablePtr $ SPCE.getIpeId iid) tbl+  SPCT.UserMessage msg -> Just $ UserMessageData $ Text.pack msg+  SPCT.SourceLocation srcLoc -> Just $ SourceLocationData srcLoc++{- |+Get the elements of a heap profile sample collection.+-}+stackProfSamples :: StackProfSampleData -> [Metric CallStackData]+stackProfSamples = List.singleton . (.stackProfSample)

@@ -12,7 +12,7 @@    -- ** Thread Labels   ThreadLabel (..),-  processThreadLabels,+  processThreadLabelData,    -- ** Thread State Spans   ThreadState (..),@@ -24,18 +24,18 @@   processThreadStateSpans', ) where -import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Trans.Class (MonadTrans (..)) import Data.Char (isSpace) import Data.Machine (Is (..), PlanT, Process, ProcessT, await, construct, repeatedly, yield) import Data.Maybe (isNothing) import Data.Text (Text) import Data.Text qualified as T import Data.Void (Void)+import GHC.Eventlog.Live.Data.Severity (Severity (..)) import GHC.Eventlog.Live.Data.Span (duration)-import GHC.Eventlog.Live.Logger (logWarning)+import GHC.Eventlog.Live.Logger (Logger, writeLog) import GHC.Eventlog.Live.Machine.Core (liftRouter) import GHC.Eventlog.Live.Machine.WithStartTime (WithStartTime (..), tryGetTimeUnixNano)-import GHC.Eventlog.Live.Verbosity (Verbosity) import GHC.RTS.Events (Event (..), EventInfo, ThreadId, ThreadStopStatus (..), Timestamp) import GHC.RTS.Events qualified as E import Text.Printf (printf)@@ -57,8 +57,8 @@ {- | This machine processes `E.ThreadLabel` events and yields t`ThreadLabel` values. -}-processThreadLabels :: Process (WithStartTime Event) ThreadLabel-processThreadLabels = repeatedly go+processThreadLabelData :: Process (WithStartTime Event) ThreadLabel+processThreadLabelData = repeatedly go  where   go =     await >>= \i -> case i.value.evSpec of@@ -145,8 +145,8 @@ with a `E.StopThread` event with the `ThreadFinished` status. -} processThreadStateSpans ::-  (MonadIO m) =>-  Verbosity ->+  (Monad m) =>+  Logger m ->   ProcessT m (WithStartTime Event) ThreadStateSpan processThreadStateSpans =   processThreadStateSpans' tryGetTimeUnixNano (.value) (const id)@@ -157,13 +157,13 @@ -} processThreadStateSpans' ::   forall m s t.-  (MonadIO m) =>+  (Monad m) =>   (s -> Maybe Timestamp) ->   (s -> Event) ->   (s -> ThreadStateSpan -> t) ->-  Verbosity ->+  Logger m ->   ProcessT m s t-processThreadStateSpans' timeUnixNano getEvent setThreadStateSpan verbosity =+processThreadStateSpans' timeUnixNano getEvent setThreadStateSpan logger =   liftRouter measure spawn  where   getEventTime = (.evTime) . getEvent@@ -258,12 +258,17 @@           --           -- If the current event is any other event, then...           | otherwise -> do-              -- ...emit an error, and...-              logWarning verbosity . T.pack $-                printf-                  "Thread %d: Unexpected event %s"-                  thread-                  (showEventInfo (getEventInfo j))+              -- ...emit a warning, and...+              let msg =+                    T.pack $+                      printf+                        "Thread %d: Unexpected event %s\n\+                        \This happens once per running thread when connecting to a running process,\n\+                        \but should not happen multiple times per thread."+                        thread+                        (showEventInfo (getEventInfo j))+              lift $ writeLog logger WARN $ msg+               --               -- This case may trigger for any event that isn't `E.RunThread`               -- or `E.StopThread` and for any `E.StopThread` event that comes

@@ -1,3 +1,4 @@+{-# LANGUAGE ImplicitParams #-} {-# LANGUAGE OverloadedStrings #-}  {- |@@ -8,116 +9,342 @@ -} module GHC.Eventlog.Live.Machine.Core (   -- * Ticks-  Tick (..),-  batchByTick,-  batchToTick,-  batchListToTick,+  TickInfo (..),+  HasTickInfo,+  Tick (Item, Tick, TickWithInfo, tickInfo),+  fanoutTick,+  fanoutTickCC,+  mergeWithTickCC,   batchByTickList,+  batchByTicksList,+  batchByTick,+  batchByTicks,   dropTick,   onlyTick,-  aggregateByTick,   liftTick,-  liftBatch, -  -- * Debug-  counterBy,-  counterByTick,-   -- * Routers   liftRouter,    -- * Event sorting   sortByBatch,-  sortByBatchTick,+  sortByTick,+  sortByTicks,    -- * Delimiting-  between,   delimit,+  betweenEach,+  betweenFirst,    -- * Validation   validateInput,   validateOrder,+  validateTicks, ) where  import Control.Monad (when)-import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Trans.Class (MonadTrans (..))+import Control.Monad.Trans.Control (MonadBaseControl) import Data.DList qualified as D-import Data.Foldable (for_)+import Data.Foldable (Foldable (..), for_) import Data.Function (on) import Data.Functor ((<&>)) import Data.HashMap.Strict (HashMap) import Data.HashMap.Strict qualified as M import Data.Hashable (Hashable (..))+import Data.Kind (Constraint) import Data.List qualified as L-import Data.Machine (Is (..), MachineT (..), Moore (..), PlanT, Process, ProcessT, Step (..), asParts, await, construct, encased, mapping, repeatedly, starve, stopped, yield, (~>))+import Data.Machine (Is (..), MachineT (..), Moore (..), PlanT, Process, ProcessT, SourceT, Step (..), asParts, await, construct, encased, mapping, repeatedly, starve, stopped, yield, (~>))+import Data.Machine.Concurrent qualified as CC+import Data.Machine.Fanout (fanout) import Data.Maybe (fromMaybe) import Data.Semigroup (Max (..))-import Data.Text (Text) import Data.Text qualified as T-import GHC.Eventlog.Live.Logger (logDebug, logError, logWarning)-import GHC.Eventlog.Live.Verbosity (Verbosity, verbosityDebug, verbosityError, verbosityWarning)-import GHC.RTS.Events (Event (..), Timestamp)-import GHC.RTS.Events qualified as E+import Data.Void (Void)+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (Logger, writeLog) import Text.Printf (printf) +{- $setup+>>> :set -XFlexibleContexts+>>> :set -XImplicitParams+>>> :set -XImportQualifiedPost+>>> :set -XLambdaCase+>>> :set -XRankNTypes+>>> :set -XTypeApplications+>>> import Data.Functor.Identity (Identity)+>>> import Data.Machine qualified as M+>>> import Data.Machine hiding (run, runT_)+>>> import Data.Machine.Fanout (fanout)+>>> import Data.Semigroup (Sum (..))++>>> :{+run :: (HasTickInfo => M.MachineT Identity k b) -> [b]+run = let ?tickInfo = TickInfo { tick = 0 } in M.run+:}++>>> :{+runT_ :: Monad m => (HasTickInfo => MachineT m k b) -> m ()+runT_ = let ?tickInfo = TickInfo { tick = 0 } in M.runT_+:}+-}+ ------------------------------------------------------------------------------- -- Ticks -------------------------------------------------------------------------------  {- |+The type of v`Tick` information.+-}+newtype TickInfo = TickInfo+  { tick :: Word+  }++{- |+The constraint that adds information to each v`Tick`.+This should be treated as opaque.+-}+type HasTickInfo :: Constraint+type HasTickInfo = (?tickInfo :: TickInfo)++{- | The type of data on a stream of items and ticks.  The t`Tick` type is isomorphic to `Maybe` modulo strictness, but with the caveat that v`Tick` does not represent failure. -}-data Tick a = Item !a | Tick-  deriving (Eq, Functor, Foldable, Traversable, Show)+data Tick a+  = Item !a+  | (HasTickInfo) => Tick  {- |-This machine batches all items between two ticks into a list.+__Warning:__+This instance loses ticks and should only be used with `fanout` to combine+streams which pass on ticks. -}+instance (Semigroup a) => Semigroup (Tick a) where+  (<>) :: Tick a -> Tick a -> Tick a+  t@Tick <> Tick = t+  i@Item{} <> Tick = i+  Tick <> i@Item{} = i+  Item a <> Item a' = Item (a <> a')++deriving instance (Eq a) => Eq (Tick a)+deriving instance Functor Tick+deriving instance Foldable Tick+deriving instance Traversable Tick+deriving instance (Show a) => Show (Tick a)++{- |+Internal helper.+Get `TickInfo` from a t`Tick`.+-}+toTickInfo :: Tick x -> Maybe TickInfo+toTickInfo Item{} = Nothing+toTickInfo Tick = Just ?tickInfo++{- |+Internal helper.+Lift `TickInfo` to a constraint.+-}+withTickInfo :: TickInfo -> ((HasTickInfo) => a) -> a+withTickInfo tickInfo action =+  let ?tickInfo = tickInfo in action++pattern TickWithInfo :: TickInfo -> Tick a+pattern TickWithInfo{tickInfo} <- (toTickInfo -> Just tickInfo)+  where+    TickWithInfo tickInfo = withTickInfo tickInfo Tick++{-# COMPLETE Item, TickWithInfo #-}++{- |+Variant of `fanout` for processes that act on t`Tick` streams.++==== __Examples__++>>> run $ fanoutTick [echo, echo] <~ source [Item [1], Tick, Item [2]]+[Item [1,1],Tick,Item [2,2]]+-}+fanoutTick ::+  forall m a b.+  (Monad m, Semigroup b) =>+  [ProcessT m (Tick a) (Tick b)] ->+  ProcessT m (Tick a) (Tick b)+fanoutTick processes =+  fanout+    [ fanout+        [ process ~> dropTick+        | process <- processes+        ]+        ~> mapping (D.singleton . Item)+    , onlyTick+        ~> mapping D.singleton+    ]+    ~> asParts++{- |+Variant of `fanoutTick` that runs processes concurrently.+-}+fanoutTickCC ::+  forall m a b.+  (MonadBaseControl IO m, Semigroup b) =>+  [ProcessT m (Tick a) (Tick b)] ->+  ProcessT m (Tick a) (Tick b)+fanoutTickCC processes =+  fanout+    [ CC.fanout+        [ process ~> dropTick+        | process <- processes+        ]+        ~> mapping (D.singleton . Item)+    , onlyTick+        ~> mapping D.singleton+    ]+    ~> asParts++{- |+Merges a stream of ticks into an existing source.+All items are discarded.+The source is run concurrently with its input.+-}+mergeWithTickCC ::+  forall m x a.+  (MonadBaseControl IO m) =>+  SourceT m a ->+  ProcessT m (Tick x) (Tick a)+mergeWithTickCC source =+  CC.scatter [onlyTick, source ~> mapping Item]++{- |+Batches items to lists.++The process @`batchByTickList`@ consumes a stream of items and ticks.+It preserves ticks but batches items between ticks to lists.++__Warning:__ This process does not yield empty batches.++==== __Examples__++>>> run $ batchByTickList <~ source [Item 1,Item 2,Tick,Item 3,Tick,Item 4,Item 5,Tick,Item 6,Tick]+[[1,2],[3],[4,5],[6]]++>>> run $ batchByTickList <~ source [Item 1,Item 2,Tick,Tick]+[[1,2]]+-} batchByTickList :: Process (Tick a) [a] batchByTickList =   mapping (fmap D.singleton)     ~> batchByTick+    ~> dropTick     ~> mapping D.toList  {- |-Generalised version of `batchByTickList`.+Batches items for a given number of ticks to lists.++The process @`batchByTicksList` n@ consumes a stream of items and ticks.+It preserves ticks but batches items for @n@ ticks to lists and yields the batch before the @n@'th tick.++__Warning:__ This process does not yield empty batches.++==== __Examples__++>>> run $ batchByTicksList 2 <~ source [Item 1,Item 2,Tick,Item 3,Tick,Item 4,Item 5,Tick,Item 6,Tick]+[[1,2,3],[4,5,6]]++>>> run $ batchByTicksList 2 <~ source [Item 1,Item 2,Tick,Tick]+[[1,2]] -}+batchByTicksList ::+  -- | The number of ticks per batch.+  Int ->+  Process (Tick a) [a]+batchByTicksList ticks =+  mapping (fmap D.singleton)+    ~> batchByTicks ticks+    ~> dropTick+    ~> mapping D.toList++{- |+Batches items via their `Semigroup` instance.++The process @`batchByTick`@ consumes a stream of items and ticks.+It preserves ticks but batches items between ticks using `sconcat`.++==== __Examples__++>>> run $ batchByTick <~ source [Item [1],Item [2],Tick,Item [3],Tick,Item [4],Item [5],Tick,Item [6],Tick]+[Item [1,2],Tick,Item [3],Tick,Item [4,5],Tick,Item [6],Tick]++>>> run $ batchByTick <~ source [Item (Sum 1),Item (Sum 2),Tick,Item (Sum 3),Tick,Item (Sum 4),Item (Sum 5),Tick,Item (Sum 6),Tick]+[Item (Sum {getSum = 3}),Tick,Item (Sum {getSum = 3}),Tick,Item (Sum {getSum = 9}),Tick,Item (Sum {getSum = 6}),Tick]+-} batchByTick ::   forall a.-  (Monoid a) => Process (Tick a) a-batchByTick = batchByTickWith mempty+  (Monoid a) => Process (Tick a) (Tick a)+batchByTick = batchByTicks 1++{- |+Batches items for a given number of ticks via their `Semigroup` instance.++The process @`batchByTicks` n@ consumes a stream of items and ticks.+It preserves ticks but batches items for @n@ ticks using `sconcat` and yields the batch before the @n@'th tick.++==== __Examples__++>>> run $ batchByTicks 2 <~ source [Item [1],Item [2],Tick,Item [3],Tick,Item [4],Item [5],Tick,Item [6],Tick]+[Tick,Item [1,2,3],Tick,Tick,Item [4,5,6],Tick]++>>> run $ batchByTicks 2 <~ source [Item (Sum 1),Item (Sum 2),Tick,Item (Sum 3),Tick,Item (Sum 4),Item (Sum 5),Tick,Item (Sum 6),Tick]+[Tick,Item (Sum {getSum = 6}),Tick,Tick,Item (Sum {getSum = 15}),Tick]+-}+batchByTicks ::+  forall a.+  (Semigroup a) =>+  -- | The number of ticks per batch.+  Int ->+  Process (Tick a) (Tick a)+batchByTicks ticks = batchByTicksWith ticks mempty  where-  batchByTickWith ::+  batchByTicksWith ::     forall m.     (Monad m) =>-    a -> MachineT m (Is (Tick a)) a-  batchByTickWith acc = MachineT $ pure $ Await onNext Refl onStop+    Int ->+    Maybe a ->+    MachineT m (Is (Tick a)) (Tick a)+  batchByTicksWith ticksRemaining maybeAcc =+    MachineT $ pure $ Await onNext Refl onStop    where-    onNext :: Tick a -> MachineT m (Is (Tick a)) a-    onNext = \case-      Item a -> batchByTickWith (a <> acc)-      Tick -> MachineT $ pure $ Yield acc batchByTick-    onStop :: MachineT m (Is (Tick a)) a-    onStop = MachineT $ pure $ Yield acc stopped+    yieldItem :: a -> ProcessT m (Tick a) (Tick a) -> ProcessT m (Tick a) (Tick a)+    yieldItem a = MachineT . pure . Yield (Item a) -{- |-This machine streams a list of items into a series of items-separated by ticks.--}-batchListToTick :: Process [a] (Tick a)-batchListToTick = batchToTick+    yieldTick :: (HasTickInfo) => ProcessT m (Tick a) (Tick a) -> ProcessT m (Tick a) (Tick a)+    yieldTick = MachineT . pure . Yield Tick -{- |-Generalised version of `batchListToTick`.--}-batchToTick :: (Foldable f) => Process (f a) (Tick a)-batchToTick = repeatedly go- where-  go = await >>= \xs -> for_ xs (yield . Item) >> yield Tick+    onNext :: Tick a -> MachineT m (Is (Tick a)) (Tick a)+    onNext = \case+      Item a -> batchByTicksWith ticksRemaining (maybeAcc <> Just a)+      Tick+        | ticksRemaining <= 1 ->+            -- Yield an `Item` if any items were accumulated.+            maybe id yieldItem maybeAcc $+              -- Yield the `Tick`.+              yieldTick $+                -- Continue with the initial state.+                batchByTicksWith ticks Nothing+        | otherwise ->+            -- Yield the `Tick`.+            yieldTick $+              -- Continue with one fewer tick remaining.+              batchByTicksWith (ticksRemaining - 1) maybeAcc +    onStop :: MachineT m (Is (Tick a)) (Tick a)+    onStop =+      -- Yield an `Item` if any items were accumulated.+      maybe id yieldItem maybeAcc $+        -- Stop.+        stopped+ {- | This machine drops all ticks. -}@@ -131,74 +358,17 @@ {- | This machine drops all items. -}-onlyTick :: Process (Tick a) ()+onlyTick :: Process (Tick a) (Tick b) onlyTick =   repeatedly $     await >>= \case-      Tick -> yield ()+      Tick -> yield Tick       Item{} -> pure () -{- |-This machine aggregates a value by tick.--The difference between `batchByTick` and `aggregateByTick` is that-`batchByTick` yields a batch on every tick whereas-`aggregateByTick` only yields a batch if there were any values.--}-aggregateByTick :: (Semigroup a) => Process (Tick a) a-aggregateByTick =-  mapping (fmap Just)-    ~> batchByTick-    ~> asParts- ------------------------------------------------------------------------------- -- Machine combinators ------------------------------------------------------------------------------- -{- |-This machine counts the number of inputs it received,-using the given function, and logs this value.--}-counterBy ::-  forall m a x.-  (MonadIO m) =>-  Verbosity ->-  Text ->-  (a -> Word) ->-  ProcessT m a x-counterBy verbosity label counter-  | verbosityDebug >= verbosity = repeatedly go-  | otherwise = stopped- where-  go :: PlanT (Is a) x m ()-  go =-    await >>= \a ->-      logDebug verbosity (T.pack (show (counter a)) <> " " <> label)--{- |-This machine counts the number of inputs it received,-and logs this value on every tick.--}-counterByTick ::-  forall m a x.-  (MonadIO m) =>-  Verbosity ->-  Text ->-  ProcessT m (Tick a) x-counterByTick verbosity label-  | verbosityDebug >= verbosity = construct $ go 0-  | otherwise = stopped- where-  go :: Word -> PlanT (Is (Tick a)) x m ()-  go count =-    await >>= \case-      Item _ -> go (count + 1)-      Tick -> logDebug verbosity (T.pack (show count) <> " " <> label) >> go 0------------------------------------------------------------------------------------ Machine combinators--------------------------------------------------------------------------------- -------------------------------------------------------------------------------- -- Lift a machine to a machine that passes on ticks unchanged @@ -212,6 +382,14 @@   [ Tick a ]                              [ Tick b ]            └─(if Item)─( ProcessT m a b )─┘ @++==== __Examples__++>>> run $ liftTick (mapping (+1)) <~ source [Item 1,Tick,Item 2,Item 3,Tick,Tick]+[Item 2,Tick,Item 3,Item 4,Tick,Tick]++>>> run $ liftTick (scan (+) 0) <~ source [Item 1,Tick,Item 2,Item 3,Tick,Tick]+[Item 0,Item 1,Tick,Item 3,Item 6,Tick,Tick] -} liftTick ::   (Monad m) =>@@ -240,133 +418,130 @@           onStop' = liftTick onStop  ----------------------------------------------------------------------------------- Lift a machine to a machine that operates on batches--{- |-Lift a machine that processes @a@s into @b@s to a machine that processes-batches of @a@s into batches of @b@s.--}-liftBatch ::-  forall m a b.-  (Monad m) =>-  ProcessT m a b ->-  ProcessT m [a] [b]-liftBatch = MachineT . running [] []- where-  -- The parent machine is running the child machine with the current batch.-  running :: [a] -> [b] -> ProcessT m a b -> m (Step (Is [a]) [b] (ProcessT m [a] [b]))-  running as bs m =-    runMachineT m >>= \case-      Stop ->-        pure Stop-      Yield b k ->-        running as (b : bs) k-      Await (onNext :: t -> ProcessT m a b) Refl onStop ->-        pure $ Yield (reverse bs) $ MachineT $ awaiting as onNext onStop--  -- The parent machine is awaiting new input.-  awaiting :: [a] -> (a -> ProcessT m a b) -> ProcessT m a b -> m (Step (Is [a]) [b] (ProcessT m [a] [b]))-  awaiting (a : as) onNext _onStop = running as [] $ onNext a-  awaiting [] onNext onStop = pure $ Await onNext' Refl onStop'-   where-    onNext' :: [a] -> ProcessT m [a] [b]-    onNext' as = MachineT $ awaiting as onNext onStop-    onStop' :: ProcessT m [a] [b]-    onStop' = exhausting onStop--  -- The parent machine is exhausting the child machine to gather its output.-  exhausting :: ProcessT m a b -> ProcessT m x [b]-  exhausting = MachineT . go []-   where-    go :: [b] -> ProcessT m a b -> m (Step (Is x) [b] (ProcessT m x [b]))-    go bs m =-      runMachineT m >>= \case-        Stop ->-          pure Stop-        Yield b k ->-          go (b : bs) k-        Await _onNext _Refl onStop ->-          pure $ Yield (reverse bs) $ MachineT $ go [] onStop---------------------------------------------------------------------------------- -- Construct a processor that spawns a separate child processor for each measure  {- |-Spawn a copy of a machine for each "measure".+Spawns a process for each measure.  Constructs the following machine:  @-    ┌─────(if measure == k0)─( ProcessT m a b )────┐-  [ a ] ──(if measure == ..)─( ProcessT m a b )─ [ b ]-    └─────(if measure == kN)─( ProcessT m a b )────┘+    ┌─────(if measure == k0)─( spawn k0 :: ProcessT m a b )────┐+  [ a ] ──(if measure == ..)─( spawn .. :: ProcessT m a b )─ [ b ]+    └─────(if measure == kN)─( spawn kN :: ProcessT m a b )────┘ @ -__Warning:__ The router does not currently garbage-collect terminated child processors.+If the spawned process for some measure stops,+then all future inputs for that measure are ignored.++__Warning:__+The router process holds on to a reference to each measure @i@ for each child+process, even after that child process has stopped.++==== __Examples__++>>> run $ liftRouter (Just . even) (\case {True -> mapping (+1); False -> echo}) <~ source [1,2,3,4,5]+[1,3,3,5,5]++>>> run $ liftRouter (Just . even) (\case {True -> echo; False -> stopped}) <~ source [1,2,3,4,5]+[2,4]++>>> run $ liftRouter (Just . even) (\case {True -> echo; False -> taking 1}) <~ source [1,2,3,4,5]+[1,2,4] -} liftRouter ::-  forall m k a b.-  (MonadIO m, Hashable k) =>+  forall m i a b.+  (Monad m, Hashable i) =>   -- | Function to measure.-  (a -> Maybe k) ->+  (a -> Maybe i) ->   -- | Function to spawn child processors.-  (k -> ProcessT m a b) ->+  (i -> ProcessT m a b) ->   ProcessT m a b liftRouter measure spawn = awaiting M.empty  where-  awaiting :: HashMap k (ProcessT m a b) -> ProcessT m a b+  awaiting :: HashMap i (Child (ProcessT m a b)) -> ProcessT m a b   awaiting st = MachineT . pure $ Await onNext Refl onStop    where     onNext :: a -> MachineT m (Is a) b     onNext a = case measure a of       Nothing -> awaiting st-      Just k -> provideThen a m $ \m' -> awaiting (M.insert k m' st)-       where-        m = fromMaybe (spawn k) (M.lookup k st)+      Just i ->+        case fromMaybe (ChildRunning $ spawn i) (M.lookup i st) of+          ChildRunning p ->+            provideThen a p $ \p' ->+              let !st' = M.insert i p' st+               in awaiting st'+          ChildStopped ->+            awaiting st+     onStop :: MachineT m (Is a) b-    onStop = foldr starve stopped (M.elems st)+    onStop = foldr starve stopped (concatMap toList . M.elems $ st) -  provideThen :: a -> ProcessT m a b -> (ProcessT m a b -> ProcessT m a b) -> ProcessT m a b-  provideThen a m k =+  provideThen :: a -> ProcessT m a b -> (Child (ProcessT m a b) -> ProcessT m a b) -> ProcessT m a b+  provideThen a p k =     MachineT $-      runMachineT m >>= \case-        Stop -> runMachineT (k stopped)-        Yield o m' -> pure (Yield o (provideThen a m' k))+      runMachineT p >>= \case+        Stop -> runMachineT (k ChildStopped)+        Yield o p' -> pure (Yield o (provideThen a p' k))         Await onNext Refl _onStop -> runMachineT (exhaustThen (onNext a) k) -  exhaustThen :: ProcessT m a b -> (ProcessT m a b -> ProcessT m a b) -> ProcessT m a b-  exhaustThen m k =+  exhaustThen :: ProcessT m a b -> (Child (ProcessT m a b) -> ProcessT m a b) -> ProcessT m a b+  exhaustThen p k =     MachineT $-      runMachineT m >>= \case-        Yield o m' -> pure (Yield o (k m'))-        m' -> runMachineT (k (encased m'))+      runMachineT p >>= \case+        Yield o p' -> pure (Yield o (k $ ChildRunning p'))+        p' -> runMachineT (k (ChildRunning $ encased p')) +{- |+Internal helper.+A wrapper for child processes spawned by `liftRouter`.+-}+data Child a+  = ChildRunning !a+  | ChildStopped+  deriving (Functor, Foldable)+ ------------------------------------------------------------------------------- -- Event stream sorting -------------------------------------------------------------------------------  {- |-Reorder events respecting ticks.+Sort items in @N@ successive batches. -This machine caches two batches worth of events, sorts them together,-and then yields only those events whose timestamp is less than or equal-to the maximum of the first batch.+If the maximum key in batch @i@ is guaranteed to be smaller than the minimum+key in batch @i + 2N@, this process produces a totally ordered stream of items.++The process @`sortByBatch` key@ caches @N@ batches of items, sorts them+together, and yields only those items whose key is less than or equal+to the maximum key in the first batch.++==== __Examples__++>>> run $ sortByBatch @Int id <~ source [[1,4],[7,2,3,5],[6,8]]+[[1,2,3,4],[5,6,7],[8]]++>>> run $ sortByBatch @Int id <~ source [[1,7],[4,2,3,5],[6,8]]+[[1,2,3,4,5,7],[6,8]] -} sortByBatch ::-  forall m a.-  (Monad m) =>-  (a -> Timestamp) ->-  ProcessT m [a] [a]-sortByBatch timestamp = sortByBatchWith Nothing+  forall a k.+  (Bounded k, Ord k) =>+  (a -> k) ->+  Process [a] [a]+sortByBatch key = sortByBatchWith Nothing  where-  sortByBatchWith :: Maybe [a] -> ProcessT m [a] [a]+  sortByBatchWith ::+    forall m.+    (Monad m) =>+    Maybe [a] ->+    ProcessT m [a] [a]   sortByBatchWith = \case     Nothing -> MachineT $ pure $ Await onNext Refl onStop      where       onNext :: [a] -> ProcessT m [a] [a]       onNext new = sortByBatchWith (Just sortedNew)        where-        sortedNew = sortByTime new+        sortedNew = sortByKey new       onStop :: ProcessT m [a] [a]       onStop = stopped     Just sortedOld -> MachineT $ pure $ Await onNext Refl onStop@@ -377,69 +552,175 @@         | otherwise = MachineT $ pure $ Yield sortedBeforeCutoff $ sortByBatchWith $ Just sortedAfterCutoff        where         -- NOTE: use of partial @maximum@ is guarded by the check @null old@.-        cutoff = getMax (foldMap (Max . timestamp) sortedOld)-        sortedNew = sortByTime new-        sorted = joinByTime sortedOld sortedNew-        (sortedBeforeCutoff, sortedAfterCutoff) = L.partition ((<= cutoff) . timestamp) sorted+        cutoff = getMax (foldMap (Max . key) sortedOld)+        sortedNew = sortByKey new+        sorted = joinByKey sortedOld sortedNew+        (sortedBeforeCutoff, sortedAfterCutoff) = L.partition ((<= cutoff) . key) sorted       onStop :: ProcessT m [a] [a]       onStop = MachineT $ pure $ Yield sortedOld $ stopped -  -- compByTime :: a -> a -> Ordering-  compByTime = compare `on` timestamp+  compByKey :: a -> a -> Ordering+  compByKey = compare `on` key -  -- sortByTime :: [a] -> [a]-  sortByTime = L.sortBy compByTime+  sortByKey :: [a] -> [a]+  sortByKey = L.sortBy compByKey -  -- joinByTime :: [a] -> [a] -> [a]-  joinByTime [] ys = ys-  joinByTime xs [] = xs-  joinByTime (x : xs) (y : ys) = case compByTime x y of-    LT -> x : joinByTime xs (y : ys)-    _ -> y : joinByTime (x : xs) ys+  joinByKey {- Sorted -} :: [a {- Sorted -}] -> [a {- Sorted -}] -> [a]+  joinByKey [] ys = ys+  joinByKey xs [] = xs+  joinByKey (x : xs) (y : ys)+    | compByKey x y == LT = x : joinByKey xs (y : ys)+    | otherwise = y : joinByKey (x : xs) ys  {- |-Variant of `sortByBatch` that operates on streams of items and ticks.+Sort items between successive ticks.++If the maximum key in batch @i@ is guaranteed to be smaller than the minimum+key in batch @i + 2@, this process produces a totally ordered stream of items.++==== __Examples__++>>> run $ sortByTick @Int id <~ source [Item 1,Item 4,Tick,Item 7,Item 2,Item 3,Item 5,Tick,Item 6,Item 8]+[Tick,Item 1,Item 2,Item 3,Item 4,Tick,Item 5,Item 6,Item 7,Item 8]++>>> run $ sortByTick @Int id <~ source [Item 1,Item 7,Tick,Item 4,Item 2,Item 3,Item 5,Tick,Item 6,Item 8]+[Tick,Item 1,Item 2,Item 3,Item 4,Item 5,Item 7,Tick,Item 6,Item 8] -}-sortByBatchTick :: (a -> Timestamp) -> Process (Tick a) (Tick a)-sortByBatchTick timestamp =-  mapping (fmap (: [])) ~> batchByTick ~> sortByBatch timestamp ~> batchListToTick+sortByTick ::+  forall a k.+  (Bounded k, Ord k) =>+  (a -> k) ->+  Process (Tick a) (Tick a)+sortByTick key =+  mapping (fmap D.singleton)+    ~> batchByTick+    ~> mapping (fmap D.toList)+    ~> liftTick (sortByBatch key ~> asParts) +{- |+Sort items between @2*K@ successive ticks.++If the maximum key in batch @i@ is guaranteed to be smaller than the minimum+key in batch @i + K@, this process produces a totally ordered stream of items.++==== __Examples__++>>> run $ sortByTicks @Int id 2 <~ source [Item 2,Tick,Item 1,Tick,Item 4,Tick,Item 3,Tick]+[Tick,Tick,Tick,Item 1,Item 2,Tick,Item 3,Item 4]++>>> run $ sortByTicks @Int id 2 <~ source [Item 1,Tick,Item 3,Tick,Item 2,Tick,Item 4,Tick]+[Tick,Tick,Tick,Item 1,Item 2,Item 3,Tick,Item 4]++>>> run $ sortByTicks @Int id 2 <~ source [Item 1,Tick,Item 4,Tick,Item 2,Tick,Item 3,Tick]+[Tick,Tick,Tick,Item 1,Item 2,Item 3,Item 4,Tick]+-}+sortByTicks ::+  forall a k.+  (Bounded k, Ord k) =>+  (a -> k) ->+  Int ->+  Process (Tick a) (Tick a)+sortByTicks key ticks =+  mapping (fmap D.singleton)+    ~> batchByTicks ticks+    ~> mapping (fmap D.toList)+    ~> liftTick (sortByBatch key ~> asParts)+ ------------------------------------------------------------------------------- -- Filtering semaphores ------------------------------------------------------------------------------- -{- | A simple delimiting t'Moore' machine,-which is opened by one constant marker and closed by the other one.+{- |+A delimiting t`Moore` machine based on constant open/close markers.++The machine @`between` o c@ consumes consumes a stream of items, and produces+a stream of `Bool` that is `False` up to and including the first occurrence of+@o@, then is `True` up to and including the first occurrence of @c@, and then+is `False` forever.++==== __Examples__++>>> run $ auto (betweenEach (2, 4)) <~ source [1, 2, 3, 4, 5]+[False,False,True,True,False,False]++>>> run $ auto (betweenEach (2, 4)) <~ source [2, 3, 4, 2, 3, 4]+[False,True,True,False,True,True,False] -}-between :: Text -> Text -> Moore Text Bool-between x y = open+betweenEach :: (Eq a) => (a, a) -> Moore a Bool+betweenEach (open, close) = beforeOpen  where-  open = Moore False open' where open' x' = if x == x' then close else open-  close = Moore True close' where close' y' = if y == y' then end else close-  end = Moore False (const end)+  beforeOpen = Moore False $ \a ->+    if a == open then betweenOpenAndClose else beforeOpen+  betweenOpenAndClose = Moore True $ \a ->+    if a == close then beforeOpen else betweenOpenAndClose --- | Delimit the event process.-delimit :: (Monad m) => Moore Text Bool -> ProcessT m Event Event-delimit = construct . go+{- |+A delimiting t`Moore` machine based on constant open/close markers.++The machine @`between` o c@ consumes consumes a stream of items, and produces+a stream of `Bool` that is `False` up to and including the first occurrence of+@o@, then is `True` up to and including the first occurrence of @c@, and then+is `False` forever.++==== __Examples__++>>> run $ auto (betweenFirst (2, 4)) <~ source [1, 2, 3, 4, 5]+[False,False,True,True,False,False]++>>> run $ auto (betweenFirst (2, 4)) <~ source [2, 3, 4, 2, 3, 4]+[False,True,True,False,False,False,False]+-}+betweenFirst :: (Eq a) => (a, a) -> Moore a Bool+betweenFirst (open, close) = beforeFirstOpen  where-  go :: (Monad m) => Moore Text Bool -> PlanT (Is Event) Event m ()-  go mm@(Moore s next) = do-    e <- await-    case evSpec e of-      -- on marker step the moore machine.-      E.UserMarker m -> do-        let mm'@(Moore s' _) = next m-        -- if current or next state is open (== True), emit the marker.-        when (s || s') $ yield e-        go mm'+  beforeFirstOpen = Moore False $ \a ->+    if a == open then betweenFirstOpenAndClose else beforeFirstOpen+  betweenFirstOpenAndClose = Moore True $ \a ->+    if a == close then afterFirstClose else betweenFirstOpenAndClose+  afterFirstClose = Moore False (const afterFirstClose) -      -- for other events, emit if the state is open.-      _ -> do-        when s $ yield e-        go mm+{- |+Filter the items in a stream based on a t`Moore` machine. +==== __Examples__++>>> run $ delimit (betweenEach (2, 4)) <~ source [1, 2, 3, 4, 5]+[2,3,4]++>>> run $ delimit (betweenEach (2, 4)) <~ source [2, 3, 4, 2, 3, 4]+[2,3,4,2,3,4]++>>> run $ delimit (betweenFirst (2, 4)) <~ source [1, 2, 3, 4, 5]+[2,3,4]++>>> run $ delimit (betweenFirst (2, 4)) <~ source [2, 3, 4, 2, 3, 4]+[2,3,4]+-}+delimit ::+  forall m a.+  (Monad m) =>+  Moore a Bool ->+  ProcessT m a a+delimit = construct . go+ where+  go ::+    Moore a Bool ->+    PlanT (Is a) a m Void+  go _st@(Moore wasOpen onNext) =+    await >>= \a -> do+      -- Feed the item to the delimiting Moore machine.+      let st'@(Moore willBeOpen _) = onNext a+      -- If the state has changed, i.e., @wasOpen /= willBeOpen@, then+      -- the current item is a marker. All markers should be yielded.+      let isMarker = wasOpen /= willBeOpen+      when (wasOpen || isMarker) $ yield a+      go st'+ ------------------------------------------------------------------------------- -- Validation+--+-- TODO: These machines, or at least the error messages that they print, are+--       specific to eventlog processing. Hence, they should be moved. -------------------------------------------------------------------------------  {- |@@ -449,62 +730,81 @@ a warning that directs the user to check that the @-l@ flag was set correctly. -} validateInput ::-  (MonadIO m) =>-  Verbosity ->+  (Monad m) =>+  Logger m ->   Int ->   ProcessT m (Tick a) x-validateInput verbosity ticks-  | verbosityWarning >= verbosity = construct $ start ticks-  | otherwise = stopped+validateInput logger ticks = construct $ start ticks  where   start remaining-    | remaining <= 0 = liftIO $ do-        logWarning verbosity . T.pack $-          printf-            "No input after %d ticks. Did you pass -l to the GHC RTS?"-            ticks+    | remaining <= 0 = do+        let msg = printf "No input after %d ticks. Did you pass -l to the GHC RTS?" ticks+        lift $ writeLog logger WARN $ T.pack msg+        pure ()     | otherwise = do-        logDebug verbosity $-          T.pack (show remaining) <> " ticks remaining."+        let msg = "Waiting for " <> T.pack (show remaining) <> " more ticks before showing input warning."+        lift $ writeLog logger DEBUG $ msg         await >>= \case-          Item{} -> do-            logDebug verbosity "Received item."-          Tick -> do-            logDebug verbosity "Received tick."+          Item{} ->+            lift $ writeLog logger DEBUG $ "Received item. Cancelled input warning."+          Tick ->             start (pred remaining)  {- | This machine validates that the inputs are received in order.  If an out-of-order input is encountered, the machine prints an error message-that directs the user to check that the @--eventlog-flush-interval@ and the-@--batch-interval@ flags are set correctly.+that directs the user to check that the @--eventlog-flush-interval@ flag is+set correctly. -} validateOrder ::-  (MonadIO m, Show a) =>-  Verbosity ->-  (a -> Timestamp) ->+  (Monad m, Ord k, Show a) =>+  Logger m ->+  (a -> k) ->   ProcessT m a x-validateOrder verbosity timestamp-  | verbosityError >= verbosity = construct $ start Nothing-  | otherwise = stopped+validateOrder logger timestamp = construct $ go Nothing  where-  start maybeOld =+  go maybeOld =     await >>= \new ->       case maybeOld of         Just old           | timestamp new < timestamp old -> do-              logError verbosity . T.pack $-                "Encountered two out-of-order inputs.\n\-                \Did you pass --eventlog-flush-interval to the GHC RTS?\n\-                \Did you set --batch-interval to be at least as big as the value of --eventlog-flush-interval?"-              logDebug verbosity . T.pack $-                printf-                  "Out-of-order inputs:\n\-                  \- %s\n\-                  \- %s"-                  (show old)-                  (show new)-              pure ()+              let msg1 =+                    "Encountered two out-of-order inputs.\n\+                    \Did you pass --eventlog-flush-interval=SECONDS to the GHC RTS?\n\+                    \Did you pass the same flag to this program?"+              lift $ writeLog logger ERROR $ T.pack msg1+              let msg2 =+                    printf+                      "Out-of-order inputs:\n\+                      \- %s\n\+                      \- %s"+                      (show old)+                      (show new)+              lift $ writeLog logger DEBUG $ T.pack msg2         _otherwise -> do-          start (Just new)+          go (Just new)++{- |+This machine validates that ticks are unique and increasing.+-}+validateTicks ::+  (Monad m) =>+  Logger m ->+  ProcessT m (Tick a) (Tick a)+validateTicks logger = construct $ go Nothing+ where+  go maybeTick =+    await >>= \case+      Item _ ->+        go maybeTick+      TickWithInfo{tickInfo = TickInfo{tick = tick'}} -> do+        for_ maybeTick $ \case+          tick+            | tick' == tick + 1 -> do+                let msg = "Saw tick " <> T.pack (show tick) <> "."+                lift $ writeLog logger TRACE $ msg+            | otherwise -> do+                let msg = "Encountered non-increasing ticks " <> T.pack (show tick) <> " and " <> T.pack (show tick') <> "."+                lift $ writeLog logger ERROR $ msg+        go (Just tick')

@@ -6,15 +6,16 @@ -} module GHC.Eventlog.Live.Machine.Decoder (   -- * Event decoding-  DecodeError (..),   decodeEvent,   decodeEventBatch, ) where -import Control.Exception (Exception, throwIO)-import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Trans.Class (MonadTrans (..)) import Data.ByteString qualified as BS import Data.Machine (Is, PlanT, ProcessT, await, construct, yield)+import Data.Text qualified as T+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (Logger, writeLog) import GHC.Eventlog.Live.Machine.Core (Tick (..), liftTick) import GHC.RTS.Events (Event) import GHC.RTS.Events.Incremental (Decoder (..), decodeEventLog)@@ -27,23 +28,26 @@  Throws a t'DecodeError' on error. -}-decodeEvent :: (MonadIO m) => ProcessT m BS.ByteString Event-decodeEvent = construct $ loop decodeEventLog+decodeEvent ::+  forall m.+  (Monad m) =>+  Logger m ->+  ProcessT m BS.ByteString Event+decodeEvent logger = construct $ loop decodeEventLog  where-  loop :: (MonadIO m) => Decoder a -> PlanT (Is BS.ByteString) a m ()+  loop :: Decoder a -> PlanT (Is BS.ByteString) a m ()   loop Done{} = pure ()   loop (Consume k) = await >>= \chunk -> loop (k chunk)   loop (Produce a d') = yield a >> loop d'-  loop (Error _ err) = liftIO $ throwIO $ DecodeError err+  loop (Error _ err) = lift $ writeLog logger ERROR $ T.pack err  {- | Parse 'Event's from a stream of 'BS.ByteString' chunks with ticks.  Throws 'DecodeError' on error. -}-decodeEventBatch :: (MonadIO m) => ProcessT m (Tick BS.ByteString) (Tick Event)-decodeEventBatch = liftTick decodeEvent--newtype DecodeError = DecodeError String deriving (Show)--instance Exception DecodeError+decodeEventBatch ::+  (Monad m) =>+  Logger m ->+  ProcessT m (Tick BS.ByteString) (Tick Event)+decodeEventBatch logger = liftTick $ decodeEvent logger

@@ -6,94 +6,58 @@ -} module GHC.Eventlog.Live.Machine.Source (   -- * Eventlog source-  sourceHandleWait,-  sourceHandleBatch,+  eventlogSourceTick,   defaultChunkSizeBytes, ) where -import Control.Exception (catch, throwIO) import Control.Monad.IO.Class (MonadIO (..)) import Data.ByteString qualified as BS-import Data.Function (fix) import Data.Machine (MachineT (..), construct, yield) import Data.Word (Word64) import GHC.Clock (getMonotonicTimeNSec)-import GHC.Eventlog.Live.Machine.Core (Tick (..))-import System.IO (Handle, hWaitForInput)-import System.IO.Error (isEOFError)------------------------------------------------------------------------------------ Socket source--{- |-A source which reads chunks from a `Handle`.-When an input is available, it yields an v`Item`.-When the timeout is reached, it yields a v`Tick`.--}-sourceHandleWait ::-  (MonadIO m) =>-  -- | The wait timeout in milliseconds.-  Int ->-  -- | The number of bytes to read.-  Int ->-  -- | The eventlog socket handle.-  Handle ->-  MachineT m k (Tick BS.ByteString)-sourceHandleWait timeoutMilli chunkSizeBytes handle =-  construct $ fix $ \loop -> do-    ready <- liftIO $ hWaitForInput' handle timeoutMilli-    case ready of-      Ready -> do-        bs <- liftIO $ BS.hGetSome handle chunkSizeBytes-        yield (Item bs)-        loop-      NotReady -> do-        yield Tick-        loop-      EOF ->-        pure ()------------------------------------------------------------------------------------ Socket source with batches+import GHC.Eventlog.Live.Machine.Core (Tick (..), TickInfo (..))+import GHC.Eventlog.Live.Source.Core (EventlogSourceData (..), EventlogSourceHandle, recv)  {- | A source which reads chunks from a `Handle`. When input is available, it yields an v`Item`. It yields a v`Tick` at each increment of the batch interval. -}-sourceHandleBatch ::+eventlogSourceTick ::   (MonadIO m) =>   -- | The batch interval in milliseconds.   Int ->   -- | The number of bytes to read.   Int ->-  -- | The eventlog socket handle.-  Handle ->+  -- | The eventlog source handle.+  EventlogSourceHandle ->   MachineT m k (Tick BS.ByteString)-sourceHandleBatch batchIntervalMs chunkSizeBytes handle = construct start+eventlogSourceTick batchIntervalMilli chunkSizeBytes h =+  construct $ start 0  where-  start = do-    startTimeMs <- liftIO getMonotonicTimeMilli-    batch startTimeMs-  batch startTimeMs = waitForInput+  batchIntervalMicro = milliToMicro batchIntervalMilli++  start tick = do+    startTimeMicro <- liftIO getMonotonicTimeMicro+    batch tick startTimeMicro++  batch tick startTimeMicro = batchLoop    where-    getRemainingTimeMilli = do-      currentTimeMilli <- liftIO getMonotonicTimeMilli-      pure $ (startTimeMs + batchIntervalMs) - currentTimeMilli-    waitForInput = do-      remainingTimeMilli <- getRemainingTimeMilli-      if remainingTimeMilli <= 0+    getRemainingTimeMicro = do+      currentTimeMicro <- liftIO getMonotonicTimeMicro+      pure $ (startTimeMicro + batchIntervalMicro) - currentTimeMicro++    batchLoop = do+      remainingTimeMicro <- getRemainingTimeMicro+      if remainingTimeMicro <= 0         then do-          yield Tick-          start+          yield TickWithInfo{tickInfo = TickInfo{tick}}+          start (tick + 1)         else do-          ready <- liftIO (hWaitForInput' handle remainingTimeMilli)-          case ready of-            Ready -> do-              chunk <- liftIO $ BS.hGetSome handle chunkSizeBytes-              yield (Item chunk) >> waitForInput-            NotReady -> waitForInput-            EOF -> pure ()+          liftIO (recv h remainingTimeMicro chunkSizeBytes) >>= \case+            EventlogSourceData chunk -> yield (Item chunk) >> batchLoop+            EventlogSourceTimeout -> batchLoop+            EventlogSourceClosed -> pure ()  {- | Eventlog chunk size in bytes.@@ -104,41 +68,23 @@  {- | Internal helper.-Return monotonic time in milliseconds, since some unspecified starting point--}-getMonotonicTimeMilli :: IO Int-getMonotonicTimeMilli = nanoToMilli <$> getMonotonicTimeNSec--{- |-Internal helper.-Convert nanoseconds to milliseconds.-The conversion from 'Word64' to 'Int' is safe.-It cannot overflow due to the division by 1_000_000.+Return monotonic time in microseconds, since some unspecified starting point -}-nanoToMilli :: Word64 -> Int-nanoToMilli = fromIntegral . (`div` 1_000_000)+getMonotonicTimeMicro :: IO Int+getMonotonicTimeMicro = nanoToMicro <$> getMonotonicTimeNSec  {- | Internal helper.-Type to represent the state of a handle.+Convert nanoseconds to microseconds.+If the size of @Int@ is at least as big as that of @Word64@, then+the conversion from 'Word64' to 'Int' is safe, due to the division by 1000. -}-data Ready = Ready | NotReady | EOF+nanoToMicro :: Word64 -> Int+nanoToMicro = fromIntegral . (`div` 1_000)  {- | Internal helper.-Wait for input from a `Handle` for a given number of milliseconds.+Convert milliseconds to microseconds. -}-hWaitForInput' ::-  -- | The handle.-  Handle ->-  -- | The timeout in milliseconds.-  Int ->-  IO Ready-hWaitForInput' handle timeoutMilli =-  catch (boolToReady <$> hWaitForInput handle timeoutMilli) handleEOFError- where-  boolToReady True = Ready-  boolToReady False = NotReady-  handleEOFError err-    | isEOFError err = pure EOF-    | otherwise = throwIO err+milliToMicro :: Int -> Int+milliToMicro = (* 1_000)

@@ -5,41 +5,32 @@ Portability : portable -} module GHC.Eventlog.Live.Options (-  EventlogSource (..),-  eventlogSourceParser,-  eventlogSocketTimeoutParser,+  eventlogSourceOptionsParser,+  eventlogSocketTimeoutSParser,   eventlogSocketTimeoutExponentParser,   heapProfBreakdownParser,   eventlogLogFileParser,-  batchIntervalParser,+  eventlogFlushIntervalSParser,   verbosityParser,   statsParser, ) where  import Control.Applicative (asum)-import Data.Char (toLower)+import GHC.Eventlog.Live.Data.Severity (Severity (..), fromSeverityString) import GHC.Eventlog.Live.Machine.Analysis.Heap (heapProfBreakdownEitherReader)-import GHC.Eventlog.Live.Verbosity (Verbosity, verbosityDebug, verbosityError, verbosityInfo, verbosityQuiet, verbosityWarning)+import GHC.Eventlog.Live.Source.Core (EventlogSourceOptions (..)) import GHC.RTS.Events (HeapProfBreakdown (..)) import Options.Applicative qualified as O-import Text.Read (readEither)+import Options.Applicative.Help.Pretty qualified as OP  -------------------------------------------------------------------------------- -- Eventlog Source  {- |-The type of eventlog sockets.--}-data EventlogSource-  = EventlogStdin-  | EventlogFile FilePath-  | EventlogSocketUnix FilePath--{- | Parser for the eventlog socket. -}-eventlogSourceParser :: O.Parser EventlogSource-eventlogSourceParser =+eventlogSourceOptionsParser :: O.Parser EventlogSourceOptions+eventlogSourceOptionsParser =   asum     [ stdinParser     , fileParser@@ -47,21 +38,21 @@     ]  where   stdinParser =-    EventlogStdin+    EventlogSourceOptionsStdin       <$ O.flag'         ()         ( O.long "eventlog-stdin"             <> O.help "Read the eventlog from stdin."         )   fileParser =-    EventlogFile+    EventlogSourceOptionsFile       <$> O.strOption         ( O.long "eventlog-file"             <> O.metavar "FILE"             <> O.help "Read the eventlog from a file."         )   socketUnixParser =-    EventlogSocketUnix+    EventlogSourceOptionsSocketUnix       <$> O.strOption         ( O.long "eventlog-socket"             <> O.metavar "SOCKET"@@ -71,13 +62,13 @@ {- | Parser for the intial timeout for exponential backoff. -}-eventlogSocketTimeoutParser :: O.Parser Double-eventlogSocketTimeoutParser =+eventlogSocketTimeoutSParser :: O.Parser Double+eventlogSocketTimeoutSParser =   O.option     O.auto     ( O.long "eventlog-socket-timeout"-        <> O.metavar "NUM"-        <> O.help "Eventlog socket connection retry timeout in microseconds."+        <> O.metavar "SECONDS"+        <> O.help "Eventlog socket connection retry timeout in seconds."         <> O.value 1     ) @@ -89,7 +80,7 @@   O.option     O.auto     ( O.long "eventlog-socket-exponent"-        <> O.metavar "NUM"+        <> O.metavar "NUMBER"         <> O.help "Eventlog socket connection retry timeout exponent."         <> O.value 1     )@@ -102,12 +93,14 @@ -} heapProfBreakdownParser :: O.Parser HeapProfBreakdown heapProfBreakdownParser =-  O.option-    (O.eitherReader heapProfBreakdownEitherReader)-    ( O.short 'h'-        <> O.metavar "Tcmdyrbi"-        <> O.help "Heap profile breakdown."-    )+  O.option (O.eitherReader heapProfBreakdownEitherReader) . mconcat $+    [ O.short 'h'+    , O.metavar "Tcmdyrbi"+    , O.helpDoc . Just . OP.vcat . fmap OP.pretty $+        [ "Heap profile breakdown."+        , "Should match the option passed to the application."+        ]+    ]  -------------------------------------------------------------------------------- -- Eventlog Log File@@ -127,67 +120,44 @@ -- Batch Interval  {- |-Parser for the batch interval.+Parser for the eventlog flush interval. -}-batchIntervalParser :: O.Parser Int-batchIntervalParser =-  O.option-    O.auto-    ( O.long "batch-interval"-        <> O.metavar "NUM"-        <> O.help "Batch interval in milliseconds."-        <> O.value defaultBatchIntervalMs-    )+eventlogFlushIntervalSParser :: O.Parser Double+eventlogFlushIntervalSParser =+  O.option O.auto . mconcat $+    [ O.long "eventlog-flush-interval"+    , O.metavar "SECONDS"+    , O.helpDoc . Just . OP.vcat . fmap OP.pretty $+        [ "Eventlog flush interval in seconds."+        , "Should match the option passed to the application."+        ]+    , O.value defaultEventlogFlushIntervalS+    ]  {- | Internal helper.-The default batch interval in milliseconds.+The default interval in which the eventlog is flushed in seconds. -}-defaultBatchIntervalMs :: Int-defaultBatchIntervalMs = 1_000+defaultEventlogFlushIntervalS :: Double+defaultEventlogFlushIntervalS = 1  -------------------------------------------------------------------------------- -- Verbosity  {- |-Parser for verbosities.-The default verbosity is `verbosityWarning`.+Parser for `Severity`+The default severity is `WARN`. -}-verbosityParser :: O.Parser Verbosity+verbosityParser :: O.Parser Severity verbosityParser =   O.option-    (O.eitherReader readEitherVerbosity)+    (O.maybeReader fromSeverityString)     ( O.short 'v'         <> O.long "verbosity"-        <> O.metavar "quiet|error|warning|info|debug|0-4"-        <> O.help "The verbosity threshold for logging."-        <> O.value verbosityWarning+        <> O.metavar "fatal|error|warning|info|debug|trace"+        <> O.help "The severity threshold for logging."+        <> O.value WARN     )--{- |-Internal helper.-Parser for verbosities by number or name.-Case insensitive.--}-readEitherVerbosity :: String -> Either String Verbosity-readEitherVerbosity rawVerbosity =-  -- try to parse the verbosity as a number...-  case readEither @Word rawVerbosity of-    -- if the verbosity string is a number, map it to a verbosity...-    Right verbosityThreshold-      | verbosityThreshold <= 0 -> Right verbosityQuiet-      | verbosityThreshold == 1 -> Right verbosityError-      | verbosityThreshold == 2 -> Right verbosityWarning-      | verbosityThreshold == 3 -> Right verbosityInfo-      | otherwise -> Right verbosityDebug-    -- otherwise, match it against the literal names of the levels...-    Left _parseError -> case toLower <$> rawVerbosity of-      "quiet" -> Right verbosityQuiet-      "error" -> Right verbosityError-      "warning" -> Right verbosityWarning-      "info" -> Right verbosityInfo-      "debug" -> Right verbosityDebug-      _otherwise -> Left $ "Could not parse verbosity '" <> rawVerbosity <> "'."  -------------------------------------------------------------------------------- -- Statistics

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

@@ -0,0 +1,207 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : GHC.Eventlog.Live.Source+Description : Utilities for running eventlog machines with sockets.+Stability   : experimental+Portability : portable+-}+module GHC.Eventlog.Live.Source (+  Tick (..),+  tryConnect,+  withEventlogSourceHandle,+  runWithEventlogSourceHandle,+  runWithEventlogSourceOptions,+) where++import Control.Concurrent (threadDelay)+import Control.Exception (Exception (..))+import Control.Exception qualified as E+import Data.Foldable (traverse_)+import Data.Machine (ProcessT, runT_, (~>))+import Data.Machine.Fanout (fanout)+import Data.Maybe (fromMaybe)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Void (Void)+import GHC.Eventlog.Live.Data.Severity (Severity (..))+import GHC.Eventlog.Live.Logger (Logger, writeLog)+import GHC.Eventlog.Live.Machine.Core+import GHC.Eventlog.Live.Machine.Decoder+import GHC.Eventlog.Live.Machine.Sink+import GHC.Eventlog.Live.Machine.Source+import GHC.Eventlog.Live.Source.Core+import GHC.RTS.Events (Event)+import Network.Socket (Socket)+import Network.Socket qualified as S+import System.IO qualified as IO+import Text.Printf (printf)++{- |+Run an event processor with `EventlogSourceOptions`.+-}+runWithEventlogSourceOptions ::+  -- | The logging action.+  Logger IO ->+  -- | The eventlog socket handle.+  EventlogSourceOptions ->+  -- | The initial timeout in seconds for exponential backoff.+  Double ->+  -- | The timeout exponent for exponential backoff.+  Double ->+  -- | The batch interval in milliseconds.+  Int ->+  -- | The number of bytes to read (defaults to 4KiB).+  Maybe Int ->+  -- | An optional file to which to stream binary eventlog data.+  Maybe FilePath ->+  -- | The event processor.+  ProcessT IO (Tick Event) Void ->+  IO ()+runWithEventlogSourceOptions logger eventlogSourceOptions timeoutExponent initialTimeoutS batchIntervalMs maybeChuckSizeBytes maybeOutputFile toEventSink = do+  withEventlogSourceHandle logger timeoutExponent initialTimeoutS eventlogSourceOptions $ \eventlogSourceHandle ->+    runWithEventlogSourceHandle logger eventlogSourceHandle batchIntervalMs maybeChuckSizeBytes maybeOutputFile toEventSink++{- |+Run an event processor with an eventlog socket handle.+-}+runWithEventlogSourceHandle ::+  -- | The logging action.+  Logger IO ->+  -- | The eventlog socket handle.+  EventlogSourceHandle ->+  -- | The batch interval in milliseconds.+  Int ->+  -- | The number of bytes to read (defaults to 4KiB).+  Maybe Int ->+  -- | An optional file to which to stream binary eventlog data.+  Maybe FilePath ->+  -- | The event processor.+  ProcessT IO (Tick Event) Void ->+  IO ()+runWithEventlogSourceHandle logger eventlogSourceHandle batchIntervalMs maybeChuckSizeBytes maybeOutputFile toEventSink = do+  let chuckSizeBytes = fromMaybe defaultChunkSizeBytes maybeChuckSizeBytes+  let fromSocket = eventlogSourceTick batchIntervalMs chuckSizeBytes eventlogSourceHandle+  case maybeOutputFile of+    Nothing ->+      runT_ $+        fromSocket ~> decodeEventBatch logger ~> toEventSink+    Just outputFile ->+      IO.withFile outputFile IO.WriteMode $ \outputHandle -> do+        runT_ $+          fromSocket+            ~> fanout+              [ fileSinkBatch outputHandle+              , decodeEventBatch logger ~> toEventSink+              ]++{- |+Run an action with a `Handle` to the eventlog described by `EventlogSourceOptions`.+-}+withEventlogSourceHandle ::+  -- | The logging action.+  Logger IO ->+  -- | The initial timeout in seconds for exponential backoff.+  Double ->+  -- | The timeout exponent for exponential backoff.+  Double ->+  -- | The eventlog socket.+  EventlogSourceOptions ->+  (EventlogSourceHandle -> IO ()) ->+  IO ()+withEventlogSourceHandle logger initialTimeoutS timeoutExponent eventlogSource action = do+  case eventlogSource of+    EventlogSourceOptionsStdin -> do+      writeLog logger INFO $+        "Reading eventlog from stdin"+      let enter = do+            maybeStdinTextEncoding <- IO.hGetEncoding IO.stdin+            IO.hSetBinaryMode IO.stdin True+            pure maybeStdinTextEncoding+      let leave maybeStdinTextEncoding = do+            traverse_ (IO.hSetEncoding IO.stdin) maybeStdinTextEncoding+            IO.hSetNewlineMode IO.stdin IO.nativeNewlineMode+      E.bracket enter leave . const . action $ EventlogSourceHandleStdin+    EventlogSourceOptionsFile eventlogFile -> do+      writeLog logger INFO $+        "Reading eventlog from " <> T.pack eventlogFile+      IO.withBinaryFile eventlogFile IO.ReadMode $ \handle ->+        action $ EventlogSourceHandleFile handle+    EventlogSourceOptionsSocketUnix eventlogSocketUnix -> do+      writeLog logger INFO $+        "Waiting to connect on " <> prettyEventlogSocketUnix eventlogSocketUnix+      E.bracket (connectRetry logger initialTimeoutS timeoutExponent eventlogSocketUnix) S.close $ \socket ->+        action $ EventlogSourceHandleSocketUnix socket++{- |+Connect to the eventlog described by `EventlogSourceOptions` with retries and non-randomised exponential backoff.+-}+connectRetry ::+  -- | The logging action.+  Logger IO ->+  -- | The initial timeout in seconds for exponential backoff.+  Double ->+  -- | The timeout exponent for exponential backoff.+  Double ->+  -- | The eventlog socket.+  FilePath ->+  IO Socket+connectRetry logger initialTimeoutS timeoutExponent eventlogSocketUnix =+  connectLoop initialTimeoutS+ where+  waitFor :: Double -> IO ()+  waitFor timeoutS = threadDelay $ round $ timeoutS * 1e6++  connectLoop :: Double -> IO Socket+  connectLoop timeoutS = do+    let connect = do+          writeLog logger DEBUG $+            "Trying to connect on " <> prettyEventlogSocketUnix eventlogSocketUnix+          handle <- tryConnect eventlogSocketUnix+          writeLog logger DEBUG $+            "Connected on " <> prettyEventlogSocketUnix eventlogSocketUnix+          pure handle+    let cleanup (e :: E.IOException) = do+          writeLog logger DEBUG $+            "Failed to connect on "+              <> prettyEventlogSocketUnix eventlogSocketUnix+              <> ": "+              <> T.pack (displayException e)+          writeLog logger DEBUG $+            "Waiting "+              <> prettyTimeoutMcs timeoutS+              <> " to retry..."+          waitFor timeoutS+          connectLoop (timeoutS * timeoutExponent)+    E.catch connect cleanup++{- |+Try to connect to a Unix socket.+-}+tryConnect :: FilePath -> IO Socket+tryConnect eventlogSocketUnix =+  E.bracketOnError (S.socket S.AF_UNIX S.Stream S.defaultProtocol) S.close $ \socket -> do+    S.connect socket (S.SockAddrUnix eventlogSocketUnix)+    -- handle <- S.socketToHandle socket IO.ReadMode+    -- IO.hSetBuffering handle IO.NoBuffering+    pure socket++{- |+Interal helper. Pretty-printer for timeout values in microseconds.+-}+prettyTimeoutMcs :: Double -> Text+prettyTimeoutMcs timeoutS+  | timeoutS > 86400 = T.pack $ printf "%.2f days" (timeoutS / 86400)+  | timeoutS > 3600 = T.pack $ printf "%.2f hours" (timeoutS / 3600)+  | timeoutS > 60 = T.pack $ printf "%.2f minutes" (timeoutS / 60)+  | timeoutS > 1 = T.pack $ printf "%.2f seconds" timeoutS+  | timeoutS > 1e-3 = T.pack $ printf "%.2f milliseconds" (timeoutS / 1e-3)+  | timeoutS > 1e-6 = T.pack $ printf "%.2f microseconds" (timeoutS / 1e-6)+  | timeoutS > 1e-9 = T.pack $ printf "%.2f nanoseconds" (timeoutS / 1e-9)+  | otherwise = T.pack $ printf "%.2f seconds" timeoutS++{- |+Internal helper. Pretty-printer for eventlog sockets.+-}+prettyEventlogSocketUnix :: FilePath -> Text+prettyEventlogSocketUnix eventlogSocketUnix = "Unix socket " <> T.pack eventlogSocketUnix

@@ -0,0 +1,114 @@+module GHC.Eventlog.Live.Source.Core (+  EventlogSourceOptions (..),+  EventlogSourceHandle (..),+  EventlogSourceData (..),+  recv,+) where++import Control.Exception (handle, throwIO)+import Data.ByteString (ByteString)+import Data.ByteString qualified as BS+import Data.Maybe (fromMaybe)+import Network.Socket (Socket)+import Network.Socket.ByteString qualified as SB+import System.IO (Handle)+import System.IO qualified as IO (stdin)+import System.IO.Error (isEOFError)+import System.Timeout (timeout)++{- |+The options for different kinds of eventlog sources.+-}+data EventlogSourceOptions+  = EventlogSourceOptionsStdin+  | EventlogSourceOptionsFile FilePath+  | EventlogSourceOptionsSocketUnix FilePath++{- |+The handles for different kinds of eventlog sources.+-}+data EventlogSourceHandle+  = EventlogSourceHandleStdin+  | EventlogSourceHandleFile Handle+  | EventlogSourceHandleSocketUnix Socket++data EventlogSourceData+  = EventlogSourceData ByteString+  | EventlogSourceTimeout+  | EventlogSourceClosed++{- |+Receive data from an `EventlogSourceHandle`.++__Warning__: The current implementation is blocking on Windows. See the documentation for `timeout`.+-}+recv ::+  -- | The handle to the eventlog source.+  EventlogSourceHandle ->+  -- | The timeout in microseconds.+  Int ->+  -- | The number of bytes to read.+  Int ->+  IO EventlogSourceData+recv = \case+  EventlogSourceHandleStdin -> recvFromHandle IO.stdin+  EventlogSourceHandleFile h -> recvFromHandle h+  EventlogSourceHandleSocketUnix s -> recvFromSocket s++-- Permit a timeout and wrap the result appropriately.++{- |+Internal helper.+Receive data from a `Socket`.+-}+recvFromSocket ::+  -- | The eventlog socket.+  Socket ->+  -- | The timeout in microseconds.+  Int ->+  -- | The number of bytes to read.+  Int ->+  IO EventlogSourceData+recvFromSocket s timeoutMicros chunkSizeBytes =+  withTimeout timeoutMicros $ do+    msg <- SB.recv s chunkSizeBytes+    if BS.null msg+      then pure EventlogSourceClosed+      else pure $ EventlogSourceData msg++{- |+Internal helper.+Receive data from a `Handle`.+-}+recvFromHandle ::+  -- | The handle to the eventlog source.+  Handle ->+  -- | The timeout in microseconds.+  Int ->+  -- | The number of bytes to read.+  Int ->+  IO EventlogSourceData+recvFromHandle h timeoutMicros chunkSizeBytes =+  withTimeout timeoutMicros $+    handleEOFError EventlogSourceClosed $+      EventlogSourceData <$> BS.hGetSome h chunkSizeBytes++{- |+Internal helper.+Allow the IO operation to timeout.+-}+withTimeout ::+  -- | The timeout in microseconds.+  Int ->+  -- | The IO action to read data from the eventlog source.+  IO EventlogSourceData ->+  IO EventlogSourceData+withTimeout timeoutMicros =+  fmap (fromMaybe EventlogSourceTimeout) . timeout timeoutMicros++{- |+Internal helper.+Recover from an EOF error with a default value.+-}+handleEOFError :: a -> IO a -> IO a+handleEOFError d = handle (\e -> if isEOFError e then pure d else throwIO e)

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

@@ -0,0 +1,125 @@+{- HLINT ignore -}+{- FOURMOLU_DISABLE -}+{-# LANGUAGE CPP, GADTs, FlexibleContexts, RankNTypes, ScopedTypeVariables,+             TupleSections #-}+-- | The primary use of concurrent machines is to establish a+-- pipelined architecture that can boost overall throughput by running+-- each stage of the pipeline at the same time. The processing, or+-- production, rate of each stage may not be identical, so facilities+-- are provided to loosen the temporal coupling between pipeline+-- stages using buffers.+--+-- This architecture also lends itself to operations where multiple+-- workers are available for procesisng inputs. If each worker is to+-- process the same set of inputs, consider 'fanout' and+-- 'fanoutSteps'. If each worker is to process a disjoint set of+-- inputs, consider 'scatter'.+module Data.Machine.Concurrent (module Data.Machine,+                                -- * Concurrent connection+                                (>~>), (<~<),+                                -- * Buffered machines+                                bufferConnect, rollingConnect,+                                -- * Concurrent processing of shared inputs+                                fanout, fanoutSteps,+                                -- * Concurrent multiple-input machines+                                wye, tee, scatter, splitSum, mergeSum,+                                splitProd) where+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Control.Applicative+#endif+import Control.Concurrent.Async.Lifted+import Control.Monad (join)+import Control.Monad.Trans.Control+import Data.Machine hiding (tee, wye)+import Data.Machine.Concurrent.AsyncStep+import Data.Machine.Concurrent.Buffer+import Data.Machine.Concurrent.Fanout+import Data.Machine.Concurrent.Scatter+import Data.Machine.Concurrent.Wye+import Data.Machine.Concurrent.Tee++-- | Build a new 'Machine' by adding a 'Process' to the output of an+-- old 'Machine'. The upstream machine is run concurrently with+-- downstream with the aim that upstream will have a yielded value+-- ready as soon as downstream awaits. This effectively creates a+-- buffer between upstream and downstream, or source and sink, that+-- can contain up to one value.+--+-- @+-- ('<~<') :: 'Process' b c -> 'Process' a b -> 'Process' a c+-- ('<~<') :: 'Process' c d -> 'Data.Machine.Tee.Tee' a b c -> 'Data.Machine.Tee.Tee' a b d+-- ('<~<') :: 'Process' b c -> 'Machine' k b -> 'Machine' k c+-- @+(<~<) :: MonadBaseControl IO m+     => ProcessT m b c -> MachineT m k b -> MachineT m k c+mp <~< ma = racers ma mp++-- | Flipped ('<~<').+(>~>) :: MonadBaseControl IO m+     => MachineT m k b -> ProcessT m b c -> MachineT m k c+ma >~> mp = mp <~< ma++infixl 7 >~>++-- | We want the first available response.+waitEither' :: MonadBaseControl IO m+            => Maybe (Async (StM m a)) -> Async (StM m b)+            -> m (Either a b)+waitEither' Nothing y = Right <$> wait y+waitEither' (Just x) y = waitEither x y++-- | Let a source and a sink chase each other, providing an effective+-- one-element buffer between the two. The idea is to run both+-- concurrently at all times so that as soon as the sink 'Await's, we+-- have a source-yielded value to provide it. This, of course,+-- involves eagerly running the source, percolating its 'Await's up+-- the chain as soon as possible.+racers :: forall m k a b. MonadBaseControl IO m+       => MachineT m k a -> ProcessT m a b -> MachineT m k b+racers src snk = MachineT . join $+                 go <$> (Just <$> asyncRun src) <*> asyncRun snk+  where go :: Maybe (AsyncStep m k a)+           -> AsyncStep m (Is a) b+           -> m (MachineStep m k b)+        go srcA snkA =+          waitEither' srcA snkA >>= \n -> case n of+            Left (Stop :: MachineStep m k a) -> go Nothing snkA+            Left (Yield o k) -> wait snkA >>= \m -> case m of+              (Stop :: MachineStep m (Is a) b) -> return Stop+              Yield o' k' -> return . Yield o' . MachineT . flushDown k' $+                             \f -> join $ go <$> (Just <$> asyncRun k)+                                             <*> asyncRun (f o)+              Await f Refl _ -> join $ go <$> (Just <$> asyncRun k)+                                          <*> asyncRun (f o)+            Left (Await g kg fg) -> asyncAwait g kg fg $+                                    MachineT . flip go snkA . Just+            Right (Stop :: MachineStep m (Is a) b) -> return Stop+            Right (Yield o k) -> asyncRun k >>=+                                 return . Yield o . MachineT . go srcA+            Right (Await f Refl ff) -> case srcA of+              Nothing -> asyncRun ff >>= go Nothing+              Just src' -> wait src' >>= \m -> case m of+                Stop -> return Stop+                Yield o k -> join $ go <$> (Just <$> asyncRun k)+                                       <*> asyncRun (f o)+                a -> feedUp (encased a) $ \o k -> join $+                       go <$> (Just <$> asyncRun k) <*> asyncRun (f o)+        -- If we have an upstream source value ready, we must flush+        -- all available values yielded by downstream until it awaits.+        flushDown :: ProcessT m a b+                  -> ((a -> ProcessT m a b) -> m (MachineStep m k b))+                  -> m (MachineStep m k b)+        flushDown m k = runMachineT m >>= \s -> case s of+          Stop -> return Stop+          Yield o m' -> return . Yield o . MachineT $ flushDown m' k+          Await f Refl _ -> k f+        -- If downstream is awaiting an input, we must pull in all+        -- necessary upstream awaits until we have a yielded value to+        -- push downstream.+        feedUp :: MachineT m k a+               -> (a -> MachineT m k a -> m (MachineStep m k b))+               -> m (MachineStep m k b)+        feedUp m k = runMachineT m >>= \s -> case s of+          Stop -> return Stop+          Yield o m' -> k o m'+          Await g kg fg -> return $ awaitStep g kg fg (MachineT . flip feedUp k)

@@ -0,0 +1,63 @@+{- HLINT ignore -}+{- FOURMOLU_DISABLE -}+{-# LANGUAGE FlexibleContexts, GADTs, RankNTypes, ScopedTypeVariables #-}+-- | Internal helpers for taking asynchronous machine steps.+module Data.Machine.Concurrent.AsyncStep where+import Control.Concurrent.Async.Lifted (Async, async, wait)+import Control.Monad.Trans.Control (MonadBaseControl, StM)+import Data.Machine++-- | Slightly more compact notation for a 'Step'.+type MachineStep m k o = Step k o (MachineT m k o)++-- | Compact notation for a 'Step' taken asynchronously.+type AsyncStep m k o = Async (StM m (MachineStep m k o))++-- | Build an 'Await' step given a continuation that provides+-- subsequent steps. @awaitStep f sel ff k@ is like applying the+-- 'Await' constructor directly, but the continuation @k@ is used to+-- continue the machine.+--+-- @awaitStep f sel ff k = Await (k . f) sel (k ff)@+awaitStep :: (a -> d) -> k' a -> d -> (d -> r) -> Step k' b r+awaitStep f sel ff k = Await (k . f) sel (k ff)++-- | Run one step of a machine as an 'Async' operation.+asyncRun :: MonadBaseControl IO m => MachineT m k o -> m (AsyncStep m k o)+asyncRun = async . runMachineT++-- | Satisfy a downstream Await by blocking on an upstream step.+stepAsync :: forall m k k' a' d b.+             MonadBaseControl IO m+           => (forall c. k c -> k' c)+           -> AsyncStep m k a'+           -> (a' -> d)+           -> d+           -> d+           -> (AsyncStep m k a' -> d -> MachineT m k' b)+           -> MachineT m k' b+stepAsync sel src f def prev go = MachineT $ wait src >>= \u -> case u of+  Stop -> go' stopped def+  Yield a k -> go' k (f a)+  Await g kg fg -> return $ awaitStep g (sel kg) fg (MachineT . flip go' prev)+  where go' :: MachineT m k a' -> d -> m (MachineStep m k' b)+        go' k d = asyncRun k >>= runMachineT . flip go d++-- | @asyncEncased f x@ launches @x@ and provides the resulting+-- 'AsyncStep' to @f@. Turn a function on 'AsyncStep' to a funciton on+-- 'MachineT'.+asyncEncased :: MonadBaseControl IO m+             => (AsyncStep m k1 o1 -> MachineT m k o)+             -> MachineT m k1 o1+             -> MachineT m k o+asyncEncased f x = MachineT $ asyncRun x >>= runMachineT . f++-- | Similar to 'awaitStep', but for continuations that want their inputs+-- to be run asynchronously.+asyncAwait :: MonadBaseControl IO m+           => (a -> MachineT m k o)+           -> k' a+           -> MachineT m k o+           -> (AsyncStep m k o -> MachineT m k1 o1)+           -> m (Step k' b (MachineT m k1 o1))+asyncAwait f sel ff = return . awaitStep f sel ff . asyncEncased

@@ -0,0 +1,138 @@+{- HLINT ignore -}+{- FOURMOLU_DISABLE -}+ {-# LANGUAGE CPP, FlexibleContexts, GADTs, ScopedTypeVariables, TupleSections #-}+-- | Place buffers between two machines. This is most useful with+-- irregular production rates.+module Data.Machine.Concurrent.Buffer (+  -- * Blocking buffers+  bufferConnect,+  -- * Non-blocking (rolling) buffers+  rollingConnect,+  -- * Internal helpers+  mediatedConnect, BufferRoom(..)+  ) where+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Control.Applicative ((<$>), (<*>))+#endif+import Control.Concurrent.Async.Lifted (wait, waitEither)+import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad (join, (>=>))+import Data.Machine.Concurrent.AsyncStep+import Data.Machine+import Data.Sequence (ViewL(..), (|>))+import qualified Data.Sequence as S+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Data.Traversable (traverse)+#endif++-- | Drain downstream until it awaits a value, then pass the awaiting+-- step to the given function.+drain :: Monad m+      => MachineStep m k a+      -> (MachineStep m k a -> m (MachineStep m k' a))+      -> m (MachineStep m k' a)+drain z k = go z+  where go Stop = return Stop+        go (Yield o kd) = Yield o . MachineT . go <$> runMachineT kd+        go aStep = k aStep++-- | Feed upstream until it yields a value, then pass the yielded+-- value and next step to the given function.+feedToBursting :: Monad m+               => MachineStep m k a+               -> (Maybe (a, MachineT m k a) -> m (MachineStep m k b))+               -> m (MachineStep m k b)+feedToBursting z k = go z+  where go Stop = k Nothing+        go (Await f kf ff) = return $+          Await (\a -> go' (f a)) kf (go' ff)+        go (Yield o kk) = k $ Just (o, kk)+        go' step = MachineT $ runMachineT step >>= go++-- | Mediate a 'MachineT' and a 'ProcessT' with a bounded capacity+-- buffer. The source machine runs concurrently with the sink process,+-- and is only blocked when the buffer is full.+bufferConnect :: MonadBaseControl IO m+              => Int -> MachineT m k b -> ProcessT m b c -> MachineT m k c+bufferConnect n = mediatedConnect S.empty snoc view+  where snoc acc x = (if S.length acc < n - 1 then Vacancy else NoVacancy) $+                       acc |> x+        view acc = case S.viewl acc of+                     EmptyL -> Nothing+                     x :< acc' -> Just (x, acc')++-- | Mediate a 'MachineT' and a 'ProcessT' with a rolling buffer. The+-- source machine runs concurrently with the sink process and is never+-- blocked. If the sink process can not keep up with upstream, yielded+-- values will be dropped.+rollingConnect :: MonadBaseControl IO m+              => Int -> MachineT m k b -> ProcessT m b c -> MachineT m k c+rollingConnect n = mediatedConnect S.empty snoc view+  where snoc acc x = Vacancy $ S.take (n-1) acc |> x+        view acc = case S.viewl acc of+                     EmptyL -> Nothing+                     x :< acc' -> Just (x, acc')++-- | Indication if the payload value is "full" or not.+data BufferRoom a = NoVacancy a | Vacancy a deriving (Eq, Ord, Show)++-- | Mediate a 'MachineT' and a 'ProcessT' with a buffer.+--+-- @mediatedConnect z snoc view source sink@ pipes @source@ into+-- @sink@ through a buffer initialized to @z@ and updated with+-- @snoc@. Upstream is blocked if @snoc@ indicates that the buffer is+-- full after adding a new element. Downstream blocks if @view@+-- indicates that the buffer is empty. Otherwise, @view@ is expected+-- to return the next element to process and an updated buffer.+mediatedConnect :: forall m t b k c. MonadBaseControl IO m+                => t -> (t -> b -> BufferRoom t) -> (t -> Maybe (b,t))+                -> MachineT m k b -> ProcessT m b c -> MachineT m k c+mediatedConnect z snoc view src0 snk0 =+  MachineT $ do srcFuture <- asyncRun src0+                snkFuture <- asyncRun snk0+                go z (Just srcFuture) snkFuture+  where -- Wait for the next available step+        go :: t+           -> Maybe (AsyncStep m k b)+           -> AsyncStep m (Is b) c+           -> m (MachineStep m k c)+        go acc src snk = maybe (Left <$> wait snk) (waitEither snk) src >>=+                           goStep acc . either (Right . (,src)) (Left . (,snk))++        -- Kick off the next step of both the source and the sink+        goAsync :: t+                -> Maybe (MachineT m k b)+                -> ProcessT m b c+                -> m (MachineStep m k c)+        goAsync acc src snk =+          join $ go acc <$> traverse asyncRun src <*> asyncRun snk++        -- Handle whichever step is ready first+        goStep :: t  -> Either (MachineStep m k b, AsyncStep m (Is b) c)+                               (MachineStep m (Is b) c, Maybe (AsyncStep m k b))+               -> m (MachineStep m k c)+        goStep acc step = case step of+          -- @src@ stepped first+          Left (Stop, snk) -> go acc Nothing snk+          Left (Await g kg fg, snk) ->+            asyncAwait g kg fg (MachineT . flip (go acc) snk . Just)+          Left (Yield o k, snk) -> case snoc acc o of+            -- add it to the right end of the buffer+            Vacancy acc' -> asyncRun k >>= flip (go acc') snk . Just+            -- buffer was full+            NoVacancy acc' ->+              let go' snk' = do src' <- asyncRun k+                                goStep acc' (Right (snk', Just src'))+              in wait snk >>= flip drain go'++          -- @snk@ stepped first+          Right (Stop, _) -> return Stop+          Right (Yield o k, src) -> do+            return $ Yield o (MachineT $ asyncRun k >>= go acc src)+          Right (Await f Refl ff, src) ->+            case view acc of+              Nothing -> maybe (goAsync acc Nothing ff) (wait >=> demandSrc) src+              Just (x, acc') -> asyncRun (f x) >>= go acc' src+            where demandSrc = flip feedToBursting go'+                  go' Nothing = goAsync acc Nothing ff+                  go' (Just (o, k)) = goAsync acc (Just k) (f o)

@@ -0,0 +1,87 @@+{- HLINT ignore -}+{- FOURMOLU_DISABLE -}+{-# LANGUAGE CPP, FlexibleContexts, GADTs, ScopedTypeVariables #-}+-- | Provide a notion of fanout wherein a single input is passed to+-- several consumers. The consumers are run concurrently.+module Data.Machine.Concurrent.Fanout (fanout, fanoutSteps) where+import Control.Arrow (first, second)+import Control.Concurrent.Async.Lifted (Async, async, wait)+import Control.Monad (foldM)+import Control.Monad.Trans.Control (MonadBaseControl, StM)+import Data.Machine (Step(..), MachineT(..), encased, stopped, ProcessT, Is(..))+import Data.Machine.Concurrent.AsyncStep (MachineStep)+import Data.Maybe (catMaybes)+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Data.Monoid (Monoid, mempty, mconcat)+#endif+import Data.Semigroup (Semigroup(sconcat))+import Data.List.NonEmpty (NonEmpty((:|)))+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Data.Coerce (coerce)+#endif++-- | Feed a value to a 'ProcessT' at an 'Await' 'Step'. If the+-- 'ProcessT' is awaiting a value, then its next step is+-- returned. Otherwise, the original process is returned.+feed :: forall m a b. MonadBaseControl IO m+     => a -> ProcessT m a b+     -> m (Async (StM m (MachineStep m (Is a) b)))+feed x m = async $ runMachineT m >>= \(v :: MachineStep m (Is a) b) ->+             case v of+               Await f Refl _ -> runMachineT (f x)+               s -> return s++-- | Like 'Data.List.mapAccumL' but with a monadic accumulating+-- function.+mapAccumLM :: MonadBaseControl IO m+           => (acc -> x -> m (acc, y)) -> acc -> [Async (StM m x)]+           -> m (acc, [y])+mapAccumLM f z = fmap (second ($ [])) . foldM aux (z,id)+  where aux (acc,ys) x = do (yielded, nxt) <- wait x >>= f acc+                            return $ (yielded, (nxt:) . ys)++-- | Exhaust a sequence of all successive 'Yield' steps taken by a+-- 'MachineT'. Returns the list of yielded values and the next+-- (non-Yield) step of the machine.+flushYields :: Monad m+            => Step k o (MachineT m k o) -> m ([o], Maybe (MachineT m k o))+flushYields = go id+  where go rs (Yield o s) = runMachineT s >>= go ((o:) . rs)+        go rs Stop = return (rs [], Nothing)+        go rs s = return (rs [], Just $ encased s)++-- | Share inputs with each of a list of processes in lockstep. Any+-- values yielded by the processes for a given input are combined into+-- a single yield from the composite process.+fanout :: (MonadBaseControl IO m, Semigroup r)+       => [ProcessT m a r] -> ProcessT m a r+fanout [] = stopped+fanout xs = encased $ Await (MachineT . aux) Refl (fanout xs)+  where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []+                   let nxt = fanout $ catMaybes xs'+                   case rs of+                     [] -> runMachineT nxt+                     (r:rs') -> return $ Yield (sconcat $ r :| rs') nxt+        yields rs Stop = return (rs,Nothing)+        yields rs y@Yield{} = first (++ rs) <$> flushYields y+        yields rs a@Await{} = return (rs, Just $ encased a)++-- | Share inputs with each of a list of processes in lockstep. If+-- none of the processes yields a value, the composite process will+-- itself yield 'mempty'. The idea is to provide a handle on steps+-- only executed for their side effects. For instance, if you want to+-- run a collection of 'ProcessT's that await but don't yield some+-- number of times, you can use 'fanOutSteps . map (fmap (const ()))'+-- followed by a 'taking' process.+fanoutSteps :: (MonadBaseControl IO m, Monoid r)+            => [ProcessT m a r] -> ProcessT m a r+fanoutSteps [] = stopped+fanoutSteps xs = encased $ Await (MachineT . aux) Refl (fanoutSteps xs)+  where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []+                   let nxt = fanoutSteps $ catMaybes xs'+                   if null rs+                   then return $ Yield mempty nxt+                   else return $ Yield (mconcat rs) nxt+        yields rs Stop = return (rs,Nothing)+        yields rs y@Yield{} = first (++rs) <$> flushYields y+        yields rs a@Await{} = return (rs, Just $ encased a)

@@ -0,0 +1,239 @@+{- HLINT ignore -}+{- FOURMOLU_DISABLE -}+{-# LANGUAGE FlexibleContexts, GADTs, TupleSections, RankNTypes,+             ScopedTypeVariables #-}+-- | Routing for splitting and merging processing pipelines.+module Data.Machine.Concurrent.Scatter (+  scatter, mergeSum, splitSum, splitProd+  ) where+import Control.Arrow ((***))+import Control.Concurrent.Async (Async, waitAny)+import Control.Concurrent.Async.Lifted (wait, waitEither, waitBoth)+import Control.Monad ((>=>))+import Control.Monad.Base (liftBase)+import Control.Monad.Trans.Control (MonadBaseControl, restoreM, StM)+import Data.Machine+import Data.Machine.Concurrent.AsyncStep++holes :: [a] -> [[a]]+holes = go id+  where go _ [] = []+        go x (y:ys) = x ys : go (x . (y:)) ys++diff :: [a] -> [(a,[a])]+diff xs = zip xs (holes xs)++waitAnyHole :: MonadBaseControl IO m => [(Async (StM m a), [b])] -> m (a, [b])+waitAnyHole xs = do (_,(s,b)) <- liftBase $ waitAny xs'+                    fmap (,b) (restoreM s)+  where xs' = map (\(a,b) -> fmap (,b) a) xs++-- | Produces values from whichever source 'MachineT' yields+-- first. This operation may also be viewed as a /gather/ operation in+-- that all values produced by the given machines are interleaved when+-- fed downstream. Note that inputs are /not/ shared. The composite+-- machine will await an input when any constituent machine awaits an+-- input. That input will be supplied to the awaiting constituent and+-- no other.+--+-- Some examples of more specific useful types @scatter@ may be used+-- at,+--+-- @+-- scatter :: [ProcessT m a b] -> ProcessT m a b+-- scatter :: [SourceT m a] -> SourceT m a+-- @+--+-- The former may be used to stream data through a collection of+-- worker 'Process'es, the latter may be used to intersperse values+-- from a collection of sources.+scatter :: MonadBaseControl IO m => [MachineT m k o] -> MachineT m k o+scatter [] = stopped+scatter sinks = MachineT $ mapM asyncRun sinks+                 >>= waitAnyHole . diff+                 >>= uncurry go+  where go :: MonadBaseControl IO m+           => MachineStep m k o+           -> [AsyncStep m k o]+           -> m (MachineStep m k o)+        go Stop [] = return Stop+        go Stop sinks' = waitAnyHole (diff sinks') >>= uncurry go+        go (Yield o k) sinks' =+          asyncRun k >>= return . Yield o . MachineT . goWait . (:sinks')+        go (Await f fk ff) sinks' =+          asyncAwait f fk ff (MachineT . goWait . (:sinks'))+        goWait :: MonadBaseControl IO m+               => [AsyncStep m k o]+               -> m (MachineStep m k o)+        goWait = waitAnyHole . diff >=> uncurry go++-- | Similar to 'Control.Arrow.|||': split the input between two+-- processes and merge their outputs.+--+-- Connect two processes to the downstream tails of a 'Machine' that+-- produces 'Either's. The two downstream consumers are run+-- concurrently when possible. When one downstream consumer stops, the+-- other is allowed to run until it stops or the upstream source+-- yields a value the remaining consumer can not handle.+--+-- @mergeSum sinkL sinkR@ produces a topology like this,+--+-- @+--                                 sinkL+--                                /      \\+--                              a          \\+--                             /            \\+--    source -- Either a b -->                -- r -->+--                             \\            /+--                              b          /+--                               \\       /+--                                 sinkR+-- @+mergeSum :: forall m a b r. MonadBaseControl IO m+         => ProcessT m a r -> ProcessT m b r -> ProcessT m (Either a b) r+mergeSum snkL snkR = MachineT $ do sl <- asyncRun snkL+                                   sr <- asyncRun snkR+                                   go sl sr+  where go :: AsyncStep m (Is a) r+           -> AsyncStep m (Is b) r+           -> m (MachineStep m (Is (Either a b)) r)+        go sl sr = waitEither sl sr >>=+                   \(s :: Either (MachineStep m (Is a) r)+                                 (MachineStep m (Is b) r)) -> case s of+          Left Stop -> wait sr >>= runMachineT . rightOnly . encased+          Right Stop -> wait sl >>= runMachineT . leftOnly . encased++          Left (Yield o k) ->+            return . Yield o . MachineT $ asyncRun k >>= flip go sr+          Right (Yield o k) ->+            return . Yield o . MachineT $ asyncRun k >>= go sl++          Left (Await f Refl ff) ->+            return $+            Await (\u -> case u of+                           Left a -> MachineT $ asyncRun (f a) >>= flip go sr+                           Right b -> MachineT $+                                      wait sr >>= forceFeed (go sl) b . encased)+                  Refl+                  (MachineT $ asyncRun ff >>= flip go sr)+          Right (Await g Refl gg) -> return $+            Await (\u -> case u of+                           Left a ->+                             MachineT $+                             wait sl >>= forceFeed (flip go sr) a . encased+                           Right b -> MachineT $ asyncRun (g b) >>= go sl)+                  Refl+                  (MachineT $ asyncRun gg >>= go sl)++-- | Similar to 'Control.Arrow.+++': split the input between two+-- processes, retagging and merging their outputs.+--+-- The two processes are run concurrently whenever possible.+splitSum :: forall m a b c d. MonadBaseControl IO m+         => ProcessT m a b -> ProcessT m c d -> ProcessT m (Either a c) (Either b d)+splitSum snkL snkR = MachineT $ do sl <- asyncRun (fmap lft snkL)+                                   sr <- asyncRun (fmap rgt snkR)+                                   go sl sr+  where lft :: b -> Either b d+        lft = Left+        rgt :: d -> Either b d+        rgt = Right+        go :: AsyncStep m (Is a) (Either b d)+           -> AsyncStep m (Is c) (Either b d)+           -> m (MachineStep m (Is (Either a c)) (Either b d))+        go sl sr = waitEither sl sr >>=+                   \(s :: Either (MachineStep m (Is a) (Either b d))+                                 (MachineStep m (Is c) (Either b d))) -> case s of+          Left Stop -> wait sr >>= runMachineT . rightOnly . encased+          Right Stop -> wait sl >>= runMachineT . leftOnly . encased++          Left (Yield o k) ->+            return . Yield o . MachineT $ asyncRun k >>= flip go sr+          Right (Yield o k) ->+            return . Yield o . MachineT $ asyncRun k >>= go sl++          Left (Await f Refl ff) ->+            return $+            Await (\u -> case u of+                           Left a -> MachineT $ asyncRun (f a) >>= flip go sr+                           Right b -> MachineT $+                                      wait sr >>= forceFeed (go sl) b . encased)+                  Refl+                  (MachineT $ asyncRun ff >>= flip go sr)+          Right (Await g Refl gg) -> return $+            Await (\u -> case u of+                           Left a ->+                             MachineT $+                             wait sl >>= forceFeed (flip go sr) a . encased+                           Right b -> MachineT $ asyncRun (g b) >>= go sl)+                  Refl+                  (MachineT $ asyncRun gg >>= go sl)++-- | @forceFeed k x p@ runs machine @p@ until it awaits, at which+-- point it is fed @x@. The result of that feeding is asynchronously+-- run, and supplied to the continuation @k@.+forceFeed :: forall m a k b. MonadBaseControl IO m+          => (AsyncStep m (Is a) b -> m (MachineStep m k b))+          -> a+          -> ProcessT m a b+          -> m (MachineStep m k b)+forceFeed go x = aux+  where aux p = runMachineT p >>= \v -> case v of+          -- Stop -> asyncRun stopped >>= go+          Stop -> return Stop+          Yield o k -> return . Yield o . MachineT $ aux k+          Await f Refl _ -> asyncRun (f x) >>= go++-- | We have a sink for the Right output of a source, so we want to+-- keep running it as long as upstream does not yield a 'Left' which+-- we can not handle. When upstream yields a 'Left', we 'stop'.+rightOnly :: Monad m => ProcessT m b r -> ProcessT m (Either a b) r+rightOnly snk = repeatedly (await >>= either (const stop) (\x -> yield x)) ~> snk++-- | We have a sink for the Left output of a source, so we want to+-- keep running it as long as upstream does not yield a 'Right' which+-- we can not handle. When upstream yields a 'Right', we 'stop'.+leftOnly :: Monad m => ProcessT m a r -> ProcessT m (Either a b) r+leftOnly snk = repeatedly (await >>= either (\x -> yield x) (const stop)) ~> snk++-- | Connect two processes to the downstream tails of a 'Machine' that+-- produces tuples. The two downstream consumers are run+-- concurrently. When one downstream consumer stops, the entire+-- pipeline is stopped.+--+-- @splitProd sink1 sink2@ produces a topology like this,+--+-- @+--                            sink1+--                           /      \\+--                         a          \\+--                        /            \\+--    source -- (a,b) -->               -- r -->+--                        \\            /+--                         b         /+--                           \\     /+--                            sink2+-- @+splitProd :: forall m a b r. MonadBaseControl IO m+          => ProcessT m a r -> ProcessT m b r -> ProcessT m (a,b) r+splitProd snk1 snk2 = MachineT $ do s1 <- asyncRun snk1+                                    s2 <- asyncRun snk2+                                    go s1 s2+  where go :: AsyncStep m (Is a) r+           -> AsyncStep m (Is b) r+           -> m (MachineStep m (Is (a,b)) r)+        go s1 s2 = waitBoth s1 s2 >>=+                   \(ss :: (MachineStep m (Is a) r, MachineStep m (Is b) r)) -> case ss of+          (Stop, _) -> return Stop+          (_, Stop) -> return Stop+          (Yield o1 k1, Yield o2 k2) ->+            return . Yield o1 . encased $ Yield o2 $ MachineT $+            do k1' <- asyncRun k1+               k2' <- asyncRun k2+               go k1' k2'+          (Yield o k, _) ->+            return . Yield o . MachineT $ asyncRun k >>= flip go s2+          (_, Yield o k) ->+            return . Yield o . MachineT $ asyncRun k >>= go s1+          (Await f Refl ff, Await g Refl gg) ->+            return $ Await (uncurry splitProd . (f***g)) Refl (splitProd ff gg)

@@ -0,0 +1,37 @@+{- HLINT ignore -}+{- FOURMOLU_DISABLE -}+{-# LANGUAGE FlexibleContexts, GADTs, ScopedTypeVariables #-}+-- | Support for machines with two inputs from which input may be+-- drawn deterministically. In contrast to "Data.Machine.Tee", the two+-- inputs are eagerly run concurrently in this implementation.+module Data.Machine.Concurrent.Tee where+import Control.Concurrent.Async.Lifted (wait)+import Control.Monad.Trans.Control (MonadBaseControl)+import Data.Machine+import Data.Machine.Concurrent.AsyncStep++-- | Compose a pair of pipes onto the front of a Tee.+tee :: forall m a a' b b' c. MonadBaseControl IO m+    => ProcessT m a a' -> ProcessT m b b' -> TeeT m a' b' c -> TeeT m a b c+tee ma mb m = MachineT $ do srcL <- asyncRun ma+                            srcR <- asyncRun mb+                            go m (Just srcL) (Just srcR)+  where go :: TeeT m a' b' c+           -> Maybe (AsyncStep m (Is a) a')+           -> Maybe (AsyncStep m (Is b) b')+           -> m (MachineStep m (T a b) c)+        go snk srcL srcR = runMachineT snk >>= \v -> case v of+          Stop -> return Stop+          Yield o k -> return . Yield o . MachineT $ go k srcL srcR+          Await f L ff -> maybe (return Stop) wait srcL >>=+                          \(u :: MachineStep m (Is a) a') -> case u of+            Stop            -> go ff Nothing srcR+            Yield a k       -> asyncRun k >>= flip (go (f a)) srcR . Just+            Await g Refl fg ->+              asyncAwait g L fg $ MachineT . flip (go (encased v)) srcR . Just+          Await f R ff -> maybe (return Stop) wait srcR >>=+                          \(u :: MachineStep m (Is b) b') -> case u of+            Stop            -> go ff srcL Nothing+            Yield b k       -> asyncRun k >>= go (f b) srcL . Just+            Await g Refl fg ->+              asyncAwait g R fg $ MachineT . go (encased v) srcL . Just

@@ -0,0 +1,115 @@+{- HLINT ignore -}+{- FOURMOLU_DISABLE -}+{-# LANGUAGE CPP, GADTs, FlexibleContexts, RankNTypes, ScopedTypeVariables,+             TupleSections #-}+-- | Support for machines with two inputs from which input may be+-- drawn deterministically or non-deterministically. In contrast to+-- "Data.Machine.Wye", the two inputs are eagerly run concurrently in+-- this implementation.+module Data.Machine.Concurrent.Wye (wye) where+#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710+import Control.Applicative+#endif+import Control.Concurrent.Async.Lifted (wait, waitEither)+import Control.Monad.Trans.Control (MonadBaseControl)+import Data.Machine hiding (wye, (~>), (<~))+import Data.Machine.Concurrent.AsyncStep++isX :: Is a c -> Y a b c+isX Refl = X++isY :: Is b c -> Y a b c+isY Refl = Y++-- | Only the 'X' input of a 'Wye' is not yet stopped, so we may employ+-- simpler dispatch logic.+wyeOnlyX :: forall a a' b b' c m. MonadBaseControl IO m+         => AsyncStep m (Is a) a' -> WyeT m a' b' c -> WyeT m a b c+wyeOnlyX src snk = MachineT $ runMachineT snk >>= \v -> case v of+  Stop -> return Stop+  Yield o k -> return $ Yield o (wyeOnlyX src k)+  Await _ Y ff -> runMachineT $ wye stopped stopped ff+  Await f X ff -> runMachineT $ stepAsync isX src f ff (encased v) wyeOnlyX+  Await f Z ff -> runMachineT $+    stepAsync isX src (f . Left) ff (encased v) wyeOnlyX++-- | Only the 'Y' input of a 'Wye' is not yet stopped, so we may+-- employ simpler dispatch logic.+wyeOnlyY :: MonadBaseControl IO m+         => AsyncStep m (Is b) b' -> WyeT m a' b' c -> WyeT m a b c+wyeOnlyY src m = MachineT $ runMachineT m >>= \v -> case v of+  Stop -> return Stop+  Yield o k -> return $ Yield o (wyeOnlyY src k)+  Await _ X ff -> runMachineT $ wye stopped stopped ff+  Await f Y ff -> runMachineT $ stepAsync isY src f ff (encased v) wyeOnlyY+  Await f Z ff ->+    runMachineT $ stepAsync isY src (f . Right) ff (encased v) wyeOnlyY++-- | Precompose a 'Process' onto each input of a 'Wye' (or 'WyeT').+--+-- When the choice of input is free (using the 'Z' input descriptor)+-- the two sources will be interleaved.+wye :: forall m a a' b b' c.+       (MonadBaseControl IO m)+    => ProcessT m a a' -> ProcessT m b b' -> WyeT m a' b' c -> WyeT m a b c+wye ma mb m = MachineT $ do srcL <- asyncRun ma+                            srcR <- asyncRun mb+                            go True m srcL srcR+  where go :: Bool+           -> WyeT m a' b' c+           -> AsyncStep m (Is a) a'+           -> AsyncStep m (Is b) b'+           -> m (MachineStep m (Y a b) c)+        go fair snk srcL srcR = runMachineT snk >>= \v -> case v of+          Stop         -> return Stop+          Yield o k    -> return . Yield o . MachineT $ go fair k srcL srcR+          Await f X ff -> wait srcL >>=+                          \(u :: MachineStep m (Is a) a') -> case u of+            Stop -> runMachineT $ wyeOnlyY srcR ff+            Yield a k -> asyncRun k >>= flip (go fair (f a)) srcR+            Await g Refl fg ->+              asyncAwait g X fg $ MachineT . flip (go fair (encased v)) srcR+          Await f Y ff -> wait srcR >>=+                          \(u :: MachineStep m (Is b) b') -> case u of+            Stop -> runMachineT $ wyeOnlyX srcL ff+            Yield b k -> asyncRun k >>= go fair (f b) srcL+            Await h Refl fh ->+              asyncAwait h Y fh $ MachineT . go fair (encased v) srcL++          -- Wait for whoever yields first+          Await f Z _  ->+            waitFair fair srcL srcR+            >>= \(u :: Either (MachineStep m (Is a) a')+                              (MachineStep m (Is b) b')) -> case u of+            Left (Yield a k) ->+              asyncRun k >>= \srcL' -> go (not fair) (f $ Left a) srcL' srcR+            Right (Yield b k) ->+              asyncRun k >>= \srcR' -> go (not fair) (f $ Right b) srcL srcR'+            Left Stop -> runMachineT $ wyeOnlyY srcR (encased v)+            Right Stop -> runMachineT $ wyeOnlyX srcL (encased v)++            -- The first source to respond wants to await, see what+            -- the other source has to offer.+            Left la@(Await g Refl fg) ->+              wait srcR >>= \(w :: MachineStep m (Is b) b') -> case w of+                Stop -> asyncAwait g X fg $ \l' -> wyeOnlyX l' (encased v)+                Yield b k -> runMachineT $ wye (encased la) k (f $ Right b)+                ra@(Await h Refl fh) -> return $+                  Await (\c -> case c of+                                 Left a -> wye (g a) (encased ra) (encased v)+                                 Right b -> wye (encased la) (h b) (encased v))+                        Z+                        (wye fg fh $ encased v)+            Right ra@(Await h Refl fh) ->+              wait srcL >>= \(w :: MachineStep m (Is a) a') -> case w of+                Stop -> asyncAwait h Y fh $ \r' -> wyeOnlyY r' (encased v)+                Yield a k -> runMachineT $ wye k (encased ra) (f $ Left a)+                la@(Await g Refl fg) -> return $+                  Await (\c -> case c of+                                 Left a -> wye (g a) (encased ra) (encased v)+                                 Right b -> wye (encased la) (h b) (encased v))+                        Z+                        (wye fg fh $ encased v)+          where waitFair True l r = waitEither l r+                waitFair False l r = either Right Left <$> waitEither r l+

@@ -0,0 +1,24 @@+{- HLINT ignore -}+{- FOURMOLU_DISABLE -}+-- | Slow producers down to run at desired rates.+module Data.Machine.Regulated where+import Control.Concurrent (threadDelay)+import Control.Monad (when)+import Control.Monad.IO.Class (MonadIO(..))+import Data.Machine.Plan+import Data.Machine.Process+import Data.Machine.Type+import Data.Time.Clock (getCurrentTime, diffUTCTime)++-- | A pass-through process rate-limited to the given inter-step+-- period in seconds. This may be used to slow down an upstream+-- producer; it can not speed things up.+regulated :: MonadIO m => Double -> ProcessT m a a+regulated target = construct $ liftIO getCurrentTime >>= go 0+  where go dt prevT =+          do await >>= \x -> yield x+             t <- liftIO getCurrentTime+             let e = target - realToFrac (diffUTCTime t prevT)+                 dt' = dt + 0.5 * e+             when (dt' > 0) (liftIO . threadDelay . round $ dt' * 1000000)+             go dt' t