From 0cd2a3904c14a5a2248b435b4c0c411119517e4b Mon Sep 17 00:00:00 2001 From: "Evgeny @ SimpleX Chat" <259188159+evgeny-simplex@users.noreply.github.com> Date: Sun, 12 Jul 2026 15:06:58 +0000 Subject: [PATCH] use workers --- src/Simplex/Chat.hs | 6 ++- src/Simplex/Chat/Controller.hs | 21 ++++++++ src/Simplex/Chat/Library/Commands.hs | 75 +++++++++++++++++++++++--- src/Simplex/Chat/Library/Subscriber.hs | 24 ++++----- tests/ChatTests/Groups.hs | 12 +++-- 5 files changed, 114 insertions(+), 24 deletions(-) diff --git a/src/Simplex/Chat.hs b/src/Simplex/Chat.hs index 617b95a822..6fef7c70c8 100644 --- a/src/Simplex/Chat.hs +++ b/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, diff --git a/src/Simplex/Chat/Controller.hs b/src/Simplex/Chat/Controller.hs index 3849bed9af..88f7fca2a0 100644 --- a/src/Simplex/Chat/Controller.hs +++ b/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) diff --git a/src/Simplex/Chat/Library/Commands.hs b/src/Simplex/Chat/Library/Commands.hs index f09b4704ad..32d5c852a5 100644 --- a/src/Simplex/Chat/Library/Commands.hs +++ b/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' diff --git a/src/Simplex/Chat/Library/Subscriber.hs b/src/Simplex/Chat/Library/Subscriber.hs index 85118ce2ca..6de966aa63 100644 --- a/src/Simplex/Chat/Library/Subscriber.hs +++ b/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 diff --git a/tests/ChatTests/Groups.hs b/tests/ChatTests/Groups.hs index a3752881d9..39146b93ad 100644 --- a/tests/ChatTests/Groups.hs +++ b/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 ->