From 129fb2ed74436ae646bc0a9fde097c61e90e9b9a Mon Sep 17 00:00:00 2001 From: Karl Knutsson Date: Mon, 16 Feb 2026 13:55:27 +0100 Subject: [PATCH 1/3] cleanup ns --- network-mux/demo/mux-leios-demo.sh | 1 + 1 file changed, 1 insertion(+) diff --git a/network-mux/demo/mux-leios-demo.sh b/network-mux/demo/mux-leios-demo.sh index 55adad00276..6f1b9077bb5 100755 --- a/network-mux/demo/mux-leios-demo.sh +++ b/network-mux/demo/mux-leios-demo.sh @@ -41,6 +41,7 @@ cleanup_netns() { # kill server and tcpdump sudo kill -9 $pid done + sudo ip netns delete ns$i done # for i in 1 2 3; do sudo ip netns del ns$i; done From bec421bf6733d1c2cae7f1283f129087558ab1d2 Mon Sep 17 00:00:00 2001 From: Karl Knutsson Date: Tue, 17 Feb 2026 11:02:05 +0100 Subject: [PATCH 2/3] mux-leios-demo: exit if cabal build fails Exit if cabal build fails. --- network-mux/demo/mux-leios-demo.sh | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/network-mux/demo/mux-leios-demo.sh b/network-mux/demo/mux-leios-demo.sh index 6f1b9077bb5..2f474acdf7e 100755 --- a/network-mux/demo/mux-leios-demo.sh +++ b/network-mux/demo/mux-leios-demo.sh @@ -164,7 +164,10 @@ setup_bridge() { setup_bridge -cabal build exe:mux-leios-demo +if ! cabal build exe:mux-leios-demo; then + echo "cabal build failed; exiting." >&2 + exit 1 +fi CMD=$(cabal list-bin exe:mux-leios-demo) # For debuging throuput shaping From 085ae9ae233888d8bd61a13681c0b61fab26beb1 Mon Sep 17 00:00:00 2001 From: Karl Knutsson Date: Mon, 16 Feb 2026 09:20:14 +0100 Subject: [PATCH 3/3] Add a micro second timestamp in each payload Add a micro second timestamp in each payload for latency calculations. If a message is too small to contain a timestamp no timestamp is included and its tracked as "missing". Use tdigest for calculating min, max, mean, median, p90, p95 and p99 stddev delay in ms. Shows delat/latency stats, as speed and total speed. --- network-mux/demo/mux-leios-demo.hs | 227 ++++++++++++++++++++++++++--- network-mux/demo/mux-leios-demo.sh | 2 +- network-mux/network-mux.cabal | 4 +- 3 files changed, 213 insertions(+), 20 deletions(-) diff --git a/network-mux/demo/mux-leios-demo.hs b/network-mux/demo/mux-leios-demo.hs index 8dd7ca05677..b8086f3e44e 100644 --- a/network-mux/demo/mux-leios-demo.hs +++ b/network-mux/demo/mux-leios-demo.hs @@ -12,16 +12,25 @@ -- module Main (main) where +import Data.Binary.Get qualified as Bin +import Data.Binary.Put qualified as Bin import Data.ByteString (ByteString) +import Data.ByteString qualified as BS import Data.ByteString.Char8 qualified as BSC +import Data.ByteString.Lazy qualified as BL import Data.IP (IP) import Data.IP qualified as IP +import Data.TDigest (TDigest, insert, maximumValue, mean, minimumValue, quantile, stddev, tdigest) +import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds) +import Data.Word (Word64) +import Text.Printf (printf) import Text.Read (readMaybe) import Control.Concurrent (forkIO) -import Control.Concurrent.STM (atomically) +import Control.Concurrent.STM (TVar, atomically, modifyTVar', newTVarIO, readTVar) import Control.Exception import Control.Monad +import Control.Monad.Class.MonadTime.SI (getCurrentTime) import Control.Tracer import System.Environment qualified as SysEnv @@ -43,6 +52,14 @@ data ClientType = Sequential | Bursty unusedValue :: a unusedValue = error "unused" +data LatencyStats = LatencyStats { + lsCount :: !Int, + lsDigest :: !(TDigest 5), + lsRespBytes :: !Word64, + lsFirstRespTsUs :: !Word64, + lsLastRxUs :: !Word64 + } + main :: IO () main = do args <- SysEnv.getArgs @@ -214,8 +231,8 @@ serverWorkerSequential bearer len1 len2 = do go :: Char -> ReqRespServer ByteString ByteString IO Int go c = ReqRespServer { - recvMsgReq = \(!_) -> - let msg = BSC.replicate n c in + recvMsgReq = \(!_) -> do + msg <- makeTimestampedPayload n c pure (msg, go (succ c)), recvMsgDone = pure n } @@ -252,13 +269,14 @@ serverWorkerBursty bearer (n1, n2) len1 len2 = do :: Int -> Int -> ReqRespServerBurst ByteString ByteString IO Int - serverReqResp n len = ReqRespServerBurst $ \_ -> pure (go n minBound) + serverReqResp n len = ReqRespServerBurst $ \_ -> go n minBound where - go :: Int -> Char -> ReqRespServerLoop ByteString IO Int - go m c | m > 0 = - SendMsgResp (BSC.replicate len c) (return $ go (m-1) (succ c)) + go :: Int -> Char -> IO (ReqRespServerLoop ByteString IO Int) + go m c | m > 0 = do + msg <- makeTimestampedPayload len c + return $ SendMsgResp msg (go (m - 1) (succ c)) | otherwise = - SendMsgDoneServer (pure n) + return $ SendMsgDoneServer (pure n) -- @@ -301,6 +319,10 @@ clientWorkerSequential -- ^ number of requests to send over `MiniProtocolNum 3` -> IO () clientWorkerSequential bearer len n1 n2 = do + praosStats <- newTVarIO newLatencyStats + praosMissing <- newTVarIO 0 + leiosStats <- newTVarIO newLatencyStats + leiosMissing <- newTVarIO 0 mux <- Mx.new Mx.nullTracers (protocols InitiatorDirectionOnly) void $ forkIO $ do awaitResult1 <- @@ -309,38 +331,51 @@ clientWorkerSequential bearer len n1 n2 = do (MiniProtocolNum 2) InitiatorDirectionOnly StartEagerly - (\chan -> runClientBin (reqrespTracer "client:praos") chan (clientReqResp '0' n1)) + (\chan -> runClientBin (reqrespTracer "client:praos") chan + (clientReqResp praosStats praosMissing '0' n1)) awaitResult2 <- runMiniProtocol mux (MiniProtocolNum 3) InitiatorDirectionOnly StartEagerly - (\chan -> runClientBin (reqrespTracer "client:leios") chan (clientReqResp '1' n2)) + (\chan -> runClientBin (reqrespTracer "client:leios") chan + (clientReqResp leiosStats leiosMissing '1' n2)) -- wait for both mini-protocols to finish results <- atomically $ (,) <$> awaitResult1 <*> awaitResult2 debugPutStrLn_ $ "client results: " ++ show results + reportStats "praos" praosStats praosMissing + reportStats "leios" leiosStats leiosMissing `finally` Mx.stop mux Mx.run mux bearer where clientReqResp - :: Char + :: TVar LatencyStats + -> TVar Int + -> Char -> Int -> ReqRespClient ByteString ByteString IO Int - clientReqResp c n = go n + clientReqResp statsVar missingVar c n = go n where !msg = BSC.replicate len c go :: Int -> ReqRespClient ByteString ByteString IO Int go m | m <= 0 = SendMsgDone (pure n) - go m = SendMsgReq msg (\_ -> pure $ go (m-1)) + go m = + SendMsgReq msg (\rsp -> do + recordLatency statsVar missingVar rsp + pure $ go (m-1)) clientWorkerBursty :: Mx.Bearer IO -> IO () clientWorkerBursty bearer = do + praosStats <- newTVarIO newLatencyStats + praosMissing <- newTVarIO 0 + leiosStats <- newTVarIO newLatencyStats + leiosMissing <- newTVarIO 0 mux <- Mx.new Mx.nullTracers (protocols InitiatorDirectionOnly) void $ forkIO $ do awaitResult1 <- @@ -349,28 +384,184 @@ clientWorkerBursty bearer = do (MiniProtocolNum 2) InitiatorDirectionOnly StartEagerly - (\chan -> runClientBurstBin (reqrespTracer "client:praos") chan clientReqResp) + (\chan -> runClientBurstBin (reqrespTracer "client:praos") chan + (clientReqResp praosStats praosMissing)) awaitResult2 <- runMiniProtocol mux (MiniProtocolNum 3) InitiatorDirectionOnly StartEagerly - (\chan -> runClientBurstBin (reqrespTracer "client:leios") chan clientReqResp) + (\chan -> runClientBurstBin (reqrespTracer "client:leios") chan + (clientReqResp leiosStats leiosMissing)) -- wait for both mini-protocols to finish results <- atomically $ (,) <$> awaitResult1 <*> awaitResult2 debugPutStrLn_ $ "client results: " ++ show results + reportStats "praos" praosStats praosMissing + reportStats "leios" leiosStats leiosMissing `finally` Mx.stop mux Mx.run mux bearer where clientReqResp - :: ReqRespClientBurst ByteString ByteString IO Int - clientReqResp = SendMsgReqBurst (BSC.replicate 10 '\NUL') (go 0) + :: TVar LatencyStats + -> TVar Int + -> ReqRespClientBurst ByteString ByteString IO Int + clientReqResp statsVar missingVar = SendMsgReqBurst (BSC.replicate 10 '\NUL') (go 0) where go :: Int -> ReqRespClientLoop ByteString IO Int go !count = AwaitResp { handleMsgDone = pure count - , handleMsgResp = \(!_) -> pure (go (count + 1)) + , handleMsgResp = \(!rsp) -> do + recordLatency statsVar missingVar rsp + return $ go (count + 1) } + +newLatencyStats :: LatencyStats +newLatencyStats = + LatencyStats { + lsCount = 0 + , lsDigest = tdigest [] + , lsRespBytes = 0 + , lsFirstRespTsUs = 0 + , lsLastRxUs = 0 + } + +updateLatencyStats :: Word64 -> Word64 -> Word64 -> Word64 -> LatencyStats -> LatencyStats +updateLatencyStats us bytes respTs nowUs LatencyStats{lsCount, lsDigest, lsRespBytes, lsFirstRespTsUs} = + let !lsCount' = lsCount + 1 + !lsDigest' = insert (fromIntegral us :: Double) lsDigest + !lsRespBytes' = + if lsRespBytes <= 0 + then bytes + else lsRespBytes + !lsFirstRespTsUs' = + if lsFirstRespTsUs <= 0 + then respTs + else lsFirstRespTsUs + in + LatencyStats { + lsCount = lsCount' + , lsDigest = lsDigest' + , lsRespBytes = lsRespBytes' + , lsFirstRespTsUs = lsFirstRespTsUs' + , lsLastRxUs = nowUs + } + +timestampSize :: Int +timestampSize = 8 + +getWallClockTimeUs :: IO Word64 +getWallClockTimeUs = do + now <- getCurrentTime + let ns = floor (utcTimeToPOSIXSeconds now * 1_000_000) :: Integer + return $ fromInteger ns + +encodeTimeStamp :: Word64 -> ByteString +encodeTimeStamp ts = + BL.toStrict (Bin.runPut (Bin.putWord64be ts)) + +extractTimestamp :: ByteString -> Maybe Word64 +extractTimestamp bs + | BS.length bs < timestampSize = Nothing + | otherwise = + Just $ Bin.runGet Bin.getWord64be + (BL.fromStrict (BS.take timestampSize bs)) + +makeTimestampedPayload :: Int -> Char -> IO ByteString +makeTimestampedPayload len c + | len < timestampSize = pure (BSC.replicate len c) + | otherwise = do + ts <- getWallClockTimeUs + let tsBytes = encodeTimeStamp ts + return $ tsBytes <> BSC.replicate (len - timestampSize) c + +updateLastRx :: Word64 -> LatencyStats -> LatencyStats +updateLastRx nowUs stats = stats { lsLastRxUs = nowUs } + +recordLatency :: TVar LatencyStats -> TVar Int -> ByteString -> IO () +recordLatency statsVar missingVar rsp = do + now <- getWallClockTimeUs + case extractTimestamp rsp of + Nothing -> atomically $ do + modifyTVar' missingVar succ + modifyTVar' statsVar (updateLastRx now) + Just ts -> do + let latency = max 0 $ now - ts + bytes = fromIntegral (BS.length rsp) + atomically $ modifyTVar' statsVar (updateLatencyStats latency bytes ts now) + +reportStats :: String -> TVar LatencyStats -> TVar Int -> IO () +reportStats label statsVar missingVar = do + (stats, missing) <- atomically $ (,) <$> readTVar statsVar <*> readTVar missingVar + let count = lsCount stats + respBytes = lsRespBytes stats + firstRespTsUs = lsFirstRespTsUs stats + lastRxUs = lsLastRxUs stats + durationUs :: Word64 + durationUs = + if firstRespTsUs > 0 && lastRxUs > firstRespTsUs + then lastRxUs - firstRespTsUs + else 0 + durationMs :: Double + durationMs = fromIntegral durationUs / 1000 + overallMbps :: Double + overallMbps = + if durationUs > 0 && respBytes > 0 && count > 0 + then (fromIntegral respBytes * fromIntegral count * 8) / fromIntegral durationUs + else 0 + bitsPerResp :: Double + bitsPerResp = fromIntegral (respBytes * 8) + speedFromUs :: Real a => a -> Double + speedFromUs us = + let usD = realToFrac us :: Double in + if usD <= 0 || respBytes == 0 + then 0 + else bitsPerResp / usD + if count <= 0 + then do + printf "%s latency: count=0 missing=%d\n" label missing + printf "%s speed: count=0 min=0.000Mbps mean=0.000Mbps median=0.000Mbps p90=0.000Mbps p95=0.000Mbps p99=0.000Mbps max=0.000Mbps stddev=0.000Mbps overall=%.3fMbps duration=%.3fms\n" + label overallMbps durationMs + else do + let td = lsDigest stats + minUs = minimumValue td + maxUs = maximumValue td + stddevUs = case stddev td of + Nothing -> 0 + Just s -> s + meanUs = case mean td of + Nothing -> 0 + Just m -> m + medianUs = quantileUs 0.5 td + p90Us = quantileUs 0.9 td + p95Us = quantileUs 0.95 td + p99Us = quantileUs 0.99 td + minMs = toMs minUs + maxMs = toMs maxUs + stddevMs = toMs stddevUs + meanMs = toMs meanUs + medianMs = toMs medianUs + p90Ms = toMs p90Us + p95Ms = toMs p95Us + p99Ms = toMs p99Us + minSp = speedFromUs maxUs + maxSp = speedFromUs minUs + stddevSp = speedFromUs stddevUs + meanSp = speedFromUs meanUs + medianSp = speedFromUs medianUs + p90Sp = speedFromUs p90Us + p95Sp = speedFromUs p95Us + p99Sp = speedFromUs p99Us + printf "%s latency: count=%d min=%.3fms mean=%.3fms median=%.3fms p90=%.3fms p95=%.3fms p99=%.3fms max=%.3fms stddev=%.3fms missing=%d\n" + label count minMs meanMs medianMs p90Ms p95Ms p99Ms maxMs stddevMs missing + printf "%s speed: count=%d min=%.3fMbps mean=%.3fMbps median=%.3fMbps p90=%.3fMbps p95=%.3fMbps p99=%.3fMbps max=%.3fMbps stddev=%.3fMbps\n" + label count minSp meanSp medianSp p90Sp p95Sp p99Sp maxSp stddevSp + printf "%s: overall=%.3fMbps duration=%.3fs\n" + label overallMbps (durationMs / 1000) + where + toMs us = us / 1000 + quantileUs q td = case quantile q td of + Nothing -> 0 + Just w -> w diff --git a/network-mux/demo/mux-leios-demo.sh b/network-mux/demo/mux-leios-demo.sh index 2f474acdf7e..4a7097cc27c 100755 --- a/network-mux/demo/mux-leios-demo.sh +++ b/network-mux/demo/mux-leios-demo.sh @@ -13,7 +13,7 @@ LEIOS_BLOCK_SIZE=$((($MUX_SDU - 5) * 1000)) PRAOS_BLOCK_SIZE=$((($MUX_SDU - 5) * 10)) # client's configuration -NUM_LEIOS_REQUESTS=20 +NUM_LEIOS_REQUESTS=10 NUM_PRAOS_REQUESTS=1000 # $(($NUM_LEIOS_REQUESTS * $LEIOS_BLOCK_SIZE / $PRAOS_BLOCK_SIZE)) # network shaping parameters diff --git a/network-mux/network-mux.cabal b/network-mux/network-mux.cabal index 6c1d55cd9f0..9f6e09ff3ce 100644 --- a/network-mux/network-mux.cabal +++ b/network-mux/network-mux.cabal @@ -220,13 +220,15 @@ executable mux-leios-demo bytestring, cborg, contra-tracer, - io-classes, + io-classes:{io-classes, si-timers}, iproute, network, network-mux, primitive, serialise, stm, + tdigest, + time, ghc-options: -rtsopts