use workers

src/Simplex/Chat.hs

@@ -170,7 +170,7 @@ newChatController
     ccVar <- newEmptyTMVarIO
     let processEvent _ t = do
           cc <- atomically $ readTMVar ccVar
-          runReaderT (processAgentEvent t) cc
+          queueAgentEvent cc t
     runExceptT (getSMPAgentClient aCfg {tbqSize} servers agentStore processEvent)
       >>= mapM (mkChatController config randomPresetServers randomAgentServers ccVar)
     where
@@ -200,6 +200,8 @@ newChatController
         deliveryTaskWorkers <- TM.emptyIO
         deliveryJobWorkers <- TM.emptyIO
         relayRequestWorkers <- TM.emptyIO
+        entityWorkers <- TM.emptyIO
+        entityWorkerSeq <- newTVarIO 0
         relayGroupLinkChecksAsync <- newTVarIO Nothing
         webPreviewState <- forM webPreviewConfig $ \_ -> newWebPreviewState
         chatRelayTests <- TM.emptyIO
@@ -246,6 +248,8 @@ newChatController
                   deliveryTaskWorkers,
                   deliveryJobWorkers,
                   relayRequestWorkers,
+                  entityWorkers,
+                  entityWorkerSeq,
                   relayGroupLinkChecksAsync,
                   webPreviewState,
                   chatRelayTests,

src/Simplex/Chat/Controller.hs

@@ -5,6 +5,7 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
+{-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
@@ -76,6 +77,7 @@ import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, NetworkConfig, ServerCfg
 import Simplex.Messaging.Agent.Lock
 import Simplex.Messaging.Agent.RetryInterval (RetryInterval (..))
 import Simplex.Messaging.Agent.Protocol
+import Simplex.Messaging.Session (SessionVar (..))
 import Simplex.Messaging.Agent.Store.Common (DBStore, withTransaction, withTransactionPriority)
 import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation, UpMigration)
 import Simplex.Messaging.Agent.Store.DB (SQLError)
@@ -305,6 +307,8 @@ data ChatController = ChatController
     deliveryTaskWorkers :: TMap DeliveryWorkerKey Worker,
     deliveryJobWorkers :: TMap DeliveryWorkerKey Worker,
     relayRequestWorkers :: TMap Int Worker, -- single global worker with key 1 is used to fit into existing worker management framework
+    entityWorkers :: TMap (Maybe ChatLockEntity) (SessionVar EntityWorker), -- per-entity agent event processing; Nothing key is the general worker
+    entityWorkerSeq :: TVar Int,
     relayGroupLinkChecksAsync :: TVar (Maybe (Async ())),
     webPreviewState :: Maybe WebPreviewState,
     chatRelayTests :: TMap ConnId RelayTest,
@@ -321,6 +325,23 @@ data ChatController = ChatController
     contactMergeEnabled :: TVar Bool
   }
 
+-- agent event with its chat lock entity resolved once (in the callback), threaded to the handler.
+-- ResolvedEvent shares AEvent's AEntity index, so the resolved value and the event always match.
+data ResolvedEvent (e :: AEntity) where
+  REConn :: Maybe ChatLockEntity -> ResolvedEvent AEConn -- Nothing: DEL_RCVQS / DEL_CONNS / empty-conn ERR
+  RESndFile :: Maybe ChatRef -> FileTransferId -> ResolvedEvent AESndFile
+  RERcvFile :: Maybe ChatRef -> FileTransferId -> ResolvedEvent AERcvFile
+  RENone :: ResolvedEvent AENone
+
+data ResolvedEvt = forall e. AEntityI e => ResolvedEvt (ResolvedEvent e) (AEvent e)
+
+type ResolvedTransmission = (ACorrId, AEntityId, ResolvedEvt)
+
+data EntityWorker = EntityWorker
+  { eventQ :: TQueue ResolvedTransmission,
+    workerThreadId :: Weak ThreadId
+  }
+
 data HelpSection = HSMain | HSFiles | HSGroups | HSContacts | HSMyAddress | HSIncognito | HSMarkdown | HSMessages | HSRemote | HSSettings | HSDatabase
   deriving (Show)
 

src/Simplex/Chat/Library/Commands.hs

@@ -113,6 +113,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (base64P)
 import Simplex.Messaging.Protocol (AProtoServerWithAuth (..), AProtocolType (..), MsgFlags (..), NameRecord (..), NtfServer, ProtoServerWithAuth (..), ProtocolServer, ProtocolType (..), ProtocolTypeI (..), SProtocolType (..), SubscriptionMode (..), UserProtocol, userProtocol)
 import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
