mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-29 09:58:41 +00:00
use multiple agent queues for concurrency
This commit is contained in:
Evgeny @ SimpleX Chat
@@ -348,7 +348,7 @@ xftpDeleteRcvFiles' c rcvFileEntityIds = do
|
||||
batchFiles f rcvFiles = withStoreBatch' c $ \db -> map (\RcvFile {rcvFileId} -> f db rcvFileId) rcvFiles
|
||||
|
||||
notify :: forall m e. (MonadIO m, AEntityI e) => AgentClient -> AEntityId -> AEvent e -> m ()
|
||||
notify c entId cmd = atomically $ writeTBQueue (subQ c) ("", entId, AEvt (sAEntity @e) cmd)
|
||||
notify c entId cmd = liftIO $ notifyEvent c ("", entId, AEvt (sAEntity @e) cmd)
|
||||
|
||||
xftpSendFile' :: AgentClient -> UserId -> CryptoFile -> Int -> AM SndFileId
|
||||
xftpSendFile' c userId file numRecipients = do
|
||||
|
||||
@@ -256,12 +256,12 @@ import UnliftIO.STM
|
||||
type AE a = ExceptT AgentErrorType IO a
|
||||
|
||||
-- | Creates an SMP agent client instance
|
||||
getSMPAgentClient :: AgentConfig -> InitialAgentServers -> DBStore -> Bool -> AE AgentClient
|
||||
getSMPAgentClient :: AgentConfig -> InitialAgentServers -> DBStore -> Bool -> (ATransmission -> IO ()) -> AE AgentClient
|
||||
getSMPAgentClient = getSMPAgentClient_ 1
|
||||
{-# INLINE getSMPAgentClient #-}
|
||||
|
||||
getSMPAgentClient_ :: Int -> AgentConfig -> InitialAgentServers -> DBStore -> Bool -> AE AgentClient
|
||||
getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp, netCfg, useServices, presetServers} store backgroundMode = do
|
||||
getSMPAgentClient_ :: Int -> AgentConfig -> InitialAgentServers -> DBStore -> Bool -> (ATransmission -> IO ()) -> AE AgentClient
|
||||
getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp, netCfg, useServices, presetServers} store backgroundMode processEvent = do
|
||||
-- This error should be prevented in the app
|
||||
when (any id useServices && sessionMode netCfg == TSMEntity) $ throwE $ CMD PROHIBITED "newAgentClient"
|
||||
liftIO $ newSMPAgentEnv cfg store >>= runReaderT runAgent
|
||||
@@ -272,7 +272,8 @@ getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp, netC
|
||||
notices <- liftIO $ withTransaction store (`getClientNotices` presetServers) `catchAll_` pure []
|
||||
env <- ask
|
||||
let processMsg c t = subscriber c t `runReaderT` env
|
||||
c@AgentClient {acThread} <- liftIO $ newAgentClient clientId initServers currentTs notices processMsg env
|
||||
c@AgentClient {acThread, generalQ} <- liftIO $ newAgentClient clientId initServers currentTs notices processEvent processMsg env
|
||||
void $ liftIO $ forkIO $ connWorkerLoop c generalQ
|
||||
unless backgroundMode $ do
|
||||
t <- runAgentThreads c `forkFinally` const (liftIO $ disconnectAgentClient c)
|
||||
atomically . writeTVar acThread . Just =<< mkWeakThreadId t
|
||||
@@ -287,10 +288,10 @@ getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp, netC
|
||||
run c "logServersStats" $ logServersStats c
|
||||
]
|
||||
`E.finally` saveServersStats c
|
||||
run AgentClient {subQ, acThread} name a =
|
||||
run c'@AgentClient {acThread} name a =
|
||||
a `E.catchAny` \e -> whenM (isJust <$> readTVarIO acThread) $ do
|
||||
logError $ "Agent thread " <> name <> " crashed: " <> tshow e
|
||||
atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR $ CRITICAL True $ show e)
|
||||
liftIO $ notifyEvent c' ("", "", AEvt SAEConn $ ERR $ CRITICAL True $ show e)
|
||||
|
||||
logServersStats :: AgentClient -> AM' ()
|
||||
logServersStats c = do
|
||||
@@ -303,19 +304,19 @@ logServersStats c = do
|
||||
liftIO $ threadDelay' int
|
||||
|
||||
saveServersStats :: AgentClient -> AM' ()
|
||||
saveServersStats c@AgentClient {subQ, smpServersStats, xftpServersStats, ntfServersStats} = do
|
||||
saveServersStats c@AgentClient {smpServersStats, xftpServersStats, ntfServersStats} = do
|
||||
sss <- mapM (liftIO . getAgentSMPServerStats) =<< readTVarIO smpServersStats
|
||||
xss <- mapM (liftIO . getAgentXFTPServerStats) =<< readTVarIO xftpServersStats
|
||||
nss <- mapM (liftIO . getAgentNtfServerStats) =<< readTVarIO ntfServersStats
|
||||
let stats = AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss, ntfServersStats = OptionalMap nss}
|
||||
tryAllErrors' (withStore' c (`updateServersStats` stats)) >>= \case
|
||||
Left e -> atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e)
|
||||
Left e -> liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e)
|
||||
Right () -> pure ()
|
||||
|
||||
restoreServersStats :: AgentClient -> AM' ()
|
||||
restoreServersStats c@AgentClient {smpServersStats, xftpServersStats, ntfServersStats, srvStatsStartedAt} = do
|
||||
tryAllErrors' (withStore c getServersStats) >>= \case
|
||||
Left e -> atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e)
|
||||
Left e -> liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e)
|
||||
Right (startedAt, Nothing) -> atomically $ writeTVar srvStatsStartedAt startedAt
|
||||
Right (startedAt, Just AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss, ntfServersStats = OptionalMap nss}) -> do
|
||||
atomically $ writeTVar srvStatsStartedAt startedAt
|
||||
@@ -820,8 +821,8 @@ deleteUser' c@AgentClient {smpServersStats, xftpServersStats} userId delSMPQueue
|
||||
lift $ saveServersStats c
|
||||
where
|
||||
delUser =
|
||||
whenM (withStore' c (`deleteUserWithoutConns` userId)) . atomically $
|
||||
writeTBQueue (subQ c) ("", "", AEvt SAENone $ DEL_USER userId)
|
||||
whenM (withStore' c (`deleteUserWithoutConns` userId)) . liftIO $
|
||||
notifyEvent c ("", "", AEvt SAENone $ DEL_USER userId)
|
||||
|
||||
setUserService' :: AgentClient -> UserId -> Bool -> AM ()
|
||||
setUserService' c userId enable = do
|
||||
@@ -1324,7 +1325,7 @@ startJoinInvitation c userId connId sq_ enableNtfs cReqUri pqSup =
|
||||
getSndRatchet db connId v >>= \case
|
||||
Right r -> pure $ Right $ snd r
|
||||
Left e -> do
|
||||
nonBlockingWriteTBQueue (subQ c) ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "no snd ratchet " <> show e))
|
||||
notifyEvent c ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "no snd ratchet " <> show e))
|
||||
runExceptT $ createRatchet_ db g maxSupported pqSupport e2eRcvParams
|
||||
pure (cData, sq, e2eSndParams, Nothing)
|
||||
_ -> do
|
||||
@@ -1418,7 +1419,7 @@ joinConnSrv c nm userId connId enableNtfs cReqUri@CRContactUri {} cInfo pqSup su
|
||||
getRatchetX3dhKeys db connId >>= \case
|
||||
Right keys -> pure $ CR.mkRcvE2ERatchetParams (maxVersion e2eVR) keys
|
||||
Left e -> do
|
||||
nonBlockingWriteTBQueue (subQ c) ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "no rcv ratchet " <> show e))
|
||||
notifyEvent c ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "no rcv ratchet " <> show e))
|
||||
let pqEnc = CR.initialPQEncryption False pqInitKeys
|
||||
(pk1, pk2, pKem, e2eRcvParams) <- liftIO $ CR.generateRcvE2EParams g (maxVersion e2eVR) pqEnc
|
||||
createRatchetX3dhKeys db connId pk1 pk2 pKem
|
||||
@@ -1430,7 +1431,7 @@ joinConnSrv c nm userId connId enableNtfs cReqUri@CRContactUri {} cInfo pqSup su
|
||||
delInvSL :: AgentClient -> ConnId -> SMPServerWithAuth -> SMP.LinkId -> AM ()
|
||||
delInvSL c connId srv lnkId =
|
||||
withStore' c (\db -> deleteInvShortLink db (protoServer srv) lnkId) `catchE` \e ->
|
||||
liftIO $ nonBlockingWriteTBQueue (subQ c) ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "error deleting short link " <> show e))
|
||||
liftIO $ notifyEvent c ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "error deleting short link " <> show e))
|
||||
|
||||
joinConnSrvAsync :: AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> SMPServerWithAuth -> AM SndQueueSecured
|
||||
joinConnSrvAsync c userId connId enableNtfs inv@CRInvitationUri {} cInfo pqSupport subMode srv = do
|
||||
@@ -1603,8 +1604,8 @@ subscribeConnections_ c conns = do
|
||||
notifyResultError rs = do
|
||||
let actual = M.size rs
|
||||
expected = length conns
|
||||
when (actual /= expected) . atomically $
|
||||
writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ INTERNAL $ "subscribeConnections result size: " <> show actual <> ", expected " <> show expected)
|
||||
when (actual /= expected) . liftIO $
|
||||
notifyEvent c ("", "", AEvt SAEConn $ ERR $ INTERNAL $ "subscribeConnections result size: " <> show actual <> ", expected " <> show expected)
|
||||
|
||||
subscribeAllConnections' :: AgentClient -> Bool -> Maybe UserId -> AM ()
|
||||
subscribeAllConnections' c onlyNeeded activeUserId_ = handleErr $ do
|
||||
@@ -1651,7 +1652,7 @@ subscribeAllConnections' c onlyNeeded activeUserId_ = handleErr $ do
|
||||
Just SSErrorQueueCount {expectedQueueCount = n, subscribedQueueCount = n'} | n > 0 && n' == 0 -> unassocQueues
|
||||
_ -> pure True
|
||||
Left e -> do
|
||||
atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR e)
|
||||
liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR e)
|
||||
if clientServiceError e
|
||||
then False <$ withStore' c (\db -> unassocUserServerRcvQueueSubs' db userId srv)
|
||||
else pure True
|
||||
@@ -1860,7 +1861,7 @@ getAsyncCmdWorker hasWork c connId server =
|
||||
data CommandCompletion = CCMoved | CCCompleted
|
||||
|
||||
runCommandProcessing :: AgentClient -> ConnId -> Maybe SMPServer -> Worker -> AM ()
|
||||
runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
|
||||
runCommandProcessing c connId server_ Worker {doWork} = do
|
||||
ri <- asks $ messageRetryInterval . config -- different retry interval?
|
||||
forever $ do
|
||||
atomically $ endAgentOperation c AOSndNetwork
|
||||
@@ -1872,7 +1873,7 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
|
||||
runProcessCmd ri cmd = do
|
||||
pending <- newTVarIO []
|
||||
processCmd ri cmd pending
|
||||
mapM_ (atomically . writeTBQueue subQ) . reverse =<< readTVarIO pending
|
||||
mapM_ (liftIO . notifyEvent c) . reverse =<< readTVarIO pending
|
||||
processCmd :: RetryInterval -> PendingCommand -> TVar [ATransmission] -> AM ()
|
||||
processCmd ri PendingCommand {cmdId, corrId, userId, command} pendingCmds = case command of
|
||||
AClientCommand cmd -> case cmd of
|
||||
@@ -2028,9 +2029,7 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
|
||||
internalErr s = cmdError $ INTERNAL $ s <> ": " <> show (agentCommandTag command)
|
||||
cmdError e = notify (ERR e) >> withStore' c (`deleteCommand` cmdId)
|
||||
notify :: forall e. AEntityI e => AEvent e -> AM ()
|
||||
notify cmd =
|
||||
let t = (corrId, connId, AEvt (sAEntity @e) cmd)
|
||||
in atomically $ ifM (isFullTBQueue subQ) (modifyTVar' pendingCmds (t :)) (writeTBQueue subQ t)
|
||||
notify cmd = atomically $ modifyTVar' pendingCmds ((corrId, connId, AEvt (sAEntity @e) cmd) :)
|
||||
-- ^ ^ ^ async command processing /
|
||||
|
||||
enqueueMessages :: AgentClient -> ConnData -> NonEmpty SndQueue -> MsgFlags -> AMessage -> AM (AgentMsgId, PQEncryption)
|
||||
@@ -2163,7 +2162,7 @@ submitPendingMsg c sq = do
|
||||
void $ getDeliveryWorker True c sq
|
||||
|
||||
runSmpQueueMsgDelivery :: AgentClient -> SndQueue -> (Worker, TMVar ()) -> AM ()
|
||||
runSmpQueueMsgDelivery c@AgentClient {subQ} sq@SndQueue {userId, connId, server, queueMode} (Worker {doWork}, qLock) = do
|
||||
runSmpQueueMsgDelivery c sq@SndQueue {userId, connId, server, queueMode} (Worker {doWork}, qLock) = do
|
||||
AgentConfig {messageRetryInterval = ri, messageTimeout, helloTimeout, quotaExceededTimeout} <- asks config
|
||||
forever $ do
|
||||
atomically $ endAgentOperation c AOSndNetwork
|
||||
@@ -2331,7 +2330,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} sq@SndQueue {userId, connId, server,
|
||||
delMsgKeep :: Bool -> InternalId -> AM ()
|
||||
delMsgKeep keepForReceipt msgId = withStore' c $ \db -> deleteSndMsgDelivery db connId sq msgId keepForReceipt
|
||||
notify :: forall e. AEntityI e => AEvent e -> AM ()
|
||||
notify cmd = atomically $ writeTBQueue subQ ("", connId, AEvt (sAEntity @e) cmd)
|
||||
notify cmd = liftIO $ notifyEvent c ("", connId, AEvt (sAEntity @e) cmd)
|
||||
notifyDel :: AEntityI e => InternalId -> AEvent e -> AM ()
|
||||
notifyDel msgId cmd = notify cmd >> delMsg msgId
|
||||
connError msgId = notifyDel msgId . ERR . (`CONN` "")
|
||||
@@ -2353,7 +2352,7 @@ retrySndOp c loop = do
|
||||
withConnLockNotify :: AgentClient -> ConnId -> Text -> AM (Maybe ATransmission) -> AM ()
|
||||
withConnLockNotify c connId name action = do
|
||||
t_ <- withConnLock c connId name action
|
||||
forM_ t_ $ atomically . writeTBQueue (subQ c)
|
||||
forM_ t_ $ liftIO . notifyEvent c
|
||||
|
||||
ackMessage' :: AgentClient -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> AM ()
|
||||
ackMessage' c connId msgId rcptInfo_ = withConnLockNotify c connId "ackMessage" $ do
|
||||
@@ -2570,7 +2569,7 @@ prepareDeleteConnections_ getConnections c waitDelivery connIds = do
|
||||
unsubNtfConnIds connIds' = do
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ writeTBQueue (ntfSubQ ns) (NSCDeleteSub, connIds')
|
||||
notify = atomically . writeTBQueue (subQ c)
|
||||
notify = liftIO . notifyEvent c
|
||||
|
||||
deleteConnQueues :: AgentClient -> NetworkRequestMode -> Bool -> Bool -> [RcvQueue] -> AM' (Map ConnId (Either AgentErrorType ()))
|
||||
deleteConnQueues c nm waitDelivery ntf rqs = do
|
||||
@@ -2604,7 +2603,7 @@ deleteConnQueues c nm waitDelivery ntf rqs = do
|
||||
-- attempts and successes are counted in deleteQueues function
|
||||
atomically $ incSMPServerStat c userId server connDeleted
|
||||
pure ((rq, Right ()), Just (Just e))
|
||||
notify = when ntf . atomically . writeTBQueue (subQ c)
|
||||
notify = when ntf . liftIO . notifyEvent c
|
||||
connResults :: [(RcvQueue, Either AgentErrorType ())] -> Map ConnId (Either AgentErrorType ())
|
||||
connResults = M.map snd . foldl' addResult M.empty
|
||||
where
|
||||
@@ -2640,8 +2639,8 @@ deleteConnections_ getConnections ntf waitDelivery c nm connIds = do
|
||||
notifyResultError rs = do
|
||||
let actual = M.size rs
|
||||
expected = length connIds
|
||||
when (actual /= expected) . atomically $
|
||||
writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ INTERNAL $ "deleteConnections result size: " <> show actual <> ", expected " <> show expected)
|
||||
when (actual /= expected) . liftIO $
|
||||
notifyEvent c ("", "", AEvt SAEConn $ ERR $ INTERNAL $ "deleteConnections result size: " <> show actual <> ", expected " <> show expected)
|
||||
|
||||
getConnectionServers' :: AgentClient -> ConnId -> AM ConnectionStats
|
||||
getConnectionServers' c connId = do
|
||||
@@ -2941,23 +2940,23 @@ suspendAgent :: AgentClient -> Int -> IO ()
|
||||
suspendAgent c 0 = do
|
||||
atomically $ writeTVar (agentState c) ASSuspended
|
||||
mapM_ suspend agentOperations
|
||||
notifyEvent c ("", "", AEvt SAENone SUSPENDED)
|
||||
where
|
||||
suspend opSel = atomically $ modifyTVar' (opSel c) $ \s -> s {opSuspended = True}
|
||||
suspendAgent c@AgentClient {agentState = as} maxDelay = do
|
||||
state <-
|
||||
atomically $ do
|
||||
writeTVar as ASSuspending
|
||||
suspendOperation c AONtfNetwork $ pure ()
|
||||
suspendOperation c AORcvNetwork $
|
||||
suspendOperation c AOMsgDelivery $
|
||||
suspendSendingAndDatabase c
|
||||
readTVar as
|
||||
(state, suspended) <- atomically $ do
|
||||
writeTVar as ASSuspending
|
||||
void $ suspendOperation c AONtfNetwork $ pure False
|
||||
suspended <- suspendOperation c AORcvNetwork $
|
||||
suspendOperation c AOMsgDelivery $
|
||||
suspendSendingAndDatabase c
|
||||
(,suspended) <$> readTVar as
|
||||
when suspended $ notifyEvent c ("", "", AEvt SAENone SUSPENDED)
|
||||
when (state == ASSuspending) . void . forkIO $ do
|
||||
threadDelay maxDelay
|
||||
-- liftIO $ putStrLn "suspendAgent after timeout"
|
||||
atomically . whenSuspending c $ do
|
||||
-- unsafeIOToSTM $ putStrLn $ "in timeout: suspendSendingAndDatabase"
|
||||
suspended' <- atomically . whenSuspendingB c $
|
||||
suspendSendingAndDatabase c
|
||||
when suspended' $ notifyEvent c ("", "", AEvt SAENone SUSPENDED)
|
||||
|
||||
execAgentStoreSQL :: AgentClient -> Text -> AE [Text]
|
||||
execAgentStoreSQL c sql = withAgentEnv c $ withStore' c (`execSQL` sql)
|
||||
@@ -2983,16 +2982,16 @@ getNextSMPServer c userId = getNextServer c userId storageSrvs
|
||||
{-# INLINE getNextSMPServer #-}
|
||||
|
||||
subscriber :: AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> AM' ()
|
||||
subscriber c@AgentClient {subQ} t = run $
|
||||
subscriber c t = run $
|
||||
agentOperationBracket c AORcvNetwork waitUntilActive $
|
||||
processSMPTransmissions c t
|
||||
where
|
||||
run a = a `catchOwn` \e -> notify $ CRITICAL True $ "subscriber error: " <> show e
|
||||
notify err = atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR err)
|
||||
notify err = liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR err)
|
||||
|
||||
|
||||
cleanupManager :: AgentClient -> AM' ()
|
||||
cleanupManager c@AgentClient {subQ} = do
|
||||
cleanupManager c = do
|
||||
AgentConfig {initialCleanupDelay, cleanupInterval = int, storedMsgDataTTL = ttl, cleanupBatchSize = limit} <-
|
||||
asks config
|
||||
liftIO $ threadDelay' initialCleanupDelay
|
||||
@@ -3060,7 +3059,7 @@ cleanupManager c@AgentClient {subQ} = do
|
||||
rcvFilesTTL <- asks $ rcvFilesTTL . config
|
||||
withStore' c (`deleteDeletedSndChunkReplicasExpired` rcvFilesTTL)
|
||||
notify :: forall e. AEntityI e => AEntityId -> AEvent e -> AM ()
|
||||
notify entId cmd = atomically $ writeTBQueue subQ ("", entId, AEvt (sAEntity @e) cmd)
|
||||
notify entId cmd = liftIO $ notifyEvent c ("", entId, AEvt (sAEntity @e) cmd)
|
||||
|
||||
data ACKd = ACKd | ACKPending
|
||||
|
||||
@@ -3068,7 +3067,7 @@ data ACKd = ACKd | ACKPending
|
||||
-- It cannot be finally, as sometimes it needs to be ACK+DEL,
|
||||
-- and sometimes ACK has to be sent from the consumer.
|
||||
processSMPTransmissions :: AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> AM' ()
|
||||
processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandleParams {thAuth, sessionId = sessId}, ts) = do
|
||||
processSMPTransmissions c (tSess@(userId, srv, _), THandleParams {thAuth, sessionId = sessId}, ts) = do
|
||||
upConnIds <- newTVarIO []
|
||||
serviceRQs <- newTVarIO ([] :: [RcvQueue])
|
||||
forM_ ts $ \(entId, t) -> case t of
|
||||
@@ -3141,14 +3140,14 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
|
||||
(atomically $ putTMVar (clientNoticesLock c) ())
|
||||
(processClientNotices c tSess [(rcvQueueSub rq, notice_)])
|
||||
notify' :: forall e m. (AEntityI e, MonadIO m) => ConnId -> AEvent e -> m ()
|
||||
notify' connId msg = atomically $ writeTBQueue subQ ("", connId, AEvt (sAEntity @e) msg)
|
||||
notify' connId msg = liftIO $ notifyEvent c ("", connId, AEvt (sAEntity @e) msg)
|
||||
notifyErr :: ConnId -> SMPClientError -> AM' ()
|
||||
notifyErr connId = notify' connId . ERR . protocolClientError SMP (B.unpack $ strEncode srv)
|
||||
runProcessSMP :: RcvQueue -> Connection c -> ConnData -> BrokerMsg -> AM ()
|
||||
runProcessSMP rq conn cData msg = do
|
||||
pending <- newTVarIO []
|
||||
processSMP rq conn cData msg pending
|
||||
mapM_ (atomically . writeTBQueue subQ) . reverse =<< readTVarIO pending
|
||||
mapM_ (liftIO . notifyEvent c) . reverse =<< readTVarIO pending
|
||||
processSMP :: forall c. RcvQueue -> Connection c -> ConnData -> BrokerMsg -> TVar [ATransmission] -> AM ()
|
||||
processSMP
|
||||
rq@RcvQueue {rcvId = rId, queueMode, e2ePrivKey, e2eDhSecret, status, smpClientVersion = agreedClientVerion}
|
||||
@@ -3355,9 +3354,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
|
||||
notify :: forall e m. (AEntityI e, MonadIO m) => AEvent e -> m ()
|
||||
notify = notify_ connId
|
||||
notify_ :: forall e m. (AEntityI e, MonadIO m) => ConnId -> AEvent e -> m ()
|
||||
notify_ connId' msg =
|
||||
let t = ("", connId', AEvt (sAEntity @e) msg)
|
||||
in atomically $ ifM (isFullTBQueue subQ) (modifyTVar' pendingMsgs (t :)) (writeTBQueue subQ t)
|
||||
notify_ connId' msg = atomically $ modifyTVar' pendingMsgs (("", connId', AEvt (sAEntity @e) msg) :)
|
||||
|
||||
prohibited :: Text -> AM ()
|
||||
prohibited s = do
|
||||
|
||||
@@ -157,6 +157,7 @@ module Simplex.Messaging.Agent.Client
|
||||
suspendOperation,
|
||||
notifySuspended,
|
||||
whenSuspending,
|
||||
whenSuspendingB,
|
||||
withStore,
|
||||
withStore',
|
||||
withStoreBatch,
|
||||
@@ -165,6 +166,8 @@ module Simplex.Messaging.Agent.Client
|
||||
storeError,
|
||||
notifySub,
|
||||
notifySub',
|
||||
notifyEvent,
|
||||
connWorkerLoop,
|
||||
userServers,
|
||||
pickServer,
|
||||
getNextServer,
|
||||
@@ -334,10 +337,20 @@ type NtfTransportSession = TransportSession NtfResponse
|
||||
|
||||
type XFTPTransportSession = TransportSession FileResponse
|
||||
|
||||
data EventWorker = EventWorker
|
||||
{ eventQ :: TBQueue ATransmission,
|
||||
workerThreadId :: Weak ThreadId
|
||||
}
|
||||
|
||||
type EventWorkerVar = SessionVar EventWorker
|
||||
|
||||
data AgentClient = AgentClient
|
||||
{ acThread :: TVar (Maybe (Weak ThreadId)),
|
||||
active :: TVar Bool,
|
||||
subQ :: TBQueue ATransmission,
|
||||
processEvent :: ATransmission -> IO (),
|
||||
generalQ :: TBQueue ATransmission,
|
||||
connWorkers :: TMap ConnId EventWorkerVar,
|
||||
connWorkerSeq :: TVar Int,
|
||||
processServerMsg :: AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO (),
|
||||
smpServers :: TMap UserId (UserServers 'PSMP),
|
||||
smpClients :: TMap SMPTransportSession SMPClientVar,
|
||||
@@ -419,7 +432,8 @@ getAgentWorker' toW fromW name hasWork c@AgentClient {agentEnv} key ws work = do
|
||||
t <- liftIO getSystemTime
|
||||
let maxRestarts = maxWorkerRestartsPerMin $ config agentEnv
|
||||
-- worker may terminate because it was deleted from the map (getWorker returns Nothing), then it won't restart
|
||||
restart <- atomically $ getWorker >>= maybe (pure False) (shouldRestart e_ (toW w) t maxRestarts)
|
||||
(restart, notify_) <- atomically $ getWorker >>= maybe (pure (False, Nothing)) (shouldRestart e_ (toW w) t maxRestarts)
|
||||
forM_ notify_ $ liftIO . notifyEvent c
|
||||
when restart runWork
|
||||
shouldRestart e_ Worker {workerId = wId, doWork, action, restarts} t maxRestarts w'
|
||||
| wId == workerId (toW w') = do
|
||||
@@ -427,24 +441,21 @@ getAgentWorker' toW fromW name hasWork c@AgentClient {agentEnv} key ws work = do
|
||||
isActive <- readTVar $ active c
|
||||
checkRestarts isActive $ updateRestartCount t rc
|
||||
| otherwise =
|
||||
pure False -- there is a new worker in the map, no action
|
||||
pure (False, Nothing) -- there is a new worker in the map, no action
|
||||
where
|
||||
checkRestarts isActive rc
|
||||
| isActive && restartCount rc < maxRestarts = do
|
||||
writeTVar restarts rc
|
||||
hasWorkToDo' doWork
|
||||
void $ tryPutTMVar action Nothing
|
||||
notifyErr INTERNAL
|
||||
pure True
|
||||
pure (True, Just $ notifyMsg rc INTERNAL)
|
||||
| otherwise = do
|
||||
TM.delete key ws
|
||||
when isActive $ notifyErr $ CRITICAL True
|
||||
pure False
|
||||
where
|
||||
notifyErr err = do
|
||||
let e = either ((", error: " <>) . show) (\_ -> ", no error") e_
|
||||
msg = "Worker " <> name <> " for " <> show key <> " terminated " <> show (restartCount rc) <> " times" <> e
|
||||
writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err msg)
|
||||
pure (False, if isActive then Just (notifyMsg rc $ CRITICAL True) else Nothing)
|
||||
notifyMsg rc err =
|
||||
let e = either ((", error: " <>) . show) (\_ -> ", no error") e_
|
||||
msg = "Worker " <> name <> " for " <> show key <> " terminated " <> show (restartCount rc) <> " times" <> e
|
||||
in ("", "", AEvt SAEConn $ ERR $ err msg)
|
||||
|
||||
newWorker :: AgentClient -> STM Worker
|
||||
newWorker c = do
|
||||
@@ -505,14 +516,16 @@ data UserNetworkType = UNNone | UNCellular | UNWifi | UNEthernet | UNOther
|
||||
deriving (Eq, Show)
|
||||
|
||||
-- | Creates an SMP agent client instance that receives commands and sends responses via 'TBQueue's.
|
||||
newAgentClient :: Int -> InitialAgentServers -> UTCTime -> Map (Maybe SMPServer) (Maybe SystemSeconds) -> (AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> Env -> IO AgentClient
|
||||
newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, useServices, presetDomains, presetServers} currentTs notices processServerMsg agentEnv = do
|
||||
newAgentClient :: Int -> InitialAgentServers -> UTCTime -> Map (Maybe SMPServer) (Maybe SystemSeconds) -> (ATransmission -> IO ()) -> (AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> Env -> IO AgentClient
|
||||
newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, useServices, presetDomains, presetServers} currentTs notices processEvent processServerMsg agentEnv = do
|
||||
let cfg = config agentEnv
|
||||
qSize = tbqSize cfg
|
||||
proxySessTs <- newTVarIO =<< getCurrentTime
|
||||
acThread <- newTVarIO Nothing
|
||||
active <- newTVarIO True
|
||||
subQ <- newTBQueueIO qSize
|
||||
generalQ <- newTBQueueIO qSize
|
||||
connWorkers <- TM.emptyIO
|
||||
connWorkerSeq <- newTVarIO 0
|
||||
smpServers <- newTVarIO $ M.map mkUserServers smp
|
||||
smpClients <- TM.emptyIO
|
||||
useClientServices <- newTVarIO useServices
|
||||
@@ -551,7 +564,10 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, useServices
|
||||
AgentClient
|
||||
{ acThread,
|
||||
active,
|
||||
subQ,
|
||||
processEvent,
|
||||
generalQ,
|
||||
connWorkers,
|
||||
connWorkerSeq,
|
||||
processServerMsg,
|
||||
smpServers,
|
||||
smpClients,
|
||||
@@ -834,7 +850,7 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess = do
|
||||
handleNotify = E.handleAny $ notifySub' c "" . ERR . INTERNAL . show
|
||||
|
||||
notifySub' :: forall e m. (AEntityI e, MonadIO m) => AgentClient -> ConnId -> AEvent e -> m ()
|
||||
notifySub' c connId cmd = liftIO $ nonBlockingWriteTBQueue (subQ c) (B.empty, connId, AEvt (sAEntity @e) cmd)
|
||||
notifySub' c connId cmd = liftIO $ notifyEvent c (B.empty, connId, AEvt (sAEntity @e) cmd)
|
||||
{-# INLINE notifySub' #-}
|
||||
|
||||
notifySub :: MonadIO m => AgentClient -> AEvent 'AENone -> m ()
|
||||
@@ -862,7 +878,7 @@ getNtfServerClient c@AgentClient {active, ntfClients, workerSeq, proxySessTs, pr
|
||||
clientDisconnected :: NtfClientVar -> NtfClient -> IO ()
|
||||
clientDisconnected v client = do
|
||||
atomically $ removeSessVar v tSess ntfClients
|
||||
atomically $ writeTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent DISCONNECT client)
|
||||
notifyEvent c ("", "", AEvt SAENone $ hostEvent DISCONNECT client)
|
||||
logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
|
||||
|
||||
getXFTPServerClient :: AgentClient -> XFTPTransportSession -> AM XFTPClient
|
||||
@@ -886,7 +902,7 @@ getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq, proxySessTs,
|
||||
clientDisconnected :: XFTPClientVar -> XFTPClient -> IO ()
|
||||
clientDisconnected v client = do
|
||||
atomically $ removeSessVar v tSess xftpClients
|
||||
atomically $ writeTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent DISCONNECT client)
|
||||
notifyEvent c ("", "", AEvt SAENone $ hostEvent DISCONNECT client)
|
||||
logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
|
||||
|
||||
waitForProtocolClient ::
|
||||
@@ -925,7 +941,7 @@ newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v =
|
||||
Right client -> do
|
||||
logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
|
||||
atomically $ putTMVar (sessionVar v) (Right client)
|
||||
liftIO $ nonBlockingWriteTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent CONNECT client)
|
||||
liftIO $ notifyEvent c ("", "", AEvt SAENone $ hostEvent CONNECT client)
|
||||
pure client
|
||||
Left e -> do
|
||||
ei <- asks $ persistErrorInterval . config
|
||||
@@ -1053,6 +1069,30 @@ withConnLock' _ "" _ = id
|
||||
withConnLock' AgentClient {connLocks} connId name = withLockMap connLocks connId name
|
||||
{-# INLINE withConnLock' #-}
|
||||
|
||||
notifyEvent :: AgentClient -> ATransmission -> IO ()
|
||||
notifyEvent c t@(_, connId, _)
|
||||
| B.null connId = atomically $ writeTBQueue (generalQ c) t
|
||||
| otherwise = do
|
||||
q <- getOrCreateConnWorker c connId
|
||||
atomically $ writeTBQueue q t
|
||||
|
||||
getOrCreateConnWorker :: AgentClient -> ConnId -> IO (TBQueue ATransmission)
|
||||
getOrCreateConnWorker c@AgentClient {connWorkers, connWorkerSeq} connId = do
|
||||
ts <- getCurrentTime
|
||||
atomically (getSessVar connWorkerSeq connId connWorkers ts) >>= \case
|
||||
Left v -> do
|
||||
q <- newTBQueueIO 64
|
||||
tId <- mkWeakThreadId =<< forkIO (connWorkerLoop c q)
|
||||
atomically $ putTMVar (sessionVar v) EventWorker {eventQ = q, workerThreadId = tId}
|
||||
pure q
|
||||
Right v -> eventQ <$> atomically (readTMVar $ sessionVar v)
|
||||
|
||||
connWorkerLoop :: AgentClient -> TBQueue ATransmission -> IO ()
|
||||
connWorkerLoop AgentClient {processEvent} q = forever $ do
|
||||
t <- atomically $ readTBQueue q
|
||||
processEvent t `E.catchAny` \e ->
|
||||
logError $ "connWorkerLoop error: " <> tshow e
|
||||
|
||||
withInvLock :: AgentClient -> ByteString -> Text -> AM a -> AM a
|
||||
withInvLock c key name = ExceptT . withInvLock' c key name . runExceptT
|
||||
{-# INLINE withInvLock #-}
|
||||
@@ -1739,7 +1779,7 @@ resubscribeClientService c tSess@(userId, srv, _) serviceSub =
|
||||
r <$ withStore' c (\db -> removeRcvServiceAssocs db userId srv)
|
||||
_ -> pure r
|
||||
Left e -> do
|
||||
atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR e)
|
||||
liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR e)
|
||||
when (clientServiceError e) $ do
|
||||
atomically $ SS.deleteServiceSub tSess $ currentSubs c
|
||||
unassocSubscribeQueues
|
||||
@@ -2265,7 +2305,7 @@ withWork_ c doWork getWork action =
|
||||
noWork = liftIO $ noWorkToDo doWork
|
||||
notifyErr err e = do
|
||||
logError $ "withWork_ error: " <> tshow e
|
||||
atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
|
||||
liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR $ err $ show e)
|
||||
|
||||
withWorkItems :: (AnyStoreError e', MonadIO m) => AgentClient -> TMVar () -> ExceptT e m (Either e' [Either e' a]) -> (NonEmpty a -> ExceptT e m ()) -> ExceptT e m ()
|
||||
withWorkItems c doWork getWork action = do
|
||||
@@ -2290,7 +2330,7 @@ withWorkItems c doWork getWork action = do
|
||||
noWork = liftIO $ noWorkToDo doWork
|
||||
notifyErr err e = do
|
||||
logError $ "withWorkItems error: " <> tshow e
|
||||
atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
|
||||
liftIO $ notifyEvent c ("", "", AEvt SAEConn $ ERR $ err $ show e)
|
||||
|
||||
noWorkToDo :: TMVar () -> IO ()
|
||||
noWorkToDo = void . atomically . tryTakeTMVar
|
||||
@@ -2304,9 +2344,9 @@ hasWorkToDo' :: TMVar () -> STM ()
|
||||
hasWorkToDo' = void . (`tryPutTMVar` ())
|
||||
{-# INLINE hasWorkToDo' #-}
|
||||
|
||||
endAgentOperation :: AgentClient -> AgentOperation -> STM ()
|
||||
endAgentOperation :: AgentClient -> AgentOperation -> STM Bool
|
||||
endAgentOperation c op = endOperation c op $ case op of
|
||||
AONtfNetwork -> pure ()
|
||||
AONtfNetwork -> pure False
|
||||
AORcvNetwork ->
|
||||
suspendOperation c AOMsgDelivery $
|
||||
suspendSendingAndDatabase c
|
||||
@@ -2318,36 +2358,37 @@ endAgentOperation c op = endOperation c op $ case op of
|
||||
AODatabase ->
|
||||
notifySuspended c
|
||||
|
||||
suspendSendingAndDatabase :: AgentClient -> STM ()
|
||||
suspendSendingAndDatabase :: AgentClient -> STM Bool
|
||||
suspendSendingAndDatabase c =
|
||||
suspendOperation c AOSndNetwork $
|
||||
suspendOperation c AODatabase $
|
||||
notifySuspended c
|
||||
|
||||
suspendOperation :: AgentClient -> AgentOperation -> STM () -> STM ()
|
||||
suspendOperation :: AgentClient -> AgentOperation -> STM Bool -> STM Bool
|
||||
suspendOperation c op endedAction = do
|
||||
n <- stateTVar (agentOpSel op c) $ \s -> (opsInProgress s, s {opSuspended = True})
|
||||
-- unsafeIOToSTM $ putStrLn $ "suspendOperation_ " <> show op <> " " <> show n
|
||||
when (n == 0) $ whenSuspending c endedAction
|
||||
if n == 0 then whenSuspendingB c endedAction else pure False
|
||||
|
||||
notifySuspended :: AgentClient -> STM ()
|
||||
notifySuspended :: AgentClient -> STM Bool
|
||||
notifySuspended c = do
|
||||
-- unsafeIOToSTM $ putStrLn "notifySuspended"
|
||||
writeTBQueue (subQ c) ("", "", AEvt SAENone SUSPENDED)
|
||||
writeTVar (agentState c) ASSuspended
|
||||
pure True
|
||||
|
||||
endOperation :: AgentClient -> AgentOperation -> STM () -> STM ()
|
||||
endOperation :: AgentClient -> AgentOperation -> STM Bool -> STM Bool
|
||||
endOperation c op endedAction = do
|
||||
(suspended, n) <- stateTVar (agentOpSel op c) $ \s ->
|
||||
let n = max 0 (opsInProgress s - 1)
|
||||
in ((opSuspended s, n), s {opsInProgress = n})
|
||||
-- unsafeIOToSTM $ putStrLn $ "endOperation: " <> show op <> " " <> show suspended <> " " <> show n
|
||||
when (suspended && n == 0) $ whenSuspending c endedAction
|
||||
if suspended && n == 0 then whenSuspendingB c endedAction else pure False
|
||||
|
||||
whenSuspending :: AgentClient -> STM () -> STM ()
|
||||
whenSuspending c = whenM ((== ASSuspending) <$> readTVar (agentState c))
|
||||
{-# INLINE whenSuspending #-}
|
||||
|
||||
whenSuspendingB :: AgentClient -> STM Bool -> STM Bool
|
||||
whenSuspendingB c action =
|
||||
ifM ((== ASSuspending) <$> readTVar (agentState c)) action (pure False)
|
||||
|
||||
beginAgentOperation :: AgentClient -> AgentOperation -> STM ()
|
||||
beginAgentOperation c op = do
|
||||
let opVar = agentOpSel op c
|
||||
@@ -2361,7 +2402,9 @@ agentOperationBracket :: MonadUnliftIO m => AgentClient -> AgentOperation -> (Ag
|
||||
agentOperationBracket c op check action =
|
||||
E.bracket
|
||||
(liftIO (check c) >> atomically (beginAgentOperation c op))
|
||||
(\_ -> atomically $ endAgentOperation c op)
|
||||
(\_ -> do
|
||||
suspended <- atomically $ endAgentOperation c op
|
||||
when suspended $ liftIO $ notifyEvent c ("", "", AEvt SAENone SUSPENDED))
|
||||
(const action)
|
||||
|
||||
waitUntilForeground :: AgentClient -> IO ()
|
||||
@@ -2834,9 +2877,9 @@ data ClientInfo
|
||||
deriving (Show)
|
||||
|
||||
getAgentQueuesInfo :: AgentClient -> IO AgentQueuesInfo
|
||||
getAgentQueuesInfo AgentClient {subQ, smpClients} = do
|
||||
getAgentQueuesInfo AgentClient {smpClients} = do
|
||||
let msgQInfo = TBQueueInfo {qLength = 0, qFull = False}
|
||||
subQInfo <- atomically $ getTBQueueInfo subQ
|
||||
subQInfo = TBQueueInfo {qLength = 0, qFull = False}
|
||||
smpClientsMap <- readTVarIO smpClients
|
||||
let smpClientsMap' = M.mapKeys (decodeLatin1 . strEncode) smpClientsMap
|
||||
smpClientsQueues <- mapM getClientQueuesInfo smpClientsMap'
|
||||
|
||||
@@ -502,9 +502,9 @@ workerInternalError c connId internalErrStr = do
|
||||
|
||||
-- TODO change error
|
||||
notifyInternalError :: MonadIO m => AgentClient -> ConnId -> String -> m ()
|
||||
notifyInternalError AgentClient {subQ} connId internalErrStr = do
|
||||
notifyInternalError c connId internalErrStr = do
|
||||
logError $ T.pack internalErrStr
|
||||
liftIO $ nonBlockingWriteTBQueue subQ ("", connId, AEvt SAEConn $ ERR $ INTERNAL internalErrStr)
|
||||
liftIO $ notifyEvent c ("", connId, AEvt SAEConn $ ERR $ INTERNAL internalErrStr)
|
||||
|
||||
notifyInternalError' :: MonadIO m => AgentClient -> String -> m ()
|
||||
notifyInternalError' c = notifyInternalError c ""
|
||||
|
||||
Reference in New Issue
Block a user