diff --git a/bench/MemBench.hs b/bench/MemBench.hs new file mode 100644 index 000000000..e1d432f79 --- /dev/null +++ b/bench/MemBench.hs @@ -0,0 +1,452 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedLists #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE PatternSynonyms #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} +{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-} + +-- | Memory-leak load driver for the SMP server. +-- +-- Starts an in-process SMP server (beta.2 code) and hammers a chosen command +-- path in a churn loop, printing GHC live-heap residency (measured after a +-- forced major GC) every checkpoint. A path whose residency climbs with the +-- iteration count is leaking; a flat path is clean. +-- +-- Usage: smp-mem-bench +-- phases: plain | svc | svcrace | ntf +module Main (main) where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (concurrently_, mapConcurrently_, withAsync) +import Control.Logger.Simple (LogConfig (..), LogLevel (..), setLogLevel, withGlobalLogging) +import Control.Concurrent.STM +import Control.Monad +import Crypto.Random (ChaChaDRG) +import qualified Data.ByteString.Char8 as B +import Data.ByteString.Char8 (ByteString) +import Data.List.NonEmpty (NonEmpty (..)) +import qualified Data.X509.Validation as XV +import GHC.Stats +import SMPClient +import qualified Simplex.Messaging.Crypto as C +import Simplex.Messaging.Protocol +import Simplex.Messaging.Server.Env.STM (AStoreType (..), ServerConfig (notificationExpiration)) +import Simplex.Messaging.Server.Expiration (ExpirationConfig (..)) +import Simplex.Messaging.Server.MsgStore.Types (SMSType (..), SQSType (..)) +import Simplex.Messaging.Transport +import Simplex.Messaging.Transport.Credentials (genCredentials, tlsCredentials) +import System.Environment (getArgs, lookupEnv) +import System.Mem (performMajorGC) +import System.Timeout (timeout) +import Text.Printf (printf) + +type H = THandleSMP TLS 'TClient + +-- store config: PostgreSQL (matches production) when built with -fserver_postgres, else journal +benchCfg :: AServerConfig +#if defined(dbServerPostgres) +benchCfg = cfgMS (ASType SQSPostgres SMSPostgres) +#else +benchCfg = cfg +#endif + +-- command helpers (copied from tests/ServerTests.hs) ------------------------ + +pattern Resp :: CorrId -> QueueId -> BrokerMsg -> Transmission (Either ErrorType BrokerMsg) +pattern Resp corrId queueId command <- (corrId, queueId, Right command) + +pattern New :: RcvPublicAuthKey -> RcvPublicDhKey -> Command 'Creator +pattern New rPub dhPub = NEW (NewQueueReq rPub dhPub Nothing SMSubscribe (Just (QRMessaging Nothing)) Nothing) + +pattern New0 :: RcvPublicAuthKey -> RcvPublicDhKey -> Command 'Creator +pattern New0 rPub dhPub = NEW (NewQueueReq rPub dhPub Nothing SMOnlyCreate (Just (QRMessaging Nothing)) Nothing) + +pattern Ids :: RecipientId -> SenderId -> RcvPublicDhKey -> BrokerMsg +pattern Ids rId sId srvDh <- IDS (QIK rId sId srvDh _sndSecure _linkId Nothing Nothing) + +pattern Ids_ :: RecipientId -> SenderId -> RcvPublicDhKey -> ServiceId -> BrokerMsg +pattern Ids_ rId sId srvDh serviceId <- IDS (QIK rId sId srvDh _sndSecure _linkId (Just serviceId) Nothing) + +pattern Msg :: MsgId -> MsgBody -> BrokerMsg +pattern Msg msgId body <- MSG RcvMessage {msgId, msgBody = EncRcvMsgBody body} + +_SEND :: MsgBody -> Command 'Sender +_SEND = SEND noMsgFlags + +_SEND' :: MsgBody -> Command 'Sender +_SEND' = SEND MsgFlags {notification = True} + +sendRecv :: forall p. PartyI p => H -> (Maybe TAuthorizations, ByteString, EntityId, Command p) -> IO (Transmission (Either ErrorType BrokerMsg)) +sendRecv h@THandle {params} (sgn, corrId, qId, cmd) = do + let TransmissionForAuth {tToSend} = encodeTransmissionForAuth params (CorrId corrId, qId, cmd) + Right () <- tPut1 h (sgn, tToSend) + tGet1 h + +signSendRecv :: forall p. PartyI p => H -> C.APrivateAuthKey -> (ByteString, EntityId, Command p) -> IO (Transmission (Either ErrorType BrokerMsg)) +signSendRecv h pk t = do + [r] <- signSendRecv_ h pk Nothing t + pure r + +serviceSignSendRecv :: forall p. PartyI p => H -> C.APrivateAuthKey -> C.PrivateKeyEd25519 -> (ByteString, EntityId, Command p) -> IO (Transmission (Either ErrorType BrokerMsg)) +serviceSignSendRecv h pk serviceKey t = do + [r] <- signSendRecv_ h pk (Just serviceKey) t + pure r + +signSendRecv_ :: forall p. PartyI p => H -> C.APrivateAuthKey -> Maybe C.PrivateKeyEd25519 -> (ByteString, EntityId, Command p) -> IO (NonEmpty (Transmission (Either ErrorType BrokerMsg))) +signSendRecv_ h@THandle {params} (C.APrivateAuthKey a pk) serviceKey_ (corrId, qId, cmd) = do + let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth params (CorrId corrId, qId, cmd) + Right () <- tPut1 h (authorize tForAuth, tToSend) + tGetClient h + where + authorize t = (,(`C.sign'` t) <$> serviceKey_) <$> case a of + C.SEd25519 -> Just . TASignature . C.ASignature C.SEd25519 $ C.sign' pk t' + C.SEd448 -> Just . TASignature . C.ASignature C.SEd448 $ C.sign' pk t' + C.SX25519 -> (\THAuthClient {peerServerPubKey = k} -> TAAuthenticator $ C.cbAuthenticate k pk (C.cbNonce corrId) t') <$> thAuth params +#if !MIN_VERSION_base(4,18,0) + _sx448 -> undefined +#endif + where + t' = case (serviceKey_, thAuth params >>= clientService) of + (Just _, Just THClientService {serviceCertHash = XV.Fingerprint fp}) -> fp <> t + _ -> t + +tPut1 :: H -> SentRawTransmission -> IO (Either TransportError ()) +tPut1 h t = do + [r] <- tPut h [Right t] + pure r + +tGet1 :: H -> IO (Transmission (Either ErrorType BrokerMsg)) +tGet1 h = do + [r] <- tGetClient h + pure r + +-- read and discard any pending transmissions until quiet, to resync after a race +drainAll :: H -> IO () +drainAll h = timeout 40000 (tGet1 h) >>= maybe (pure ()) (const $ drainAll h) + +-- measurement --------------------------------------------------------------- + +liveBytesMiB :: IO Double +liveBytesMiB = do + performMajorGC + s <- getRTSStats + pure $ fromIntegral (gcdetails_live_bytes (gc s)) / (1024 * 1024) + +report :: String -> Int -> Double -> Double -> IO () +report phase i base cur = + printf "%-8s iter=%7d live=%9.1f MiB delta=%+9.1f MiB (%+.4f KiB/iter)\n" + phase i cur (cur - base) (if i == 0 then 0 else (cur - base) * 1024 / fromIntegral i) + +withCheckpoints :: String -> Int -> Int -> (Int -> IO ()) -> IO () +withCheckpoints phase iters cp step = do + base <- liveBytesMiB + report phase 0 base base + forM_ ([1 .. iters] :: [Int]) $ \i -> do + step i + when (i `mod` cp == 0) $ liveBytesMiB >>= report phase i base + +-- phases -------------------------------------------------------------------- + +-- regular recipient: create+subscribe, send, receive, ack, delete (churn) +runPlain :: TVar ChaChaDRG -> Int -> Int -> IO () +runPlain g iters cp = + testSMPClient @TLS $ \recip -> + testSMPClient @TLS $ \sndr -> + withCheckpoints "plain" iters cp $ \i -> do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _srvDh) <- signSendRecv recip rKey (corr, NoEntity, New rPub dhPub) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND "hello") + Resp _ _ (Msg mId _) <- tGet1 recip + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, ACK mId) + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, DEL) + pure () + +-- recipient-service: create queue as service, send, service receives, ack, delete (churn) +runSvc :: TVar ChaChaDRG -> Int -> Int -> IO () +runSvc g iters cp = do + creds <- genCredentials g Nothing (0, 2400) "localhost" + let (_fp, tlsCred) = tlsCredentials (creds :| []) + serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g + testSMPClient @TLS $ \sndr -> + testSMPServiceClient @TLS (tlsCred, serviceKeys) $ \sh -> + withCheckpoints "svc" iters cp $ \i -> do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids_ rId sId _srvDh _serviceId) <- serviceSignSendRecv sh rKey servicePK (corr, NoEntity, New rPub dhPub) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND "hello") + Resp _ _ (Msg mId _) <- tGet1 sh + Resp _ _ OK <- signSendRecv sh rKey (corr, rId, ACK mId) + Resp _ _ OK <- signSendRecv sh rKey (corr, rId, DEL) + pure () + +-- recipient-service with concurrent SEND vs DEL to probe the TOCTOU orphan +runSvcRace :: TVar ChaChaDRG -> Int -> Int -> IO () +runSvcRace g iters cp = do + creds <- genCredentials g Nothing (0, 2400) "localhost" + let (_fp, tlsCred) = tlsCredentials (creds :| []) + serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g + testSMPClient @TLS $ \sndr -> + testSMPServiceClient @TLS (tlsCred, serviceKeys) $ \sh -> + withCheckpoints "svcrace" iters cp $ \i -> do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids_ rId sId _srvDh _serviceId) <- serviceSignSendRecv sh rKey servicePK (corr, NoEntity, New rPub dhPub) + -- race an in-flight SEND (creates delivery Sub) against queue deletion + concurrently_ + (void $ sendRecv sndr (Nothing, corr, sId, _SEND "hello")) + (void $ signSendRecv sh rKey (corr, rId, DEL)) + drainAll sh + +-- notifications: enable ntf on live queues and send ntf-flagged messages without draining +runNtf :: TVar ChaChaDRG -> Int -> Int -> IO () +runNtf g iters cp = + testSMPClient @TLS $ \recip -> + testSMPClient @TLS $ \sndr -> + withCheckpoints "ntf" iters cp $ \i -> do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + (nPub, _nKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (rcvNtfPubDh, _dhNtfPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _srvDh) <- signSendRecv recip rKey (corr, NoEntity, New rPub dhPub) + Resp _ _ (NID _nId _) <- signSendRecv recip rKey (corr, rId, NKEY nPub rcvNtfPubDh) + -- ntf-flagged send stores a notification; no notifier subscribed -> stays in ntfStore + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND' "hello") + Resp _ _ (Msg mId _) <- tGet1 recip + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, ACK mId) + pure () + +-- reusable steps ------------------------------------------------------------ + +genKeys :: TVar ChaChaDRG -> IO (RcvPublicAuthKey, C.APrivateAuthKey, RcvPublicDhKey) +genKeys g = do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + pure (rPub, rKey, dhPub) + +plainStep :: TVar ChaChaDRG -> H -> H -> Int -> IO () +plainStep g recip sndr i = do + (rPub, rKey, dhPub) <- genKeys g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _) <- signSendRecv recip rKey (corr, NoEntity, New rPub dhPub) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND "hello") + Resp _ _ (Msg mId _) <- tGet1 recip + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, ACK mId) + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, DEL) + pure () + +ntfStep :: TVar ChaChaDRG -> H -> H -> Int -> IO () +ntfStep g recip sndr i = do + (rPub, rKey, dhPub) <- genKeys g + (nPub, _nKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (rcvNtfPubDh, _dh :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _) <- signSendRecv recip rKey (corr, NoEntity, New rPub dhPub) + Resp _ _ (NID _ _) <- signSendRecv recip rKey (corr, rId, NKEY nPub rcvNtfPubDh) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND' "hello") + Resp _ _ (Msg mId _) <- tGet1 recip + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, ACK mId) + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, DEL) + pure () + +-- concurrency/scale: many client connections running a mixed workload +runConc :: TVar ChaChaDRG -> Int -> Int -> IO () +runConc g iters _cp = do + let nW = 8 + per = max 1 (iters `div` nW) + base <- liveBytesMiB + report "conc" 0 base base + counter <- newTVarIO (0 :: Int) + let target = nW * per + monitor = do + threadDelay 2000000 + c <- readTVarIO counter + cur <- liveBytesMiB + report "conc" c base cur + when (c < target) monitor + worker w = + testSMPClient @TLS $ \recip -> + testSMPClient @TLS $ \sndr -> + forM_ ([1 .. per] :: [Int]) $ \i -> do + (if i `mod` 3 == 0 then ntfStep else plainStep) g recip sndr (w * per + i) + atomically $ modifyTVar' counter (+ 1) + withAsync monitor $ \_ -> mapConcurrently_ worker ([0 .. nW - 1] :: [Int]) + cur <- liveBytesMiB + report "conc" target base cur + +-- service SUBS reconnect: a fixed set of service queues, reconnect+resubscribe each iteration +runSvcSubs :: TVar ChaChaDRG -> Int -> Int -> IO () +runSvcSubs g iters cp = do + creds <- genCredentials g Nothing (0, 2400) "localhost" + let (_fp, tlsCred) = tlsCredentials (creds :| []) + serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g + let aServicePK = C.APrivateAuthKey C.SEd25519 servicePK + nQueues = 20 + -- create nQueues associated with the service (persisted in the store) + (serviceId, rIds) <- testSMPServiceClient @TLS (tlsCred, serviceKeys) $ \sh -> do + xs <- forM ([1 .. nQueues] :: [Int]) $ \j -> do + (rPub, rKey, dhPub) <- genKeys g + Resp _ _ (Ids_ rId _sId _ sid) <- serviceSignSendRecv sh rKey servicePK (B.pack ("c" <> show j), NoEntity, New rPub dhPub) + pure (rId, sid) + pure (snd (head xs), map fst xs) + let idsHash = queueIdsHash rIds + base <- liveBytesMiB + report "svcsubs" 0 base base + -- each iteration: fresh service connection, SUBS to resubscribe all queues, then disconnect + forM_ ([1 .. iters] :: [Int]) $ \i -> do + testSMPServiceClient @TLS (tlsCred, serviceKeys) $ \sh -> + void $ signSendRecv_ sh aServicePK Nothing (B.pack (show i), serviceId, SUBS (fromIntegral nQueues) idsHash) + when (i `mod` cp == 0) $ liveBytesMiB >>= report "svcsubs" i base + +-- GET path: create-only (unsubscribed) queue, send, GET (poll), ack, delete +runGet :: TVar ChaChaDRG -> Int -> Int -> IO () +runGet g iters cp = + testSMPClient @TLS $ \recip -> + testSMPClient @TLS $ \sndr -> + withCheckpoints "getp" iters cp $ \i -> do + (rPub, rKey, dhPub) <- genKeys g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _) <- signSendRecv recip rKey (corr, NoEntity, New0 rPub dhPub) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND "hello") + Resp _ _ (Msg mId _) <- signSendRecv recip rKey (corr, rId, GET) + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, ACK mId) + Resp _ _ OK <- signSendRecv recip rKey (corr, rId, DEL) + pure () + +-- forkDeliver blocked in SubPending: a subscriber that never reads its sndQ. +-- Deliveries to a full sndQ fork a deliverThread that blocks forever -> threads/subs_thread grow. +runStuck :: TVar ChaChaDRG -> Int -> Int -> IO () +runStuck g iters cp = + testSMPClient @TLS $ \sndr -> + testSMPClient @TLS $ \recip -> do + -- phase 1: create + subscribe `iters` queues (recip reads only the IDS responses, no messages yet) + qs <- forM ([1 .. iters] :: [Int]) $ \i -> do + (rPub, rKey, dhPub) <- genKeys g + Resp _ _ (Ids _rId sId _) <- signSendRecv recip rKey (B.pack ('c' : show i), NoEntity, New rPub dhPub) + pure sId + -- phase 2: send one message to each; recip never reads -> server delivery threads block on the full sndQ + base <- liveBytesMiB + report "stuck" 0 base base + forM_ (zip ([1 ..] :: [Int]) qs) $ \(i, sId) -> do + _ <- sendRecv sndr (Nothing, B.pack ('s' : show i), sId, _SEND "x") + when (i `mod` cp == 0) $ liveBytesMiB >>= report "stuck" i base + threadDelay 20000000 -- hold the subscriber open so diagnostics can sample the blocked threads + +-- serviceLocks + services never evicted: connect as a messaging service with a fresh certificate +-- each iteration. getCreateService (run in the handshake) adds a services row + serviceLocks entry +-- that is never removed -> store_rcvServices grows. +runCertChurn :: TVar ChaChaDRG -> Int -> Int -> IO () +runCertChurn g iters cp = do + base <- liveBytesMiB + report "certchurn" 0 base base + forM_ ([1 .. iters] :: [Int]) $ \i -> do + creds <- genCredentials g Nothing (0, 2400) "localhost" + let (_fp, tlsCred) = tlsCredentials (creds :| []) + serviceKeys <- atomically $ C.generateKeyPair g + testSMPServiceClient @TLS (tlsCred, serviceKeys) $ \_sh -> pure () + when (i `mod` cp == 0) $ liveBytesMiB >>= report "certchurn" i base + +-- LINK path coverage: create a queue with short-link data, update it (LSET), secure it via the +-- link (LKEY), delete the link data (LDEL), delete the queue. Exercises the links map on create +-- and delete. (Not a leak repro: the links-not-removed-on-delete bug only bites useCache=True.) +runLink :: TVar ChaChaDRG -> Int -> Int -> IO () +runLink g iters cp = + testSMPClient @TLS $ \r -> + testSMPClient @TLS $ \s -> + withCheckpoints "link" iters cp $ \i -> do + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + C.CbNonce corrId <- atomically $ C.randomCbNonce g + let sId = EntityId $ B.take 24 $ C.sha3_384 corrId -- sender ID must be derived from corrId + ld = (EncDataBytes "fixed data", EncDataBytes "user data") + qrd = QRMessaging $ Just (sId, ld) + Resp _ NoEntity (IDS (QIK rId _sId _srvDh _qm (Just lnkId) _svc _ntf)) <- + signSendRecv r rKey (corrId, NoEntity, NEW (NewQueueReq rPub dhPub Nothing SMSubscribe (Just qrd) Nothing)) + Resp _ _ OK <- signSendRecv r rKey (B.pack ('a' : show i), rId, LSET lnkId ld) + Resp _ _ (LNK _sId2 _ld') <- signSendRecv s sKey (B.pack ('b' : show i), lnkId, LKEY sPub) + Resp _ _ OK <- signSendRecv r rKey (B.pack ('c' : show i), rId, LDEL) + Resp _ _ OK <- signSendRecv r rKey (B.pack ('d' : show i), rId, DEL) + pure () + +-- Note: the two proxy leaks are NOT reproducible in this load bench and are intentionally not +-- included. The sentCommands/PFWD-timeout leak needs a relay that keeps the session up but drops +-- RFWD (a mock relay). The empty-SessionVar leak is a precise disconnect-during-connect race +-- (reproduced deterministically by the SMPProxyTests unit test, not by a load loop). Both are +-- observable on a live proxy via the LEAKDIAG proxy_sentCommands / proxy_smpClients counters. + +-- NtfStore key-retention leak: store notifications in many notifier queues, then let them expire. +-- With the fix, deleteExpiredNtfs removes the emptied outer keys, so LEAKDIAG ntfStore_keys rises +-- during creation then falls to ~0 after expiry; without it, the empty keys are retained. +-- Run with the short-expiry config (see main) so expiry fires within the run. +runNtfExp :: TVar ChaChaDRG -> Int -> Int -> IO () +runNtfExp g iters _cp = + testSMPClient @TLS $ \recip -> + testSMPClient @TLS $ \sndr -> do + forM_ ([1 .. iters] :: [Int]) $ \i -> do + (rPub, rKey, dhPub) <- genKeys g + (nPub, _nKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (rcvNtfPubDh, _dh :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let corr = B.pack (show i) + Resp _ _ (Ids rId sId _) <- signSendRecv recip rKey (corr, NoEntity, New0 rPub dhPub) + Resp _ _ (NID _nId _) <- signSendRecv recip rKey (corr, rId, NKEY nPub rcvNtfPubDh) + Resp _ _ OK <- sendRecv sndr (Nothing, corr, sId, _SEND' "hi") + pure () + base <- liveBytesMiB + report "ntfexp" iters base base + -- hold while notifications expire; LEAKDIAG samples ntfStore_keys over this window + forM_ ([1 .. 12] :: [Int]) $ \k -> threadDelay 5000000 >> (liveBytesMiB >>= report "ntfexp" (iters * 10 + k) base) + +-- store config selectable via BENCHSTORE env: pgmsg (default, useCache=False) | pgjournal (useCache=True) | journal +srvStoreCfg :: Maybe String -> AServerConfig +srvStoreCfg = \case +#if defined(dbServerPostgres) + Just "pgjournal" -> cfgMS (ASType SQSPostgres SMSJournal) + Just "journal" -> cfgMS (ASType SQSMemory SMSJournal) + _ -> cfgMS (ASType SQSPostgres SMSPostgres) +#else + _ -> cfg +#endif + +main :: IO () +main = do + args <- getArgs + let (phase, iters) = case args of + (p : n : _) -> (p, read n) + [p] -> (p, 20000) + _ -> ("svc", 20000) + cp = max 1 (iters `div` 20) + g <- C.newRandom + storeEnv <- lookupEnv "BENCHSTORE" + -- ntfexp uses a short notification-expiration so deleteExpiredNtfs fires within the run + let srvCfg = case phase of + "ntfexp" -> updateCfg (srvStoreCfg storeEnv) $ \c -> c {notificationExpiration = ExpirationConfig {ttl = 2, checkInterval = 3}} + _ -> srvStoreCfg storeEnv + setLogLevel LogInfo + withGlobalLogging LogConfig {lc_file = Nothing, lc_stderr = True} $ + withSmpServerConfigOn (transport @TLS) srvCfg testPort $ \_ -> do + threadDelay 250000 + case phase of + "plain" -> runPlain g iters cp + "svc" -> runSvc g iters cp + "svcrace" -> runSvcRace g iters cp + "ntf" -> runNtf g iters cp + "conc" -> runConc g iters cp + "svcsubs" -> runSvcSubs g iters cp + "getp" -> runGet g iters cp + "stuck" -> runStuck g iters cp + "certchurn" -> runCertChurn g iters cp + "link" -> runLink g iters cp + "ntfexp" -> runNtfExp g iters cp + _ -> error $ "unknown phase: " <> phase diff --git a/simplexmq.cabal b/simplexmq.cabal index b017dc706..7c4fe4d61 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -652,3 +652,47 @@ test-suite simplexmq-test if flag(client_postgres) || flag(server_postgres) build-depends: postgresql-simple ==0.7.* + +executable smp-mem-bench + if flag(client_library) + buildable: False + main-is: MemBench.hs + other-modules: + SMPClient + Util + hs-source-dirs: + bench + tests + default-extensions: + StrictData + ghc-options: -Wall -Wno-unused-imports -Wno-unused-top-binds -Wno-name-shadowing -threaded -rtsopts "-with-rtsopts=-T" + build-depends: + base + , async + , bytestring + , containers + , crypton + , crypton-x509 + , crypton-x509-store + , crypton-x509-validation + , directory + , hspec ==2.11.* + , hspec-core ==2.11.* + , mtl + , network + , process + , simple-logger + , simplexmq + , sqlcipher-simple + , stm + , text + , time + , tls >=1.9.0 && <1.10 + , transformers + , unliftio + , unliftio-core + if flag(server_postgres) + cpp-options: -DdbServerPostgres + build-depends: + postgresql-simple ==0.7.* + default-language: Haskell2010 diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index dc811fa0b..5cd268f21 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/src/Simplex/Messaging/Client.hs @@ -35,6 +35,7 @@ module Simplex.Messaging.Client ProxiedRelay (..), getProtocolClient, closeProtocolClient, + pClientSentCommandsCount, protocolClientServer, protocolClientServer', transportHost', @@ -167,6 +168,7 @@ import Simplex.Messaging.Protocol import Simplex.Messaging.Protocol.Types import Simplex.Messaging.Server.QueueStore.QueueInfo import Simplex.Messaging.SimplexName (SimplexDomain) +import qualified Data.Map.Strict as M import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport @@ -737,6 +739,10 @@ useWebPort cfg presetDomains ProtocolServer {host = h :| _} = case smpWebPortSer SWPPreset -> isPresetDomain presetDomains h SWPOff -> False +-- | Count of in-flight (awaiting-response) commands on a client - for leak diagnostics. +pClientSentCommandsCount :: ProtocolClient v err msg -> IO Int +pClientSentCommandsCount ProtocolClient {client_ = PClient {sentCommands}} = M.size <$> readTVarIO sentCommands + isPresetDomain :: [HostName] -> TransportHost -> Bool isPresetDomain presetDomains = \case THDomainName h -> any (`isSuffixOf` h) presetDomains diff --git a/src/Simplex/Messaging/Client/Agent.hs b/src/Simplex/Messaging/Client/Agent.hs index e41d7a811..582658601 100644 --- a/src/Simplex/Messaging/Client/Agent.hs +++ b/src/Simplex/Messaging/Client/Agent.hs @@ -31,6 +31,8 @@ module Simplex.Messaging.Client.Agent removeActiveSubs, removePendingSub, removePendingSubs, + AgentLeakStats (..), + getAgentLeakStats, ) where @@ -158,6 +160,33 @@ data SMPClientAgent p = SMPClientAgent type OwnServer = Bool +-- | Sizes of every per-server/per-session map in the client agent, plus total in-flight +-- forwarded commands - for leak diagnostics on the proxy path. +data AgentLeakStats = AgentLeakStats + { alSmpClients :: Int, + alSmpSessions :: Int, + alActiveServiceSubs :: Int, + alActiveQueueSubs :: Int, + alPendingServiceSubs :: Int, + alPendingQueueSubs :: Int, + alSmpSubWorkers :: Int, + alSentCommands :: Int + } + +getAgentLeakStats :: SMPClientAgent p -> IO AgentLeakStats +getAgentLeakStats SMPClientAgent {smpClients, smpSessions, activeServiceSubs, activeQueueSubs, pendingServiceSubs, pendingQueueSubs, smpSubWorkers} = do + alSmpClients <- msize smpClients + sess <- readTVarIO smpSessions + alActiveServiceSubs <- msize activeServiceSubs + alActiveQueueSubs <- msize activeQueueSubs + alPendingServiceSubs <- msize pendingServiceSubs + alPendingQueueSubs <- msize pendingQueueSubs + alSmpSubWorkers <- msize smpSubWorkers + alSentCommands <- foldM (\ !a (_, c) -> (a +) <$> pClientSentCommandsCount c) 0 (M.elems sess) + pure AgentLeakStats {alSmpClients, alSmpSessions = M.size sess, alActiveServiceSubs, alActiveQueueSubs, alPendingServiceSubs, alPendingQueueSubs, alSmpSubWorkers, alSentCommands} + where + msize m = M.size <$> readTVarIO m + newSMPClientAgent :: SParty p -> SMPClientAgentConfig -> Maybe DBService -> TVar ChaChaDRG -> IO (SMPClientAgent p) newSMPClientAgent agentParty agentCfg@SMPClientAgentConfig {msgQSize, agentQSize} dbService randomDrg = do active <- newTVarIO True diff --git a/src/Simplex/Messaging/Protocol.hs b/src/Simplex/Messaging/Protocol.hs index 8ef5bb04f..8730597f3 100644 --- a/src/Simplex/Messaging/Protocol.hs +++ b/src/Simplex/Messaging/Protocol.hs @@ -1546,11 +1546,14 @@ queueIdHash = IdsHash . C.md5Hash . unEntityId {-# INLINE queueIdHash #-} addServiceSubs :: (Int64, IdsHash) -> (Int64, IdsHash) -> (Int64, IdsHash) -addServiceSubs (n', idsHash') (n, idsHash) = (n + n', idsHash <> idsHash') +addServiceSubs (n', idsHash') (n, idsHash) = + let !n'' = n + n' + !h = idsHash <> idsHash' + in (n'', h) subtractServiceSubs :: (Int64, IdsHash) -> (Int64, IdsHash) -> (Int64, IdsHash) subtractServiceSubs (n', idsHash') (n, idsHash) - | n > n' = (n - n', idsHash <> idsHash') -- concat is a reversible xor: (x `xor` y) `xor` y == x + | n > n' = let !n'' = n - n'; !h = idsHash <> idsHash' in (n'', h) -- concat is a reversible xor: (x `xor` y) `xor` y == x | otherwise = (0, mempty) data ProtocolErrorType = PECmdSyntax | PECmdUnknown | PESession | PEBlock diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 36ac471f9..082f5c4d1 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -98,7 +98,7 @@ import qualified Network.TLS as TLS import Numeric.Natural (Natural) import Simplex.Messaging.Agent.Lock import Simplex.Messaging.Client (ProtocolClient (thParams), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, forwardSMPTransmission, smpProxyError, temporaryClientError) -import Simplex.Messaging.Client.Agent (OwnServer, SMPClientAgent (..), SMPClientAgentEvent (..), closeSMPClientAgent, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient) +import Simplex.Messaging.Client.Agent (AgentLeakStats (..), OwnServer, SMPClientAgent (..), SMPClientAgentEvent (..), closeSMPClientAgent, getAgentLeakStats, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String @@ -128,6 +128,7 @@ import Simplex.Messaging.Transport.Server import Simplex.Messaging.Util import Simplex.Messaging.Version import System.Environment (lookupEnv) +import Text.Read (readMaybe) import System.Exit (exitFailure, exitSuccess) import System.IO (hPrint, hPutStrLn, hSetNewlineMode, universalNewlineMode) import System.Mem.Weak (deRefWeak) @@ -174,6 +175,23 @@ data ClientSubAction type PrevClientSub s = (Client s, ClientSubAction, (EntityId, BrokerMsg)) +-- accumulator for per-client leak diagnostics (summed across all connected clients) +data ClientAgg = ClientAgg + { aggEndThreads :: !Int, -- Weak ThreadId registrations in endThreads (forkClient leak) + aggEndThreadSeq :: !Int, -- total forkClient forks ever (fork rate) + aggProcThreads :: !Int, + aggSubs :: !Int, -- entries in per-client subscriptions map + aggNoSub :: !Int, + aggPending :: !Int, -- SubPending: forked delivery threads not yet completed (blocked-thread candidates) + aggThread :: !Int, -- SubThread: live delivery threads + aggProhibit :: !Int, + aggNtfSubs :: !Int, + aggSvcSubsCount :: !Int, + aggRcvQ :: !Int, + aggSndQ :: !Int, -- full sndQ => forkDeliver blocks (leak trigger) + aggMsgQ :: !Int + } + smpServer :: forall s. MsgStoreClass s => TMVar Bool -> ServerConfig s -> Maybe AttachHTTP -> M s () smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOptions} attachHTTP_ = do s <- asks server @@ -191,6 +209,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt ( serverThread "server subscribers" s subscribers subscriptions serviceSubsCount (Just cancelSub) : serverThread "server ntfSubscribers" s ntfSubscribers ntfSubscriptions ntfServiceSubsCount Nothing : deliverNtfsThread s + : leakDiagnosticsThread s : sendPendingEvtsThread s : receiveFromProxyAgent pa : expireNtfsThread cfg @@ -496,6 +515,98 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt printMessageStats "STORE: messages" msgStats Left e -> logError $ "STORE: expireOldMessages, error expiring messages, " <> tshow e + -- Periodic comprehensive leak diagnostics: sizes of every growable structure in the + -- server, summed across clients, plus the proxy agent. Interval seconds via SMP_LEAKDIAG_SEC + -- (default 60). A single greppable "LEAKDIAG ..." line per interval; whichever counter grows + -- monotonically over time is the leak. + leakDiagnosticsThread :: Server s -> M s () + leakDiagnosticsThread srv = do + secStr <- liftIO $ lookupEnv "SMP_LEAKDIAG_SEC" + let sec = max 5 $ fromMaybe 60 (secStr >>= readMaybe) + ms <- asks msgStore + ns <- asks ntfStore + ProxyAgent {smpAgent} <- asks proxyAgent + labelMyThread "leakDiagnosticsThread" + -- never let a diagnostics error crash the server (this thread is in raceAny_) + liftIO $ forever $ do + threadDelay $ sec * 1000000 + tryAny (logLeakStats srv ms ns smpAgent) >>= either (logError . ("LEAKDIAG error: " <>) . tshow) (const $ pure ()) + + logLeakStats :: Server s -> s -> NtfStore -> SMPClientAgent 'Sender -> IO () + logLeakStats srv ms (NtfStore nsv) smpAgent = do +#if MIN_VERSION_base(4,18,0) + nThreads <- length <$> listThreads +#else + let nThreads = 0 :: Int +#endif + cls <- IM.elems <$> getServerClients srv + cc <- foldM accClient emptyAgg cls + (smpQ, smpS, smpC, smpT, smpP) <- subAgg (subscribers srv) + (ntfQ, ntfS, ntfC, ntfT, ntfP) <- subAgg (ntfSubscribers srv) + ntfMap <- readTVarIO nsv + ntfMsgs <- foldM (\ !a v -> (a +) . length <$> readTVarIO v) (0 :: Int) (M.elems ntfMap) + EntityCounts {queueCount, notifierCount, rcvServiceCount, ntfServiceCount, rcvServiceQueuesCount, ntfServiceQueuesCount} <- getEntityCounts @(StoreQueue s) (queueStore ms) + LoadedQueueCounts {loadedQueueCount, loadedNotifierCount, openJournalCount, queueLockCount, notifierLockCount} <- loadedQueueCounts ms + AgentLeakStats {alSmpClients, alSmpSessions, alActiveServiceSubs, alActiveQueueSubs, alPendingServiceSubs, alPendingQueueSubs, alSmpSubWorkers, alSentCommands} <- getAgentLeakStats smpAgent + logNote $ + T.concat + [ "LEAKDIAG", + f "threads" nThreads, f "clients" (length cls), + f "endThreads" (aggEndThreads cc), f "endThreadSeq" (aggEndThreadSeq cc), f "procThreads" (aggProcThreads cc), + f "subs" (aggSubs cc), f "subs_nosub" (aggNoSub cc), f "subs_pending" (aggPending cc), f "subs_thread" (aggThread cc), f "subs_prohibit" (aggProhibit cc), + f "ntfSubsClient" (aggNtfSubs cc), f "svcSubsCount" (aggSvcSubsCount cc), + f "rcvQ" (aggRcvQ cc), f "sndQ" (aggSndQ cc), f "msgQ" (aggMsgQ cc), + f "smp_qSubscribers" smpQ, f "smp_svcSubscribers" smpS, f "smp_subClients" smpC, f "smp_totalSvcSubs" smpT, f "smp_pendingEvents" smpP, + f "ntf_qSubscribers" ntfQ, f "ntf_svcSubscribers" ntfS, f "ntf_subClients" ntfC, f "ntf_totalSvcSubs" ntfT, f "ntf_pendingEvents" ntfP, + f "ntfStore_keys" (M.size ntfMap), f "ntfStore_msgs" ntfMsgs, + f "store_queues" queueCount, f "store_notifiers" notifierCount, f "store_rcvServices" rcvServiceCount, f "store_ntfServices" ntfServiceCount, f "store_rcvSvcQueues" rcvServiceQueuesCount, f "store_ntfSvcQueues" ntfServiceQueuesCount, + f "loaded_queues" loadedQueueCount, f "loaded_notifiers" loadedNotifierCount, f "open_journals" openJournalCount, f "queue_locks" queueLockCount, f "notifier_locks" notifierLockCount, + f "proxy_smpClients" alSmpClients, f "proxy_smpSessions" alSmpSessions, f "proxy_activeSvcSubs" alActiveServiceSubs, f "proxy_activeQSubs" alActiveQueueSubs, f "proxy_pendingSvcSubs" alPendingServiceSubs, f "proxy_pendingQSubs" alPendingQueueSubs, f "proxy_subWorkers" alSmpSubWorkers, f "proxy_sentCommands" alSentCommands + ] + where + f :: Show a => Text -> a -> Text + f k v = " " <> k <> "=" <> tshow v + emptyAgg = ClientAgg 0 0 0 0 0 0 0 0 0 0 0 0 0 + accClient !agg Client {subscriptions, ntfSubscriptions, serviceSubsCount, procThreads, endThreads, endThreadSeq, rcvQ, sndQ, msgQ} = do + et <- IM.size <$> readTVarIO endThreads + es <- readTVarIO endThreadSeq + pt <- readTVarIO procThreads + subs <- readTVarIO subscriptions + (no, pe, th, pr) <- foldM accSub (0, 0, 0, 0) (M.elems subs) + nt <- M.size <$> readTVarIO ntfSubscriptions + sc <- fst <$> readTVarIO serviceSubsCount + (rl, sl, ml) <- atomically $ (,,) <$> lengthTBQueue rcvQ <*> lengthTBQueue sndQ <*> lengthTBQueue msgQ + pure + agg + { aggEndThreads = aggEndThreads agg + et, + aggEndThreadSeq = aggEndThreadSeq agg + es, + aggProcThreads = aggProcThreads agg + pt, + aggSubs = aggSubs agg + M.size subs, + aggNoSub = aggNoSub agg + no, + aggPending = aggPending agg + pe, + aggThread = aggThread agg + th, + aggProhibit = aggProhibit agg + pr, + aggNtfSubs = aggNtfSubs agg + nt, + aggSvcSubsCount = aggSvcSubsCount agg + fromIntegral sc, + aggRcvQ = aggRcvQ agg + fromIntegral rl, + aggSndQ = aggSndQ agg + fromIntegral sl, + aggMsgQ = aggMsgQ agg + fromIntegral ml + } + accSub (no, pe, th, pr) Sub {subThread} = case subThread of + ServerSub t -> + readTVarIO t >>= \case + NoSub -> pure (no + 1, pe, th, pr) + SubPending -> pure (no, pe + 1, th, pr) + SubThread _ -> pure (no, pe, th + 1, pr) + ProhibitSub -> pure (no, pe, th, pr + 1) + subAgg ServerSubscribers {queueSubscribers, serviceSubscribers, subClients, totalServiceSubs, pendingEvents} = do + q <- M.size <$> getSubscribedClients queueSubscribers + s' <- M.size <$> getSubscribedClients serviceSubscribers + sc <- IS.size <$> readTVarIO subClients + ts <- fst <$> readTVarIO totalServiceSubs + pe <- IM.size <$> readTVarIO pendingEvents + pure (q, s', sc, ts, pe) + expireNtfsThread :: ServerConfig s -> M s () expireNtfsThread ServerConfig {notificationExpiration = expCfg} = do ns <- asks ntfStore diff --git a/src/Simplex/Messaging/Server/NtfStore.hs b/src/Simplex/Messaging/Server/NtfStore.hs index b73fd4860..711072303 100644 --- a/src/Simplex/Messaging/Server/NtfStore.hs +++ b/src/Simplex/Messaging/Server/NtfStore.hs @@ -33,13 +33,10 @@ data MsgNtf = MsgNtf } storeNtf :: NtfStore -> NotifierId -> MsgNtf -> IO () -storeNtf (NtfStore ns) nId ntf = do - TM.lookupIO nId ns >>= atomically . maybe newNtfs (`modifyTVar'` (ntf :)) +storeNtf (NtfStore ns) nId ntf = -- TODO [ntfdb] coalesce messages here once the client is updated to process multiple messages -- for single notification. - -- when (isJust prevNtf) $ incStat $ msgNtfReplaced stats - where - newNtfs = TM.lookup nId ns >>= maybe (TM.insertM nId (newTVar [ntf]) ns) (`modifyTVar'` (ntf :)) + atomically $ TM.lookup nId ns >>= maybe (TM.insertM nId (newTVar [ntf]) ns) (`modifyTVar'` (ntf :)) deleteNtfs :: NtfStore -> NotifierId -> IO Int deleteNtfs (NtfStore ns) nId = atomically (TM.lookupDelete nId ns) >>= maybe (pure 0) (fmap length . readTVarIO) @@ -48,18 +45,17 @@ deleteExpiredNtfs :: NtfStore -> Int64 -> IO Int deleteExpiredNtfs (NtfStore ns) old = foldM (\expired -> fmap (expired +) . expireQueue) 0 . M.keys =<< readTVarIO ns where - expireQueue nId = TM.lookupIO nId ns >>= maybe (pure 0) expire - expire v = readTVarIO v >>= \case - [] -> pure 0 - _ -> - atomically $ readTVar v >>= \case - [] -> pure 0 - -- check the last message first, it is the earliest - ntfs | systemSeconds (ntfTs $ last $ ntfs) < old -> do + expireQueue nId = atomically $ TM.lookup nId ns >>= maybe (pure 0) (expire nId) + expire nId v = readTVar v >>= \case + [] -> TM.delete nId ns >> pure 0 + -- check the last message first, it is the earliest + ntfs + | systemSeconds (ntfTs $ last ntfs) < old -> do let !ntfs' = filter (\MsgNtf {ntfTs = ts} -> systemSeconds ts >= old) ntfs - writeTVar v ntfs' - pure $! length ntfs - length ntfs' - _ -> pure 0 + if null ntfs' + then TM.delete nId ns >> pure (length ntfs) + else writeTVar v ntfs' >> pure (length ntfs - length ntfs') + | otherwise -> pure 0 data NtfLogRecord = NLRv1 NotifierId MsgNtf