diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 79404f1f9..122b2a28c 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -11,7 +11,6 @@ {-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE OverloadedLists #-} {-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TupleSections #-} @@ -99,7 +98,7 @@ 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 (NetworkConfig (..), NetworkRequestMode (..), ProtocolClient (thParams), ProtocolClientConfig (..), ProtocolClientError (..), SMPClient, SMPClientError, clientHandlers, forwardSMPTransmission, netTimeoutInt, smpProxyError, temporaryClientError, pattern NRMInteractive) +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 @@ -1222,6 +1221,9 @@ updateSlotWaitStats ServerStats {cmdSlotWaits, cmdSlotWaitMs, cmdSlotWaitMaxMs, atomicModifyIORef'_ cmdSlotWaitMaxMs (max waitMs) unless taken $ incStat cmdSlotWaitTimeouts +forwardedCmdWait :: ServerConfig s -> Int +forwardedCmdWait = (`netTimeoutInt` NRMBackground) . tcpTimeout . networkConfig . smpCfg . smpAgentCfg + microseconds :: SystemTime -> Int64 microseconds (MkSystemTime s ns) = s * 1000000 + fromIntegral (ns `div` 1000) @@ -1507,12 +1509,12 @@ client 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 NRMInteractive (sendResp timeoutErr) $ cmdAction >>= sendResp + 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) -> NetworkRequestMode -> M s () -> M s () -> M s (Maybe a) - forkCmd_ concurrency nm onTimeout cmd = do + 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. @@ -1529,7 +1531,7 @@ client wait = do limit <- asks (concurrency . config) ifM (atomically $ takeSlot limit) (pure True) $ do - clientTimeout <- asks $ (`netTimeoutInt` nm) . tcpTimeout . networkConfig . smpCfg . smpAgentCfg . config + clientTimeout <- asks $ waitTimeout . config waitStarted <- liftIO getSystemTime expired <- liftIO $ registerDelay clientTimeout taken <- atomically $ (True <$ (takeSlot limit >>= check)) `orElse` (False <$ (readTVar expired >>= check)) @@ -2218,7 +2220,7 @@ client Nothing -> pure $ Just (corrId', entId', ERR (NAME NO_RESOLVER)) -- The proxy waits for forwarded commands with the background timeout. -- If no slot is free before it, there is no response, as the proxy has already timed out. - Just nenv -> forkCmd_ serverResolverConcurrency NRMBackground (pure ()) $ do + Just nenv -> forkCmd_ serverResolverConcurrency forwardedCmdWait (pure ()) $ do msg <- resolveNameMsg (thVersion clntTHParams) nenv d rres <- either ERR id <$> runExceptT (encodeResp (corrId', entId', msg)) atomically $ writeTBQueue sndQ ([(corrId, NoEntity, rres)], []) diff --git a/src/Simplex/Messaging/Server/Env/STM.hs b/src/Simplex/Messaging/Server/Env/STM.hs index d06c56e4e..7f6e71749 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, @@ -206,6 +207,7 @@ data ServerConfig s = ServerConfig -- 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 @@ -257,6 +259,9 @@ defaultProxyClientConcurrency = 32 defaultNameResolverConcurrency :: Int defaultNameResolverConcurrency = 1000 +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/tests/RSLVTests.hs b/tests/RSLVTests.hs index 1731f8886..a52aa2a86 100644 --- a/tests/RSLVTests.hs +++ b/tests/RSLVTests.hs @@ -27,7 +27,6 @@ import NamesResolverServer (memCfg, memCfg2, memProxyCfg, withNames) import qualified NamesResolverServer as NRS import SMPClient import Simplex.Messaging.Client -import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (..)) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding.String (strDecode) import SMPNamesTests (availableBody, registeredBody, reservedBody, resolved, testNameRecord, testPricing) @@ -331,11 +330,7 @@ testRslvSlotWaitTimeout = slotWaitCfg :: Int -> Int -> AServerConfig slotWaitCfg port clientTimeout = - updateCfg (withNames port memCfg) $ \c -> - let aCfg = smpAgentCfg c - cCfg = smpCfg aCfg - nt = NetworkTimeout {backgroundTimeout = clientTimeout, interactiveTimeout = clientTimeout} - in c {serverClientConcurrency = 4, prometheusInterval = Just 1, smpAgentCfg = aCfg {smpCfg = cCfg {networkConfig = (networkConfig cCfg) {tcpTimeout = nt}}}} + updateCfg (withNames port memCfg) $ \c -> c {serverClientConcurrency = 4, prometheusInterval = Just 1, maxClientCmdWait = clientTimeout} data SlotWaitMetrics = SlotWaitMetrics {waits :: Int, waitTimeouts :: Int, waitMaxMs :: Int, waitSumMs :: Int} 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