debug deadlock

Author: evgeny-simplex

src/Simplex/Messaging/Agent.hs

@@ -145,10 +145,12 @@ module Simplex.Messaging.Agent
 where
 
 import Control.Applicative ((<|>))
+import Control.Concurrent (myThreadId)
 import Control.Concurrent.STM (retry)
 import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
+import Debug.Trace (traceIO)
 import Control.Monad.Reader
 import Control.Monad.Trans.Except
 import Crypto.Random (ChaChaDRG)
@@ -1841,6 +1843,9 @@ sendMessagesB_ c reqs connIds = withConnLocks c connIds "sendMessages" $ do
 
 enqueueCommand :: AgentClient -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> AM ()
 enqueueCommand c corrId connId server aCommand = do
+  liftIO $ do
+    tid <- myThreadId
+    traceIO $ "ENQUEUE_CMD connId=" <> show connId <> " cmd=" <> show (agentCommandTag aCommand) <> " tid=" <> show tid
   withStore c $ \db -> createCommand db corrId connId server aCommand
   lift . void $ getAsyncCmdWorker True c connId server
 
@@ -1869,7 +1874,13 @@ runCommandProcessing c connId server_ Worker {doWork} = do
   ri <- asks $ messageRetryInterval . config -- different retry interval?
   forever $ do
     endAgentOp c AOSndNetwork
+    liftIO $ do
+      tid <- myThreadId
+      traceIO $ "ASYNC_CMD wait connId=" <> show connId <> " tid=" <> show tid
     lift $ waitForWork doWork
+    liftIO $ do
+      tid <- myThreadId
+      traceIO $ "ASYNC_CMD woke connId=" <> show connId <> " tid=" <> show tid
     liftIO $ throwWhenInactive c
     atomically $ beginAgentOperation c AOSndNetwork
     withWork c doWork (\db -> getPendingServerCommand db connId server_) $ runProcessCmd (riFast ri)
@@ -2161,7 +2172,10 @@ getDeliveryWorker hasWork c sq =
       pure (w, retryLock)
 
 submitPendingMsg :: AgentClient -> SndQueue -> AM' ()
-submitPendingMsg c sq = do
+submitPendingMsg c sq@SndQueue {connId = sqConnId} = do
+  liftIO $ do
+    tid <- myThreadId
+    traceIO $ "SUBMIT_PENDING connId=" <> show sqConnId <> " tid=" <> show tid
   atomically $ modifyTVar' (msgDeliveryOp c) $ \s -> s {opsInProgress = opsInProgress s + 1}
   void $ getDeliveryWorker True c sq
 
