use monoid for queue processing

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

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 = 

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