diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index e2cef6f50..c26a8b0a6 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -98,8 +98,8 @@ import Network.Socket (ServiceName, Socket, socketToHandle) 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 (NetworkConfig (..), NetworkRequestMode (..), ProtocolClient (thParams), ProtocolClientConfig (..), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, forwardSMPTransmission, netTimeoutInt, smpProxyError, temporaryClientError) +import Simplex.Messaging.Client.Agent (OwnServer, SMPClientAgent (..), SMPClientAgentConfig (..), SMPClientAgentEvent (..), closeSMPClientAgent, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String @@ -1212,7 +1212,23 @@ sendAfterDelay c@Client {sndQ, clientTHParams = THandleParams {sessionId}} stats | otherwise = write where write = atomically $ writeTBQueue sndQ rs - microseconds (MkSystemTime s ns) = s * 1000000 + fromIntegral (ns `div` 1000) + +updateSlotWaitStats :: ServerStats -> SystemTime -> Bool -> IO () +updateSlotWaitStats ServerStats {cmdSlotWaits, cmdSlotWaitMs, cmdSlotWaitMaxMs, cmdSlotWaitTimeouts} since taken = do + waitMs <- fromIntegral . (`div` 1000) . subtract (microseconds since) . microseconds <$> getSystemTime + incStat cmdSlotWaits + atomicModifyIORef'_ cmdSlotWaitMs (+ waitMs) + atomicModifyIORef'_ cmdSlotWaitMaxMs (max waitMs) + unless taken $ incStat cmdSlotWaitTimeouts + +rslvTooManyRequests :: BrokerMsg +rslvTooManyRequests = ERR $ NAME $ RESOLVER "too many requests" + +forwardedCmdWait :: ServerConfig s -> Int +forwardedCmdWait = (`netTimeoutInt` NRMBackground) . tcpTimeout . networkConfig . smpCfg . smpAgentCfg + +microseconds :: SystemTime -> Int64 +microseconds (MkSystemTime s ns) = s * 1000000 + fromIntegral (ns `div` 1000) send :: Transport c => MVar (THandleSMP c 'TServer) -> Client s -> IO () send th c@Client {sndQ, msgQ, clientTHParams = THandleParams {sessionId}} = do @@ -1491,21 +1507,45 @@ client Nothing -> inc False pRequests >> inc False pErrorsConnect $> Just (ERR $ PROXY NO_SESSION) where forkProxiedCmd :: M s BrokerMsg -> M s (Maybe BrokerMsg) - forkProxiedCmd = forkCmd serverClientConcurrency corrId (EntityId sessId) - -- Run a slow command on a thread - forkCmd :: (ServerConfig s -> Int) -> CorrId -> EntityId -> M s BrokerMsg -> M s (Maybe a) - forkCmd concurrency corrId entId cmdAction = do - bracket_ wait signal . forkClient clnt (B.unpack $ "client $" <> encode sessionId <> " cmd") $ - -- commands MUST be processed under a reasonable timeout or the client would halt - cmdAction >>= \t -> atomically $ writeTBQueue sndQ ([(corrId, entId, t)], []) + forkProxiedCmd = forkCmd serverClientConcurrency corrId (EntityId sessId) (ERR $ smpProxyError PCEResponseTimeout) + -- Run a slow command on a thread, responding with timeoutErr if it was not started before the client timeout + forkCmd :: (ServerConfig s -> Int) -> CorrId -> EntityId -> BrokerMsg -> M s BrokerMsg -> M s (Maybe a) + forkCmd concurrency corrId entId timeoutErr cmdAction = + -- commands MUST be processed under a reasonable timeout or the client would halt + forkCmd_ concurrency maxClientCmdWait (sendResp timeoutErr) $ cmdAction >>= sendResp + where + sendResp t = atomically $ writeTBQueue sndQ ([(corrId, entId, t)], []) + -- Run a slow command on a thread that sends its response + forkCmd_ :: (ServerConfig s -> Int) -> (ServerConfig s -> Int) -> M s () -> M s () -> M s (Maybe a) + forkCmd_ concurrency waitTimeout onTimeout cmd = do + let label = B.unpack $ "client $" <> encode sessionId <> " cmd" + -- Exceptions are masked so the slot is never lost between taking it and starting the thread. + -- The thread frees the slot when the command finishes; the caller frees it only if forking failed. + mask $ \restore -> + ifM + wait + (forkClient clnt label (restore cmd `finally` signal) `onException` signal) + (restore onTimeout) pure Nothing where + -- Takes a free slot at once if there is one. Otherwise waits for a slot until the client + -- timeout, after which the client no longer expects the response. + -- Other commands of this client are not processed while it waits. wait = do limit <- asks (concurrency . config) - atomically $ do - used <- readTVar procThreads - when (used >= limit) retry - writeTVar procThreads $! used + 1 + ifM (atomically $ takeSlot limit) (pure True) $ do + clientTimeout <- asks $ waitTimeout . config + waitStarted <- liftIO getSystemTime + expired <- liftIO $ registerDelay clientTimeout + taken <- atomically $ ifM (takeSlot limit) (pure True) (ifM (readTVar expired) (pure False) retry) + stats <- asks serverStats + liftIO $ updateSlotWaitStats stats waitStarted taken + pure taken + takeSlot limit = do + used <- readTVar procThreads + let free = used < limit + when free $ writeTVar procThreads $! used + 1 + pure free signal = atomically $ modifyTVar' procThreads (\t -> t - 1) rslvNamesEnv :: M s (Maybe NamesEnv) rslvNamesEnv = do @@ -1515,7 +1555,7 @@ client Nothing -> incStat (rslvDisabled st) $> Nothing Just nenv -> pure (Just nenv) -- Runs on a forked thread so RSLV does not block other commands; - -- concurrency is limited by serverResolverConcurrency in forkCmd. + -- concurrency is limited in forkCmd by serverClientConcurrency for direct and serverResolverConcurrency for forwarded RSLV. resolveNameMsg :: VersionSMP -> NamesEnv -> NameQuery -> M s BrokerMsg resolveNameMsg v nenv q = do st <- asks (rslvStats . serverStats) @@ -1545,7 +1585,7 @@ client Cmd SProxyService (RFWD encBlock) -> (response . (corrId, NoEntity,) =<<) <$> processForwardedCommand encBlock Cmd SResolver (RSLV d) -> rslvNamesEnv >>= \case Nothing -> pure $ response (corrId, NoEntity, ERR (NAME NO_RESOLVER)) - Just nenv -> forkCmd serverResolverConcurrency corrId NoEntity (resolveNameMsg (thVersion thParams') nenv d) + Just nenv -> forkCmd serverClientConcurrency corrId NoEntity rslvTooManyRequests (resolveNameMsg (thVersion thParams') nenv d) Cmd SSenderLink command -> case command of LKEY k -> withQueue $ \q qr -> checkMode QMMessaging qr $ secureQueue_ q k $>> getQueueLink_ q qr LGET -> withQueue $ \q qr -> checkContact qr $ getQueueLink_ q qr @@ -2181,9 +2221,12 @@ client Right t''@(_, (corrId', entId', cmd')) -> case cmd' of Cmd SResolver (RSLV d) -> lift $ rslvNamesEnv >>= \case Nothing -> pure $ Just (corrId', entId', ERR (NAME NO_RESOLVER)) - Just nenv -> forkCmd serverResolverConcurrency corrId NoEntity $ do - msg <- resolveNameMsg (thVersion clntTHParams) nenv d - either ERR id <$> runExceptT (encodeResp (corrId', entId', msg)) + Just nenv -> do + let sendResolved msg = do + rres <- either ERR id <$> runExceptT (encodeResp (corrId', entId', msg)) + atomically $ writeTBQueue sndQ ([(corrId, NoEntity, rres)], []) + forkCmd_ serverResolverConcurrency forwardedCmdWait (sendResolved rslvTooManyRequests) $ + resolveNameMsg (thVersion clntTHParams) nenv d >>= sendResolved -- INTERNAL because processCommand never returns Nothing for sender commands; -- `fst` drops the empty message only returned for SUB. _ -> Just . maybe (corrId', entId', ERR INTERNAL) fst <$> lift (processCommand Nothing (Right (M.empty, M.empty, M.empty)) t'') diff --git a/src/Simplex/Messaging/Server/Env/STM.hs b/src/Simplex/Messaging/Server/Env/STM.hs index 2c1843e51..79bfe21f4 100644 --- a/src/Simplex/Messaging/Server/Env/STM.hs +++ b/src/Simplex/Messaging/Server/Env/STM.hs @@ -68,6 +68,7 @@ module Simplex.Messaging.Server.Env.STM defaultInactiveClientExpiration, defaultProxyClientConcurrency, defaultNameResolverConcurrency, + defaultMaxClientCmdWait, defaultAuthErrorDelayMs, defaultMaxJournalMsgCount, defaultMaxJournalStateLines, @@ -201,11 +202,12 @@ data ServerConfig s = ServerConfig smpAgentCfg :: SMPClientAgentConfig, allowSMPProxy :: Bool, -- auth is the same with `newQueueBasicAuth` serverClientConcurrency :: Int, - -- | max concurrent name resolutions per connection, enforced in forkCmd. + -- | max concurrent forwarded name resolutions per proxy connection, enforced in forkCmd. -- Much higher than serverClientConcurrency: forwarded RSLVs from many clients -- aggregate over a single proxy->relay connection (only servers send proxied -- requests), so bounding them by the per-client limit would throttle unduly. serverResolverConcurrency :: Int, + maxClientCmdWait :: Int, -- | public-namespace resolver config; Nothing disables the names role namesConfig :: Maybe NamesConfig, -- | server public information @@ -252,10 +254,13 @@ defaultInactiveClientExpiration = } defaultProxyClientConcurrency :: Int -defaultProxyClientConcurrency = 32 +defaultProxyClientConcurrency = 500 defaultNameResolverConcurrency :: Int -defaultNameResolverConcurrency = 1000 +defaultNameResolverConcurrency = 5000 + +defaultMaxClientCmdWait :: Int +defaultMaxClientCmdWait = 60000000 -- microseconds defaultAuthErrorDelayMs :: Int64 defaultAuthErrorDelayMs = 50 diff --git a/src/Simplex/Messaging/Server/Main.hs b/src/Simplex/Messaging/Server/Main.hs index 4a3c909ed..64b4af287 100644 --- a/src/Simplex/Messaging/Server/Main.hs +++ b/src/Simplex/Messaging/Server/Main.hs @@ -611,6 +611,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath = allowSMPProxy = True, serverClientConcurrency = readIniDefault defaultProxyClientConcurrency "PROXY" "client_concurrency" ini, serverResolverConcurrency = readIniDefault defaultNameResolverConcurrency "NAMES" "resolver_concurrency" ini, + maxClientCmdWait = defaultMaxClientCmdWait, namesConfig = readNamesConfig ini, information = serverPublicInfo ini, startOptions diff --git a/src/Simplex/Messaging/Server/Main/Init.hs b/src/Simplex/Messaging/Server/Main/Init.hs index b82061ca3..d2938821c 100644 --- a/src/Simplex/Messaging/Server/Main/Init.hs +++ b/src/Simplex/Messaging/Server/Main/Init.hs @@ -154,7 +154,7 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds = \# `socks_mode` can be 'onion' for SOCKS proxy to be used for .onion destination hosts only (default)\n\ \# or 'always' to be used for all destination hosts (can be used if it is an .onion server).\n\ \# socks_mode = onion\n\n\ - \# Limit number of threads a client can spawn to process proxy commands in parrallel.\n" + \# Limit number of threads a client can spawn to process proxy commands and name resolutions in parallel.\n" <> ("# client_concurrency = " <> tshow defaultProxyClientConcurrency) <> "\n\n\ \[NAMES]\n\ @@ -169,7 +169,7 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds = \# resolver_auth: basic :\n\ \# resolver_timeout_ms: 3000\n\ \# resolver_max_response_bytes: 16000\n\ - \# Max concurrent name resolutions per connection (forwarded RSLVs from many\n\ + \# Max concurrent forwarded name resolutions per proxy connection (forwarded RSLVs from many\n\ \# clients share one proxy connection, so this is much higher than PROXY client_concurrency).\n" <> ("# resolver_concurrency = " <> tshow defaultNameResolverConcurrency) <> "\n\n\ diff --git a/src/Simplex/Messaging/Server/Prometheus.hs b/src/Simplex/Messaging/Server/Prometheus.hs index 864a2014f..940407569 100644 --- a/src/Simplex/Messaging/Server/Prometheus.hs +++ b/src/Simplex/Messaging/Server/Prometheus.hs @@ -127,6 +127,10 @@ prometheusMetrics sm rtm ts = _rcvServicesSubMsg, _rcvServicesSubDuplicate, _authDelayExceeded, + _cmdSlotWaits, + _cmdSlotWaitMs, + _cmdSlotWaitMaxMs, + _cmdSlotWaitTimeouts, _qCount, _msgCount, _ntfCount, @@ -197,6 +201,19 @@ prometheusMetrics sm rtm ts = \# HELP simplex_smp_auth_delay_exceeded Responses with AUTH errors sent without delay, as processing took longer than the delay.\n\ \# TYPE simplex_smp_auth_delay_exceeded counter\n\ \simplex_smp_auth_delay_exceeded " <> mshow _authDelayExceeded <> "\n# authDelayExceeded\n\ + \\n\ + \# HELP simplex_smp_cmd_slot_wait_ms Time commands waited for a free slot of the connection concurrency limit, in ms.\n\ + \# TYPE simplex_smp_cmd_slot_wait_ms summary\n\ + \simplex_smp_cmd_slot_wait_ms_sum " <> mshow _cmdSlotWaitMs <> "\n# cmdSlotWaitMs\n\ + \simplex_smp_cmd_slot_wait_ms_count " <> mshow _cmdSlotWaits <> "\n# cmdSlotWaits\n\ + \\n\ + \# HELP simplex_smp_cmd_slot_wait_ms_max The longest wait for a free command slot, in ms.\n\ + \# TYPE simplex_smp_cmd_slot_wait_ms_max gauge\n\ + \simplex_smp_cmd_slot_wait_ms_max " <> mshow _cmdSlotWaitMaxMs <> "\n# cmdSlotWaitMaxMs\n\ + \\n\ + \# HELP simplex_smp_cmd_slot_wait_timeouts Commands not processed because the client timed out while they waited for a free slot.\n\ + \# TYPE simplex_smp_cmd_slot_wait_timeouts counter\n\ + \simplex_smp_cmd_slot_wait_timeouts " <> mshow _cmdSlotWaitTimeouts <> "\n# cmdSlotWaitTimeouts\n\ \\n" subscriptions = "# Subscriptions\n\ diff --git a/src/Simplex/Messaging/Server/Stats.hs b/src/Simplex/Messaging/Server/Stats.hs index c1c599822..a1c9dfa29 100644 --- a/src/Simplex/Messaging/Server/Stats.hs +++ b/src/Simplex/Messaging/Server/Stats.hs @@ -129,6 +129,11 @@ data ServerStats = ServerStats rcvServicesSubMsg :: IORef Int, rcvServicesSubDuplicate :: IORef Int, authDelayExceeded :: IORef Int, + -- commands that waited for a free slot in forkCmd, total and max wait in ms, and waits that timed out + cmdSlotWaits :: IORef Int, + cmdSlotWaitMs :: IORef Int, + cmdSlotWaitMaxMs :: IORef Int, + cmdSlotWaitTimeouts :: IORef Int, qCount :: IORef Int, msgCount :: IORef Int, ntfCount :: IORef Int, @@ -192,6 +197,10 @@ data ServerStatsData = ServerStatsData _rcvServicesSubMsg :: Int, _rcvServicesSubDuplicate :: Int, _authDelayExceeded :: Int, + _cmdSlotWaits :: Int, + _cmdSlotWaitMs :: Int, + _cmdSlotWaitMaxMs :: Int, + _cmdSlotWaitTimeouts :: Int, _qCount :: Int, _msgCount :: Int, _ntfCount :: Int, @@ -257,6 +266,10 @@ newServerStats ts = do rcvServicesSubMsg <- newIORef 0 rcvServicesSubDuplicate <- newIORef 0 authDelayExceeded <- newIORef 0 + cmdSlotWaits <- newIORef 0 + cmdSlotWaitMs <- newIORef 0 + cmdSlotWaitMaxMs <- newIORef 0 + cmdSlotWaitTimeouts <- newIORef 0 qCount <- newIORef 0 msgCount <- newIORef 0 ntfCount <- newIORef 0 @@ -319,6 +332,10 @@ newServerStats ts = do rcvServicesSubMsg, rcvServicesSubDuplicate, authDelayExceeded, + cmdSlotWaits, + cmdSlotWaitMs, + cmdSlotWaitMaxMs, + cmdSlotWaitTimeouts, qCount, msgCount, ntfCount, @@ -383,6 +400,10 @@ getServerStatsData s = do _rcvServicesSubMsg <- readIORef $ rcvServicesSubMsg s _rcvServicesSubDuplicate <- readIORef $ rcvServicesSubDuplicate s _authDelayExceeded <- readIORef $ authDelayExceeded s + _cmdSlotWaits <- readIORef $ cmdSlotWaits s + _cmdSlotWaitMs <- readIORef $ cmdSlotWaitMs s + _cmdSlotWaitMaxMs <- readIORef $ cmdSlotWaitMaxMs s + _cmdSlotWaitTimeouts <- readIORef $ cmdSlotWaitTimeouts s _qCount <- readIORef $ qCount s _msgCount <- readIORef $ msgCount s _ntfCount <- readIORef $ ntfCount s @@ -445,6 +466,10 @@ getServerStatsData s = do _rcvServicesSubMsg, _rcvServicesSubDuplicate, _authDelayExceeded, + _cmdSlotWaits, + _cmdSlotWaitMs, + _cmdSlotWaitMaxMs, + _cmdSlotWaitTimeouts, _qCount, _msgCount, _ntfCount, @@ -510,6 +535,10 @@ setServerStats s d = do writeIORef (rcvServicesSubMsg s) $! _rcvServicesSubMsg d writeIORef (rcvServicesSubDuplicate s) $! _rcvServicesSubDuplicate d writeIORef (authDelayExceeded s) $! _authDelayExceeded d + writeIORef (cmdSlotWaits s) $! _cmdSlotWaits d + writeIORef (cmdSlotWaitMs s) $! _cmdSlotWaitMs d + writeIORef (cmdSlotWaitMaxMs s) $! _cmdSlotWaitMaxMs d + writeIORef (cmdSlotWaitTimeouts s) $! _cmdSlotWaitTimeouts d writeIORef (qCount s) $! _qCount d writeIORef (msgCount s) $! _msgCount d writeIORef (ntfCount s) $! _ntfCount d @@ -713,6 +742,10 @@ instance StrEncoding ServerStatsData where _rcvServicesSubMsg = 0, _rcvServicesSubDuplicate = 0, _authDelayExceeded = 0, + _cmdSlotWaits = 0, + _cmdSlotWaitMs = 0, + _cmdSlotWaitMaxMs = 0, + _cmdSlotWaitTimeouts = 0, _qCount, _msgCount = 0, _ntfCount = 0, diff --git a/tests/RSLVTests.hs b/tests/RSLVTests.hs index 9263b6953..a52aa2a86 100644 --- a/tests/RSLVTests.hs +++ b/tests/RSLVTests.hs @@ -11,11 +11,14 @@ module RSLVTests (rslvTests) where +import Control.Concurrent (threadDelay) +import Control.Monad (forM_) import Control.Monad.Trans.Except (ExceptT, runExceptT) import qualified Data.ByteString.Char8 as B import qualified Data.ByteString.Lazy as LB import Data.IORef (IORef, readIORef) import Data.List.NonEmpty (NonEmpty (..)) +import qualified Data.List.NonEmpty as L import Data.Text (Text) import Data.Text.Encoding (encodeUtf8) import Data.Time.Clock (getCurrentTime) @@ -47,6 +50,7 @@ import Simplex.Messaging.Protocol tPut, ) import qualified Simplex.Messaging.Protocol as SMP +import Simplex.Messaging.Server.Env.STM (ServerConfig (..)) import Simplex.Messaging.SimplexName (SimplexDomain) import Simplex.Messaging.Transport import Simplex.Messaging.Version (mkVersionRange) @@ -102,6 +106,10 @@ rslvTests = do it "RSLV sends the 2LD as its hash" testRslvSendsTheHash it "a name with subnames is sent as text" testSubnameKeepsItsLabels it "a record naming a different name is rejected" testRslvWrongName + describe "RSLV resource use" $ do + it "one connection has at most client_concurrency lookups in flight" testRslvConnectionCap + it "a lookup waiting for a slot runs when a slot is released" testRslvSlotWait + it "a lookup waiting for a slot fails when the client times out" testRslvSlotWaitTimeout -- | /v2/resolve answers 200, 400 or 502, so a 404 is a resolver that predates -- the route, not a name that does not exist. @@ -275,5 +283,84 @@ testRslvWrongName = Left (PCEUnexpectedResponse _) -> pure () _ -> expectationFailure $ "expected Left (PCEUnexpectedResponse ..), got: " <> show r +-- The resolver answers after 3s and the server gives up after 1s, so a request +-- that reached the resolver stays in flight for the whole check. +testRslvConnectionCap :: IO () +testRslvConnectionCap = + NRS.withResolverServerDelayed 3000 (NRS.resolveResp status200 "{}") $ \port reqs -> + withSmpServerConfigOn (transport @TLS) (updateCfg (withNames port memCfg) $ \c -> c {serverClientConcurrency = connCap}) testPort $ const $ + testSMPClient @TLS $ \h -> do + sendRslvs h "cap" 16 + threadDelay 800000 + length <$> resolvePaths reqs `shouldReturn` connCap + recvResponses h 16 `shouldReturn` replicate 16 (Right (ERR (NAME (RESOLVER "timeout")))) + where + connCap = 4 + +-- 4 lookups take the slots for the 1s resolver timeout, the 5th waits for a slot. +testRslvSlotWait :: IO () +testRslvSlotWait = + NRS.withResolverServerDelayed 3000 (NRS.resolveResp status200 "{}") $ \port reqs -> + withSmpServerConfigOn (transport @TLS) (slotWaitCfg port 10000000) testPort $ const $ + testSMPClient @TLS $ \h -> do + sendRslvs h "wait" 5 + recvResponses h 5 `shouldReturn` replicate 5 (Right (ERR (NAME (RESOLVER "timeout")))) + length <$> resolvePaths reqs `shouldReturn` 5 + threadDelay 1500000 + SlotWaitMetrics {waits, waitTimeouts, waitMaxMs, waitSumMs} <- readSlotWaitMetrics + (waits, waitTimeouts) `shouldBe` (1, 0) + waitMaxMs `shouldSatisfy` (\t -> t >= 500 && t < 2000) + waitSumMs `shouldBe` waitMaxMs + +-- The client times out after 0.2s, so the 2 waiting lookups fail before the slots are released. +-- The connection does not process further commands while one waits, so the 6th waits after the 5th times out. +testRslvSlotWaitTimeout :: IO () +testRslvSlotWaitTimeout = + NRS.withResolverServerDelayed 3000 (NRS.resolveResp status200 "{}") $ \port reqs -> + withSmpServerConfigOn (transport @TLS) (slotWaitCfg port 200000) testPort $ const $ + testSMPClient @TLS $ \h -> do + sendRslvs h "wait" 6 + recvResponses h 6 `shouldReturn` (replicate 2 (Right (ERR (NAME (RESOLVER "too many requests")))) <> replicate 4 (Right (ERR (NAME (RESOLVER "timeout"))))) + threadDelay 1500000 + length <$> resolvePaths reqs `shouldReturn` 4 + SlotWaitMetrics {waits, waitTimeouts, waitMaxMs, waitSumMs} <- readSlotWaitMetrics + (waits, waitTimeouts) `shouldBe` (2, 2) + waitMaxMs `shouldSatisfy` (\t -> t >= 200 && t < 1000) + waitSumMs `shouldSatisfy` (>= 400) + +slotWaitCfg :: Int -> Int -> AServerConfig +slotWaitCfg port clientTimeout = + updateCfg (withNames port memCfg) $ \c -> c {serverClientConcurrency = 4, prometheusInterval = Just 1, maxClientCmdWait = clientTimeout} + +data SlotWaitMetrics = SlotWaitMetrics {waits :: Int, waitTimeouts :: Int, waitMaxMs :: Int, waitSumMs :: Int} + +readSlotWaitMetrics :: IO SlotWaitMetrics +readSlotWaitMetrics = do + ms <- map words . lines <$> readFile testPrometheusMetricsFile + -- metric lines are "name value timestamp" + let metric name = case [v | n : v : _ <- ms, n == name] of + [v] -> pure $ read v + _ -> fail $ "no metric " <> name + SlotWaitMetrics + <$> metric "simplex_smp_cmd_slot_wait_ms_count" + <*> metric "simplex_smp_cmd_slot_wait_timeouts" + <*> metric "simplex_smp_cmd_slot_wait_ms_max" + <*> metric "simplex_smp_cmd_slot_wait_ms_sum" + +-- | One RSLV per block, so no batch limit applies. +sendRslvs :: THandleSMP TLS 'TClient -> String -> Int -> IO () +sendRslvs h@THandle {params} prefix n = + forM_ [1 .. n] $ \i -> do + let TransmissionForAuth {tToSend} = encodeTransmissionForAuth params (CorrId (B.pack $ prefix <> show i), NoEntity, Cmd SResolver (RSLV (NQDomain (domain "alice.simplex")))) + [Right ()] <- tPut h (Right (Nothing, tToSend) :| []) + pure () + +recvResponses :: THandleSMP TLS 'TClient -> Int -> IO [Either ErrorType BrokerMsg] +recvResponses h n + | n <= 0 = pure [] + | otherwise = do + rs <- map (\(_, _, r) -> r) . L.toList <$> tGetClient h + (rs <>) <$> recvResponses h (n - length rs) + runExceptT' :: Show e => ExceptT e IO a -> IO a runExceptT' a = runExceptT a >>= either (fail . show) pure diff --git a/tests/SMPClient.hs b/tests/SMPClient.hs index 2db4a9819..dd84bd6e9 100644 --- a/tests/SMPClient.hs +++ b/tests/SMPClient.hs @@ -278,6 +278,7 @@ cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg -> allowSMPProxy = False, serverClientConcurrency = 2, serverResolverConcurrency = defaultNameResolverConcurrency, + maxClientCmdWait = defaultMaxClientCmdWait, namesConfig = Nothing, information = Nothing, startOptions = defaultStartOptions