Merge pull request #538 from symbiont-io/journal-latency

feat(sut): try to add latency to metrics (still not working)

doc/demo-journal/slides-journal.md

@@ -104,7 +104,7 @@ nocite: |
 
 # Design of the journal
 
-* Heavily inspired by Martin "LMAX" Thompson et al's Aeron
+* Heavily inspired by Martin "LMAX" Thompson et al's Aeron log buffer
 * Three (virtual) files (clean, active, dirty)
 * Circular buffer implemented on top of `mmap`ed byte array
 * `recv` zero-copied straight to byte array (and persisted)

src/sut/dumblog/bench/Common.hs

@@ -39,10 +39,10 @@ rEAD_FREQUENCY :: Int
 rEAD_FREQUENCY = 80
 
 nUM_OF_CLIENTS :: Int
-nUM_OF_CLIENTS = 8
+nUM_OF_CLIENTS = 400
 
 iTERATIONS :: Int
-iTERATIONS = 10_300
+iTERATIONS = 200
 
 vALUE_TO_WRITE :: ByteString
 vALUE_TO_WRITE = LBS.pack "Dumblog"

src/sut/dumblog/src/Dumblog/Journal/Codec.hs

@@ -1,24 +1,41 @@
-{-# LANGUAGE DeriveGeneric, DeriveFunctor#-}
+{-# LANGUAGE DeriveFunctor #-}
+{-# LANGUAGE DeriveGeneric #-}
+{-# LANGUAGE DerivingStrategies #-}
+
 module Dumblog.Journal.Codec where
 
 import Data.Binary (Binary)
 import qualified Data.Binary as Binary
 import Data.ByteString (ByteString)
 import qualified Data.ByteString.Lazy as LBS
+import Data.Int (Int64)
+import Data.Time (UTCTime, getCurrentTime, nominalDiffTimeToSeconds)
+import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds)
+import GHC.Generics (Generic)
 
-import GHC.Generics(Generic)
+------------------------------------------------------------------------
 
 data Envelope a = Envelope
-  { eLength :: Int
-  , eContent :: a
-  } deriving (Functor, Generic)
+  { eLength :: !Int
+  , eContent :: !a
+  , eArrival :: !Int64 -- Nano seconds since epoch.
+  } deriving stock (Functor, Generic)
 
 instance Binary a => Binary (Envelope a) where
 
 encode :: Binary a => Envelope a -> ByteString
 encode e = LBS.toStrict (Binary.encode e)
 
--- this is guaranteed not to copy the bytestring
--- but we should probably allow this to fail
+-- This is guaranteed not to copy the bytestring but we should probably
+-- allow this to fail.
 decode :: Binary a => ByteString -> Envelope a
 decode bs = Binary.decode $ LBS.fromStrict bs
+
+nanosSinceEpoch :: UTCTime -> Int64
+nanosSinceEpoch =
+  floor . (1e9 *) . nominalDiffTimeToSeconds . utcTimeToPOSIXSeconds
+
+getCurrentNanosSinceEpoch :: IO Int64
+getCurrentNanosSinceEpoch = do
+  now <- getCurrentTime
+  return (nanosSinceEpoch now)

src/sut/dumblog/src/Dumblog/Journal/FrontEnd.hs

@@ -16,14 +16,14 @@ import qualified Network.Wai as Wai
 import Network.Wai.Handler.Warp
 import System.Timeout (timeout)
 
-import Journal.Types (Journal)
 import Journal.Internal.Metrics (incrCounter)
 import qualified Journal.MP as Journal
+import Journal.Types (Journal)
 
 import Dumblog.Journal.Blocker
 import Dumblog.Journal.Codec
-import Dumblog.Journal.Types
 import Dumblog.Journal.Metrics
+import Dumblog.Journal.Types
 
 ------------------------------------------------------------------------
 
@@ -53,7 +53,8 @@ httpFrontend journal metrics (FrontEndInfo blocker) req respond = do
       respond $ Wai.responseLBS status400 [] err
     Right cmd -> do
       key <- newKey blocker
-      let env = encode (Envelope (sequenceNumber key) cmd)
+      now <- return 0 -- getCurrentNanosSinceEpoch
+      let env = encode (Envelope (sequenceNumber key) cmd now)
       res <- Journal.appendBS journal env
       res' <- case res of
         Left err -> do

src/sut/dumblog/src/Dumblog/Journal/Main.hs

@@ -102,7 +102,7 @@ collectAll jour = do
       pure []
     Just entry -> do
       putStrLn "[collect] Found an entry"
-      let Envelope _key cmd = decode entry
+      let Envelope _key cmd _arrivalTime = decode entry
       cmds <- collectAll jour
       pure $ cmd : cmds
 

src/sut/dumblog/src/Dumblog/Journal/Metrics.hs

@@ -1,16 +1,19 @@
 module Dumblog.Journal.Metrics where
 
-import Journal.Internal.Metrics (MetricsSchema, Metrics) -- should maybe be moved to separate package
+import Journal.Internal.Metrics (Metrics, MetricsSchema)
 import qualified Journal.Internal.Metrics as Metrics
 
+------------------------------------------------------------------------
+
 data DumblogCounters
   = CurrentNumberTransactions
   | QueueDepth
   | ErrorsEncountered
   deriving (Eq, Show, Enum, Bounded)
 
 data DumblogHistograms
-  = ServiceTimeReads
+  = Latency
+  | ServiceTimeReads
   | ServiceTimeWrites
   deriving (Eq, Show, Enum, Bounded)
 

src/sut/dumblog/src/Dumblog/Journal/Types.hs

@@ -1,11 +1,14 @@
 {-# LANGUAGE DeriveGeneric #-}
+
 module Dumblog.Journal.Types where
 
 import Data.Binary (Binary)
 import Data.ByteString (ByteString)
 import qualified Data.ByteString.Lazy as LBS
 import GHC.Generics (Generic)
 
+------------------------------------------------------------------------
+
 data Command
   = Write ByteString
   | Read Int

src/sut/dumblog/src/Dumblog/Journal/Worker.hs

@@ -62,7 +62,10 @@ worker journal metrics (WorkerInfo blocker snapshotFile eventCount untilSnapshot
         { Nothing -> return (ev, s)
         ; Just entry -> do
             Metrics.decrCounter_ metrics QueueDepth 1
-            let Envelope key cmd = decode entry
+            let Envelope key cmd arrivalTime = decode entry
+            -- now <- getCurrentNanosSinceEpoch
+            -- Convert from nano s to µs with `* 10^3`.
+            -- Metrics.measure metrics Latency (realToFrac ((now - arrivalTime) * 1000))
             timeIt metrics cmd $ do
               {- // in case of decode error
                   Metrics.incrCounter metrics ErrorsEncountered 1

src/sut/dumblog/src/Dumblog/Metrics/Main.hs

@@ -14,7 +14,7 @@ import Text.Printf (printf)
 import Dumblog.Journal.Main
 import Dumblog.Journal.Metrics
 import Journal (journalMetadata)
-import Journal.Internal.Metrics
+import Journal.Internal.Metrics hiding (Latency)
 import Journal.Types
 
 ------------------------------------------------------------------------
@@ -46,10 +46,12 @@ metricsMain = do
       metrics <- newMetrics dumblogSchema dUMBLOG_METRICS
       eMeta   <- journalMetadata dUMBLOG_JOURNAL dumblogOptions
 
+      -- Only needed on MacOS it seems.
       msyncMetrics metrics
       either (const (return ())) msyncMetadata eMeta
 
-      putStrLn ansiClearScreen
+      putStr (ansiClearScreen ++ ansiGoto 1 1)
+      -- displayLatency metrics
       displayServiceTime metrics
       displayQueueDepth metrics
       ts' <- displayThroughput metrics ts
@@ -64,6 +66,27 @@ metricsMain = do
 ansiClearScreen :: String
 ansiClearScreen = "\ESC[2J"
 
+ansiGoto :: Int -> Int -> String
+ansiGoto x y    = "\ESC[" ++ show y ++ ";" ++ show x ++ "H"
+
+displayLatency :: DumblogMetrics -> IO ()
+displayLatency metrics = do
+  mMin  <- percentile metrics Latency 0
+  mMed  <- percentile metrics Latency 50
+  m90   <- percentile metrics Latency 90
+  m99   <- percentile metrics Latency 99
+  m999  <- percentile metrics Latency 99.9
+  m9999 <- percentile metrics Latency 99.99
+  mMax  <- percentile metrics Latency 100
+  printf "%-25.25s\n" "Latency"
+  printf "  min   %10.2f µs\n" (fromMaybe 0 mMin)
+  printf "  med   %10.2f µs\n" (fromMaybe 0 mMed)
+  printf "  90    %10.2f µs\n" (fromMaybe 0 m90)
+  printf "  99    %10.2f µs\n" (fromMaybe 0 m99)
+  printf "  99.9  %10.2f µs\n" (fromMaybe 0 m999)
+  printf "  99.99 %10.2f µs\n" (fromMaybe 0 m9999)
+  printf "  max   %10.2f µs\n" (fromMaybe 0 mMax)
+
 displayServiceTime :: DumblogMetrics -> IO ()
 displayServiceTime metrics = do
   mMin  <- percentile metrics ServiceTimeWrites 0
@@ -92,7 +115,7 @@ displayServiceTime metrics = do
   readSum  <- realToFrac <$> metricsSum metrics ServiceTimeReads  :: IO Double
   writeCnt <- count metrics ServiceTimeWrites
   readCnt  <- count metrics ServiceTimeReads
-  printf "  sum %10.2f s%15.2fs\n" (writeSum / 1e6) (readSum / 1e6)
+  printf "  sum   %10.2f s %15.2fs\n" (writeSum / 1e6) (readSum / 1e6)
   let totalCnt :: Double
       totalCnt = realToFrac (writeCnt + readCnt)
   printf "  count %7d (%2.0f%%) %10d (%2.0f%%)\n"
@@ -164,5 +187,8 @@ displayErrors metrics = do
 
 displayUtilisation :: DumblogMetrics -> ThroughputState -> IO ()
 displayUtilisation metrics ts = do
-  serviceTimeAvg <- metricsAvg metrics ServiceTimeWrites -- XXX: what about reads?
-  printf "\nUtilisation: %.2f\n" (throughputAvg ts * serviceTimeAvg * 1e-6)
+  serviceTimeWAvg <- metricsAvg metrics ServiceTimeWrites
+  serviceTimeRAvg <- metricsAvg metrics ServiceTimeReads
+  printf "\nUtilisation: %.2f\n"
+    -- Throughput uses seconds and service time uses µs, hence the `* 10^-6`.
+    (throughputAvg ts * (serviceTimeWAvg + serviceTimeRAvg) * 1e-6)

src/sut/dumblog/src/Dumblog/SQLite/FrontEnd.hs

@@ -65,5 +65,5 @@ runFrontEnd queue port mReady = runSettings settings (httpFrontend queue)
     settings
       = setPort port
       $ maybe id (\ready -> setBeforeMainLoop (putMVar ready ())) mReady
-      $ setOnClose (\addr -> putStrLn ("closing: " ++ show addr))
+      -- $ setOnClose (\addr -> putStrLn ("closing: " ++ show addr))
       $ defaultSettings