concurrent read messages

epoberezkin · epoberezkin · commit 05a32e64913c · 2025-09-09T21:48:40.000+01:00
diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs
@@ -126,7 +126,7 @@ import System.Environment (lookupEnv)
 import System.Exit (exitFailure, exitSuccess)
 import System.IO (hPrint, hPutStrLn, hSetNewlineMode, universalNewlineMode)
 import System.Mem.Weak (deRefWeak)
-import UnliftIO (timeout)
+import UnliftIO (mapConcurrently, timeout)
 import UnliftIO.Concurrent
 import UnliftIO.Directory (doesFileExist, renameFile)
 import UnliftIO.Exception
@@ -2114,10 +2114,32 @@ exportMessages :: forall s. MsgStoreClass s => Bool -> MsgStore s -> FilePath ->
 exportMessages tty st f drainMsgs = do
   logNote $ "saving messages to file " <> T.pack f
   run $ case st of
-    StoreMemory ms -> exportMessages_ ms $ getMsgs ms
-    StoreJournal ms -> exportMessages_ ms $ getJournalMsgs ms
+    StoreMemory ms -> exportMessages_ getMsgs ms
+    StoreJournal ms -> exportMessages_' getJournalMsgs ms
   where
-    exportMessages_ ms get = fmap (\(Sum n) -> n) . unsafeWithAllMsgQueues tty ms . saveQueueMsgs get
+    exportMessages_ get ms h =
+      fmap (\(Sum n) -> n) $ unsafeWithAllMsgQueues tty ms $ \q ->
+        get ms q >>= saveQueueMsgs h q
+    exportMessages_' get ms h = do
+      qv <- newIORef []
+      Sum n <- unsafeWithAllMsgQueues tty ms $ \q -> do
+        qs <- (q :) <$> readIORef qv
+        if length qs < 8
+          then writeIORef qv qs $> Sum 0
+          else do
+            writeIORef qv []
+            saveQueues qs
+      qs <- readIORef qv
+      if null qs
+        then pure n
+        else do
+          Sum n' <- saveQueues qs
+          pure $ n + n'
+      where
+        saveQueues qs = do
+          let qs' = reverse qs
+          qMsgs <- zip qs' <$> mapConcurrently (get ms) qs'
+          mconcat <$> mapM (uncurry $ saveQueueMsgs h) qMsgs
     run :: (Handle -> IO Int) -> IO ()
     run a = liftIO $ withFile f WriteMode $ tryAny . a >=> \case
       Right n -> logNote $ "messages saved: " <> tshow n
@@ -2130,9 +2152,8 @@ exportMessages tty st f drainMsgs = do
         Nothing -> getJournalQueueMessages ms q
     getMsgs :: MsgStoreClass s' => s' -> StoreQueue s' -> IO [Message]
     getMsgs ms q = unsafeRunStore q "saveQueueMsgs" $ getQueueMessages_ drainMsgs q =<< getMsgQueue ms q False
-    saveQueueMsgs :: (StoreQueue s -> IO [Message]) -> Handle -> StoreQueue s -> IO (Sum Int)
-    saveQueueMsgs get h q = do
-      msgs <- get q
+    saveQueueMsgs :: Handle -> StoreQueue s -> [Message] -> IO (Sum Int)
+    saveQueueMsgs h q msgs = do
       unless (null msgs) $ 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')
diff --git a/tests/CoreTests/MsgStoreTests.hs b/tests/CoreTests/MsgStoreTests.hs
@@ -54,7 +54,7 @@ msgStoreTests = do
   around (withMsgStore testSMTStoreConfig) $ describe "STM message store" someMsgStoreTests
   around (withMsgStore $ testJournalStoreCfg MQStoreCfg) $ describe "Journal message store" $ do
     someMsgStoreTests
-    it "should export and import journal store" testExportImportStore
+    fit "should export and import journal store" testExportImportStore
     describe "queue state" $ do
       it "should restore queue state from the last line" testQueueState
       it "should recover when message is written and state is not" testMessageState
