From 57c0326ddddef8470845cf09cbd692c2fbdf6e28 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Wed, 23 Oct 2024 15:46:48 +0100 Subject: [PATCH] use monoid for queue processing --- src/Simplex/Messaging/Server.hs | 38 ++++++++++--------- .../Messaging/Server/MsgStore/Journal.hs | 18 ++++----- .../Messaging/Server/MsgStore/Types.hs | 6 +-- 3 files changed, 33 insertions(+), 29 deletions(-) diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 084e75303..7b539ec69 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -68,6 +68,7 @@ import Data.List.NonEmpty (NonEmpty (..), (<|)) import qualified Data.List.NonEmpty as L import qualified Data.Map.Strict as M import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing) +import Data.Semigroup (Sum (..)) import qualified Data.Text as T import Data.Text.Encoding (decodeLatin1) import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime) @@ -148,6 +149,14 @@ data MessageStats = MessageStats storedQueues :: Int } +instance Monoid MessageStats where + mempty = MessageStats 0 0 0 + {-# INLINE mempty #-} + +instance Semigroup MessageStats where + MessageStats a b c <> MessageStats x y z = MessageStats (a + x) (b + y) (c + z) + {-# INLINE (<>) #-} + newMessageStats :: MessageStats newMessageStats = MessageStats 0 0 0 @@ -378,12 +387,12 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT liftIO $ forever $ do threadDelay' interval old <- expireBeforeEpoch expCfg - deleted <- withActiveMsgQueues ms (\_ -> expireQueueMsgs stats old) 0 + Sum deleted <- withActiveMsgQueues ms $ \_ -> expireQueueMsgs stats old logInfo $ "STORE: expireMessagesThread, expired " <> tshow deleted <> " messages" where - expireQueueMsgs stats old q !acc = + expireQueueMsgs stats old q = runExceptT (deleteExpiredMsgs q True old) >>= \case - Right deleted -> (acc + deleted) <$ atomicModifyIORef'_ (msgExpired stats) (+ deleted) + Right deleted -> Sum deleted <$ atomicModifyIORef'_ (msgExpired stats) (+ deleted) Left _ -> pure 0 expireNtfsThread :: ServerConfig -> M () @@ -1739,10 +1748,10 @@ saveServerMessages drainMsgs = exportMessages :: MsgStoreClass s => s -> FilePath -> Bool -> IO () exportMessages ms f drainMsgs = do logInfo $ "saving messages to file " <> T.pack f - total <- liftIO $ withFile f WriteMode $ \h -> withAllMsgQueues ms (saveQueueMsgs h) 0 + Sum total <- liftIO $ withFile f WriteMode $ withAllMsgQueues ms . saveQueueMsgs logInfo $ "messages saved: " <> tshow total where - saveQueueMsgs h rId q !acc = getQueueMessages drainMsgs q >>= \msgs -> (acc + length msgs) <$ BLD.hPutBuilder h (encodeMessages rId msgs) + saveQueueMsgs h rId q = getQueueMessages drainMsgs q >>= \msgs -> Sum (length msgs) <$ BLD.hPutBuilder h (encodeMessages rId msgs) encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n') processServerMessages :: M MessageStats @@ -1758,20 +1767,15 @@ processServerMessages = do AMS SMSJournal ms -> case old_ of Just old -> do logInfo "expiring journal store messages..." - withAllMsgQueues ms (\_ -> processExpireQueue old) newMessageStats + withAllMsgQueues ms $ \_ -> processExpireQueue old Nothing -> do logInfo "validating journal store messages..." - withAllMsgQueues ms (\_ -> processValidateQueue) newMessageStats + withAllMsgQueues ms $ \_ -> processValidateQueue where - processExpireQueue old q MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues} = + processExpireQueue old q = runExceptT expireQueue >>= \case - Right (stored', expired') -> - pure - MessageStats - { storedMsgsCount = storedMsgsCount + stored', - expiredMsgsCount = expiredMsgsCount + expired', - storedQueues = storedQueues + 1 - } + Right (storedMsgsCount, expiredMsgsCount) -> + pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = 1} Left e -> do logError $ "failed expiring messages in queue " <> T.pack (queueDirectory $ queue q) <> ": " <> tshow e exitFailure @@ -1782,8 +1786,8 @@ processServerMessages = do liftIO $ logQueueState q liftIO $ closeMsgQueue q pure (stored'', expired'') - processValidateQueue q stats@MessageStats {storedMsgsCount, storedQueues} = - getQueueSize q >>= \stored' -> pure stats {storedMsgsCount = storedMsgsCount + stored', storedQueues = storedQueues + 1} + processValidateQueue q = + getQueueSize q >>= \storedMsgsCount -> pure mempty {storedMsgsCount, storedQueues = 1} importMessages :: forall s. MsgStoreClass s => s -> FilePath -> Maybe Int64 -> IO MessageStats importMessages ms f old_ = do diff --git a/src/Simplex/Messaging/Server/MsgStore/Journal.hs b/src/Simplex/Messaging/Server/MsgStore/Journal.hs index cd0fe3923..327800eb8 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Journal.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Journal.hs @@ -214,28 +214,28 @@ instance MsgStoreClass JournalMsgStore where -- It is used to export storage to a single file and also to expire messages and validate all queues when server is started. -- TODO this function requires case-sensitive file system, because it uses queue directory as recipient ID. -- It can be made to support case-insensite FS by supporting more than one queue per directory, by getting recipient ID from state file name. - withAllMsgQueues :: JournalMsgStore -> (RecipientId -> JournalMsgQueue -> a -> IO a) -> a -> IO a - withAllMsgQueues ms@JournalMsgStore {config} action res = ifM (doesDirectoryExist storePath) processStore (pure res) + withAllMsgQueues :: Monoid a => JournalMsgStore -> (RecipientId -> JournalMsgQueue -> IO a) -> IO a + withAllMsgQueues ms@JournalMsgStore {config} action = ifM (doesDirectoryExist storePath) processStore (pure mempty) where processStore = do closeMsgStore ms lock <- createLockIO -- the same lock is used for all queues - (!count, !res') <- foldQueues 0 (processQueue lock) (0, res) ("", storePath) + (!count, !res) <- foldQueues 0 (processQueue lock) (0, mempty) ("", storePath) progress count putStrLn "" - pure res' + pure res JournalStoreConfig {storePath, pathParts} = config - processQueue queueLock (!i :: Int, !acc) (queueId, dir) = do + processQueue queueLock (!i :: Int, !r) (queueId, dir) = do when (i `mod` 100 == 0) $ progress i let statePath = msgQueueStatePath dir queueId q <- openMsgQueue ms JMQueue {queueDirectory = dir, queueLock, statePath} - acc' <- case strDecode $ B.pack queueId of - Right rId -> action rId q acc + r' <- case strDecode $ B.pack queueId of + Right rId -> action rId q Left e -> do putStrLn ("Error: message queue directory " <> dir <> " is invalid: " <> e) exitFailure closeMsgQueue q - pure (i + 1, acc') + pure (i + 1, r <> r') progress i = do putStr $ "Processed: " <> show i <> " queues\r" IO.hFlush stdout @@ -253,7 +253,7 @@ instance MsgStoreClass JournalMsgStore where (Nothing <$ putStrLn ("Error: path " <> path' <> " is not a directory, skipping")) logQueueStates :: JournalMsgStore -> IO () - logQueueStates ms = withActiveMsgQueues ms (\_ q _ -> logQueueState q) () + logQueueStates ms = withActiveMsgQueues ms $ \_ -> logQueueState logQueueState :: JournalMsgQueue -> IO () logQueueState q = diff --git a/src/Simplex/Messaging/Server/MsgStore/Types.hs b/src/Simplex/Messaging/Server/MsgStore/Types.hs index cd480b0ae..296a1d791 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Types.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Types.hs @@ -24,7 +24,7 @@ class Monad (StoreMonad s) => MsgStoreClass s where newMsgStore :: MsgStoreConfig s -> IO s closeMsgStore :: s -> IO () activeMsgQueues :: s -> TMap RecipientId (MsgQueue s) - withAllMsgQueues :: s -> (RecipientId -> MsgQueue s -> a -> IO a) -> a -> IO a + withAllMsgQueues :: Monoid a => s -> (RecipientId -> MsgQueue s -> IO a) -> IO a logQueueStates :: s -> IO () logQueueState :: MsgQueue s -> IO () getMsgQueue :: s -> RecipientId -> ExceptT ErrorType IO (MsgQueue s) @@ -46,8 +46,8 @@ data SMSType :: MSType -> Type where data AMSType = forall s. AMSType (SMSType s) -withActiveMsgQueues :: MsgStoreClass s => s -> (RecipientId -> MsgQueue s -> a -> IO a) -> a -> IO a -withActiveMsgQueues st f a = readTVarIO (activeMsgQueues st) >>= M.foldrWithKey (\k v -> (>>= f k v)) (pure a) +withActiveMsgQueues :: (MsgStoreClass s, Monoid a) => s -> (RecipientId -> MsgQueue s -> IO a) -> IO a +withActiveMsgQueues st f = readTVarIO (activeMsgQueues st) >>= M.foldrWithKey (\k v -> (>>= \r -> r `seq` (r <>) <$> f k v)) (pure mempty) tryPeekMsg :: MsgStoreClass s => MsgQueue s -> ExceptT ErrorType IO (Maybe Message) tryPeekMsg mq = isolateQueue mq "tryPeekMsg" $ tryPeekMsg_ mq