From 19aef52135b288dc92c4a2ec6d0be4b8283649ab Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com> Date: Sat, 22 Oct 2022 16:36:32 +0100 Subject: [PATCH] debug agent locks --- simplexmq.cabal | 1 + src/Simplex/Messaging/Agent.hs | 37 ++++++++++++++++++--------- src/Simplex/Messaging/Agent/Client.hs | 34 ++++++++++++++---------- src/Simplex/Messaging/Agent/Lock.hs | 24 +++++++++++++++++ 4 files changed, 70 insertions(+), 26 deletions(-) create mode 100644 src/Simplex/Messaging/Agent/Lock.hs diff --git a/simplexmq.cabal b/simplexmq.cabal index cb39de6e9..4fddcc1e3 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -37,6 +37,7 @@ library Simplex.Messaging.Agent Simplex.Messaging.Agent.Client Simplex.Messaging.Agent.Env.SQLite + Simplex.Messaging.Agent.Lock Simplex.Messaging.Agent.NtfSubSupervisor Simplex.Messaging.Agent.Protocol Simplex.Messaging.Agent.QueryString diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 41e25303f..886b4ccaf 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -74,6 +74,7 @@ module Simplex.Messaging.Agent activateAgent, suspendAgent, execAgentStoreSQL, + debugAgentLocks, logConnection, ) where @@ -86,6 +87,7 @@ import Control.Monad.Reader import Crypto.Random (MonadRandom) import Data.Bifunctor (bimap, first, second) import Data.ByteString.Char8 (ByteString) +import qualified Data.ByteString.Char8 as B import Data.Composition ((.:), (.:.), (.::)) import Data.Functor (($>)) import Data.List (deleteFirstsBy) @@ -288,6 +290,9 @@ suspendAgent c = withAgentEnv c . suspendAgent' c execAgentStoreSQL :: AgentErrorMonad m => AgentClient -> Text -> m [Text] execAgentStoreSQL c = withAgentEnv c . execAgentStoreSQL' c +debugAgentLocks :: AgentErrorMonad m => AgentClient -> m AgentLocks +debugAgentLocks c = withAgentEnv c $ debugAgentLocks' c + withAgentEnv :: AgentClient -> ReaderT Env m a -> m a withAgentEnv c = (`runReaderT` agentEnv c) @@ -498,7 +503,7 @@ createReplyQueue c ConnData {connId, enableNtfs} SndQueue {smpClientVersion} srv -- | Approve confirmation (LET command) in Reader monad allowConnection' :: AgentMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m () -allowConnection' c connId confId ownConnInfo = withConnLock c connId $ do +allowConnection' c connId confId ownConnInfo = withConnLock c connId "allowConnection" $ do withStore c (`getConn` connId) >>= \case SomeConn _ (RcvConnection _ rq@RcvQueue {server, rcvId, e2ePrivKey, smpClientVersion = v}) -> do senderKey <- withStore c $ \db -> runExceptT $ do @@ -512,7 +517,7 @@ allowConnection' c connId confId ownConnInfo = withConnLock c connId $ do -- | Accept contact (ACPT command) in Reader monad acceptContact' :: AgentMonad m => AgentClient -> ConnId -> Bool -> InvitationId -> ConnInfo -> m ConnId -acceptContact' c connId enableNtfs invId ownConnInfo = withConnLock c connId $ do +acceptContact' c connId enableNtfs invId ownConnInfo = withConnLock c connId "acceptContact" $ do Invitation {contactConnId, connReq} <- withStore c (`getInvitation` invId) withStore c (`getConn` contactConnId) >>= \case SomeConn _ ContactConnection {} -> do @@ -656,7 +661,7 @@ getNotificationMessage' c nonce encNtfInfo = do -- | Send message to the connection (SEND command) in Reader monad sendMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> MsgFlags -> MsgBody -> m AgentMsgId -sendMessage' c connId msgFlags msg = withConnLock c connId $ do +sendMessage' c connId msgFlags msg = withConnLock c connId "sendMessage" $ do withStore c (`getConn` connId) >>= \case SomeConn _ (DuplexConnection cData _ sq) -> enqueueMsg cData sq SomeConn _ (SndConnection cData sq) -> enqueueMsg cData sq @@ -740,9 +745,9 @@ runCommandProcessing c@AgentClient {subQ} server = do _ -> notify $ ERR $ INTERNAL $ "unsupported async command " <> show (aCommandTag cmd) AInternalCommand cmd -> case server of Just _srv -> case cmd of - ICAckDel _rId srvMsgId msgId -> tryWithLock $ ack _rId srvMsgId >> withStore' c (\db -> deleteMsg db connId msgId) - ICAck _rId srvMsgId -> tryWithLock $ ack _rId srvMsgId - ICAllowSecure _rId senderKey -> tryWithLock $ do + ICAckDel _rId srvMsgId msgId -> tryWithLock "ICAckDel" $ ack _rId srvMsgId >> withStore' c (\db -> deleteMsg db connId msgId) + ICAck _rId srvMsgId -> tryWithLock "ICAck" $ ack _rId srvMsgId + ICAllowSecure _rId senderKey -> tryWithLock "ICAllowSecure" $ do (SomeConn _ conn, AcceptedConfirmation {senderConf, ownConnInfo}) <- withStore c $ \db -> runExceptT $ (,) <$> ExceptT (getConn db connId) <*> ExceptT (getAcceptedConfirmation db connId) case conn of @@ -750,7 +755,7 @@ runCommandProcessing c@AgentClient {subQ} server = do secure rq senderKey mapM_ (connectReplyQueues c cData ownConnInfo) (L.nonEmpty $ smpReplyQueues senderConf) _ -> throwError $ INTERNAL $ "incorrect connection type " <> show (internalCmdTag cmd) - ICDuplexSecure _rId senderKey -> tryWithLock $ do + ICDuplexSecure _rId senderKey -> tryWithLock "ICDuplexSecure" $ do SomeConn _ conn <- withStore c (`getConn` connId) case conn of DuplexConnection cData rq sq -> do @@ -775,7 +780,7 @@ runCommandProcessing c@AgentClient {subQ} server = do | temporaryAgentError e || e == BROKER HOST -> retrySndOp c loop | otherwise -> notify (ERR e) >> withStore' c (`deleteCommand` cmdId) Right () -> withStore' c (`deleteCommand` cmdId) - tryWithLock = tryCommand . withConnLock c connId + tryWithLock name = tryCommand . withConnLock c connId name notify cmd = atomically $ writeTBQueue subQ (corrId, connId, cmd) withNextSrv :: TVar [SMPServer] -> [SMPServer] -> (SMPServer -> m ()) -> m () withNextSrv usedSrvs initUsed action = do @@ -948,7 +953,7 @@ retrySndOp c loop = do loop ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m () -ackMessage' c connId msgId = withConnLock c connId $ do +ackMessage' c connId msgId = withConnLock c connId "ackMessage" $ do withStore c (`getConn` connId) >>= \case SomeConn _ (DuplexConnection _ rq _) -> ack rq SomeConn _ (RcvConnection _ rq) -> ack rq @@ -971,7 +976,7 @@ ackQueueMessage c rq srvMsgId = -- | Suspend SMP agent connection (OFF command) in Reader monad suspendConnection' :: AgentMonad m => AgentClient -> ConnId -> m () -suspendConnection' c connId = withConnLock c connId $ do +suspendConnection' c connId = withConnLock c connId "suspendConnection" $ do withStore c (`getConn` connId) >>= \case SomeConn _ (DuplexConnection _ rq _) -> suspendQueue c rq SomeConn _ (RcvConnection _ rq) -> suspendQueue c rq @@ -981,7 +986,7 @@ suspendConnection' c connId = withConnLock c connId $ do -- | Delete SMP agent connection (DEL command) in Reader monad deleteConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m () -deleteConnection' c connId = withConnLock c connId $ do +deleteConnection' c connId = withConnLock c connId "deleteConnection" $ do withStore c (`getConn` connId) >>= \case SomeConn _ (DuplexConnection _ rq _) -> delete rq SomeConn _ (RcvConnection _ rq) -> delete rq @@ -1245,6 +1250,14 @@ suspendAgent' c@AgentClient {agentState = as} maxDelay = do execAgentStoreSQL' :: AgentMonad m => AgentClient -> Text -> m [Text] execAgentStoreSQL' c sql = withStore' c (`execSQL` sql) +debugAgentLocks' :: AgentMonad m => AgentClient -> m AgentLocks +debugAgentLocks' AgentClient {connLocks = cs, reconnectLocks = rs} = do + connLocks <- getLocks cs + srvLocks <- getLocks rs + pure AgentLocks {connLocks, srvLocks} + where + getLocks ls = atomically $ M.mapKeys (B.unpack . strEncode) . M.mapMaybe id <$> (mapM tryReadTMVar =<< readTVar ls) + getSMPServer :: AgentMonad m => AgentClient -> m SMPServer getSMPServer c = readTVarIO (smpServers c) >>= pickServer @@ -1281,7 +1294,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm _ -> atomically $ writeTBQueue subQ ("", "", ERR $ CONN NOT_FOUND) where processSMP :: Connection c -> ConnData -> RcvQueue -> m () - processSMP conn cData@ConnData {connId, duplexHandshake} rq@RcvQueue {e2ePrivKey, e2eDhSecret, status} = withConnLock c connId $ + processSMP conn cData@ConnData {connId, duplexHandshake} rq@RcvQueue {e2ePrivKey, e2eDhSecret, status} = withConnLock c connId "processSMP" $ case cmd of SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId} -> handleNotifyAck $ do SMP.ClientRcvMsgBody {msgTs = srvTs, msgFlags, msgBody} <- decryptSMPMessage v rq msg diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index ca670d612..552a545e2 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -1,6 +1,7 @@ {-# LANGUAGE AllowAmbiguousTypes #-} {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} @@ -56,6 +57,7 @@ module Simplex.Messaging.Agent.Client AgentOperation (..), AgentOpState (..), AgentState (..), + AgentLocks (..), agentOperations, agentOperationBracket, waitUntilActive, @@ -80,6 +82,8 @@ import Control.Logger.Simple import Control.Monad.Except import Control.Monad.IO.Unlift import Control.Monad.Reader +import Data.Aeson (ToJSON) +import qualified Data.Aeson as J import Data.Bifunctor (bimap, first, second) import Data.ByteString.Base64 import Data.ByteString.Char8 (ByteString) @@ -96,7 +100,9 @@ import qualified Data.Set as S import Data.Text.Encoding import Data.Word (Word16) import qualified Database.SQLite.Simple as DB +import GHC.Generics (Generic) import Simplex.Messaging.Agent.Env.SQLite +import Simplex.Messaging.Agent.Lock import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.RetryInterval import Simplex.Messaging.Agent.Store @@ -177,9 +183,9 @@ data AgentClient = AgentClient agentState :: TVar AgentState, getMsgLocks :: TMap (SMPServer, SMP.RecipientId) (TMVar ()), -- locks to prevent concurrent operations with connection - connLocks :: TMap ConnId (TMVar ()), + connLocks :: TMap ConnId Lock, -- locks to prevent concurrent reconnections to SMP servers - reconnectLocks :: TMap SMPServer (TMVar ()), + reconnectLocks :: TMap SMPServer Lock, reconnections :: TVar [Async ()], asyncClients :: TVar [Async ()], clientId :: Int, @@ -205,6 +211,11 @@ data AgentOpState = AgentOpState {opSuspended :: Bool, opsInProgress :: Int} data AgentState = ASActive | ASSuspending | ASSuspended deriving (Eq, Show) +data AgentLocks = AgentLocks {connLocks :: Map String String, srvLocks :: Map String String} + deriving (Show, Generic) + +instance ToJSON AgentLocks where toEncoding = J.genericToEncoding J.defaultOptions + newAgentClient :: InitialAgentServers -> Env -> STM AgentClient newAgentClient InitialAgentServers {smp, ntf, netCfg} agentEnv = do let qSize = tbqSize $ config agentEnv @@ -305,7 +316,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} srv = do reconnectClient :: m () reconnectClient = - withLock_ (reconnectLocks c) srv $ + withLockMap_ (reconnectLocks c) srv "reconnect" $ atomically (TM2.lookup1 srv (pendingSubs c) >>= mapM readTVar) >>= mapM_ resubscribe where @@ -446,19 +457,14 @@ closeProtocolServerClients c clientsSel = cancelActions :: (Foldable f, Monoid (f (Async ()))) => TVar (f (Async ())) -> IO () cancelActions as = readTVarIO as >>= mapM_ (forkIO . uninterruptibleCancel) >> atomically (writeTVar as mempty) -withConnLock :: MonadUnliftIO m => AgentClient -> ConnId -> m a -> m a -withConnLock _ "" = id -withConnLock AgentClient {connLocks} connId = withLock_ connLocks connId +withConnLock :: MonadUnliftIO m => AgentClient -> ConnId -> String -> m a -> m a +withConnLock _ "" _ = id +withConnLock AgentClient {connLocks} connId name = withLockMap_ connLocks connId name -withLock_ :: (Ord k, MonadUnliftIO m) => TMap k (TMVar ()) -> k -> m a -> m a -withLock_ locks connId = - E.bracket - (atomically $ getLock >>= \l -> takeTMVar l $> l) - (atomically . (`putTMVar` ())) - . const +withLockMap_ :: (Ord k, MonadUnliftIO m) => TMap k Lock -> k -> String -> m a -> m a +withLockMap_ locks key = withGetLock $ TM.lookup key locks >>= maybe newLock pure where - getLock = TM.lookup connId locks >>= maybe newLock pure - newLock = newTMVar () >>= \l -> TM.insert connId l locks $> l + newLock = newEmptyTMVar >>= \l -> TM.insert key l locks $> l withClient_ :: forall a m msg. (AgentMonad m, ProtocolServerClient msg) => AgentClient -> ProtoServer msg -> (ProtocolClient msg -> m a) -> m a withClient_ c srv action = (getProtocolServerClient c srv >>= action) `catchError` logServerError diff --git a/src/Simplex/Messaging/Agent/Lock.hs b/src/Simplex/Messaging/Agent/Lock.hs new file mode 100644 index 000000000..eca04d0aa --- /dev/null +++ b/src/Simplex/Messaging/Agent/Lock.hs @@ -0,0 +1,24 @@ +{-# LANGUAGE NamedFieldPuns #-} + +module Simplex.Messaging.Agent.Lock where + +import Control.Monad (void) +import Control.Monad.IO.Unlift +import Data.Functor (($>)) +import qualified UnliftIO.Exception as E +import UnliftIO.STM + +type Lock = TMVar String + +withLock :: MonadUnliftIO m => TMVar String -> String -> m a -> m a +withLock lock name = + E.bracket_ + (atomically $ putTMVar lock name) + (void . atomically $ takeTMVar lock) + +withGetLock :: MonadUnliftIO m => STM Lock -> String -> m a -> m a +withGetLock getLock name a = + E.bracket + (atomically $ getLock >>= \l -> putTMVar l name $> l) + (atomically . takeTMVar) + (const a)