debug agent locks

2026-08-28 11:34:36 +00:00 · 2022-10-22 16:43:14 +01:00
parent d06bb2b9c5
commit 19aef52135
4 changed files with 70 additions and 26 deletions
@@ -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
@@ -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
@@ -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
@@ -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)