From 1b5a9f3b0ca82e0fe729c10c213ca1d76473febb Mon Sep 17 00:00:00 2001 From: Evgeny Date: Sat, 15 Mar 2025 20:11:30 +0000 Subject: [PATCH] smp server: do not cache all queues from database while processing expirations (#1483) * smp server: expire only active queues * version * do not cache all queues while processing expirations * refactor * foldWithOptions_ * version * use shared lock when expiring all queues * use TMVar * comment * rename * remove fold options * do not create locks in the Map for temporarily loaded queues * fix * revert version --- simplexmq.cabal | 1 + src/Simplex/Messaging/Agent/Lock.hs | 1 + src/Simplex/Messaging/Server.hs | 51 ++++----- .../Messaging/Server/MsgStore/Journal.hs | 104 +++++++++++++----- .../Server/MsgStore/Journal/SharedLock.hs | 43 ++++++++ src/Simplex/Messaging/Server/MsgStore/STM.hs | 13 ++- .../Messaging/Server/MsgStore/Types.hs | 20 ++-- .../Messaging/Server/QueueStore/Postgres.hs | 48 ++++---- tests/CoreTests/MsgStoreTests.hs | 2 +- 9 files changed, 186 insertions(+), 97 deletions(-) create mode 100644 src/Simplex/Messaging/Server/MsgStore/Journal/SharedLock.hs diff --git a/simplexmq.cabal b/simplexmq.cabal index 457394f72..e081f67ca 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -228,6 +228,7 @@ library Simplex.Messaging.Server.Main Simplex.Messaging.Server.MsgStore Simplex.Messaging.Server.MsgStore.Journal + Simplex.Messaging.Server.MsgStore.Journal.SharedLock Simplex.Messaging.Server.MsgStore.STM Simplex.Messaging.Server.MsgStore.Types Simplex.Messaging.Server.NtfStore diff --git a/src/Simplex/Messaging/Agent/Lock.hs b/src/Simplex/Messaging/Agent/Lock.hs index 3c087499c..43b2358fd 100644 --- a/src/Simplex/Messaging/Agent/Lock.hs +++ b/src/Simplex/Messaging/Agent/Lock.hs @@ -6,6 +6,7 @@ module Simplex.Messaging.Agent.Lock withLock', withGetLock, withGetLocks, + getPutLock, ) where diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index f0b97529a..d816118d5 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -97,7 +97,7 @@ import Simplex.Messaging.Server.Control import Simplex.Messaging.Server.Env.STM as Env import Simplex.Messaging.Server.Expiration import Simplex.Messaging.Server.MsgStore -import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgStore, JournalQueue, closeMsgQueue) +import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgStore, JournalQueue) import Simplex.Messaging.Server.MsgStore.STM import Simplex.Messaging.Server.MsgStore.Types import Simplex.Messaging.Server.NtfStore @@ -404,11 +404,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt old <- expireBeforeEpoch expCfg now <- systemSeconds <$> getSystemTime msgStats@MessageStats {storedMsgsCount = stored, expiredMsgsCount = expired} <- - withAllMsgQueues False ms $ expireQueueMsgs now ms old + withAllMsgQueues False "idleDeleteExpiredMsgs" ms $ expireQueueMsgs now ms old atomicWriteIORef (msgCount stats) stored atomicModifyIORef'_ (msgExpired stats) (+ expired) printMessageStats "STORE: messages" msgStats - expireQueueMsgs now ms old q = fmap (fromRight newMessageStats) . runExceptT $ do + expireQueueMsgs now ms old q = do (expired_, stored) <- idleDeleteExpiredMsgs now ms q old pure MessageStats {storedMsgsCount = stored, expiredMsgsCount = fromMaybe 0 expired_, storedQueues = 1} @@ -1806,19 +1806,18 @@ exportMessages :: MsgStoreClass s => Bool -> s -> FilePath -> Bool -> IO () exportMessages tty ms f drainMsgs = do logInfo $ "saving messages to file " <> T.pack f liftIO $ withFile f WriteMode $ \h -> - tryAny (withAllMsgQueues tty ms $ saveQueueMsgs h) >>= \case + tryAny (unsafeWithAllMsgQueues tty ms $ saveQueueMsgs h) >>= \case Right (Sum total) -> logInfo $ "messages saved: " <> tshow total Left e -> do logError $ "error exporting messages: " <> tshow e exitFailure where saveQueueMsgs h q = do - let rId = recipientId q - runExceptT (getQueueMessages drainMsgs ms q) >>= \case - Right msgs -> Sum (length msgs) <$ BLD.hPutBuilder h (encodeMessages rId msgs) - Left e -> do - logError $ "STORE: saveQueueMsgs, error exporting messages from queue " <> decodeLatin1 (strEncode rId) <> ", " <> tshow e - exitFailure + msgs <- + unsafeRunStore q "saveQueueMsgs" $ + getQueueMessages_ drainMsgs q =<< getMsgQueue ms q False + BLD.hPutBuilder h $ encodeMessages (recipientId q) msgs + pure $ Sum $ length msgs encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n') processServerMessages :: StartOptions -> M (Maybe MessageStats) @@ -1838,33 +1837,23 @@ processServerMessages StartOptions {skipWarnings} = do | expire = Just <$> case old_ of Just old -> do logInfo "expiring journal store messages..." - withAllMsgQueues False ms $ processExpireQueue old + run $ processExpireQueue old Nothing -> do logInfo "validating journal store messages..." - withAllMsgQueues False ms $ processValidateQueue + run processValidateQueue | otherwise = logWarn "skipping message expiration" $> Nothing where + run a = unsafeWithAllMsgQueues False ms a `catchAny` \_ -> exitFailure processExpireQueue :: Int64 -> JournalQueue s -> IO MessageStats - processExpireQueue old q = - runExceptT expireQueue >>= \case - Right (storedMsgsCount, expiredMsgsCount) -> - pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = 1} - Left e -> do - logError $ "STORE: processExpireQueue, failed expiring messages in queue, " <> tshow e - exitFailure - where - expireQueue = do - expired'' <- deleteExpiredMsgs ms q old - stored'' <- getQueueSize ms q - liftIO $ closeMsgQueue q - pure (stored'', expired'') + processExpireQueue old q = unsafeRunStore q "processExpireQueue" $ do + mq <- getMsgQueue ms q False + expiredMsgsCount <- deleteExpireMsgs_ old q mq + storedMsgsCount <- getQueueSize_ mq + pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = 1} processValidateQueue :: JournalQueue s -> IO MessageStats - processValidateQueue q = - runExceptT (getQueueSize ms q) >>= \case - Right storedMsgsCount -> pure newMessageStats {storedMsgsCount, storedQueues = 1} - Left e -> do - logError $ "STORE: processValidateQueue, failed opening message queue, " <> tshow e - exitFailure + processValidateQueue q = unsafeRunStore q "processValidateQueue" $ do + storedMsgsCount <- getQueueSize_ =<< getMsgQueue ms q False + pure newMessageStats {storedMsgsCount, storedQueues = 1} importMessages :: forall s. MsgStoreClass s => Bool -> s -> FilePath -> Maybe Int64 -> Bool -> IO MessageStats importMessages tty ms f old_ skipWarnings = do diff --git a/src/Simplex/Messaging/Server/MsgStore/Journal.hs b/src/Simplex/Messaging/Server/MsgStore/Journal.hs index 74ebac2d3..369194030 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Journal.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Journal.hs @@ -55,6 +55,7 @@ import Control.Monad.Trans.Except import qualified Data.Attoparsec.ByteString.Char8 as A import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B +import Data.Either (fromRight) import Data.Functor (($>)) import Data.Int (Int64) import Data.List (intercalate, sort) @@ -65,10 +66,11 @@ import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, getCurrentTime) import Data.Time.Clock.System (SystemTime (..), getSystemTime) import Data.Time.Format.ISO8601 (iso8601Show, iso8601ParseM) import GHC.IO (catchAny) -import Simplex.Messaging.Agent.Client (getMapLock, withLockMap) +import Simplex.Messaging.Agent.Client (getMapLock) import Simplex.Messaging.Agent.Lock import Simplex.Messaging.Encoding.String import Simplex.Messaging.Protocol +import Simplex.Messaging.Server.MsgStore.Journal.SharedLock import Simplex.Messaging.Server.MsgStore.Types import Simplex.Messaging.Server.QueueStore import Simplex.Messaging.Server.QueueStore.Postgres @@ -87,6 +89,7 @@ data JournalMsgStore s = JournalMsgStore { config :: JournalStoreConfig s, random :: TVar StdGen, queueLocks :: TMap RecipientId Lock, + sharedLock :: TMVar RecipientId, queueStore_ :: QStore s, expireBackupsBefore :: UTCTime } @@ -138,6 +141,7 @@ data QStoreCfg s where data JournalQueue (s :: QSType) = JournalQueue { recipientId' :: RecipientId, queueLock :: Lock, + sharedLock :: TMVar RecipientId, -- To avoid race conditions and errors when restoring queues, -- Nothing is written to TVar when queue is deleted. queueRec' :: TVar (Maybe QueueRec), @@ -276,7 +280,8 @@ instance StoreQueueClass (JournalQueue s) where msgQueue = msgQueue' {-# INLINE msgQueue #-} withQueueLock :: JournalQueue s -> String -> IO a -> IO a - withQueueLock = withLock' . queueLock + withQueueLock JournalQueue {recipientId', queueLock, sharedLock} = + withLockWaitShared recipientId' queueLock sharedLock {-# INLINE withQueueLock #-} instance QueueStoreClass (JournalQueue s) (QStore s) where @@ -316,6 +321,27 @@ instance QueueStoreClass (JournalQueue s) (QStore s) where deleteStoreQueue = withQS deleteStoreQueue {-# INLINE deleteStoreQueue #-} +mkTempQueue :: JournalMsgStore s -> RecipientId -> QueueRec -> IO (JournalQueue s) +mkTempQueue ms rId qr = createLockIO >>= makeQueue_ ms rId qr +{-# INLINE mkTempQueue #-} + +makeQueue_ :: JournalMsgStore s -> RecipientId -> QueueRec -> Lock -> IO (JournalQueue s) +makeQueue_ JournalMsgStore {sharedLock} rId qr queueLock = do + queueRec' <- newTVarIO $ Just qr + msgQueue' <- newTVarIO Nothing + activeAt <- newTVarIO 0 + queueState <- newTVarIO Nothing + pure $ + JournalQueue + { recipientId' = rId, + queueLock, + sharedLock, + queueRec', + msgQueue', + activeAt, + queueState + } + instance MsgStoreClass (JournalMsgStore s) where type StoreMonad (JournalMsgStore s) = StoreIO s type QueueStore (JournalMsgStore s) = QStore s @@ -326,9 +352,10 @@ instance MsgStoreClass (JournalMsgStore s) where newMsgStore config@JournalStoreConfig {queueStoreCfg} = do random <- newTVarIO =<< newStdGen queueLocks <- TM.emptyIO + sharedLock <- newEmptyTMVarIO queueStore_ <- newQueueStore @(JournalQueue s) queueStoreCfg expireBackupsBefore <- addUTCTime (- expireBackupsAfter config) <$> getCurrentTime - pure JournalMsgStore {config, random, queueLocks, queueStore_, expireBackupsBefore} + pure JournalMsgStore {config, random, queueLocks, sharedLock, queueStore_, expireBackupsBefore} closeMsgStore :: JournalMsgStore s -> IO () closeMsgStore ms = do @@ -341,10 +368,32 @@ instance MsgStoreClass (JournalMsgStore s) where withActiveMsgQueues :: Monoid a => JournalMsgStore s -> (JournalQueue s -> IO a) -> IO a withActiveMsgQueues = withQS withLoadedQueues . queueStore_ - withAllMsgQueues :: Monoid a => Bool -> JournalMsgStore s -> (JournalQueue s -> IO a) -> IO a - withAllMsgQueues tty ms action = case queueStore_ ms of - MQStore st -> withLoadedQueues st action - PQStore st -> foldQueues tty st (mkQueue ms) action + -- This function can only be used in server CLI commands or before server is started. + -- It does not cache queues and is NOT concurrency safe. + unsafeWithAllMsgQueues :: Monoid a => Bool -> JournalMsgStore s -> (JournalQueue s -> IO a) -> IO a + unsafeWithAllMsgQueues tty ms action = case queueStore_ ms of + MQStore st -> withLoadedQueues st run + PQStore st -> foldQueueRecs tty st $ uncurry (mkTempQueue ms) >=> run + where + run q = do + r <- action q + closeMsgQueue q + pure r + + -- This function is concurrency safe, it is used to expire queues. + withAllMsgQueues :: forall a. Monoid a => Bool -> String -> JournalMsgStore s -> (JournalQueue s -> StoreIO s a) -> IO a + withAllMsgQueues tty op ms@JournalMsgStore {queueLocks, sharedLock} action = case queueStore_ ms of + MQStore st -> + withLoadedQueues st $ \q -> + run $ isolateQueue q op $ action q + PQStore st -> + foldQueueRecs tty st $ \(rId, qr) -> do + q <- mkTempQueue ms rId qr + withSharedWaitLock rId queueLocks sharedLock $ + run $ tryStore' op rId $ unStoreIO $ action q + where + run :: ExceptT ErrorType IO a -> IO a + run = fmap (fromRight mempty) . runExceptT logQueueStates :: JournalMsgStore s -> IO () logQueueStates ms = withActiveMsgQueues ms $ unStoreIO . logQueueState @@ -361,20 +410,11 @@ instance MsgStoreClass (JournalMsgStore s) where mkQueue :: JournalMsgStore s -> RecipientId -> QueueRec -> IO (JournalQueue s) mkQueue ms rId qr = do - queueLock <- atomically $ getMapLock (queueLocks ms) rId - queueRec' <- newTVarIO $ Just qr - msgQueue' <- newTVarIO Nothing - activeAt <- newTVarIO 0 - queueState <- newTVarIO Nothing - pure $ - JournalQueue - { recipientId' = rId, - queueLock, - queueRec', - msgQueue', - activeAt, - queueState - } + lock <- atomically $ getMapLock (queueLocks ms) rId + makeQueue_ ms rId qr lock + + getLoadedQueue :: JournalMsgStore s -> JournalQueue s -> StoreIO s (JournalQueue s) + getLoadedQueue ms sq = StoreIO $ fromMaybe sq <$> TM.lookupIO (recipientId sq) (loadedQueues $ queueStore_ ms) getMsgQueue :: JournalMsgStore s -> JournalQueue s -> Bool -> StoreIO s (JournalMsgQueue s) getMsgQueue ms@JournalMsgStore {random} q'@JournalQueue {recipientId' = rId, msgQueue'} forWrite = @@ -546,8 +586,11 @@ instance MsgStoreClass (JournalMsgStore s) where $>>= \hs -> updateReadPos q mq logState len hs $> Just () isolateQueue :: JournalQueue s -> String -> StoreIO s a -> ExceptT ErrorType IO a - isolateQueue JournalQueue {recipientId' = rId, queueLock} op a = - tryStore' op rId $ withLock' queueLock op $ unStoreIO a + isolateQueue sq op = tryStore' op (recipientId' sq) . withQueueLock sq op . unStoreIO + + unsafeRunStore :: JournalQueue s -> String -> StoreIO s a -> IO a + unsafeRunStore sq op a = + unStoreIO a `E.catch` \e -> storeError op (recipientId' sq) e >> E.throwIO e updateActiveAt :: JournalQueue s -> IO () updateActiveAt q = atomically . writeTVar (activeAt q) . systemSeconds =<< getSystemTime @@ -556,15 +599,16 @@ tryStore' :: String -> RecipientId -> IO a -> ExceptT ErrorType IO a tryStore' op rId = tryStore op rId . fmap Right tryStore :: forall a. String -> RecipientId -> IO (Either ErrorType a) -> ExceptT ErrorType IO a -tryStore op rId a = ExceptT $ E.mask_ $ E.try a >>= either storeErr pure - where - storeErr :: E.SomeException -> IO (Either ErrorType a) - storeErr e = - let e' = intercalate ", " [op, B.unpack $ strEncode rId, show e] - in logError ("STORE: " <> T.pack e') $> Left (STORE e') +tryStore op rId a = ExceptT $ E.mask_ $ a `E.catch` storeError op rId + +storeError :: String -> RecipientId -> E.SomeException -> IO (Either ErrorType a) +storeError op rId e = + let e' = intercalate ", " [op, B.unpack $ strEncode rId, show e] + in logError ("STORE: " <> T.pack e') $> Left (STORE e') isolateQueueId :: String -> JournalMsgStore s -> RecipientId -> IO (Either ErrorType a) -> ExceptT ErrorType IO a -isolateQueueId op ms rId = tryStore op rId . withLockMap (queueLocks ms) rId op +isolateQueueId op JournalMsgStore {queueLocks, sharedLock} rId = + tryStore op rId . withLockMapWaitShared rId queueLocks sharedLock op openMsgQueue :: JournalMsgStore s -> JMQueue -> Bool -> IO (JournalMsgQueue s) openMsgQueue ms@JournalMsgStore {config} q@JMQueue {queueDirectory = dir, statePath} forWrite = do diff --git a/src/Simplex/Messaging/Server/MsgStore/Journal/SharedLock.hs b/src/Simplex/Messaging/Server/MsgStore/Journal/SharedLock.hs new file mode 100644 index 000000000..4e09f3895 --- /dev/null +++ b/src/Simplex/Messaging/Server/MsgStore/Journal/SharedLock.hs @@ -0,0 +1,43 @@ +module Simplex.Messaging.Server.MsgStore.Journal.SharedLock + ( withLockWaitShared, + withLockMapWaitShared, + withSharedWaitLock, + ) +where + +import Control.Concurrent.STM +import qualified Control.Exception as E +import Control.Monad +import Simplex.Messaging.Agent.Lock +import Simplex.Messaging.Agent.Client (getMapLock) +import Simplex.Messaging.Protocol (RecipientId) +import Simplex.Messaging.TMap (TMap) +import qualified Simplex.Messaging.TMap as TM +import Simplex.Messaging.Util (($>>), ($>>=)) + +-- wait until shared lock with passed ID is released and take lock +withLockWaitShared :: RecipientId -> Lock -> TMVar RecipientId -> String -> IO a -> IO a +withLockWaitShared rId lock shared name = + E.bracket_ + (atomically $ waitShared rId shared >> putTMVar lock name) + (void $ atomically $ takeTMVar lock) + +-- wait until shared lock with passed ID is released and take lock from Map for this ID +withLockMapWaitShared :: RecipientId -> TMap RecipientId Lock -> TMVar RecipientId -> String -> IO a -> IO a +withLockMapWaitShared rId locks shared name a = + E.bracket + (atomically $ waitShared rId shared >> getPutLock (getMapLock locks) rId name) + (atomically . takeTMVar) + (const a) + +waitShared :: RecipientId -> TMVar RecipientId -> STM () +waitShared rId shared = tryReadTMVar shared >>= mapM_ (\rId' -> when (rId == rId') retry) + +-- wait until lock with passed ID in Map is released and take shared lock for this ID +withSharedWaitLock :: RecipientId -> TMap RecipientId Lock -> TMVar RecipientId -> IO a -> IO a +withSharedWaitLock rId locks shared = + E.bracket_ + (atomically $ waitLock >> putTMVar shared rId) + (atomically $ takeTMVar shared) + where + waitLock = TM.lookup rId locks $>>= tryReadTMVar $>> retry diff --git a/src/Simplex/Messaging/Server/MsgStore/STM.hs b/src/Simplex/Messaging/Server/MsgStore/STM.hs index ff7a93db8..43a41d7ca 100644 --- a/src/Simplex/Messaging/Server/MsgStore/STM.hs +++ b/src/Simplex/Messaging/Server/MsgStore/STM.hs @@ -80,7 +80,9 @@ instance MsgStoreClass STMMsgStore where {-# INLINE closeMsgStore #-} withActiveMsgQueues = withLoadedQueues . queueStore_ {-# INLINE withActiveMsgQueues #-} - withAllMsgQueues _ = withLoadedQueues . queueStore_ + unsafeWithAllMsgQueues _ = withLoadedQueues . queueStore_ + {-# INLINE unsafeWithAllMsgQueues #-} + withAllMsgQueues _tty _op ms action = withLoadedQueues (queueStore_ ms) $ atomically . action {-# INLINE withAllMsgQueues #-} logQueueStates _ = pure () {-# INLINE logQueueStates #-} @@ -92,6 +94,10 @@ instance MsgStoreClass STMMsgStore where mkQueue _ rId qr = STMQueue rId <$> newTVarIO (Just qr) <*> newTVarIO Nothing {-# INLINE mkQueue #-} + getLoadedQueue :: STMMsgStore -> STMQueue -> STM STMQueue + getLoadedQueue _ = pure + {-# INLINE getLoadedQueue #-} + getMsgQueue :: STMMsgStore -> STMQueue -> Bool -> STM STMMsgQueue getMsgQueue _ STMQueue {msgQueue'} _ = readTVar msgQueue' >>= maybe newQ pure where @@ -168,3 +174,8 @@ instance MsgStoreClass STMMsgStore where isolateQueue :: STMQueue -> String -> STM a -> ExceptT ErrorType IO a isolateQueue _ _ = liftIO . atomically + {-# INLINE isolateQueue #-} + + unsafeRunStore :: STMQueue -> String -> STM a -> IO a + unsafeRunStore _ _ = atomically + {-# INLINE unsafeRunStore #-} diff --git a/src/Simplex/Messaging/Server/MsgStore/Types.hs b/src/Simplex/Messaging/Server/MsgStore/Types.hs index 420517ac4..514b67d7b 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Types.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Types.hs @@ -35,13 +35,16 @@ class (Monad (StoreMonad s), QueueStoreClass (StoreQueue s) (QueueStore s)) => M newMsgStore :: MsgStoreConfig s -> IO s closeMsgStore :: s -> IO () withActiveMsgQueues :: Monoid a => s -> (StoreQueue s -> IO a) -> IO a - withAllMsgQueues :: Monoid a => Bool -> s -> (StoreQueue s -> IO a) -> IO a + -- This function can only be used in server CLI commands or before server is started. + unsafeWithAllMsgQueues :: Monoid a => Bool -> s -> (StoreQueue s -> IO a) -> IO a + withAllMsgQueues :: Monoid a => Bool -> String -> s -> (StoreQueue s -> StoreMonad s a) -> IO a logQueueStates :: s -> IO () logQueueState :: StoreQueue s -> StoreMonad s () queueStore :: s -> QueueStore s -- message store methods mkQueue :: s -> RecipientId -> QueueRec -> IO (StoreQueue s) + getLoadedQueue :: s -> StoreQueue s -> StoreMonad s (StoreQueue s) getMsgQueue :: s -> StoreQueue s -> Bool -> StoreMonad s (MsgQueue (StoreQueue s)) getPeekMsgQueue :: s -> StoreQueue s -> StoreMonad s (Maybe (MsgQueue (StoreQueue s), Message)) @@ -56,6 +59,7 @@ class (Monad (StoreMonad s), QueueStoreClass (StoreQueue s) (QueueStore s)) => M tryPeekMsg_ :: StoreQueue s -> MsgQueue (StoreQueue s) -> StoreMonad s (Maybe Message) tryDeleteMsg_ :: StoreQueue s -> MsgQueue (StoreQueue s) -> Bool -> StoreMonad s () isolateQueue :: StoreQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a + unsafeRunStore :: StoreQueue s -> String -> StoreMonad s a -> IO a data MSType = MSMemory | MSJournal @@ -82,10 +86,6 @@ getQueueRec st party qId = getQueue st party qId $>>= (\q -> maybe (Left AUTH) (Right . (q,)) <$> readTVarIO (queueRec q)) -getQueueMessages :: MsgStoreClass s => Bool -> s -> StoreQueue s -> ExceptT ErrorType IO [Message] -getQueueMessages drainMsgs st q = withPeekMsgQueue st q "getQueueMessages" $ maybe (pure []) (getQueueMessages_ drainMsgs q . fst) -{-# INLINE getQueueMessages #-} - getQueueSize :: MsgStoreClass s => s -> StoreQueue s -> ExceptT ErrorType IO Int getQueueSize st q = withPeekMsgQueue st q "getQueueSize" $ maybe (pure 0) (getQueueSize_ . fst) {-# INLINE getQueueSize #-} @@ -124,10 +124,12 @@ deleteExpiredMsgs st q old = -- closed and idle queues will be closed after expiration -- returns (expired count, queue size after expiration) -idleDeleteExpiredMsgs :: MsgStoreClass s => Int64 -> s -> StoreQueue s -> Int64 -> ExceptT ErrorType IO (Maybe Int, Int) -idleDeleteExpiredMsgs now st q old = - isolateQueue q "idleDeleteExpiredMsgs" $ - withIdleMsgQueue now st q (deleteExpireMsgs_ old q) +idleDeleteExpiredMsgs :: MsgStoreClass s => Int64 -> s -> StoreQueue s -> Int64 -> StoreMonad s (Maybe Int, Int) +idleDeleteExpiredMsgs now st q old = do + -- Use cached queue if available. + -- Also see the comment in loadQueue in PostgresQueueStore + q' <- getLoadedQueue st q + withIdleMsgQueue now st q' $ deleteExpireMsgs_ old q' deleteExpireMsgs_ :: MsgStoreClass s => Int64 -> StoreQueue s -> MsgQueue (StoreQueue s) -> StoreMonad s Int deleteExpireMsgs_ old q mq = do diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs b/src/Simplex/Messaging/Server/QueueStore/Postgres.hs index 0d5796298..0b6216b35 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs +++ b/src/Simplex/Messaging/Server/QueueStore/Postgres.hs @@ -23,7 +23,6 @@ module Simplex.Messaging.Server.QueueStore.Postgres PostgresStoreCfg (..), batchInsertQueues, foldQueueRecs, - foldQueues, ) where @@ -166,11 +165,26 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where loadRcvQueue = loadQueue " WHERE recipient_id = ?" $ \_ -> pure () loadSndQueue = loadQueue " WHERE sender_id = ?" $ \rId -> TM.insert qId rId senders loadNtfQueue = loadQueue " WHERE notifier_id = ?" $ \_ -> pure () -- do NOT cache ref - ntf subscriptions are rare - loadQueue condition insertRef = runExceptT $ loadQueueRec >>= liftIO . cachedOrLoadedQueue st mkQ insertRef - where - loadQueueRec = + loadQueue condition insertRef = + runExceptT $ do + (rId, qRec) <- withDB "getQueue_" st $ \db -> firstRow rowToQueueRec AUTH $ DB.query db (queueRecQuery <> condition <> " AND deleted_at IS NULL") (Only qId) + liftIO $ do + sq <- mkQ rId qRec -- loaded queue + -- This lock prevents the scenario when the queue is added to cache, + -- while another thread is proccessing the same queue in withAllMsgQueues + -- without adding it to cache, possibly trying to open the same files twice. + -- Alse see comment in idleDeleteExpiredMsgs. + withQueueLock sq "getQueue_" $ atomically $ + -- checking the cache again for concurrent reads, + -- use previously loaded queue if exists. + TM.lookup rId queues >>= \case + Just sq' -> pure sq' + Nothing -> do + insertRef rId + TM.insert rId sq queues + pure sq secureQueue :: PostgresQueueStore q -> q -> SndPublicAuthKey -> IO (Either ErrorType ()) secureQueue st sq sKey = @@ -311,18 +325,15 @@ insertQueueQuery = VALUES (?,?,?,?,?,?,?,?,?,?,?) |] -foldQueues :: Monoid a => Bool -> PostgresQueueStore q -> (RecipientId -> QueueRec -> IO q) -> (q -> IO a) -> IO a -foldQueues tty st mkQ f = - foldQueueRecs tty st $ cachedOrLoadedQueue st mkQ (\_ -> pure ()) >=> f - foldQueueRecs :: Monoid a => Bool -> PostgresQueueStore q -> ((RecipientId, QueueRec) -> IO a) -> IO a foldQueueRecs tty st f = do (n, r) <- withConnection (dbStore st) $ \db -> - DB.fold_ db (queueRecQuery <> " WHERE deleted_at IS NULL") (0 :: Int, mempty) $ \(!i, !acc) row -> do + DB.fold_ db (queueRecQuery <> " WHERE deleted_at IS NULL") (0 :: Int, mempty) $ \(i, acc) row -> do r <- f $ rowToQueueRec row - let i' = i + 1 - when (tty && i' `mod` 100000 == 0) $ putStr (progress i <> "\r") >> hFlush stdout - pure (i', acc <> r) + let !i' = i + 1 + !acc' = acc <> r + when (tty && i' `mod` 100000 == 0) $ putStr (progress i' <> "\r") >> hFlush stdout + pure (i', acc') when tty $ putStrLn $ progress n pure r where @@ -338,19 +349,6 @@ queueRecQuery = FROM msg_queues |] -cachedOrLoadedQueue :: PostgresQueueStore q -> (RecipientId -> QueueRec -> IO q) -> (RecipientId -> STM ()) -> (RecipientId, QueueRec) -> IO q -cachedOrLoadedQueue PostgresQueueStore {queues} mkQ insertRef (rId, qRec) = do - sq <- liftIO $ mkQ rId qRec -- loaded queue - atomically $ - -- checking the cache again for concurrent reads, - -- use previously loaded queue if exists. - TM.lookup rId queues >>= \case - Just sq' -> pure sq' - Nothing -> do - insertRef rId - TM.insert rId sq queues - pure sq - type QueueRecRow = (RecipientId, RcvPublicAuthKey, RcvDhSecret, SenderId, Maybe SndPublicAuthKey, SenderCanSecure, Maybe NotifierId, Maybe NtfPublicAuthKey, Maybe RcvNtfDhSecret, ServerEntityStatus, Maybe RoundedSystemTime) queueRecToRow :: (RecipientId, QueueRec) -> QueueRecRow diff --git a/tests/CoreTests/MsgStoreTests.hs b/tests/CoreTests/MsgStoreTests.hs index bbb7a7e8c..e73527ad4 100644 --- a/tests/CoreTests/MsgStoreTests.hs +++ b/tests/CoreTests/MsgStoreTests.hs @@ -453,7 +453,7 @@ testExpireIdleQueues = do old <- expireBeforeEpoch ExpirationConfig {ttl = 1, checkInterval = 1} -- no old messages now <- systemSeconds <$> getSystemTime - (expired_, stored) <- runRight $ idleDeleteExpiredMsgs now ms q old + (expired_, stored) <- runRight $ isolateQueue q "" $ idleDeleteExpiredMsgs now ms q old expired_ `shouldBe` Just 0 stored `shouldBe` 0 (Nothing, False) <- readQueueState ms statePath