From 4310a69391fc791015d86a7b8449156b5e441ffa Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Sat, 4 May 2024 09:38:01 +0100 Subject: [PATCH] revert not expiring sending subs --- src/Simplex/Messaging/Client.hs | 49 ++++++++++----------------------- 1 file changed, 15 insertions(+), 34 deletions(-) diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index 3564990c7..0d8ec0df8 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/src/Simplex/Messaging/Client.hs @@ -113,7 +113,6 @@ import Simplex.Messaging.Transport.WebSockets (WS) import Simplex.Messaging.Util (bshow, diffToMicroseconds, raceAny_, threadDelay', whenM) import Simplex.Messaging.Version import System.Timeout (timeout) -import UnliftIO (pooledMapConcurrentlyN) -- | 'SMPClient' is a handle used to send commands to a specific SMP server. -- @@ -136,7 +135,7 @@ data PClient v err msg = PClient timeoutErrorCount :: TVar Int, clientCorrId :: TVar ChaChaDRG, sentCommands :: TMap CorrId (Request err msg), - sndQ :: TBQueue (Maybe (TVar Bool), ByteString), + sndQ :: TBQueue (TVar Bool, ByteString), rcvQ :: TBQueue (NonEmpty (SignedTransmission err msg)), msgQ :: Maybe (TBQueue (ServerTransmission v msg)) } @@ -407,9 +406,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize send :: Transport c => ProtocolClient v err msg -> THandle v c 'TClient -> IO () send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= sendActive where - sendActive (Nothing, s) = send_ s - sendActive (Just active, s) = whenM (readTVarIO active) $ send_ s - send_ = void . tPutLog h + sendActive (active, s) = whenM (readTVarIO active) $ void $ tPutLog h s receive :: Transport c => ProtocolClient v err msg -> THandle v c 'TClient -> IO () receive ProtocolClient {client_ = PClient {rcvQ, lastReceived, timeoutErrorCount}} h = forever $ do @@ -534,10 +531,7 @@ createSMPQueue c (rKey, rpKey) dhKey auth subMode = subscribeSMPQueue :: SMPClient -> RcvPrivateAuthKey -> RecipientId -> ExceptT SMPClientError IO () subscribeSMPQueue c@ProtocolClient {client_ = PClient {sendPings}} rpKey rId = do liftIO . atomically $ writeTVar sendPings True - -- We are not expiring sending subscriptions even if it expires to increase chances of subscription - -- succeeding without retries - the subscription is registered in the agent when uncorrelated MSG response - -- is received via the active client for pending queue subscription - it also prevents unnecessary retries. - sendProtocolCommand' False c (Just rpKey) rId (Cmd SRecipient SUB) >>= \case + sendSMPCommand c (Just rpKey) rId SUB >>= \case OK -> pure () cmd@MSG {} -> liftIO $ writeSMPMessage c rId cmd r -> throwE . PCEUnexpectedResponse $ bshow r @@ -546,8 +540,7 @@ subscribeSMPQueue c@ProtocolClient {client_ = PClient {sendPings}} rpKey rId = d subscribeSMPQueues :: SMPClient -> NonEmpty (RcvPrivateAuthKey, RecipientId) -> IO (NonEmpty (Either SMPClientError ())) subscribeSMPQueues c@ProtocolClient {client_ = PClient {sendPings}} qs = do atomically $ writeTVar sendPings True - -- See comment in subscribeSMPQueue - sendProtocolCommands' False c cs >>= mapM (processSUBResponse c) + sendProtocolCommands c cs >>= mapM (processSUBResponse c) where cs = L.map (\(rpKey, rId) -> (Just rpKey, rId, Cmd SRecipient SUB)) qs @@ -696,13 +689,9 @@ type PCTransmission err msg = (Either TransportError SentRawTransmission, Reques -- | Send multiple commands with batching and collect responses sendProtocolCommands :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg)) -sendProtocolCommands = sendProtocolCommands' True -{-# INLINE sendProtocolCommands #-} - -sendProtocolCommands' :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => Bool -> ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg)) -sendProtocolCommands' expire c@ProtocolClient {thParams = THandleParams {batch, blockSize}} cs = do +sendProtocolCommands c@ProtocolClient {thParams = THandleParams {batch, blockSize}} cs = do bs <- batchTransmissions' batch blockSize <$> mapM (mkTransmission c) cs - validate . concat =<< mapM (sendBatch c expire) bs + validate . concat =<< mapM (sendBatch c) bs where validate :: [Response err msg] -> IO (NonEmpty (Response err msg)) validate rs @@ -719,36 +708,28 @@ sendProtocolCommands' expire c@ProtocolClient {thParams = THandleParams {batch, streamProtocolCommands :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> ([Response err msg] -> IO ()) -> IO () streamProtocolCommands c@ProtocolClient {thParams = THandleParams {batch, blockSize}} cs cb = do bs <- batchTransmissions' batch blockSize <$> mapM (mkTransmission c) cs - mapM_ (cb <=< sendBatch c True) bs + mapM_ (cb <=< sendBatch c) bs -sendBatch :: ProtocolClient v err msg -> Bool -> TransportBatch (Request err msg) -> IO [Response err msg] -sendBatch c@ProtocolClient {client_ = PClient {rcvConcurrency, sndQ}} expire b = do +sendBatch :: ProtocolClient v err msg -> TransportBatch (Request err msg) -> IO [Response err msg] +sendBatch c@ProtocolClient {client_ = PClient {sndQ}} b = do case b of TBError e Request {entityId} -> do putStrLn "send error: large message" pure [Response entityId $ Left $ PCETransportError e] TBTransmissions s n rs | n > 0 -> do - active <- mkActive_ expire + active <- newTVarIO True atomically $ writeTBQueue sndQ (active, s) mapConcurrently (getResponse c active) rs | otherwise -> pure [] TBTransmission s r -> do - active <- mkActive_ expire + active <- newTVarIO True atomically $ writeTBQueue sndQ (active, s) (: []) <$> getResponse c active r -mkActive_ :: Bool -> IO (Maybe (TVar Bool)) -mkActive_ True = Just <$> newTVarIO True -mkActive_ False = pure Nothing - -- | Send Protocol command sendProtocolCommand :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> Maybe C.APrivateAuthKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg -sendProtocolCommand = sendProtocolCommand' True -{-# INLINE sendProtocolCommand #-} - -sendProtocolCommand' :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => Bool -> ProtocolClient v err msg -> Maybe C.APrivateAuthKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg -sendProtocolCommand' expire c@ProtocolClient {client_ = PClient {sndQ}, thParams = THandleParams {batch, blockSize}} pKey entId cmd = +sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THandleParams {batch, blockSize}} pKey entId cmd = ExceptT $ uncurry sendRecv =<< mkTransmission c (pKey, entId, cmd) where -- two separate "atomically" needed to avoid blocking @@ -758,7 +739,7 @@ sendProtocolCommand' expire c@ProtocolClient {client_ = PClient {sndQ}, thParams Right t | B.length s > blockSize - 2 -> pure . Left $ PCETransportError TELargeMsg | otherwise -> do - active <- mkActive_ expire + active <- newTVarIO True atomically (writeTBQueue sndQ (active, s)) response <$> getResponse c active r where @@ -767,13 +748,13 @@ sendProtocolCommand' expire c@ProtocolClient {client_ = PClient {sndQ}, thParams | otherwise = tEncode t -- TODO switch to timeout or TimeManager that supports Int64 -getResponse :: ProtocolClient v err msg -> Maybe (TVar Bool) -> Request err msg -> IO (Response err msg) +getResponse :: ProtocolClient v err msg -> TVar Bool -> Request err msg -> IO (Response err msg) getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount, sentCommands}} active Request {corrId, entityId, responseVar} = do response <- timeout tcpTimeout (atomically (takeTMVar responseVar)) >>= \case Just r -> atomically (writeTVar timeoutErrorCount 0) $> r Nothing -> do - atomically (mapM_ (`writeTVar` False) active >> TM.delete corrId sentCommands) + atomically (writeTVar active False >> TM.delete corrId sentCommands) atomically $ modifyTVar' timeoutErrorCount (+ 1) pure $ Left PCEResponseTimeout pure Response {entityId, response}