@@ -2170,12 +2184,21 @@ runSmpQueueMsgDelivery c sq@SndQueue {userId, connId, server, queueMode} (Worker
   AgentConfig {messageRetryInterval = ri, messageTimeout, helloTimeout, quotaExceededTimeout} <- asks config
   forever $ do
     endAgentOp c AOSndNetwork
+    liftIO $ do
+      tid <- myThreadId
+      traceIO $ "DELIVERY wait connId=" <> show connId <> " tid=" <> show tid
     lift $ waitForWork doWork
+    liftIO $ do
+      tid <- myThreadId
+      traceIO $ "DELIVERY woke connId=" <> show connId <> " tid=" <> show tid
     liftIO $ throwWhenInactive c
     liftIO $ throwWhenNoDelivery c sq
     atomically $ beginAgentOperation c AOSndNetwork
     withWork c doWork (\db -> getPendingQueueMsg db connId sq) $
       \(rq_, PendingMsgData {msgId, msgType, msgBody, pqEncryption, msgFlags, msgRetryState, internalTs, internalSndId, prevMsgHash, pendingMsgPrepData_}) -> do
+        liftIO $ do
+          tid <- myThreadId
+          traceIO $ "DELIVERY send connId=" <> show connId <> " msgType=" <> show msgType <> " tid=" <> show tid
         endAgentOp c AOMsgDelivery -- this operation begins in submitPendingMsg
         let mId = unId msgId
             ri' = maybe id updateRetryInterval2 msgRetryState ri

src/Simplex/Messaging/Agent/Client.hs

@@ -195,7 +195,8 @@ module Simplex.Messaging.Agent.Client
 where
 
 import Control.Applicative ((<|>))
-import Control.Concurrent (ThreadId, killThread)
+import Control.Concurrent (ThreadId, killThread, myThreadId)
+import Debug.Trace (traceIO)
 import Control.Concurrent.Async (Async, uninterruptibleCancel)
 import Control.Concurrent.STM (retry)
 import Control.Exception (AsyncException (..), BlockedIndefinitelyOnSTM (..))
@@ -1077,9 +1078,11 @@ nonBlockingNotifyEvent :: AgentClient -> ATransmission -> IO ()
 nonBlockingNotifyEvent = notifyEvent_ nonBlockingWriteTBQueue
 
 notifyEvent_ :: (TBQueue ATransmission -> ATransmission -> IO ()) -> AgentClient -> ATransmission -> IO ()
-notifyEvent_ write c t@(_, connId, _)
+notifyEvent_ write c t@(_, connId, AEvt _ evt)
   | B.null connId = write (generalQ c) t
   | otherwise = do
+      tid <- myThreadId
+      traceIO $ "NOTIFY connId=" <> show connId <> " evt=" <> show (aEventTag evt) <> " tid=" <> show tid
       q <- getOrCreateConnWorker c connId
       write q t
 
@@ -1096,9 +1099,12 @@ getOrCreateConnWorker c@AgentClient {connWorkers, connWorkerSeq} connId = do
 
 connWorkerLoop :: AgentClient -> TBQueue ATransmission -> IO ()
 connWorkerLoop AgentClient {processEvent} q = forever $ do
-  t <- atomically $ readTBQueue q
+  t@(_, connId, _) <- atomically $ readTBQueue q
+  tid <- myThreadId
+  traceIO $ "CALLBACK fire connId=" <> show connId <> " tid=" <> show tid
   processEvent t `E.catchAny` \e ->
     logError $ "connWorkerLoop error: " <> tshow e
+  traceIO $ "CALLBACK done connId=" <> show connId <> " tid=" <> show tid
 
 withInvLock :: AgentClient -> ByteString -> Text -> AM a -> AM a
 withInvLock c key name = ExceptT . withInvLock' c key name . runExceptT

src/Simplex/Messaging/Agent/Lock.hs

@@ -10,13 +10,16 @@ module Simplex.Messaging.Agent.Lock
   )
 where
 
+import Control.Concurrent (myThreadId)
 import Control.Monad (void)
 import Control.Monad.Except (ExceptT (..), runExceptT)
 import Control.Monad.IO.Unlift
 import Data.Functor (($>))
 import Data.Set (Set)
 import qualified Data.Set as S
 import Data.Text (Text)
+import qualified Data.Text as T
+import Debug.Trace (traceIO)
 import UnliftIO.Async (forConcurrently)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -44,15 +47,30 @@ withLock' lock name =
 withGetLock :: MonadUnliftIO m => (k -> STM Lock) -> k -> Text -> m a -> m a
 withGetLock getLock key name a =
   E.bracket
-    (atomically $ getPutLock getLock key name)
-    (atomically . takeTMVar)
+    (do liftIO $ do
+          tid <- myThreadId
+          traceIO $ "LOCK acquire " <> T.unpack name <> " tid=" <> show tid
+        atomically $ getPutLock getLock key name)
+    (\l -> do
+        atomically $ takeTMVar l
+        liftIO $ do
+          tid <- myThreadId
+          traceIO $ "LOCK release " <> T.unpack name <> " tid=" <> show tid)
     (const a)
 
 withGetLocks :: MonadUnliftIO m => (k -> STM Lock) -> Set k -> Text -> m a -> m a
 withGetLocks getLock keys name = E.bracket holdLocks releaseLocks . const
   where
-    holdLocks = forConcurrently (S.toList keys) $ \key -> atomically $ getPutLock getLock key name
-    releaseLocks = mapM_ (atomically . takeTMVar)
+    holdLocks = do
+      liftIO $ do
+        tid <- myThreadId
+        traceIO $ "LOCKS acquire " <> T.unpack name <> " (" <> show (S.size keys) <> " keys) tid=" <> show tid
+      forConcurrently (S.toList keys) $ \key -> atomically $ getPutLock getLock key name
+    releaseLocks ls = do
+      mapM_ (atomically . takeTMVar) ls
+      liftIO $ do
+        tid <- myThreadId
+        traceIO $ "LOCKS release " <> T.unpack name <> " (" <> show (S.size keys) <> " keys) tid=" <> show tid
 
 -- getLock and putTMVar can be in one transaction on the assumption that getLock doesn't write in case the lock already exists,
 -- and in case it is created and added to some shared resource (we use TMap) it also helps avoid contention for the newly created lock.