mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-09-01 20:18:26 +00:00
remove subscribed clients from map
This commit is contained in:
Evgeny Poberezkin
@@ -260,9 +260,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
|
||||
upsertSubscribedClient qId c queueSubscribers
|
||||
| otherwise = do
|
||||
removeWhenNoSubs c
|
||||
lookupRemoveSubscribedClient qId queueSubscribers
|
||||
lookupDeleteSubscribedClient qId queueSubscribers
|
||||
-- do not insert client if it is already disconnected, but send END to any other client
|
||||
updateSubDisconnected = lookupRemoveSubscribedClient qId queueSubscribers
|
||||
updateSubDisconnected = lookupDeleteSubscribedClient qId queueSubscribers
|
||||
clientToBeNotified ac@(AClient _ _ Client {clientId, connected})
|
||||
| clntId == clientId = pure Nothing
|
||||
| otherwise = (\yes -> if yes then Just ((qId, subEvt), ac) else Nothing) <$> readTVar connected
|
||||
@@ -585,9 +585,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
|
||||
pure RealTimeMetrics {socketStats, threadsCount, clientsCount, smpSubs, ntfSubs, loadedCounts}
|
||||
where
|
||||
getSubscribersMetrics ServerSubscribers {queueSubscribers, subClients} = do
|
||||
(storedSubs, subsCount) <- getSubscribedClients queueSubscribers
|
||||
subsCount <- M.size <$> getSubscribedClients queueSubscribers
|
||||
subClientsCount <- IS.size <$> readTVarIO subClients
|
||||
pure RTSubscriberMetrics {subsCount, subVarsCount = M.size storedSubs, subClientsCount}
|
||||
pure RTSubscriberMetrics {subsCount, subClientsCount}
|
||||
|
||||
runClient :: Transport c => C.APrivateSignKey -> TProxy c -> c -> M ()
|
||||
runClient signKey tp h = do
|
||||
@@ -782,10 +782,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
|
||||
where
|
||||
putActiveClientsInfo :: String -> SubscribedClients -> Bool -> IO ()
|
||||
putActiveClientsInfo protoName clients showIds = do
|
||||
(storedSubs, subsCount) <- getSubscribedClients clients
|
||||
hPutStrLn h $ protoName <> " subscription vars: " <> show (M.size storedSubs)
|
||||
hPutStrLn h $ protoName <> " subscriptions: " <> show subsCount
|
||||
clnts <- countSubClients storedSubs
|
||||
activeSubs <- getSubscribedClients clients
|
||||
hPutStrLn h $ protoName <> " subscriptions: " <> show (M.size activeSubs)
|
||||
clnts <- countSubClients activeSubs
|
||||
hPutStrLn h $ protoName <> " subscribed clients: " <> show (IS.size clnts) <> (if showIds then " " <> show (IS.toList clnts) else "")
|
||||
where
|
||||
countSubClients :: M.Map QueueId (TVar (Maybe AClient)) -> IO IS.IntSet
|
||||
@@ -931,7 +930,7 @@ clientDisconnected c@Client {clientId, subscriptions, ntfSubscriptions, connecte
|
||||
where
|
||||
updateSubscribers :: M.Map QueueId a -> ServerSubscribers -> IO ()
|
||||
updateSubscribers subs ServerSubscribers {queueSubscribers, subClients} = do
|
||||
mapM_ (\qId -> removeSubcribedClient qId c queueSubscribers) (M.keys subs)
|
||||
mapM_ (\qId -> deleteSubcribedClient qId c queueSubscribers) (M.keys subs)
|
||||
atomically $ modifyTVar' subClients $ IS.delete clientId
|
||||
|
||||
cancelSub :: Sub -> IO ()
|
||||
|
||||
@@ -49,8 +49,8 @@ module Simplex.Messaging.Server.Env.STM
|
||||
getSubscribedClients,
|
||||
getSubscribedClient,
|
||||
upsertSubscribedClient,
|
||||
lookupRemoveSubscribedClient,
|
||||
removeSubcribedClient,
|
||||
lookupDeleteSubscribedClient,
|
||||
deleteSubcribedClient,
|
||||
sameClientId,
|
||||
sameClient,
|
||||
clientId',
|
||||
@@ -314,48 +314,40 @@ data ServerSubscribers = ServerSubscribers
|
||||
-- any STM transaction that reads subscribed client will re-evaluate in this case.
|
||||
-- The subscriptions that were made at any point are not removed -
|
||||
-- this is a better trade-off with intermittently connected mobile clients.
|
||||
data SubscribedClients = SubscribedClients {subscribedClients :: TMap EntityId (TVar (Maybe AClient)), subcribedCount :: TVar Int}
|
||||
data SubscribedClients = SubscribedClients (TMap EntityId (TVar (Maybe AClient)))
|
||||
|
||||
getSubscribedClients :: SubscribedClients -> IO (Map EntityId (TVar (Maybe AClient)), Int)
|
||||
getSubscribedClients (SubscribedClients cs cnt) = (,) <$> readTVarIO cs <*> readTVarIO cnt
|
||||
getSubscribedClients :: SubscribedClients -> IO (Map EntityId (TVar (Maybe AClient)))
|
||||
getSubscribedClients (SubscribedClients cs) = readTVarIO cs
|
||||
|
||||
getSubscribedClient :: EntityId -> SubscribedClients -> IO (Maybe (TVar (Maybe AClient)))
|
||||
getSubscribedClient entId (SubscribedClients cs _) = TM.lookupIO entId cs
|
||||
getSubscribedClient entId (SubscribedClients cs) = TM.lookupIO entId cs
|
||||
{-# INLINE getSubscribedClient #-}
|
||||
|
||||
-- insert subscribed and current client, return previously subscribed client if it is different
|
||||
upsertSubscribedClient :: EntityId -> AClient -> SubscribedClients -> STM (Maybe AClient)
|
||||
upsertSubscribedClient entId ac@(AClient _ _ c) (SubscribedClients cs cnt) =
|
||||
upsertSubscribedClient entId ac@(AClient _ _ c) (SubscribedClients cs) =
|
||||
TM.lookup entId cs >>= \case
|
||||
Nothing -> do
|
||||
TM.insertM entId (newTVar $ Just ac) cs
|
||||
modifyTVar' cnt (+ 1)
|
||||
pure Nothing
|
||||
Nothing -> Nothing <$ TM.insertM entId (newTVar (Just ac)) cs
|
||||
Just cv ->
|
||||
readTVar cv >>= \case
|
||||
Just c'
|
||||
| sameClientId c c' -> pure Nothing
|
||||
| otherwise -> Just c' <$ writeTVar cv (Just ac)
|
||||
Nothing -> do
|
||||
writeTVar cv (Just ac)
|
||||
modifyTVar' cnt (+ 1)
|
||||
pure Nothing
|
||||
Just c' | sameClientId c c' -> pure Nothing
|
||||
c_ -> c_ <$ writeTVar cv (Just ac)
|
||||
|
||||
-- lookup and delete currently subscribed client
|
||||
lookupRemoveSubscribedClient :: EntityId -> SubscribedClients -> STM (Maybe AClient)
|
||||
lookupRemoveSubscribedClient entId (SubscribedClients cs cnt) =
|
||||
TM.lookup entId cs $>>= (`swapTVar` Nothing) >>= mapM (<$ modifyTVar' cnt (subtract 1))
|
||||
lookupDeleteSubscribedClient :: EntityId -> SubscribedClients -> STM (Maybe AClient)
|
||||
lookupDeleteSubscribedClient entId (SubscribedClients cs) =
|
||||
TM.lookupDelete entId cs $>>= (`swapTVar` Nothing)
|
||||
|
||||
removeSubcribedClient :: EntityId -> Client s -> SubscribedClients -> IO ()
|
||||
removeSubcribedClient entId c (SubscribedClients cs cnt) =
|
||||
deleteSubcribedClient :: EntityId -> Client s -> SubscribedClients -> IO ()
|
||||
deleteSubcribedClient entId c (SubscribedClients cs) =
|
||||
-- lookup of the subscribed client TVar can be in separate transaction,
|
||||
-- as long as the client is read in the same transaction -
|
||||
-- it prevents removing the next subscribed client and also avoids STM contention for the Map.
|
||||
TM.lookupIO entId cs >>= mapM_ (\cv -> atomically $ whenM (sameClient c cv) $ remove cv)
|
||||
TM.lookupIO entId cs >>= mapM_ (\cv -> atomically $ whenM (sameClient c cv) $ delete cv)
|
||||
where
|
||||
remove cv = do
|
||||
delete cv = do
|
||||
writeTVar cv Nothing
|
||||
modifyTVar' cnt (subtract 1)
|
||||
TM.delete entId cs
|
||||
|
||||
sameClientId :: Client s -> AClient -> Bool
|
||||
sameClientId Client {clientId} ac = clientId == clientId' ac
|
||||
@@ -436,7 +428,7 @@ deleteServerClient cId Server {clients} = atomically $ modifyTVar' (serverClient
|
||||
newServerSubscribers :: IO ServerSubscribers
|
||||
newServerSubscribers = do
|
||||
subQ <- newTQueueIO
|
||||
queueSubscribers <- SubscribedClients <$> TM.emptyIO <*> newTVarIO 0
|
||||
queueSubscribers <- SubscribedClients <$> TM.emptyIO
|
||||
subClients <- newTVarIO IS.empty
|
||||
pendingEvents <- newTVarIO IM.empty
|
||||
pure ServerSubscribers {subQ, queueSubscribers, subClients, pendingEvents}
|
||||
|
||||
@@ -40,7 +40,6 @@ data RealTimeMetrics = RealTimeMetrics
|
||||
|
||||
data RTSubscriberMetrics = RTSubscriberMetrics
|
||||
{ subsCount :: Int,
|
||||
subVarsCount :: Int,
|
||||
subClientsCount :: Int
|
||||
}
|
||||
|
||||
@@ -373,10 +372,6 @@ prometheusMetrics sm rtm ts =
|
||||
\# TYPE simplex_smp_subscribtion_total gauge\n\
|
||||
\simplex_smp_subscribtion_total " <> mshow (subsCount smpSubs) <> "\n# smp.subsCount\n\
|
||||
\\n\
|
||||
\# HELP simplex_smp_subscribtion_vars_total Total SMP subscription vars\n\
|
||||
\# TYPE simplex_smp_subscribtion_vars_total gauge\n\
|
||||
\simplex_smp_subscribtion_vars_total " <> mshow (subVarsCount smpSubs) <> "\n# smp.subVarsCount\n\
|
||||
\\n\
|
||||
\# HELP simplex_smp_subscribtion_clients_total Subscribed clients\n\
|
||||
\# TYPE simplex_smp_subscribtion_clients_total gauge\n\
|
||||
\simplex_smp_subscribtion_clients_total " <> mshow (subClientsCount smpSubs) <> "\n# smp.subClientsCount\n\
|
||||
@@ -385,10 +380,6 @@ prometheusMetrics sm rtm ts =
|
||||
\# TYPE simplex_smp_subscription_ntf_total gauge\n\
|
||||
\simplex_smp_subscription_ntf_total " <> mshow (subsCount ntfSubs) <> "\n# ntf.subsCount\n\
|
||||
\\n\
|
||||
\# HELP simplex_smp_subscription_ntf_vars_total Total notification subscripbtion vars (from ntf server)\n\
|
||||
\# TYPE simplex_smp_subscription_ntf_vars_total gauge\n\
|
||||
\simplex_smp_subscription_ntf_vars_total " <> mshow (subVarsCount ntfSubs) <> "\n# ntf.subVarsCount\n\
|
||||
\\n\
|
||||
\# HELP simplex_smp_subscription_ntf_clients_total Total subscribed NTF servers\n\
|
||||
\# TYPE simplex_smp_subscription_ntf_clients_total gauge\n\
|
||||
\simplex_smp_subscription_ntf_clients_total " <> mshow (subClientsCount ntfSubs) <> "\n# ntf.subClientsCount\n\
|
||||
|
||||
Reference in New Issue
Block a user