+import Simplex.Messaging.Session (SessionVar (..), withGetSessVar')
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport.Client (defaultSocksProxyWithAuth)
 import Simplex.Messaging.Util
@@ -122,10 +123,11 @@ import Simplex.RemoteControl.Types (RCCtrlAddress (..))
 import System.Exit (ExitCode, exitSuccess)
 import System.FilePath (takeExtension, takeFileName, (</>))
 import System.IO (Handle, IOMode (..))
+import System.Mem.Weak (deRefWeak)
 import System.Random (randomRIO)
 import System.Timeout (timeout)
 import UnliftIO.Async
-import UnliftIO.Concurrent (forkIO, threadDelay)
+import UnliftIO.Concurrent (forkIO, forkIOWithUnmask, killThread, mkWeakThreadId, threadDelay)
 import UnliftIO.Directory
 import qualified UnliftIO.Exception as E
 import UnliftIO.IO (hClose)
@@ -296,10 +298,11 @@ restoreCalls = do
   atomically $ writeTVar calls callsMap
 
 stopChatController :: ChatController -> IO ()
-stopChatController ChatController {smpAgent, chatRunning, subscribeAsync, sndFiles, rcvFiles, expireCIFlags, remoteHostSessions, remoteCtrlSession} = do
+stopChatController ChatController {smpAgent, chatRunning, subscribeAsync, entityWorkers, sndFiles, rcvFiles, expireCIFlags, remoteHostSessions, remoteCtrlSession} = do
   readTVarIO remoteHostSessions >>= mapM_ (cancelRemoteHost False . snd)
   atomically (stateTVar remoteCtrlSession (,Nothing)) >>= mapM_ (cancelRemoteCtrl False . snd)
   disconnectAgentClient smpAgent
+  readTVarIO entityWorkers >>= mapM_ stopEntityWorker
   readTVarIO subscribeAsync >>= mapM_ (void . forkIO . uninterruptibleCancel)
   closeFiles sndFiles
   closeFiles rcvFiles
@@ -308,7 +311,10 @@ stopChatController ChatController {smpAgent, chatRunning, subscribeAsync, sndFil
     forM_ keys $ \k -> TM.insert k False expireCIFlags
     writeTVar chatRunning False
     writeTVar subscribeAsync Nothing
+    TM.clear entityWorkers -- drop the killed workers so a restart (/_start) creates fresh ones
   where
+    stopEntityWorker v =
+      atomically (tryReadTMVar $ sessionVar v) >>= mapM_ (\EntityWorker {workerThreadId} -> deRefWeak workerThreadId >>= mapM_ killThread)
     closeFiles :: TVar (Map Int64 Handle) -> IO ()
     closeFiles files = do
       fs <- readTVarIO files
@@ -5024,12 +5030,65 @@ setAllExpireCIFlags b = do
     keys <- M.keys <$> readTVar expireFlags
     forM_ keys $ \k -> TM.insert k b expireFlags
 
-processAgentEvent :: ATransmission -> CM' ()
-processAgentEvent (corrId, entId, AEvt e msg) = run $ case e of
-  SAENone -> processAgentMessageNoConn msg
-  SAEConn -> processAgentMessage corrId entId msg
-  SAERcvFile -> processAgentMsgRcvFile corrId entId msg
-  SAESndFile -> processAgentMsgSndFile corrId entId msg
+-- agent event callback: resolve the event's lock entity, then hand it to that entity's worker.
+-- runs on the agent thread (under its connection lock), so it must not block - the worker queue is unbounded.
+queueAgentEvent :: ChatController -> ATransmission -> IO ()
+queueAgentEvent cc t =
+  runReaderT (runExceptT $ resolveEvent t) cc >>= \case
+    Right rt -> getEntityWorker cc (workerKey rt) >>= \q -> atomically (writeTQueue q rt)
+    Left e -> runReaderT (eToView' e) cc
+
+-- resolves the event's chat lock entity once, used both to pick the worker and (threaded through) by the handler.
+-- a lookup failure is surfaced by queueAgentEvent and the event is not enqueued (not ACKed either, so it is redelivered).
+resolveEvent :: ATransmission -> CM ResolvedTransmission
+resolveEvent (corrId, entId, AEvt sae ev) = (corrId,entId,) <$> case sae of
+  SAEConn -> (\me -> ResolvedEvt (REConn me) ev) <$> connLockEntity
+  SAESndFile -> (\(cRef, fId) -> ResolvedEvt (RESndFile cRef fId) ev) <$> withStore (`getXFTPSndFileDBIds` AgentSndFileId entId)
+  SAERcvFile -> (\(cRef, fId) -> ResolvedEvt (RERcvFile cRef fId) ev) <$> withStore (`getXFTPRcvFileDBIds` AgentRcvFileId entId)
+  SAENone -> pure $ ResolvedEvt RENone ev
+  where
+    connLockEntity
+      | B.null entId = pure Nothing -- DEL_RCVQS / DEL_CONNS / empty-conn ERR are entity-less
+      | otherwise = Just <$> critical entId (withStore (`getChatLockEntity` AgentConnId entId))
+
+workerKey :: ResolvedTransmission -> Maybe ChatLockEntity
+workerKey (_, _, ResolvedEvt re _) = case re of
+  REConn me -> me
+  RESndFile cRef fId -> Just $ fileEntity cRef fId
+  RERcvFile cRef fId -> Just $ fileEntity cRef fId
+  RENone -> Nothing
+
+-- file events run on the chat entity's worker; files with no chat entity use CLFile
+fileEntity :: Maybe ChatRef -> FileTransferId -> ChatLockEntity
+fileEntity cRef fId = case cRef of
+  Just (ChatRef CTDirect cId _) -> CLContact cId
+  Just (ChatRef CTGroup gId _) -> CLGroup gId
+  _ -> CLFile fId
+
+getEntityWorker :: ChatController -> Maybe ChatLockEntity -> IO (TQueue ResolvedTransmission)
+getEntityWorker cc@ChatController {entityWorkerSeq, entityWorkers} key = do
+  ts <- getCurrentTime
+  withGetSessVar' entityWorkerSeq key entityWorkers ts createWorker existingWorker
+  where
+    createWorker v = do
+      q <- newTQueueIO
+      -- forkIOWithUnmask (not forkIO): the callback runs under uninterruptibleMask, an inherited mask makes the worker unkillable
+      tId <- mkWeakThreadId =<< forkIOWithUnmask (\unmask -> unmask $ entityWorkerLoop cc q)
+      atomically $ putTMVar (sessionVar v) EntityWorker {eventQ = q, workerThreadId = tId}
+      pure q
+    existingWorker v = eventQ <$> atomically (readTMVar $ sessionVar v)
+
+entityWorkerLoop :: ChatController -> TQueue ResolvedTransmission -> IO ()
+entityWorkerLoop cc q = forever $
+  -- unmasked while blocked on the queue (killable at teardown), uninterruptible while processing so each event completes atomically
+  atomically (readTQueue q) >>= \rt -> E.uninterruptibleMask_ $ runReaderT (processAgentEvent rt) cc
+
+processAgentEvent :: ResolvedTransmission -> CM' ()
+processAgentEvent (corrId, entId, ResolvedEvt re ev) = run $ case re of
+  RENone -> processAgentMessageNoConn ev
+  REConn me -> processAgentMessage me corrId entId ev
+  RERcvFile cRef fId -> processAgentMsgRcvFile (cRef, fId) corrId entId ev
+  RESndFile cRef fId -> processAgentMsgSndFile (cRef, fId) corrId entId ev
   where
     run action = action `catchAllOwnErrors'` eToView'
 

src/Simplex/Chat/Library/Subscriber.hs

@@ -118,23 +118,24 @@ verifyGroupSig key publicGroupId memberId signatures signedBody =
   let prefix = smpEncode CBGroup <> smpEncode (publicGroupId, memberId)
    in all (\case (MsgSignature KRMember sig) -> C.verify (C.APublicVerifyKey C.SEd25519 key) sig (prefix <> signedBody)) signatures
 
-processAgentMessage :: ACorrId -> ConnId -> AEvent 'AEConn -> CM ()
-processAgentMessage _ _ (DEL_RCVQS delQs) =
+-- the lock entity is resolved by the caller (resolveEvent) and threaded in; entity-less events (DEL_*, empty-conn ERR) ignore it
+processAgentMessage :: Maybe ChatLockEntity -> ACorrId -> ConnId -> AEvent 'AEConn -> CM ()
+processAgentMessage _ _ _ (DEL_RCVQS delQs) =
   toView $ CEvtAgentRcvQueuesDeleted $ L.map rcvQ delQs
   where
     rcvQ (connId, server, rcvId, err_) = DeletedRcvQueue (AgentConnId connId) server (AgentQueueId rcvId) err_
-processAgentMessage _ _ (DEL_CONNS connIds) =
+processAgentMessage _ _ _ (DEL_CONNS connIds) =
   toView $ CEvtAgentConnsDeleted $ L.map AgentConnId connIds
-processAgentMessage _ "" (ERR e) =
+processAgentMessage _ _ "" (ERR e) =
   eToView $ chatErrorAgent e
-processAgentMessage corrId connId msg = do
-  lockEntity <- critical connId (withStore (`getChatLockEntity` AgentConnId connId))
+processAgentMessage (Just lockEntity) corrId connId msg =
   withEntityLock "processAgentMessage" lockEntity $ do
     cxt <- chatStoreCxt
     -- getUserByAConnId never throws logical errors, only SEDBBusyError can be thrown here
     critical connId (withStore' (`getUserByAConnId` AgentConnId connId)) >>= \case
       Just user -> processAgentMessageConn cxt user corrId connId msg `catchAllErrors` eToView
       _ -> throwChatError $ CENoConnectionUser (AgentConnId connId)
+processAgentMessage Nothing _ connId _ = throwChatError $ CENoConnectionUser (AgentConnId connId)
 
 -- CRITICAL error will be shown to the user as alert with restart button in Android/desktop apps.
 -- SEDBBusyError will only be thrown on IO exceptions or SQLError during DB queries,
@@ -177,9 +178,9 @@ processAgentMessageNoConn = \case
     errsEvent :: [(ConnId, AgentErrorType)] -> CM ()
     errsEvent = toView . CEvtChatErrors . map (\(cId, e) -> ChatErrorAgent e (AgentConnId cId) Nothing)
 
-processAgentMsgSndFile :: ACorrId -> SndFileId -> AEvent 'AESndFile -> CM ()
-processAgentMsgSndFile _corrId aFileId msg = do
-  (cRef_, fileId) <- withStore (`getXFTPSndFileDBIds` AgentSndFileId aFileId)
+-- (cRef_, fileId) are resolved by the caller (resolveEvent) and threaded in
+processAgentMsgSndFile :: (Maybe ChatRef, FileTransferId) -> ACorrId -> SndFileId -> AEvent 'AESndFile -> CM ()
+processAgentMsgSndFile (cRef_, fileId) _corrId aFileId msg =
   withEntityLock_ cRef_ . withFileLock "processAgentMsgSndFile" fileId $
     withStore' (`getUserByASndFileId` AgentSndFileId aFileId) >>= \case
       Just user -> process user fileId `catchAllErrors` eToView
@@ -317,9 +318,8 @@ agentFileError = \case
       SMP.TRANSPORT TEVersion -> srvErr SrvErrVersion
       e -> srvErr . SrvErrOther $ tshow e
 
-processAgentMsgRcvFile :: ACorrId -> RcvFileId -> AEvent 'AERcvFile -> CM ()
-processAgentMsgRcvFile _corrId aFileId msg = do
-  (cRef_, fileId) <- withStore (`getXFTPRcvFileDBIds` AgentRcvFileId aFileId)
+processAgentMsgRcvFile :: (Maybe ChatRef, FileTransferId) -> ACorrId -> RcvFileId -> AEvent 'AERcvFile -> CM ()
+processAgentMsgRcvFile (cRef_, fileId) _corrId aFileId msg =
   withEntityLock_ cRef_ . withFileLock "processAgentMsgRcvFile" fileId $
     withStore' (`getUserByARcvFileId` AgentRcvFileId aFileId) >>= \case
       Just user -> process user fileId `catchAllErrors` eToView

tests/ChatTests/Groups.hs

@@ -2984,7 +2984,7 @@ testGroupLinkMemberRole =
             bob <## "#team: you joined the group"
         ]
 
-      threadDelay 100000
+      threadDelay 500000
 
       alice ##> "/ms team"
       alice
@@ -3154,7 +3154,7 @@ testGroupLinkHostProfileReceived =
             bob <## "#team: you joined the group"
         ]
 
-      threadDelay 100000
+      threadDelay 500000
 
       aliceImage <- getProfilePictureByName bob "alice"
       aliceImage `shouldBe` Just profileImage
@@ -3218,7 +3218,13 @@ testGLinkRejectBlockedName =
       bob <## "#team: joining the group..."
       bob <## "#team: join rejected, reason: GRRBlockedName"
 
-      threadDelay 100000
+      -- bob's contact is removed by async reject/handshake/delete events (~0.2-0.4s via per-entity workers); poll instead of a fixed delay
+      let waitRejected n = do
+            ps <- getContactProfiles alice
+            if "bob" `notElem` ps || (n :: Int) <= 0
+              then pure ()
+              else threadDelay 10000 >> waitRejected (n - 1)
+      waitRejected 300
 
       alice `hasContactProfiles` ["alice"]
       memCount <- withCCTransaction alice $ \db ->